Explain why a go2rtc stream change was rejected

A stream source with an undefined or malformed {FRIGATE_*} placeholder
was reported as a restricted source, which reads like a security block
when the user made a typo. The config API now reports unknown variables
and invalid placeholders separately from restricted sources.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Makefile

@@ -1,7 +1,7 @@
 default_target: local
 
 COMMIT_HASH := $(shell git log -1 --pretty=format:"%h"|tail -1)
-VERSION = 0.18.1
+VERSION = 0.18.2
 IMAGE_REPO ?= ghcr.io/blakeblackshear/frigate
 GITHUB_REF_NAME ?= $(shell git rev-parse --abbrev-ref HEAD)
 BOARDS= #Initialized empty

docker/main/rootfs/usr/local/nginx/conf/auth_location.conf

@@ -15,6 +15,9 @@ location /auth {
     # Pass info about the request
     proxy_set_header X-Original-Method $request_method;
     proxy_set_header X-Original-URL $request_uri;
+    # set by the go2rtc stream locations; client headers are not forwarded, so
+    # this cannot be spoofed
+    proxy_set_header X-Go2rtc-Stream-Proxy $go2rtc_stream_proxy;
     proxy_set_header X-Server-Port $server_port;
     proxy_set_header Content-Length "";
     # Pass along auth related info

docker/main/rootfs/usr/local/nginx/conf/nginx.conf

@@ -202,6 +202,8 @@ http {
 
         # frigate lovelace card uses this path
         location /live/mse/api/ws {
+            # tells /auth to apply the go2rtc stream checks
+            set $go2rtc_stream_proxy 1;
             include auth_request.conf;
             limit_except GET {
                 deny  all;
@@ -211,6 +213,8 @@ http {
         }
 
         location /live/webrtc/api/ws {
+            # tells /auth to apply the go2rtc stream checks
+            set $go2rtc_stream_proxy 1;
             include auth_request.conf;
             limit_except GET {
                 deny  all;
@@ -231,6 +235,8 @@ http {
 
         # integration uses this to add webrtc candidate
         location /api/go2rtc/webrtc {
+            # tells /auth to apply the go2rtc stream checks
+            set $go2rtc_stream_proxy 1;
             include auth_request.conf;
             limit_except POST {
                 deny  all;

docs/docs/frigate/installation.md

@@ -78,9 +78,9 @@ Users of the Snapcraft build of Docker cannot use storage locations outside your
 
 Frigate utilizes shared memory to store frames during processing. The default `shm-size` provided by Docker is **64MB**.
 
-The default shm size of **128MB** is fine for setups with **2 cameras** detecting at **720p**. If Frigate is exiting with "Bus error" messages, it is likely because you have too many high resolution cameras and you need to specify a higher shm size, using [`--shm-size`](https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources) (or [`service.shm_size`](https://docs.docker.com/compose/compose-file/compose-file-v2/#shm_size) in Docker Compose). If raising the shm size does not help, check your [process and file limits](#process-and-file-limits) as well.
+Frigate needs about **210MB** for **2 cameras** detecting at **720p**. If Frigate is exiting with "Bus error" messages, it is likely because you have too many high resolution cameras and you need to specify a higher shm size, using [`--shm-size`](https://docs.docker.com/engine/reference/run/#runtime-constraints-on-resources) (or [`service.shm_size`](https://docs.docker.com/compose/compose-file/compose-file-v2/#shm_size) in Docker Compose). If raising the shm size does not help, check your [process and file limits](#process-and-file-limits) as well.
 
-The Frigate container also stores logs in shm, which can take up to **40MB**, so make sure to take this into account in your math as well.
+The Frigate container also stores logs and an nginx cache in shm, which can take up to **50MB**. The calculator below includes this, along with room for 2 cameras added while Frigate is running. Its result holds about 25 frames per camera, a little above the 20 frame minimum Frigate warns below. Frigate keeps up to 50 when there is room, which gives slow processes more time to read each frame.
 
 <ShmCalculator/>
 

docs/docs/frigate/updating.md

@@ -5,7 +5,7 @@ title: Updating
 
 # Updating Frigate
 
-The current stable version of Frigate is **0.18.0**. The release notes and any breaking changes for this version can be found on the [Frigate GitHub releases page](https://github.com/blakeblackshear/frigate/releases/tag/v0.18.0).
+The current stable version of Frigate is **0.18.1**. The release notes and any breaking changes for this version can be found on the [Frigate GitHub releases page](https://github.com/blakeblackshear/frigate/releases/tag/v0.18.1).
 
 Keeping Frigate up to date ensures you benefit from the latest features, performance improvements, and bug fixes. The update process varies slightly depending on your installation method (Docker, Home Assistant App, etc.). Below are instructions for the most common setups.
 
@@ -31,21 +31,21 @@ If you’re running Frigate via Docker (recommended method), follow these steps:
 
 2. **Update and Pull the Latest Image**:
    - If using Docker Compose:
-     - Edit your `docker-compose.yml` file to specify the desired version tag (e.g., `0.18.0` instead of `0.17.1`). For example:
+     - Edit your `docker-compose.yml` file to specify the desired version tag (e.g., `0.18.1` instead of `0.17.1`). For example:
        ```yaml
        services:
          frigate:
-           image: ghcr.io/blakeblackshear/frigate:0.18.0
+           image: ghcr.io/blakeblackshear/frigate:0.18.1
        ```
      - Then pull the image:
        ```bash
-       docker pull ghcr.io/blakeblackshear/frigate:0.18.0
+       docker pull ghcr.io/blakeblackshear/frigate:0.18.1
        ```
      - **Note for `stable` Tag Users**: If your `docker-compose.yml` uses the `stable` tag (e.g., `ghcr.io/blakeblackshear/frigate:stable`), you don’t need to update the tag manually. The `stable` tag always points to the latest stable release after pulling.
    - If using `docker run`:
-     - Pull the image with the appropriate tag (e.g., `0.18.0`, `0.18.0-tensorrt`, or `stable`):
+     - Pull the image with the appropriate tag (e.g., `0.18.1`, `0.18.1-tensorrt`, or `stable`):
        ```bash
-       docker pull ghcr.io/blakeblackshear/frigate:0.18.0
+       docker pull ghcr.io/blakeblackshear/frigate:0.18.1
        ```
 
 3. **Start the Container**:
@@ -100,7 +100,7 @@ If an update causes issues:
 1. Stop Frigate.
 2. Restore your backed-up config file and database.
 3. Revert to the previous image version:
-   - For Docker: Specify an older tag (e.g., `ghcr.io/blakeblackshear/frigate:0.17.1`) in your `docker run` command.
+   - For Docker: Specify an older tag (e.g., `ghcr.io/blakeblackshear/frigate:0.18.0`) in your `docker run` command.
    - For Docker Compose: Edit your `docker-compose.yml`, specify the older version tag (e.g., `ghcr.io/blakeblackshear/frigate:0.16.4`), and re-run `docker compose up -d`.
    - For Home Assistant: Restore from the app/addon backup you took before you updated.
 4. Verify the old version is running again.

docs/src/components/ShmCalculator/index.jsx

@@ -2,42 +2,53 @@ import React, { useState, useEffect } from "react";
 import Admonition from "@theme/Admonition";
 import styles from "./styles.module.css";
 
+// follows calculate_shm_requirements() in frigate/util/services.py, but sizes
+// for 25 frames so float rounding can't land under its 20 frame warning
+const MB = 1048576;
+const RESERVED_MB = 50; // logs + nginx cache
+const TARGET_FRAMES = 25;
+const DEFAULT_FRAMES = 50;
+
+const frameMb = (width, height) =>
+  Math.round(((width * height * 1.5 + 270480) / MB) * 10) / 10;
+
+// room for 2 cameras added at runtime, sized as 720p
+const addedCamerasMb = 2 * frameMb(1280, 720);
+
 const ShmCalculator = () => {
   const [width, setWidth] = useState(1280);
   const [height, setHeight] = useState(720);
   const [cameraCount, setCameraCount] = useState(1);
-  const [result, setResult] = useState("26.32MB");
-  const [singleCameraShm, setSingleCameraShm] = useState("26.32MB");
-  const [totalShm, setTotalShm] = useState("26.32MB");
+  const [result, setResult] = useState("-");
+  const [singleCameraShm, setSingleCameraShm] = useState("-");
+  const [addedCamerasShm, setAddedCamerasShm] = useState("-");
+  const [fullRingShm, setFullRingShm] = useState("-");
 
   const calculate = () => {
     if (!width || !height || !cameraCount) {
       setResult("Please enter valid values");
       setSingleCameraShm("-");
-      setTotalShm("-");
+      setAddedCamerasShm("-");
+      setFullRingShm("-");
       return;
     }
 
-    // Single camera base SHM calculation (excluding logs)
-    // Formula: (width * height * 1.5 * 20 + 270480) / 1048576
-    const singleCameraBase =
-      (width * height * 1.5 * 20 + 270480) / 1048576;
-    setSingleCameraShm(`${singleCameraBase.toFixed(2)}mb`);
+    const cameraFramesMb = cameraCount * frameMb(width, height) + addedCamerasMb;
 
-    // Total SHM calculation (multiple cameras, including logs)
-    const totalBase = singleCameraBase * cameraCount;
-    const finalResult = totalBase + 40; // Default includes logs +40mb
+    setSingleCameraShm(formatMb(frameMb(width, height) * TARGET_FRAMES));
+    setAddedCamerasShm(formatMb(addedCamerasMb * TARGET_FRAMES));
+    setResult(formatMb(Math.round(RESERVED_MB + cameraFramesMb * TARGET_FRAMES)));
+    setFullRingShm(
+      formatMb(Math.round(RESERVED_MB + cameraFramesMb * DEFAULT_FRAMES)),
+    );
+  };
 
-    setTotalShm(`${(totalBase + 40).toFixed(2)}mb`);
-
-    // Format result
-    if (finalResult < 1) {
-      setResult(`${(finalResult * 1024).toFixed(2)}kb`);
-    } else if (finalResult >= 1024) {
-      setResult(`${(finalResult / 1024).toFixed(2)}gb`);
-    } else {
-      setResult(`${finalResult.toFixed(2)}mb`);
+  const formatMb = (value) => {
+    if (value >= 1024) {
+      return `${(value / 1024).toFixed(2)}gb`;
     }
+
+    return `${Number(value.toFixed(2))}mb`;
   };
 
   const formatWithUnit = (value) => {
@@ -147,19 +158,27 @@ const ShmCalculator = () => {
           </div>
           <div className={styles.formulaDisplay}>
             <p>
-              <strong>Single Camera:</strong> {formatWithUnit(singleCameraShm)}
+              <strong>Per Camera:</strong> {formatWithUnit(singleCameraShm)}
             </p>
             <p>
-              <strong>Formula:</strong> (width × height × 1.5 × 20 + 270480) ÷
-              1048576
+              <strong>Formula:</strong> (width × height × 1.5 + 270480) ÷
+              1048576 × {TARGET_FRAMES} frames
             </p>
-            {cameraCount > 1 && (
-              <p>
-                <strong>Total ({cameraCount} cameras):</strong> {formatWithUnit(totalShm)}
-              </p>
-            )}
             <p>
-              <strong>With Logs:</strong> + 40<span className={styles.unit}>mb</span>
+              <strong>Room for 2 Added Cameras:</strong>{" "}
+              {formatWithUnit(addedCamerasShm)}
+            </p>
+            <p>
+              <strong>Logs and Cache:</strong> + {RESERVED_MB}
+              <span className={styles.unit}>mb</span>
+            </p>
+            <p>
+              <strong>Birdseye Restream:</strong> + 8
+              <span className={styles.unit}>mb</span> if enabled
+            </p>
+            <p>
+              <strong>For {DEFAULT_FRAMES} Frames:</strong>{" "}
+              {formatWithUnit(fullRingShm)}
             </p>
           </div>
         </div>

frigate/api/app.py

@@ -73,6 +73,8 @@ from frigate.util.config import (
 )
 from frigate.util.schema import get_config_schema
 from frigate.util.services import (
+    Go2rtcStreamIssue,
+    find_restricted_go2rtc_stream_changes,
     get_nvidia_driver_info,
     process_logs,
     restart_frigate,
@@ -522,6 +524,63 @@ def config_raw():
         )
 
 
+def _go2rtc_streams(raw_config: str) -> Any:
+    """Load go2rtc.streams with the same loader Frigate and go2rtc startup use.
+
+    Raises ruamel.yaml.YAMLError when the config cannot be parsed.
+    """
+    loaded = ruamel.yaml.YAML().load(raw_config)
+    go2rtc = loaded.get("go2rtc") if isinstance(loaded, dict) else None
+    return go2rtc.get("streams") if isinstance(go2rtc, dict) else None
+
+
+def _restricted_go2rtc_streams_response(
+    old_raw_config: str, new_raw_config: str
+) -> JSONResponse | None:
+    """Reject config changes that add restricted go2rtc stream sources.
+
+    The same sources are rejected by PUT /go2rtc/streams, and go2rtc would
+    otherwise load them from the saved config on the next restart.
+    """
+    try:
+        new_streams = _go2rtc_streams(new_raw_config)
+    except ruamel.yaml.YAMLError:
+        logger.warning("Rejected config change that could not be parsed")
+        return JSONResponse(
+            content={"success": False, "message": "Unable to parse config"},
+            status_code=400,
+        )
+
+    try:
+        old_streams = _go2rtc_streams(old_raw_config)
+    except ruamel.yaml.YAMLError:
+        # with no readable saved config, every source is treated as new
+        old_streams = None
+
+    issues = find_restricted_go2rtc_stream_changes(old_streams, new_streams)
+    if not issues:
+        return None
+
+    messages = []
+    for issue, label in (
+        (Go2rtcStreamIssue.restricted, "Restricted go2rtc stream source in"),
+        (Go2rtcStreamIssue.unknown_variable, "Unknown variable in go2rtc stream"),
+        (
+            Go2rtcStreamIssue.invalid_placeholder,
+            "Invalid variable placeholder in go2rtc stream",
+        ),
+    ):
+        names = [name for name, found in issues.items() if found == issue]
+        if names:
+            messages.append(f"{label}: {', '.join(names)}")
+
+    logger.warning("Rejected config change: %s", "; ".join(messages))
+    return JSONResponse(
+        content={"success": False, "message": "; ".join(messages)},
+        status_code=400,
+    )
+
+
 @router.post("/config/save", dependencies=[Depends(require_role(["admin"]))])
 def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
     new_config = body.decode()
@@ -598,10 +657,22 @@ def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
             status_code=400,
         )
 
+    config_file = find_config_file()
+
+    try:
+        with open(config_file) as f:
+            old_raw_config = f.read()
+    except OSError:
+        old_raw_config = ""
+
+    restricted_response = _restricted_go2rtc_streams_response(
+        old_raw_config, new_config
+    )
+    if restricted_response is not None:
+        return restricted_response
+
     # Save the config to file
     try:
-        config_file = find_config_file()
-
         with open(config_file, "w") as f:
             f.write(new_config)
             f.close()
@@ -912,6 +983,13 @@ def config_set(request: Request, body: AppConfigSetBody):
                         ),
                         status_code=400,
                     )
+                restricted_response = _restricted_go2rtc_streams_response(
+                    old_raw_config, new_raw_config
+                )
+                if restricted_response is not None:
+                    with open(config_file, "w") as f:
+                        f.write(old_raw_config)
+                    return restricted_response
             except Exception as e:
                 logging.error(f"Error updating config: {e}")
                 return JSONResponse(

frigate/api/auth.py

@@ -11,8 +11,9 @@ import secrets
 import time
 from datetime import datetime
 from pathlib import Path
-from urllib.parse import parse_qs, urlparse
+from urllib.parse import parse_qs, unquote
 
+import requests
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from fastapi.responses import JSONResponse, RedirectResponse
 from joserfc import jwt
@@ -1138,11 +1139,44 @@ def _get_stream_owner_cameras(request: Request, stream_name: str) -> set[str]:
     return owner_cameras
 
 
+def _is_existing_go2rtc_stream(request: Request, stream_name: str) -> bool:
+    """Check that `stream_name` names a stream go2rtc already knows about.
+
+    go2rtc turns any `src` that is not an existing stream name into a new ad hoc
+    stream and starts the producer it describes, so the proxied live-stream
+    paths must only ever reference existing streams.
+    """
+    frigate_config = request.app.frigate_config
+
+    if _get_stream_owner_cameras(request, stream_name):
+        return True
+
+    config_streams = getattr(frigate_config.go2rtc, "streams", None) or {}
+    if stream_name in config_streams:
+        return True
+
+    # streams added at runtime through PUT /go2rtc/streams only exist in go2rtc
+    try:
+        r = requests.get("http://127.0.0.1:1984/api/streams", timeout=2)
+    except requests.RequestException:
+        logger.warning("Unable to fetch streams from go2rtc to authorize request")
+        return False
+
+    if not r.ok:
+        return False
+
+    try:
+        return stream_name in r.json()
+    except ValueError:
+        return False
+
+
 # nginx proxies these paths straight to go2rtc with authentication-only checks
 # (see auth_request.conf). Each names the desired stream via the `src` query
 # param, so the camera-level check must happen here in the `/auth` subrequest —
 # `require_go2rtc_stream_access` only guards the REST `/go2rtc/streams/{name}`
-# endpoint, not these proxied live-stream paths.
+# endpoint, not these proxied live-stream paths. The nginx locations are prefix
+# matches, so anything under these paths is proxied too.
 GO2RTC_STREAM_PROXY_PATHS = frozenset(
     {
         "/live/mse/api/ws",
@@ -1151,32 +1185,80 @@ GO2RTC_STREAM_PROXY_PATHS = frozenset(
     }
 )
 
+_URI_SCHEME_AND_AUTHORITY = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://[^/?]*")
+
+
+def _split_request_uri(request_uri: str) -> tuple[str, str]:
+    """Split a raw request URI into its path and query string.
+
+    urlparse is not used since it reads a path starting with `//` as a host.
+    """
+    uri = _URI_SCHEME_AND_AUTHORITY.sub("", request_uri, count=1)
+    path, _, query = uri.partition("?")
+    return path, query
+
+
+def _normalize_request_path(path: str) -> str:
+    """Normalize a raw request path the way nginx does before matching locations.
+
+    nginx percent-decodes the path, merges repeated slashes and resolves dot
+    segments, while `X-Original-URL` carries the raw `$request_uri`.
+    """
+    segments: list[str] = []
+    for segment in unquote(path).split("/"):
+        if segment in ("", "."):
+            continue
+        if segment == "..":
+            if segments:
+                segments.pop()
+            continue
+        segments.append(segment)
+
+    return "/" + "/".join(segments)
+
+
+def _is_go2rtc_stream_proxy_request(path: str, request: Request) -> bool:
+    # nginx marks requests routed through the go2rtc stream locations, and the
+    # normalized path catches them if that marker is ever missing
+    if request.headers.get("x-go2rtc-stream-proxy") == "1":
+        return True
+
+    normalized = _normalize_request_path(path)
+    return any(normalized.startswith(p) for p in GO2RTC_STREAM_PROXY_PATHS)
+
 
 def deny_response_for_go2rtc_stream(
     original_url: str | None, role: str | None, request: Request
 ) -> int | None:
-    """Block role-restricted users from go2rtc live streams they cannot access.
+    """Block go2rtc live-stream requests the current role may not make.
 
-    Returns 403 when any `src` stream named in `original_url` resolves to a
-    camera outside the role's allow-list (or when no `src` is provided on a
-    stream-proxy path), otherwise None. Mirrors the resolution logic in
+    Returns 403 when any `src` named in `original_url` is not an existing go2rtc
+    stream (for every role), or when it resolves to a camera outside a
+    restricted role's allow-list (or no `src` is provided for such a role),
+    otherwise None. Mirrors the resolution logic in
     `require_go2rtc_stream_access` so substream names map to their owning
     camera correctly.
     """
     if not original_url:
         return None
 
-    parsed = urlparse(original_url)
-    if parsed.path not in GO2RTC_STREAM_PROXY_PATHS:
+    path, query = _split_request_uri(original_url)
+    if not _is_go2rtc_stream_proxy_request(path, request):
         return None
 
     frigate_config = request.app.frigate_config
+    sources = parse_qs(query, keep_blank_values=True).get("src", [])
+
+    # an unknown src would make go2rtc create an ad hoc stream from it, which
+    # bypasses the restricted-source checks on PUT /go2rtc/streams
+    for src in sources:
+        if not _is_existing_go2rtc_stream(request, src):
+            return 403
 
     # admin and full-access roles (no allow-list) bypass the camera check
     if not role or not is_role_restricted(role, frigate_config):
         return None
 
-    sources = parse_qs(parsed.query).get("src", [])
     if not sources:
         # a stream-proxy request naming no stream has nothing legitimate to
         # show a restricted user

frigate/api/camera.py

@@ -46,7 +46,7 @@ from frigate.util.image import run_ffmpeg_snapshot
 from frigate.util.services import (
     analyze_record_keyframes,
     ffprobe_stream,
-    is_restricted_go2rtc_source,
+    is_restricted_go2rtc_api_source,
 )
 
 logger = logging.getLogger(__name__)
@@ -152,9 +152,9 @@ def go2rtc_camera_stream(request: Request, stream_name: str):
 )
 def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
     """Add or update a go2rtc stream configuration."""
-    if src and is_restricted_go2rtc_source(src):
+    if src and is_restricted_go2rtc_api_source(src):
         logger.warning(
-            "Rejected go2rtc stream '%s' with restricted source type (echo/expr/exec)",
+            "Rejected go2rtc stream '%s' with restricted source",
             stream_name,
         )
         return JSONResponse(
@@ -173,9 +173,9 @@ def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
             except KeyError:
                 resolved_src = src
 
-            if is_restricted_go2rtc_source(resolved_src):
+            if is_restricted_go2rtc_api_source(resolved_src):
                 logger.warning(
-                    "Rejected go2rtc stream '%s' with restricted source type (echo/expr/exec)",
+                    "Rejected go2rtc stream '%s' with restricted source",
                     stream_name,
                 )
                 return JSONResponse(

frigate/config/env.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import re
 from pathlib import Path
@@ -5,6 +6,8 @@ from typing import Annotated
 
 from pydantic import AfterValidator, ValidationInfo
 
+logger = logging.getLogger(__name__)
+
 FRIGATE_ENV_VARS = {k: v for k, v in os.environ.items() if k.startswith("FRIGATE_")}
 secrets_dir = os.environ.get("CREDENTIALS_DIRECTORY", "/run/secrets")
 # read secret files as env vars too
@@ -16,6 +19,10 @@ if os.path.isdir(secrets_dir) and os.access(secrets_dir, os.R_OK):
             )
 
 
+# Security overrides that must come from the container environment, which is
+# controlled by the operator, rather than from the config, which admins can edit
+_PROTECTED_ENV_VARS = frozenset({"GO2RTC_ALLOW_ARBITRARY_EXEC"})
+
 # Matches a FRIGATE_* identifier following an opening brace.
 _FRIGATE_IDENT_RE = re.compile(r"FRIGATE_[A-Za-z0-9_]+")
 
@@ -95,6 +102,13 @@ EnvString = Annotated[str, AfterValidator(validate_env_string)]
 def validate_env_vars(v: dict[str, str], info: ValidationInfo) -> dict[str, str]:
     if isinstance(info.context, dict) and info.context.get("install", False):
         for k, val in v.items():
+            if k.upper() in _PROTECTED_ENV_VARS:
+                logger.warning(
+                    "Ignoring %s in environment_vars, it must be set in the container environment",
+                    k,
+                )
+                continue
+
             os.environ[k] = val
             if k.startswith("FRIGATE_"):
                 FRIGATE_ENV_VARS[k] = val

frigate/test/http_api/test_http_camera_access.py

@@ -403,6 +403,77 @@ class TestGo2rtcStreamAccess(BaseTestHttp):
                 forwarded_src = mock_put.call_args.kwargs["params"]["src"]
                 assert forwarded_src == "exec:/tmp/something"
 
+    def test_add_stream_rejects_ffmpeg_argument_injection(self):
+        """PUT /go2rtc/streams must reject ffmpeg: sources whose parameters can
+        inject arbitrary ffmpeg arguments, since go2rtc expands them to exec:"""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        with patch("frigate.api.camera.requests.put") as mock_put:
+            with AuthTestClient(app) as client:
+                for src in (
+                    # shape of the source in GHSA-qjjw-3v66-279f
+                    "ffmpeg:#input='-f''h264''-i''data:placeholder''-c''copy'",
+                    "ffmpeg:#input=-re",
+                    "ffmpeg:front_door#raw=-vn",
+                    "ffmpeg:front_door#video=-vn",
+                    "ffmpeg:front_door#video=out.mp4",
+                    "ffmpeg:front_door#drawtext=x",
+                    "ffmpeg:front_door#width=1x",
+                    "ffmpeg:front_door#audio=/16000",
+                    "ffmpeg:front_door#audio=opus/",
+                    "ffmpeg:front_door#audio=opus/16000/1",
+                    "ffmpeg:'front_door'",
+                    "  ffmpeg:#input=-re",
+                ):
+                    resp = client.put("/go2rtc/streams/injected", params={"src": src})
+                    assert resp.status_code == 400, (
+                        f"Expected 400 for ffmpeg src {src!r}; got {resp.status_code}"
+                    )
+                    assert resp.json().get("success") is False
+            mock_put.assert_not_called()
+
+    def test_add_stream_allows_standard_ffmpeg_sources(self):
+        """ffmpeg: sources built by the UI and shown in the docs must still be
+        forwarded to go2rtc"""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        mock_response = type("R", (), {"ok": True, "status_code": 200, "text": "ok"})()
+        with patch(
+            "frigate.api.camera.requests.put", return_value=mock_response
+        ) as mock_put:
+            with AuthTestClient(app) as client:
+                for src in (
+                    "ffmpeg:rtsp://10.0.0.1:554/video",
+                    "ffmpeg:rtsp://10.0.0.1:554/video#video=copy#audio=copy",
+                    "ffmpeg:rtsp://10.0.0.1:554/video#video=h264#hardware",
+                    "ffmpeg:front_door#audio=opus",
+                    "ffmpeg:front_door#video=h264#hardware=vaapi",
+                    "ffmpeg:front_door#video=h264#width=1280#rotate=-90",
+                    # documented HomeKit source
+                    "ffmpeg:front_door#video=h264#width=1280#height=720#audio=opus/16000",
+                    "ffmpeg:http://10.0.0.1/flv?port=1935&app=bcs#video=copy",
+                ):
+                    resp = client.put("/go2rtc/streams/legit", params={"src": src})
+                    assert resp.status_code == 200, (
+                        f"Expected 200 for ffmpeg src {src!r}; got {resp.status_code}"
+                    )
+                    forwarded_src = mock_put.call_args.kwargs["params"]["src"]
+                    assert forwarded_src == src
+
+    def test_add_stream_allows_ffmpeg_injection_when_override_set(self):
+        """GO2RTC_ALLOW_ARBITRARY_EXEC also lifts the ffmpeg: parameter check"""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        mock_response = type("R", (), {"ok": True, "status_code": 200, "text": "ok"})()
+        with patch.dict(os.environ, {"GO2RTC_ALLOW_ARBITRARY_EXEC": "true"}):
+            with patch(
+                "frigate.api.camera.requests.put", return_value=mock_response
+            ) as mock_put:
+                with AuthTestClient(app) as client:
+                    resp = client.put(
+                        "/go2rtc/streams/legit",
+                        params={"src": "ffmpeg:front_door#raw=-vn"},
+                    )
+                assert resp.status_code == 200
+                mock_put.assert_called_once()
+
     def test_stream_alias_blocked_when_owning_camera_disallowed(self):
         """limited_user cannot access a stream alias that belongs to a camera they
         are not allowed to see."""

frigate/test/http_api/test_http_config_go2rtc_streams.py

@@ -0,0 +1,217 @@
+"""Tests for go2rtc stream source restrictions on config/save and config/set."""
+
+import copy
+import os
+import tempfile
+from unittest.mock import patch
+
+import ruamel.yaml
+
+from frigate.config.env import FRIGATE_ENV_VARS
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+# shape of the source in GHSA-qjjw-3v66-279f, with the payload replaced
+_INJECTED_SRC = "ffmpeg:#input='-f''h264''-i''data:placeholder''-c''copy'"
+
+# an unrestricted ffmpeg: source the operator already has in the file on disk
+_EXISTING_SRC = "ffmpeg:rtsp://10.0.0.1:554/video#raw=-vn"
+
+_CONFIG = {
+    "mqtt": {"host": "mqtt"},
+    "go2rtc": {"streams": {"front_door": [_EXISTING_SRC]}},
+    "cameras": {
+        "front_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+    },
+}
+
+
+def _dump(config: dict) -> str:
+    yaml = ruamel.yaml.YAML()
+    stream = tempfile.SpooledTemporaryFile(mode="w+")
+    yaml.dump(config, stream)
+    stream.seek(0)
+    return stream.read()
+
+
+class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event, Recordings, ReviewSegment])
+        self.minimal_config = _CONFIG
+        self.app = super().create_app()
+
+        fd, self.config_path = tempfile.mkstemp(suffix=".yml")
+        with os.fdopen(fd, "w") as f:
+            f.write(_dump(_CONFIG))
+        self.addCleanup(os.unlink, self.config_path)
+
+        find_patch = patch(
+            "frigate.api.app.find_config_file", return_value=self.config_path
+        )
+        find_patch.start()
+        self.addCleanup(find_patch.stop)
+
+        env_patch = patch.dict(os.environ)
+        env_patch.start()
+        self.addCleanup(env_patch.stop)
+        os.environ.pop("GO2RTC_ALLOW_ARBITRARY_EXEC", None)
+
+    def _read_config(self) -> str:
+        with open(self.config_path) as f:
+            return f.read()
+
+    def _set_streams(self, streams: dict):
+        with AuthTestClient(self.app) as client:
+            return client.put(
+                "/config/set",
+                json={
+                    "config_data": {"go2rtc": {"streams": streams}},
+                    "requires_restart": 1,
+                },
+            )
+
+    def _save_streams(self, streams: dict, prefix: str = ""):
+        config = copy.deepcopy(_CONFIG)
+        config["go2rtc"]["streams"].update(streams)
+        with AuthTestClient(self.app) as client:
+            return client.post(
+                "/config/save?save_option=saveonly",
+                content=prefix + _dump(config),
+                headers={"Content-Type": "text/plain"},
+            )
+
+    # --- config/set ---
+
+    def test_set_rejects_injected_ffmpeg_source(self):
+        before = self._read_config()
+        resp = self._set_streams({"new_stream": [_INJECTED_SRC]})
+        assert resp.status_code == 400
+        assert resp.json()["success"] is False
+        assert self._read_config() == before
+
+    def test_set_rejects_injected_string_source(self):
+        before = self._read_config()
+        resp = self._set_streams({"new_stream": "ffmpeg:front_door#raw=-vn"})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_set_rejects_injected_source_added_to_existing_stream(self):
+        before = self._read_config()
+        resp = self._set_streams({"front_door": [_EXISTING_SRC, _INJECTED_SRC]})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_set_allows_standard_sources(self):
+        resp = self._set_streams(
+            {
+                "new_stream": [
+                    "rtsp://10.0.0.2:554/video",
+                    "ffmpeg:new_stream#audio=opus",
+                ]
+            }
+        )
+        assert resp.status_code == 200, resp.text
+        assert "ffmpeg:new_stream#audio=opus" in self._read_config()
+
+    def test_set_keeps_existing_source_on_disk(self):
+        resp = self._set_streams(
+            {"front_door": [_EXISTING_SRC, "ffmpeg:front_door#audio=opus"]}
+        )
+        assert resp.status_code == 200, resp.text
+        assert _EXISTING_SRC in self._read_config()
+
+    def test_set_allows_injected_source_when_override_set(self):
+        with patch.dict(os.environ, {"GO2RTC_ALLOW_ARBITRARY_EXEC": "true"}):
+            resp = self._set_streams({"new_stream": [_INJECTED_SRC]})
+        assert resp.status_code == 200, resp.text
+
+    # --- config/save ---
+
+    def test_save_rejects_injected_ffmpeg_source(self):
+        before = self._read_config()
+        resp = self._save_streams({"new_stream": [_INJECTED_SRC]})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_rejects_exec_source(self):
+        before = self._read_config()
+        resp = self._save_streams({"new_stream": "exec:/tmp/something"})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_rejects_map_source(self):
+        before = self._read_config()
+        resp = self._save_streams({"new_stream": {"url": "rtsp://10.0.0.2/video"}})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_resubmitting_existing_config_allowed(self):
+        # the UI config editor always sends back the whole file
+        resp = self._save_streams({"new_stream": "rtsp://10.0.0.2:554/video"})
+        assert resp.status_code == 200, resp.text
+        config = self._read_config()
+        assert _EXISTING_SRC in config
+        assert "rtsp://10.0.0.2:554/video" in config
+
+    def test_save_tagged_config_with_standard_source_allowed(self):
+        # a tag on the root map still passes validation and startup parsing
+        resp = self._save_streams(
+            {"new_stream": "rtsp://10.0.0.2:554/video"}, prefix="!config\n"
+        )
+        assert resp.status_code == 200, resp.text
+
+    def test_save_tagged_config_with_injected_source_rejected(self):
+        before = self._read_config()
+        resp = self._save_streams({"new_stream": [_INJECTED_SRC]}, prefix="!config\n")
+        assert resp.status_code == 400
+        assert "Restricted go2rtc stream source" in resp.json()["message"]
+        assert self._read_config() == before
+
+    # --- {FRIGATE_*} placeholders resolve the same way as at go2rtc startup ---
+
+    def test_set_rejects_unresolvable_placeholder(self):
+        before = self._read_config()
+        with patch.dict(FRIGATE_ENV_VARS, {"FRIGATE_RTSP_USER": "user"}):
+            for src, message in (
+                (
+                    "{FRIGATE_RTSP_USER!r}fmpeg:front_door#raw=-vn",
+                    "Invalid variable placeholder in go2rtc stream: new_stream",
+                ),
+                (
+                    "{FRIGATE_RTSP_USER:>5}",
+                    "Invalid variable placeholder in go2rtc stream: new_stream",
+                ),
+                (
+                    "rtsp://{FRIGATE_UNDEFINED}@10.0.0.2:554/video",
+                    "Unknown variable in go2rtc stream: new_stream",
+                ),
+            ):
+                resp = self._set_streams({"new_stream": [src]})
+                assert resp.status_code == 400, src
+                assert resp.json()["message"] == message, src
+                assert self._read_config() == before
+
+    def test_set_reports_each_issue(self):
+        resp = self._set_streams(
+            {
+                "typo_stream": ["rtsp://{FRIGATE_UNDEFINED}@10.0.0.2:554/video"],
+                "injected_stream": [_INJECTED_SRC],
+            }
+        )
+        assert resp.status_code == 400
+        assert resp.json()["message"] == (
+            "Restricted go2rtc stream source in: injected_stream; "
+            "Unknown variable in go2rtc stream: typo_stream"
+        )
+
+    def test_set_allows_resolvable_placeholder(self):
+        src = "rtsp://{FRIGATE_RTSP_USER}@10.0.0.2:554/video"
+        with patch.dict(FRIGATE_ENV_VARS, {"FRIGATE_RTSP_USER": "user"}):
+            resp = self._set_streams({"new_stream": [src]})
+        assert resp.status_code == 200, resp.text
+        assert src in self._read_config()

frigate/test/http_api/test_http_go2rtc_proxy_auth.py

@@ -0,0 +1,73 @@
+"""Tests for the /auth subrequest on nginx-proxied go2rtc live-stream paths."""
+
+import time
+from unittest.mock import patch
+
+import requests
+from fastapi.testclient import TestClient
+from joserfc.jwk import OctKey
+
+from frigate.api.auth import create_encoded_jwt
+from frigate.models import Event, Recordings, ReviewSegment, User
+from frigate.test.http_api.base_http_test import BaseTestHttp
+
+_CONFIG = {
+    "mqtt": {"host": "mqtt"},
+    "cameras": {
+        "front_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+    },
+}
+
+# shape of the source in GHSA-93xj-hg88-ffg7, with the payload replaced
+_AD_HOC_SRC = "ffmpeg:%23input%3D'-f''h264''-i''data:placeholder''-c''copy'"
+
+
+class TestGo2rtcProxyAuth(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event, ReviewSegment, Recordings, User])
+        self.minimal_config = _CONFIG
+        self.app = super().create_app()
+        self.app.jwt_token = OctKey.import_key(b"0" * 32)
+
+        get_patch = patch(
+            "frigate.api.auth.requests.get",
+            side_effect=requests.ConnectionError("go2rtc unavailable"),
+        )
+        get_patch.start()
+        self.addCleanup(get_patch.stop)
+
+    def _auth(self, role: str, original_url: str):
+        token = create_encoded_jwt(
+            role, role, int(time.time()) + 3600, self.app.jwt_token
+        )
+        with TestClient(self.app) as client:
+            return client.get(
+                "/auth",
+                headers={
+                    "authorization": f"Bearer {token}",
+                    "x-original-url": original_url,
+                },
+            )
+
+    def test_viewer_ad_hoc_mse_source_denied(self):
+        resp = self._auth("viewer", f"/live/mse/api/ws?src={_AD_HOC_SRC}")
+        assert resp.status_code == 403
+
+    def test_viewer_ad_hoc_webrtc_sources_denied(self):
+        for path in ("/live/webrtc/api/ws", "/api/go2rtc/webrtc"):
+            resp = self._auth("viewer", f"{path}?src={_AD_HOC_SRC}")
+            assert resp.status_code == 403, path
+
+    def test_admin_ad_hoc_mse_source_denied(self):
+        resp = self._auth("admin", f"/live/mse/api/ws?src={_AD_HOC_SRC}")
+        assert resp.status_code == 403
+
+    def test_viewer_camera_stream_allowed(self):
+        resp = self._auth("viewer", "/live/mse/api/ws?src=front_door")
+        assert resp.status_code == 202
+        assert resp.headers["remote-role"] == "viewer"

frigate/test/test_env.py

@@ -49,14 +49,10 @@ class TestGo2RtcAddStreamSubstitution(unittest.TestCase):
     def test_mixed_localtime_and_frigate_var(self):
         """%{localtime\\:...} alongside {FRIGATE_USER} substitutes only the var."""
         FRIGATE_ENV_VARS["FRIGATE_USER"] = "admin"
-        src = (
-            "ffmpeg:rtsp://host/s#raw=-vf "
-            "drawtext=text=%{localtime\\:%Y-%m-%d}:user={FRIGATE_USER}"
-        )
+        src = "rtsp://host/s?text=%{localtime\\:%Y-%m-%d}&user={FRIGATE_USER}"
         self.assertEqual(
             self._call_route(src),
-            "ffmpeg:rtsp://host/s#raw=-vf "
-            "drawtext=text=%{localtime\\:%Y-%m-%d}:user=admin",
+            "rtsp://host/s?text=%{localtime\\:%Y-%m-%d}&user=admin",
         )
 
     def test_unknown_var_falls_back_to_raw_src(self):
@@ -232,6 +228,29 @@ class TestEnvVars(unittest.TestCase):
         result = validate_env_string("{FRIGATE_BROKER}")
         self.assertEqual(result, "mqtt.local")
 
+    def test_install_ignores_go2rtc_exec_override(self):
+        """The go2rtc exec override cannot be enabled from the config."""
+        from frigate.util.services import is_go2rtc_arbitrary_exec_allowed
+
+        env_patch = patch.dict(os.environ)
+        env_patch.start()
+        self.addCleanup(env_patch.stop)
+        os.environ.pop("GO2RTC_ALLOW_ARBITRARY_EXEC", None)
+        ctx = self._make_context(install=True)
+        with patch("frigate.util.services.os.path.isfile", return_value=False):
+            for key in ("GO2RTC_ALLOW_ARBITRARY_EXEC", "go2rtc_allow_arbitrary_exec"):
+                validate_env_vars({key: "true", "OTHER_VAR": "value"}, ctx)
+                self.assertNotIn(key, os.environ)
+                self.assertFalse(is_go2rtc_arbitrary_exec_allowed())
+        self.assertEqual(os.environ.get("OTHER_VAR"), "value")
+
+    def test_install_keeps_container_go2rtc_exec_override(self):
+        """An override set in the container environment is left untouched."""
+        ctx = self._make_context(install=True)
+        with patch.dict(os.environ, {"GO2RTC_ALLOW_ARBITRARY_EXEC": "true"}):
+            validate_env_vars({"GO2RTC_ALLOW_ARBITRARY_EXEC": "false"}, ctx)
+            self.assertEqual(os.environ.get("GO2RTC_ALLOW_ARBITRARY_EXEC"), "true")
+
 
 if __name__ == "__main__":
     unittest.main()

frigate/test/test_go2rtc_stream_auth.py

@@ -8,6 +8,9 @@ param, which the static-media auth in `media_auth` does not inspect.
 
 import types
 import unittest
+from unittest.mock import MagicMock, patch
+
+import requests
 
 from frigate.api.auth import deny_response_for_go2rtc_stream
 from frigate.config import FrigateConfig
@@ -20,6 +23,7 @@ _CONFIG = {
             "dual_user": ["front_door", "back_door"],
         }
     },
+    "go2rtc": {"streams": {"doorbell_audio": ["rtsp://10.0.0.4:554/audio"]}},
     "cameras": {
         "front_door": {
             "ffmpeg": {
@@ -46,7 +50,9 @@ _CONFIG = {
 
 
 def _request(config: FrigateConfig) -> types.SimpleNamespace:
-    return types.SimpleNamespace(app=types.SimpleNamespace(frigate_config=config))
+    return types.SimpleNamespace(
+        app=types.SimpleNamespace(frigate_config=config), headers={}
+    )
 
 
 class TestDenyResponseForGo2rtcStream(unittest.TestCase):
@@ -54,6 +60,20 @@ class TestDenyResponseForGo2rtcStream(unittest.TestCase):
         self.config = FrigateConfig(**_CONFIG)
         self.request = _request(self.config)
 
+        # go2rtc is not running in tests; individual tests override this
+        get_patch = patch(
+            "frigate.api.auth.requests.get",
+            side_effect=requests.ConnectionError("go2rtc unavailable"),
+        )
+        self.mock_get = get_patch.start()
+        self.addCleanup(get_patch.stop)
+
+    def _go2rtc_streams(self, *names: str) -> None:
+        response = MagicMock(ok=True)
+        response.json.return_value = {name: {} for name in names}
+        self.mock_get.side_effect = None
+        self.mock_get.return_value = response
+
     def _deny(self, url: str, role: str):
         return deny_response_for_go2rtc_stream(url, role, self.request)
 
@@ -170,6 +190,107 @@ class TestDenyResponseForGo2rtcStream(unittest.TestCase):
     def test_missing_role_bypasses(self):
         self.assertIsNone(self._deny("http://host/live/mse/api/ws?src=back_door", None))
 
+    # --- ad hoc sources are denied for every role ---
+
+    def test_ad_hoc_ffmpeg_source_denied_for_all_roles(self):
+        # shape of the source in GHSA-93xj-hg88-ffg7
+        src = "ffmpeg:%23input%3D'-f''h264''-i''data:placeholder''-c''copy'"
+        for path in (
+            "/live/mse/api/ws",
+            "/live/webrtc/api/ws",
+            "/api/go2rtc/webrtc",
+        ):
+            for role in ("viewer", "admin", "limited_user", None):
+                self.assertEqual(
+                    self._deny(f"http://host{path}?src={src}", role),
+                    403,
+                    f"{path} {role}",
+                )
+
+    def test_ad_hoc_source_with_name_param_denied(self):
+        # name= would otherwise let go2rtc replace an existing stream's source
+        self.assertEqual(
+            self._deny(
+                "http://host/live/mse/api/ws?src=ffmpeg:x%23video%3Dh264"
+                "&name=front_door",
+                "admin",
+            ),
+            403,
+        )
+
+    def test_ad_hoc_source_after_existing_src_denied(self):
+        self.assertEqual(
+            self._deny(
+                "http://host/live/mse/api/ws?src=front_door&src=exec:x", "viewer"
+            ),
+            403,
+        )
+
+    def test_unknown_stream_denied_for_admin(self):
+        self.assertEqual(
+            self._deny("http://host/live/mse/api/ws?src=nonexistent", "admin"), 403
+        )
+
+    def test_blank_src_denied_for_admin(self):
+        self.assertEqual(self._deny("http://host/live/mse/api/ws?src=", "admin"), 403)
+
+    def test_go2rtc_config_stream_allowed(self):
+        self.assertIsNone(
+            self._deny("http://host/live/mse/api/ws?src=doorbell_audio", "viewer")
+        )
+        self.mock_get.assert_not_called()
+
+    def test_runtime_go2rtc_stream_allowed(self):
+        # streams added through PUT /go2rtc/streams only exist in go2rtc
+        self._go2rtc_streams("front_door", "wizard_test")
+        self.assertIsNone(
+            self._deny("http://host/live/mse/api/ws?src=wizard_test", "admin")
+        )
+
+    def test_runtime_lookup_does_not_admit_ad_hoc_source(self):
+        self._go2rtc_streams("front_door", "wizard_test")
+        self.assertEqual(
+            self._deny("http://host/live/mse/api/ws?src=ffmpeg:wizard_test", "admin"),
+            403,
+        )
+
+    def test_go2rtc_error_fails_closed(self):
+        self.mock_get.side_effect = None
+        self.mock_get.return_value = MagicMock(ok=False)
+        self.assertEqual(
+            self._deny("http://host/live/mse/api/ws?src=wizard_test", "admin"), 403
+        )
+
+    # --- path spellings nginx normalizes before matching its location ---
+
+    def test_path_variants_are_checked(self):
+        for path in (
+            "/live/mse/api/%77s",
+            "/live/mse/api%2Fws",
+            "//live/mse/api/ws",
+            "/live//mse/api/ws",
+            "/live/mse/api/./ws",
+            "/live/other/../mse/api/ws",
+            "/live/mse/api/%2e/ws",
+            "/%6Cive/webrtc/api/ws",
+            "/api/go2rtc/%77ebrtc",
+            "/live/mse/api/ws/",
+            "/live/mse/api/ws/extra",
+            "http://host//live/mse/api/ws",
+        ):
+            self.assertEqual(self._deny(f"{path}?src=nonexistent", "viewer"), 403, path)
+
+    def test_path_variants_allow_existing_stream(self):
+        self.assertIsNone(self._deny("/live/mse/api/%77s?src=front_door", "viewer"))
+
+    def test_nginx_marker_applies_check(self):
+        self.request.headers = {"x-go2rtc-stream-proxy": "1"}
+        self.assertEqual(self._deny("/unexpected?src=nonexistent", "viewer"), 403)
+
+    def test_unrelated_paths_not_checked(self):
+        for path in ("/live/mse/api", "/api/go2rtc/streams", "/live/jsmpeg/x"):
+            self.assertIsNone(self._deny(f"{path}?src=nonexistent", "viewer"), path)
+
 
 if __name__ == "__main__":
     unittest.main()

frigate/util/services.py

@@ -12,6 +12,7 @@ import subprocess as sp
 import time
 import traceback
 from datetime import datetime
+from enum import Enum
 from typing import Any
 
 import cv2
@@ -986,6 +987,122 @@ def is_restricted_go2rtc_source(stream_source: str) -> bool:
     return not is_go2rtc_arbitrary_exec_allowed()
 
 
+# go2rtc expands an `ffmpeg:` source into an `exec:` command line built from its
+# `#key=value` parameters, so only parameters that map to fixed presets or plain
+# numbers are accepted from the API
+_GO2RTC_FFMPEG_NAME_PARAMS = frozenset({"video", "audio", "hardware"})
+_GO2RTC_FFMPEG_NUMERIC_PARAMS = frozenset(
+    {"width", "height", "rotate", "bitrate", "timeout"}
+)
+_GO2RTC_FFMPEG_FLAG_PARAMS = frozenset({"async"})
+# a preset or codec name, optionally with a sample rate (e.g. opus/16000)
+_GO2RTC_FFMPEG_NAME_VALUE = re.compile(r"^([A-Za-z0-9_]*|[A-Za-z0-9_]+/[0-9]{1,6})$")
+_GO2RTC_FFMPEG_NUMERIC_VALUE = re.compile(r"^-?[0-9]+[kKmM]?$")
+
+
+def _is_safe_go2rtc_ffmpeg_source(stream_source: str) -> bool:
+    """Check that an `ffmpeg:` source only uses allowlisted parameters."""
+    if any(c in stream_source for c in "'\"") or re.search(r"\s", stream_source):
+        return False
+
+    _, *params = stream_source.split("#")
+    for param in params:
+        key, has_value, value = param.partition("=")
+        if key in _GO2RTC_FFMPEG_NAME_PARAMS:
+            if not _GO2RTC_FFMPEG_NAME_VALUE.match(value):
+                return False
+        elif key in _GO2RTC_FFMPEG_NUMERIC_PARAMS:
+            if not _GO2RTC_FFMPEG_NUMERIC_VALUE.match(value):
+                return False
+        elif key in _GO2RTC_FFMPEG_FLAG_PARAMS:
+            if has_value:
+                return False
+        else:
+            return False
+
+    return True
+
+
+def is_restricted_go2rtc_api_source(stream_source: str) -> bool:
+    """Check if a stream source submitted through the API is restricted.
+
+    In addition to the echo/expr/exec sources, this restricts `ffmpeg:` sources
+    whose parameters could inject arbitrary ffmpeg arguments. The
+    GO2RTC_ALLOW_ARBITRARY_EXEC override lifts both restrictions.
+    """
+    source = stream_source.strip()
+    if is_restricted_go2rtc_source(source):
+        return True
+    if not source.startswith("ffmpeg:"):
+        return False
+    if _is_safe_go2rtc_ffmpeg_source(source[len("ffmpeg:") :]):
+        return False
+    return not is_go2rtc_arbitrary_exec_allowed()
+
+
+def _go2rtc_stream_sources(stream: Any) -> list[Any]:
+    return stream if isinstance(stream, list) else [stream]
+
+
+class Go2rtcStreamIssue(str, Enum):
+    """Why a go2rtc stream source was rejected."""
+
+    restricted = "restricted"
+    unknown_variable = "unknown_variable"
+    invalid_placeholder = "invalid_placeholder"
+
+
+def find_restricted_go2rtc_stream_changes(
+    old_streams: dict[str, Any] | None, new_streams: dict[str, Any] | None
+) -> dict[str, Go2rtcStreamIssue]:
+    """Return the go2rtc streams with new sources that must be rejected.
+
+    Used when the config is changed through the API. A source that is already
+    in the saved config for the same stream is left alone, so streams the
+    operator configured on disk keep working; any other source must resolve
+    its {FRIGATE_*} placeholders and pass `is_restricted_go2rtc_api_source`.
+    """
+    # imported here since frigate.config imports this module
+    from frigate.config.env import substitute_frigate_vars
+
+    old_streams = old_streams if isinstance(old_streams, dict) else {}
+    new_streams = new_streams if isinstance(new_streams, dict) else {}
+    issues: dict[str, Go2rtcStreamIssue] = {}
+
+    for name, stream in new_streams.items():
+        existing = _go2rtc_stream_sources(old_streams.get(name))
+
+        for source in _go2rtc_stream_sources(stream):
+            if source in existing:
+                continue
+
+            if not isinstance(source, str):
+                # the map form is only usable with arbitrary exec enabled
+                if is_go2rtc_arbitrary_exec_allowed():
+                    continue
+                issues[name] = Go2rtcStreamIssue.restricted
+                break
+
+            # only sources whose placeholders resolve here are accepted, so the
+            # resolved value checked below is what go2rtc loads at startup
+            try:
+                resolved_source = substitute_frigate_vars(source)
+            except KeyError:
+                issues[name] = Go2rtcStreamIssue.unknown_variable
+                break
+            except ValueError:
+                issues[name] = Go2rtcStreamIssue.invalid_placeholder
+                break
+
+            if is_restricted_go2rtc_api_source(
+                source
+            ) or is_restricted_go2rtc_api_source(resolved_source):
+                issues[name] = Go2rtcStreamIssue.restricted
+                break
+
+    return issues
+
+
 def ffprobe_stream(ffmpeg, path: str, detailed: bool = False) -> sp.CompletedProcess:
     """Run ffprobe on stream."""
     clean_path = escape_special_characters(path)