Merge remote-tracking branch 'origin/master' into dev

Brings the go2rtc security fixes from #24658 to dev. Dev-specific changes
in the merge:
- frigate/config/env.py stays deleted; the environment_vars protection for
  GO2RTC_ALLOW_ARBITRARY_EXEC moves to apply_config_env_vars in
  frigate/util/env.py, so the value never reaches os.environ
- services.py imports substitute_frigate_vars from frigate.util.env and
  maps UnknownVariableError to the unknown variable message
- go2rtc stream tests patch dev's import-time exec override snapshot and
  its FRIGATE_* variable sources
- test_http_media.py keeps dev's _insert_recording helper; the one added
  on master replaced it and broke the dev tests

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

frigate/__main__.py

@@ -15,7 +15,9 @@ from frigate.util.config import find_config_file
 
 
 def main() -> None:
-    manager = mp.Manager()
+    # Fork this manager so the forkserver does not start until the first
+    # worker process, after the config has applied environment_vars.
+    manager = mp.get_context("fork").Manager()
     faulthandler.enable()
 
     # Setup the logging thread

frigate/api/app.py

@@ -78,6 +78,8 @@ from frigate.util.live_streams import (
 from frigate.util.object_names import get_categorized_object_names
 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,
@@ -546,6 +548,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()
@@ -622,10 +681,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()
@@ -960,6 +1031,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

@@ -12,8 +12,9 @@ import threading
 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
@@ -1003,7 +1004,7 @@ def create_user(
     HASH_ITERATIONS = request.app.frigate_config.auth.hash_iterations
     config_roles = list(request.app.frigate_config.auth.roles.keys())
 
-    if not re.match("^[A-Za-z0-9._]+$", body.username):
+    if not re.match(r"^(?!\.+$)[A-Za-z0-9._]+$", body.username):
         return JSONResponse(content={"message": "Invalid username"}, status_code=400)
 
     if body.role not in config_roles:
@@ -1229,11 +1230,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",
@@ -1242,32 +1276,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

@@ -34,7 +34,11 @@ from frigate.config.camera.updater import (
     CameraConfigUpdateTopic,
 )
 from frigate.models import User
-from frigate.util.builtin import clean_camera_user_pass, get_record_segment_time
+from frigate.util.builtin import (
+    clean_camera_user_pass,
+    clear_orphaned_comments,
+    get_record_segment_time,
+)
 from frigate.util.camera_cleanup import cleanup_camera_db, cleanup_camera_files
 from frigate.util.config import find_config_file
 from frigate.util.env import UnknownVariableError, substitute_frigate_vars
@@ -47,7 +51,7 @@ from frigate.util.live_streams import (
 from frigate.util.services import (
     analyze_record_keyframes,
     ffprobe_stream,
-    is_restricted_go2rtc_source,
+    is_restricted_go2rtc_api_source,
 )
 
 logger = logging.getLogger(__name__)
@@ -153,9 +157,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(
@@ -174,9 +178,9 @@ def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
             except UnknownVariableError:
                 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(
@@ -1273,6 +1277,7 @@ async def delete_camera(
                 # Remove camera from config
                 if "cameras" in data and camera_name in data["cameras"]:
                     del data["cameras"][camera_name]
+                    clear_orphaned_comments(data["cameras"], data, "cameras")
 
                 # Remove camera from auth roles
                 auth = data.get("auth", {})

frigate/api/export.py

@@ -58,6 +58,7 @@ from frigate.api.defs.response.export_response import (
 )
 from frigate.api.defs.response.generic_response import GenericResponse
 from frigate.api.defs.tags import Tags
+from frigate.api.media_auth import deny_response_for_media_uri
 from frigate.const import CLIPS_DIR, EXPORT_DIR
 from frigate.jobs.export import (
     ExportJob,
@@ -134,6 +135,7 @@ def _validate_export_case(export_case_id: str | None) -> JSONResponse | None:
 
 
 def _sanitize_existing_image(
+    request: Request,
     image_path: str | None,
 ) -> tuple[str | None, JSONResponse | None]:
     if not image_path:
@@ -141,6 +143,15 @@ def _sanitize_existing_image(
 
     existing_image = sanitize_contained_path(image_path, CLIPS_DIR)
 
+    # CLIPS_DIR is shared by every camera, so the caller must also be allowed
+    # to read the image.
+    if existing_image is not None and deny_response_for_media_uri(
+        f"/clips/{quote(os.path.relpath(existing_image, CLIPS_DIR))}",
+        request.headers.get("remote-role"),
+        request.app.frigate_config,
+    ):
+        existing_image = None
+
     if existing_image is None:
         return None, JSONResponse(
             content={"success": False, "message": "Invalid image path"},
@@ -701,7 +712,7 @@ def export_recordings_batch(
     sanitized_images: list[str | None] = []
     for item in body.items:
         existing_image, image_validation_error = _sanitize_existing_image(
-            item.image_path
+            request, item.image_path
         )
         if image_validation_error is not None:
             return image_validation_error
@@ -849,7 +860,9 @@ def export_recording(
 
     playback_source = body.source
     friendly_name = body.name
-    existing_image, image_validation_error = _sanitize_existing_image(body.image_path)
+    existing_image, image_validation_error = _sanitize_existing_image(
+        request, body.image_path
+    )
     if image_validation_error is not None:
         return image_validation_error
 
@@ -1044,7 +1057,9 @@ def export_recording_custom(
 
     playback_source = body.source
     friendly_name = body.name
-    existing_image, image_validation_error = _sanitize_existing_image(body.image_path)
+    existing_image, image_validation_error = _sanitize_existing_image(
+        request, body.image_path
+    )
     if image_validation_error is not None:
         return image_validation_error
     ffmpeg_input_args = body.ffmpeg_input_args

frigate/api/media.py

@@ -384,8 +384,13 @@ async def get_snapshot_from_recording(
         mime_type = "png" if format == "png" else "jpeg"
         config: FrigateConfig = request.app.frigate_config
 
-        image_data = get_image_from_recording(
-            config.ffmpeg, recording.path, time_in_segment, codec, height
+        image_data = await asyncio.to_thread(
+            get_image_from_recording,
+            config.ffmpeg,
+            recording.path,
+            time_in_segment,
+            codec,
+            height,
         )
 
         if not image_data:
@@ -439,8 +444,12 @@ async def submit_recording_snapshot_to_plus(
         config: FrigateConfig = request.app.frigate_config
         recording: Recordings = recording_query.get()
         time_in_segment = frame_time - recording.start_time
-        image_data = get_image_from_recording(
-            config.ffmpeg, recording.path, time_in_segment, "png"
+        image_data = await asyncio.to_thread(
+            get_image_from_recording,
+            config.ffmpeg,
+            recording.path,
+            time_in_segment,
+            "png",
         )
 
         if not image_data:

frigate/comms/config_updater.py

@@ -1,6 +1,7 @@
 """Facilitates communication between processes."""
 
 import multiprocessing as mp
+import threading
 from _pickle import UnpicklingError
 from multiprocessing.synchronize import Event as MpEvent
 from typing import Any
@@ -18,11 +19,13 @@ class ConfigPublisher:
         self.socket = self.context.socket(zmq.PUB)
         self.socket.bind(SOCKET_PUB_SUB)
         self.stop_event: MpEvent = mp.Event()
+        self.lock = threading.Lock()
 
     def publish(self, topic: str, payload: Any) -> None:
         """There is no communication back to the processes."""
-        self.socket.send_string(topic, flags=zmq.SNDMORE)
-        self.socket.send_pyobj(payload)
+        with self.lock:
+            self.socket.send_string(topic, flags=zmq.SNDMORE)
+            self.socket.send_pyobj(payload)
 
     def stop(self) -> None:
         self.stop_event.set()

frigate/comms/embeddings_updater.py

@@ -1,6 +1,7 @@
 """Facilitates communication between processes."""
 
 import logging
+import threading
 from collections.abc import Callable
 from enum import Enum
 from typing import Any
@@ -79,14 +80,21 @@ class EmbeddingsRequestor:
         self.context = zmq.Context()
         self.socket = self.context.socket(zmq.REQ)
         self.socket.connect(SOCKET_REP_REQ)
+        self.lock = threading.Lock()
 
     def send_data(self, topic: str, data: Any) -> Any:
         """Sends data and then waits for reply."""
+        # an overlapping call fails fast so a slow reply can't stall the API
+        if not self.lock.acquire(blocking=False):
+            return ""
+
         try:
             self.socket.send_json((topic, data))
             return self.socket.recv_json()
         except zmq.ZMQError:
             return ""
+        finally:
+            self.lock.release()
 
     def stop(self) -> None:
         self.socket.close()

frigate/comms/inter_process.py

@@ -79,14 +79,16 @@ class InterProcessRequestor:
         self.context = zmq.Context()
         self.socket = self.context.socket(zmq.REQ)
         self.socket.connect(SOCKET_REP_REQ)
+        self.lock = threading.Lock()
 
     def send_data(self, topic: str, data: Any) -> Any:
         """Sends data and then waits for reply."""
-        try:
-            self.socket.send_json((topic, data))
-            return self.socket.recv_json()
-        except zmq.ZMQError:
-            return ""
+        with self.lock:
+            try:
+                self.socket.send_json((topic, data))
+                return self.socket.recv_json()
+            except zmq.ZMQError:
+                return ""
 
     def stop(self) -> None:
         self.socket.close(linger=0)

frigate/comms/webpush.py

@@ -41,6 +41,20 @@ class PushNotification:
     ttl: int = 0
 
 
+def _build_web_pushers(user: str, subs: list[dict[str, Any]]) -> list[WebPusher]:
+    """Build pushers for a user's stored subscriptions, skipping unusable ones."""
+    pushers: list[WebPusher] = []
+
+    for sub in subs:
+        # WebPusher decodes the stored keys and raises on malformed ones
+        try:
+            pushers.append(WebPusher(sub))
+        except Exception:
+            logger.warning("Skipping invalid notification subscription for %s", user)
+
+    return pushers
+
+
 class WebPushClient(Communicator):
     """Frigate wrapper for webpush client."""
 
@@ -56,6 +70,7 @@ class WebPushClient(Communicator):
             for c in self.config.cameras.values()
         }
         self.suspension_broadcaster: Callable[[str, Any, bool], None] | None = None
+        self.config_lock = threading.Lock()
         self.last_camera_notification_time: dict[str, float] = {
             c.name: 0  # type: ignore[misc]
             for c in self.config.cameras.values()
@@ -76,9 +91,9 @@ class WebPushClient(Communicator):
             User.select(User.username, User.notification_tokens).dicts().iterator()
         )
         for user in users:
-            self.web_pushers[user["username"]] = []
-            for sub in user["notification_tokens"]:
-                self.web_pushers[user["username"]].append(WebPusher(sub))
+            self.web_pushers[user["username"]] = _build_web_pushers(
+                user["username"], user["notification_tokens"]
+            )
 
         # notification and auth config updater
         self.global_config_subscriber = ConfigSubscriber("config/")
@@ -146,10 +161,7 @@ class WebPushClient(Communicator):
                     User.username == user
                 ).execute()
 
-                self.web_pushers[user] = []
-
-                for sub in user_subs:
-                    self.web_pushers[user].append(WebPusher(sub))
+                self.web_pushers[user] = _build_web_pushers(user, user_subs)
 
                 logger.info(
                     f"Cleaned up {len(expired)} notification subscriptions for {user}"
@@ -198,29 +210,30 @@ class WebPushClient(Communicator):
 
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Wrapper for publishing when client is in valid state."""
-        # check for updated global config (notifications, auth)
-        while True:
-            config_topic, config_payload = (
-                self.global_config_subscriber.check_for_update()
-            )
-            if config_topic is None:
-                break
-            if config_topic == "config/notifications" and config_payload:
-                self.config.notifications = config_payload
-            elif config_topic == "config/auth":
-                if isinstance(config_payload, AuthConfig):
-                    self.config.auth = config_payload
+        with self.config_lock:
+            # check for updated global config (notifications, auth)
+            while True:
+                config_topic, config_payload = (
+                    self.global_config_subscriber.check_for_update()
+                )
+                if config_topic is None:
+                    break
+                if config_topic == "config/notifications" and config_payload:
+                    self.config.notifications = config_payload
+                elif config_topic == "config/auth":
+                    if isinstance(config_payload, AuthConfig):
+                        self.config.auth = config_payload
+                    self._refresh_user_cameras()
+
+            updates = self.config_subscriber.check_for_updates()
+
+            if "add" in updates:
+                for camera in updates["add"]:
+                    self.suspended_cameras[camera] = 0
+                    self.last_camera_notification_time[camera] = 0
+
                 self._refresh_user_cameras()
 
-        updates = self.config_subscriber.check_for_updates()
-
-        if "add" in updates:
-            for camera in updates["add"]:
-                self.suspended_cameras[camera] = 0
-                self.last_camera_notification_time[camera] = 0
-
-            self._refresh_user_cameras()
-
         if topic == "reviews":
             decoded = json.loads(payload)
             camera = decoded["before"]["camera"]

frigate/data_processing/post/object_descriptions.py

@@ -105,6 +105,10 @@ class ObjectDescriptionProcessor(PostProcessorApi):
                         logger.debug(f"{camera} sending early request to GenAI")
 
                         self.early_request_sent[data["id"]] = True
+
+                        if camera_config.objects.genai.debug_save_thumbnails:
+                            self._save_debug_thumbnails(data["id"])
+
                         # Copy thumbnails to avoid holding references after cleanup
                         thumbnails_copy = [
                             data["thumbnail"][:] if data.get("thumbnail") else None
@@ -302,26 +306,7 @@ class ObjectDescriptionProcessor(PostProcessorApi):
         )
 
         if camera_config.objects.genai.debug_save_thumbnails and num_thumbnails > 0:
-            logger.debug(f"Saving {num_thumbnails} thumbnails for event {event_id}")
-
-            Path(os.path.join(CLIPS_DIR, f"genai-requests/{event_id}")).mkdir(
-                parents=True, exist_ok=True
-            )
-
-            for idx, data in enumerate(self.tracked_events[event_id], 1):
-                jpg_bytes: bytes | None = data["thumbnail"]
-
-                if jpg_bytes is None:
-                    logger.warning(f"Unable to save thumbnail {idx} for {event_id}.")
-                else:
-                    with open(
-                        os.path.join(
-                            CLIPS_DIR,
-                            f"genai-requests/{event_id}/{idx}.jpg",
-                        ),
-                        "wb",
-                    ) as j:
-                        j.write(jpg_bytes)
+            self._save_debug_thumbnails(event_id)
 
         # Generate the description. Call happens in a thread since it is network bound.
         threading.Thread(
@@ -337,6 +322,30 @@ class ObjectDescriptionProcessor(PostProcessorApi):
         # Clean up tracked events and early request state
         self.cleanup_event(event_id)
 
+    def _save_debug_thumbnails(self, event_id: str) -> None:
+        """Write the tracked thumbnails for an event to the genai-requests dir."""
+        tracked = self.tracked_events.get(event_id, [])
+        logger.debug(f"Saving {len(tracked)} thumbnails for event {event_id}")
+
+        Path(os.path.join(CLIPS_DIR, f"genai-requests/{event_id}")).mkdir(
+            parents=True, exist_ok=True
+        )
+
+        for idx, data in enumerate(tracked, 1):
+            jpg_bytes: bytes | None = data["thumbnail"]
+
+            if jpg_bytes is None:
+                logger.warning(f"Unable to save thumbnail {idx} for {event_id}.")
+            else:
+                with open(
+                    os.path.join(
+                        CLIPS_DIR,
+                        f"genai-requests/{event_id}/{idx}.jpg",
+                    ),
+                    "wb",
+                ) as j:
+                    j.write(jpg_bytes)
+
     def _genai_embed_description(self, event: Event, thumbnails: list[bytes]) -> None:
         """Embed the description for an event."""
         start = datetime.datetime.now().timestamp()

frigate/data_processing/real_time/custom_classification.py

@@ -193,12 +193,7 @@ class CustomStateClassificationProcessor(DeferredRealtimeProcessorApi):
         return None
 
     def process_frame(self, frame_data: dict[str, Any], frame: np.ndarray) -> None:
-        if (
-            not self.model_config.name
-            or not self.model_config.state_config
-            or not self.tensor_input_details
-            or not self.tensor_output_details
-        ):
+        if not self.model_config.name or not self.model_config.state_config:
             return
 
         if self.metrics and self.model_config.name in self.metrics.classification_cps:
@@ -541,12 +536,7 @@ class CustomObjectClassificationProcessor(DeferredRealtimeProcessorApi):
         return best_label, avg_score
 
     def process_frame(self, obj_data: dict[str, Any], frame: np.ndarray) -> None:
-        if (
-            not self.model_config.name
-            or not self.model_config.object_config
-            or not self.tensor_input_details
-            or not self.tensor_output_details
-        ):
+        if not self.model_config.name or not self.model_config.object_config:
             return
 
         if self.metrics and self.model_config.name in self.metrics.classification_cps:

frigate/test/http_api/test_http_camera_access.py

@@ -417,6 +417,77 @@ class TestGo2rtcStreamAccess(BaseTestHttp):
         assert resp.status_code == 400
         assert resp.json().get("success") is False
 
+    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("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "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,226 @@
+"""Tests for go2rtc stream source restrictions on config/save and config/set."""
+
+import contextlib
+import copy
+import os
+import tempfile
+from unittest.mock import patch
+
+import ruamel.yaml
+
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+from frigate.util.env import _CONTAINER_ENV, FRIGATE_ENV_VARS, reload_sources
+
+# 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},
+        },
+    },
+}
+
+
+@contextlib.contextmanager
+def _frigate_vars(values: dict[str, str]):
+    # config parsing rebuilds FRIGATE_ENV_VARS from its sources
+    with patch.dict(_CONTAINER_ENV, values), patch.dict(FRIGATE_ENV_VARS, values):
+        yield
+    reload_sources(warn=False)
+
+
+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)
+
+        # the override is read once at import; make sure it is off here
+        exec_patch = patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", None)
+        exec_patch.start()
+        self.addCleanup(exec_patch.stop)
+
+    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("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "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 _frigate_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 _frigate_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_export.py

@@ -4,6 +4,7 @@ import tempfile
 import zipfile
 from unittest.mock import patch
 
+from frigate.const import CLIPS_DIR
 from frigate.jobs.export import (
     ExportJob,
     get_export_job_manager,
@@ -1051,6 +1052,65 @@ class TestHttpExport(BaseTestHttp):
         assert response.status_code == 400
         assert ExportCase.select().count() == 0
 
+    def _batch_export_with_image(self, image_path: str, role: str):
+        with patch(
+            "frigate.api.export.start_export_job",
+            side_effect=lambda _config, job: job.id,
+        ) as start_export_job:
+            with AuthTestClient(self.app) as client:
+                response = client.post(
+                    "/exports/batch",
+                    headers={"remote-user": role, "remote-role": role},
+                    json={
+                        "items": [
+                            {
+                                "camera": "front_door",
+                                "start_time": 110,
+                                "end_time": 150,
+                                "image_path": image_path,
+                            }
+                        ],
+                    },
+                )
+
+        return response, start_export_job
+
+    def test_batch_export_restricted_role_rejects_unreadable_image_path(self):
+        self._insert_recording("rec-front", "front_door", 100, 400)
+        self.app.frigate_config.auth.roles["limited_user"] = ["front_door"]
+
+        for image_path in (
+            f"{CLIPS_DIR}/review/thumb-backyard-123.456-abc.webp",
+            f"{CLIPS_DIR}/thumbs/backyard/123.webp",
+            f"{CLIPS_DIR}/faces/someone/1.webp",
+            f"{CLIPS_DIR}/custom/thumb.jpg",
+        ):
+            response, _ = self._batch_export_with_image(image_path, "limited_user")
+
+            assert response.status_code == 400, image_path
+
+    def test_batch_export_restricted_role_accepts_own_camera_image_path(self):
+        self._insert_recording("rec-front", "front_door", 100, 400)
+        self.app.frigate_config.auth.roles["limited_user"] = ["front_door"]
+        image_path = f"{CLIPS_DIR}/review/thumb-front_door-123.456-abc.webp"
+
+        response, start_export_job = self._batch_export_with_image(
+            image_path, "limited_user"
+        )
+
+        assert response.status_code == 202
+        assert start_export_job.call_args.args[1].image_path == image_path
+
+    def test_batch_export_unrestricted_roles_accept_custom_image_path(self):
+        self._insert_recording("rec-front", "front_door", 100, 400)
+        image_path = f"{CLIPS_DIR}/custom/thumb.jpg"
+
+        for role in ("admin", "viewer"):
+            response, start_export_job = self._batch_export_with_image(image_path, role)
+
+            assert response.status_code == 202, role
+            assert start_export_job.call_args.args[1].image_path == image_path
+
     def test_batch_export_non_admin_can_queue(self):
         self._insert_recording("rec-front", "front_door", 100, 400)
 

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/http_api/test_http_media.py

@@ -1,5 +1,7 @@
 """Unit tests for recordings/media API endpoints."""
 
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from unittest.mock import patch
@@ -2019,3 +2021,79 @@ class TestHttpMedia(BaseTestHttp):
 
             assert response.status_code == 200
             assert response.json() == [{"start_time": 1010, "end_time": 1030}]
+
+    def test_recording_snapshot_does_not_block_event_loop(self):
+        self._insert_recording("snapshot", 1000, 1010)
+        started = threading.Event()
+        release = threading.Event()
+        finished = threading.Event()
+
+        def slow_image(*args):
+            started.set()
+            release.wait(timeout=5)
+            finished.set()
+            return b"jpeg"
+
+        with (
+            AuthTestClient(self.app) as client,
+            patch("frigate.api.media.get_image_from_recording", slow_image),
+            ThreadPoolExecutor(max_workers=1) as executor,
+        ):
+            snapshot = executor.submit(
+                client.get, "/front_door/recordings/1005/snapshot.jpg"
+            )
+            self.assertTrue(started.wait(timeout=2))
+            version = client.get("/version")
+            blocked = finished.is_set()
+            release.set()
+
+            self.assertEqual(version.status_code, 200)
+            self.assertFalse(blocked)
+            self.assertEqual(snapshot.result().status_code, 200)
+
+    def test_recording_snapshot_hit_returns_image(self):
+        self._insert_recording("snapshot", 1000, 1010)
+
+        with (
+            AuthTestClient(self.app) as client,
+            patch("frigate.api.media.get_image_from_recording", return_value=b"jpeg"),
+        ):
+            response = client.get("/front_door/recordings/1005/snapshot.jpg")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers["Content-Type"], "image/jpeg")
+        self.assertEqual(response.content, b"jpeg")
+
+    def test_recording_snapshot_misses_return_404(self):
+        self._insert_recording("snapshot", 1000, 1010)
+
+        with AuthTestClient(self.app) as client:
+            absent = client.get("/front_door/recordings/2000/snapshot.jpg")
+            rounded_miss = client.get("/front_door/recordings/1010.5/snapshot.jpg")
+
+        self.assertEqual(absent.status_code, 404)
+        self.assertEqual(rounded_miss.status_code, 404)
+
+    def test_plus_snapshot_miss_returns_404_without_image_lookup(self):
+        with (
+            AuthTestClient(self.app) as client,
+            patch("frigate.api.media.get_image_from_recording") as image,
+        ):
+            response = client.post("/front_door/plus/1005")
+
+        self.assertEqual(response.status_code, 404)
+        image.assert_not_called()
+
+    def test_plus_snapshot_hit_uploads_image(self):
+        self._insert_recording("snapshot", 1000, 1010)
+
+        with (
+            AuthTestClient(self.app) as client,
+            patch("frigate.api.media.get_image_from_recording", return_value=b"png"),
+            patch("frigate.api.media.cv2.imdecode", return_value="frame"),
+            patch.object(self.app.frigate_config.plus_api, "upload_image") as upload,
+        ):
+            response = client.post("/front_door/plus/1005")
+
+        self.assertEqual(response.status_code, 200)
+        upload.assert_called_once_with("frame", "front_door")

frigate/test/http_api/test_http_users.py

@@ -0,0 +1,33 @@
+"""Tests for user creation."""
+
+from unittest.mock import MagicMock
+
+from frigate.models import User
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+PASSWORD = "a-valid-password-123"
+
+
+class TestCreateUser(BaseTestHttp):
+    def setUp(self):
+        super().setUp([User])
+        self.app = super().create_app()
+        self.app.config_publisher = MagicMock()
+
+    def _create(self, username: str):
+        with AuthTestClient(self.app) as client:
+            return client.post(
+                "/users",
+                json={"username": username, "password": PASSWORD, "role": "viewer"},
+            )
+
+    def test_rejects_dot_only_usernames(self):
+        # browsers resolve these as URL path segments, so the per-user
+        # endpoints can never be reached for them
+        for username in (".", "..", "..."):
+            assert self._create(username).status_code == 400
+            assert User.get_or_none(User.username == username) is None
+
+    def test_accepts_username_containing_dots(self):
+        assert self._create("john.doe_1").status_code == 200
+        assert User.get_or_none(User.username == "john.doe_1") is not None

frigate/test/test_comms_thread_safety.py

@@ -0,0 +1,193 @@
+"""Tests for sharing one zmq socket wrapper across threads."""
+
+import os
+import random
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+import zmq
+
+from frigate.comms import config_updater, embeddings_updater, inter_process
+from frigate.comms.config_updater import ConfigPublisher, ConfigSubscriber
+from frigate.comms.embeddings_updater import EmbeddingsRequestor
+from frigate.comms.inter_process import InterProcessRequestor
+from frigate.comms.webpush import WebPushClient
+
+THREADS = 8
+CALLS = 100
+
+# nothing reads until the end, so the total stays under the default zmq HWM
+PUBLISH_CALLS = 100
+
+
+class TestSharedRequestor(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        self.address = f"ipc://{os.path.join(self.tmp.name, 'comms')}"
+        self.stop = threading.Event()
+        self.context = zmq.Context()
+        self.responder = self.context.socket(zmq.REP)
+        self.responder.bind(self.address)
+        self.responder_thread = threading.Thread(target=self._respond)
+        self.responder_thread.start()
+
+    def tearDown(self) -> None:
+        self.stop.set()
+        self.responder_thread.join()
+        self.responder.close(linger=0)
+        self.context.destroy(linger=0)
+        self.tmp.cleanup()
+
+    def _respond(self) -> None:
+        while not self.stop.is_set():
+            ready, _, _ = zmq.select([self.responder], [], [], 0.1)
+
+            if ready:
+                self.responder.recv_json()
+                time.sleep(random.uniform(0, 0.002))
+                self.responder.send_json(["ok"])
+
+    def _call_from_threads(self, requestor) -> list:
+        results: list = []
+
+        def call() -> None:
+            for _ in range(CALLS):
+                # jitter lands some calls on the instant another reply arrives
+                time.sleep(random.uniform(0, 0.002))
+                results.append(requestor.send_data("topic", {"key": "value"}))
+
+        threads = [threading.Thread(target=call, daemon=True) for _ in range(THREADS)]
+
+        for thread in threads:
+            thread.start()
+
+        for thread in threads:
+            thread.join(30)
+
+        self.assertFalse(any(thread.is_alive() for thread in threads))
+        return results
+
+    def test_inter_process_requestor_waits_for_overlapping_calls(self) -> None:
+        with patch.object(inter_process, "SOCKET_REP_REQ", self.address):
+            requestor = InterProcessRequestor()
+
+        results = self._call_from_threads(requestor)
+        self.assertEqual(results, [["ok"]] * THREADS * CALLS)
+        requestor.stop()
+
+    def test_embeddings_requestor_survives_overlapping_calls(self) -> None:
+        with patch.object(embeddings_updater, "SOCKET_REP_REQ", self.address):
+            requestor = EmbeddingsRequestor()
+
+        # an overlapping call may fail fast, but the socket has to stay usable
+        results = self._call_from_threads(requestor)
+        self.assertEqual(len(results), THREADS * CALLS)
+        self.assertEqual(requestor.send_data("topic", {}), ["ok"])
+        requestor.stop()
+
+
+class TestSharedConfigPublisher(unittest.TestCase):
+    def test_frames_stay_paired_across_threads(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            address = f"ipc://{os.path.join(tmp, 'config')}"
+
+            with patch.object(config_updater, "SOCKET_PUB_SUB", address):
+                publisher = ConfigPublisher()
+                subscriber = ConfigSubscriber("config/")
+
+            # a subscription that is still connecting drops messages
+            subscriber.socket.setsockopt(zmq.RCVTIMEO, 5000)
+
+            while True:
+                publisher.publish("config/ready", None)
+
+                if zmq.select([subscriber.socket], [], [], 0.05)[0]:
+                    break
+
+            def publish(index: int) -> None:
+                topic = f"config/cameras/camera_{index}"
+
+                for _ in range(PUBLISH_CALLS):
+                    publisher.publish(topic, topic)
+
+            threads = [
+                threading.Thread(target=publish, args=(i,)) for i in range(THREADS)
+            ]
+
+            for thread in threads:
+                thread.start()
+
+            for thread in threads:
+                thread.join()
+
+            publisher.publish("config/done", None)
+            received = 0
+
+            while True:
+                topic = subscriber.socket.recv_string()
+                payload = subscriber.socket.recv_pyobj()
+
+                if topic == "config/done":
+                    break
+
+                if topic != "config/ready":
+                    self.assertEqual(topic, payload)
+                    received += 1
+
+            self.assertEqual(received, THREADS * PUBLISH_CALLS)
+            subscriber.stop()
+            publisher.stop()
+
+
+class FakeConfigSubscriber:
+    """Records whether two threads read at the same time."""
+
+    def __init__(self) -> None:
+        self.reading = threading.Lock()
+        self.overlapped = False
+
+    def _read(self) -> None:
+        if not self.reading.acquire(blocking=False):
+            self.overlapped = True
+            return
+
+        time.sleep(0.001)
+        self.reading.release()
+
+    def check_for_update(self) -> tuple[None, None]:
+        self._read()
+        return (None, None)
+
+    def check_for_updates(self) -> dict:
+        self._read()
+        return {}
+
+
+class TestSharedWebPushClient(unittest.TestCase):
+    def test_config_subscribers_read_by_one_thread_at_a_time(self) -> None:
+        client = WebPushClient.__new__(WebPushClient)
+        client.config_lock = threading.Lock()
+        client.global_config_subscriber = FakeConfigSubscriber()
+        client.config_subscriber = FakeConfigSubscriber()
+
+        def publish() -> None:
+            for _ in range(20):
+                client.publish("topic", "payload")
+
+        threads = [threading.Thread(target=publish) for _ in range(THREADS)]
+
+        for thread in threads:
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+
+        self.assertFalse(client.global_config_subscriber.overlapped)
+        self.assertFalse(client.config_subscriber.overlapped)
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_env.py

@@ -54,14 +54,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):
@@ -251,6 +247,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")
+
 
 class TestVariableSources(unittest.TestCase):
     """Precedence between the sources that feed FRIGATE_ENV_VARS."""

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/test/test_process.py

@@ -0,0 +1,81 @@
+import multiprocessing as mp
+import os
+import sys
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import frigate.__main__ as frigate_main
+import frigate.log
+from frigate.config import FrigateConfig
+from frigate.util.process import FrigateProcess
+
+ENV_KEY = "FRIGATE_TEST_PROCESS_ENV"
+
+
+class EnvReportingProcess(FrigateProcess):
+    def __init__(self, stop_event, result_queue):
+        super().__init__(stop_event, 0, name="env_reporter", daemon=True)
+        self.result_queue = result_queue
+
+    def run(self) -> None:
+        self.pre_run_setup()
+        self.result_queue.put(os.environ.get(ENV_KEY))
+
+
+def _run_main_and_report(result_queue) -> None:
+    """Run frigate's main() with the config load and app faked out.
+
+    Runs in a fresh interpreter so no earlier test has started the forkserver.
+    """
+    mp.set_start_method("forkserver", force=True)
+
+    def load(**kwargs):
+        os.environ[ENV_KEY] = "applied"
+        return SimpleNamespace()
+
+    def setup_logging(manager):
+        frigate.log.log_listener = SimpleNamespace(queue=manager.Queue())
+
+    class FakeApp:
+        def __init__(self, config, manager, stop_event):
+            self.stop_event = stop_event
+
+        def start(self):
+            child_queue = mp.Queue()
+            process = EnvReportingProcess(self.stop_event, child_queue)
+            process.start()
+            result_queue.put(child_queue.get(timeout=30))
+            process.join(timeout=10)
+
+    try:
+        with (
+            patch.object(sys, "argv", ["frigate"]),
+            patch.object(FrigateConfig, "load", load),
+            patch("frigate.__main__.setup_logging", setup_logging),
+            patch("frigate.__main__.FrigateApp", FakeApp),
+        ):
+            frigate_main.main()
+    except Exception as err:
+        result_queue.put(f"error: {err!r}")
+
+
+class TestWorkerEnvironment(unittest.TestCase):
+    def test_env_applied_by_config_load_reaches_worker_processes(self):
+        ctx = mp.get_context("spawn")
+        result_queue = ctx.Queue()
+        runner = ctx.Process(target=_run_main_and_report, args=(result_queue,))
+        runner.start()
+
+        try:
+            self.assertEqual(result_queue.get(timeout=60), "applied")
+        finally:
+            runner.join(timeout=10)
+
+            if runner.is_alive():
+                runner.terminate()
+                runner.join()
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_update_yaml.py

@@ -6,7 +6,7 @@ import unittest
 
 from ruamel.yaml import YAML
 
-from frigate.util.builtin import update_yaml_file_bulk
+from frigate.util.builtin import clear_orphaned_comments, update_yaml_file_bulk
 
 
 class TestUpdateYaml(unittest.TestCase):
@@ -178,6 +178,27 @@ class TestUpdateYaml(unittest.TestCase):
         assert data["cameras"]["cam1"]["detect"]["fps"] == 5
         assert "# tuned for the pi" in self._read()
 
+    def test_delete_only_camera_with_comment_on_cameras_key(self):
+        """Deleting the only camera stays parseable when the cameras key
+        carries the default config's trailing comment."""
+        self._write(
+            "cameras: # No cameras defined, UI wizard should be used\n"
+            "  cam1:\n"
+            "    enabled: true\n"
+            "version: 0.18-0\n"
+        )
+        yaml = YAML()
+        with open(self.config_path) as f:
+            data = yaml.load(f)
+        del data["cameras"]["cam1"]
+        clear_orphaned_comments(data["cameras"], data, "cameras")
+        with open(self.config_path, "w") as f:
+            yaml.dump(data, f)
+
+        data = self._load()
+        assert data["cameras"] == {}
+        assert data["version"] == "0.18-0"
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

frigate/test/test_webpush_registration.py

@@ -1,8 +1,10 @@
 """Tests for push notification subscription validation."""
 
 import unittest
+from base64 import urlsafe_b64encode
 
 from frigate.api.notification import _validate_push_endpoint, _validate_subscription
+from frigate.comms.webpush import _build_web_pushers
 
 VALID_ENDPOINTS = [
     "https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu",
@@ -148,3 +150,16 @@ class TestValidateSubscription(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestBuildWebPushers(unittest.TestCase):
+    def test_skips_subscriptions_with_malformed_keys(self):
+        bad = _subscription(VALID_ENDPOINTS[0])
+        good = _subscription(VALID_ENDPOINTS[0])
+        good["keys"]["p256dh"] = urlsafe_b64encode(b"\x04" + bytes(64)).decode()
+
+        with self.assertLogs("frigate.comms.webpush", level="WARNING"):
+            pushers = _build_web_pushers("viewer", [bad, good])
+
+        self.assertEqual(len(pushers), 1)
+        self.assertEqual(pushers[0].receiver_key, b"\x04" + bytes(64))

frigate/util/env.py

@@ -165,12 +165,25 @@ def reload_sources(warn: bool = True) -> None:
     _rebuild(warn)
 
 
+# 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"})
+
+
 def apply_config_env_vars(values: Mapping[str, object]) -> None:
     """Install the environment_vars block as the lowest priority source.
 
-    Unprefixed keys only set os.environ.
+    Unprefixed keys only set os.environ. Protected security overrides are
+    skipped.
     """
     for key, value in values.items():
+        if key.upper() in _PROTECTED_ENV_VARS:
+            logger.warning(
+                "Ignoring %s in environment_vars, it must be set in the container environment",
+                key,
+            )
+            continue
+
         resolved = str(value)
 
         if key.startswith("FRIGATE_"):

frigate/util/services.py

@@ -14,6 +14,7 @@ import subprocess as sp
 import time
 import traceback
 from datetime import datetime
+from enum import Enum
 from typing import Any
 
 import cv2
@@ -29,6 +30,7 @@ from frigate.const import (
     SHM_FRAMES_VAR,
 )
 from frigate.util.builtin import clean_camera_user_pass, escape_special_characters
+from frigate.util.env import UnknownVariableError, substitute_frigate_vars
 
 logger = logging.getLogger(__name__)
 
@@ -511,8 +513,9 @@ def get_intel_gpu_stats(
     Each DRM client FD exposes monotonic per-engine busy counters via
     /proc/<pid>/fdinfo/<fd>. For i915 this requires kernel 6.5 or newer:
     earlier kernels omit the per-engine counters whenever GuC submission is
-    active, which is the default on 12th gen and newer. Xe has exposed them
-    since its first release. We sample twice and divide busy-time deltas by
+    active, which is the default on 12th gen and newer. i915 never exposes
+    them on GPUs older than Broadwell (Gen8). Xe has exposed them since its
+    first release. We sample twice and divide busy-time deltas by
     wall-clock to derive utilization. Render/3D and Compute are pooled into
     "compute"; Video and VideoEnhance into "dec". Overall "gpu" is the sum of
     those pools (clamped to 100%).
@@ -581,10 +584,11 @@ def get_intel_gpu_stats(
         # i915 suppresses per-client engine counters while GuC submission is
         # active on kernels older than 6.5 (kernel commit 1324680a80eb lifted
         # this), which covers stock Debian 12 and Ubuntu 22.04 on 12th gen
-        # and newer.
+        # and newer. It never publishes them for pre-Broadwell GPUs.
         logger.warning(
             "Unable to collect Intel GPU stats: found %d DRM client(s) for %s but "
-            "no per-engine counters. Kernel 6.5 or newer is required.",
+            "the kernel reports no usage for them. GPUs older than Broadwell "
+            "(5th gen) are not supported. 12th gen and newer need kernel 6.5+",
             len(snapshot_a),
             "/".join(sorted({client["driver"] for client in snapshot_a.values()})),
         )
@@ -1027,6 +1031,119 @@ 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`.
+    """
+    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 UnknownVariableError:
+                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)