Security patches (#24658)

* update version * Restrict ffmpeg: go2rtc sources added through the API go2rtc expands ffmpeg: sources into an exec: command line built from the source's #key=value parameters, so the echo/expr/exec prefix check on PUT /go2rtc/streams could be bypassed to run arbitrary ffmpeg arguments. Sources submitted through the API now only accept the ffmpeg: parameters the UI and docs use (video, audio, hardware with preset names, plus numeric width/height/rotate/bitrate/timeout and async). Quotes, whitespace, #raw, #input, #drawtext and unknown parameters are rejected unless GO2RTC_ALLOW_ARBITRARY_EXEC is set. Sources defined in the config file are unchanged. Fixes GHSA-qjjw-3v66-279f Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Only allow existing streams on proxied go2rtc live paths nginx proxies /live/mse/api/ws, /live/webrtc/api/ws and /api/go2rtc/webrtc straight to go2rtc, which creates a new ad hoc stream from any src that is not an existing stream name and starts the producer it describes. Any authenticated user, including the built-in viewer role, could use this to launch arbitrary go2rtc sources without going through the restricted-source checks on PUT /go2rtc/streams. The /auth subrequest now denies these paths unless every src names an existing stream: a camera or its live stream, a go2rtc stream from the config, or a stream go2rtc reports at runtime. This applies to every role, and the runtime lookup fails closed when go2rtc cannot be reached. Fixes GHSA-93xj-hg88-ffg7 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Ignore GO2RTC_ALLOW_ARBITRARY_EXEC in environment_vars environment_vars installs every key into the Frigate process environment at startup, so an admin could enable GO2RTC_ALLOW_ARBITRARY_EXEC from the config and turn off the restricted go2rtc source checks on the API. The override is meant to be set by the operator in the container environment (or the Home Assistant add-on options), so it is now skipped with a warning when it appears in environment_vars. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Apply go2rtc source restrictions to config/save and config/set go2rtc.streams written through config/save or config/set were only filtered at startup, which blocks echo/expr/exec and the map form but not the ffmpeg: parameters that PUT /go2rtc/streams now rejects. Saving a stream this way and restarting bypassed the API restriction. Both endpoints now reject a change that adds a go2rtc stream source that fails is_restricted_go2rtc_api_source, or a map-form source, unless GO2RTC_ALLOW_ARBITRARY_EXEC is set. Sources already in the saved config for the same stream are left alone, so streams configured on disk keep working and resubmitting the whole file from the config editor succeeds. config/set restores the previous file when it rejects a change. Follow-up to GHSA-qjjw-3v66-279f Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Match go2rtc stream-proxy paths the way nginx does /auth compared the raw $request_uri path against the three go2rtc stream-proxy paths exactly, while nginx matches its prefix locations against the normalized URI. Encoded characters, repeated slashes, dot segments or a suffix on the path skipped the stream check even though nginx still proxied the request to go2rtc. The go2rtc stream locations now set a variable that the auth subrequest forwards as X-Go2rtc-Stream-Proxy (client headers are not forwarded, so it cannot be spoofed). As a fallback, /auth also normalizes the path like nginx (percent-decoding, merging slashes, resolving dot segments) and prefix matches it, and splits the raw URI by hand since urlparse reads a leading // as a host. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Parse config stream changes with Frigate's YAML loader The go2rtc stream check on config/save and config/set parsed the config with the safe loader and treated a parse error as an empty config, while Frigate and the go2rtc startup script use the round-trip loader. Input that only the round-trip loader accepts, such as a tagged root map, hid new stream sources from the check. The check now uses the same round-trip loader and rejects the request when the new config cannot be parsed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Resolve go2rtc stream placeholders the same way everywhere The go2rtc startup script expanded stream sources with str.format(), which also supports attribute and index access, while the config API check used substitute_frigate_vars and fell back to the raw string when a placeholder could not be resolved. A placeholder the API check could not resolve could therefore expand into a restricted source at startup. The startup script now uses substitute_frigate_vars, like the rest of the config, and the API check rejects a stream source whose placeholders cannot be resolved. The startup script's own copy of the FRIGATE_* env var map is removed; the shared one also honors CREDENTIALS_DIRECTORY. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Accept codec sample rates in ffmpeg: go2rtc sources The ffmpeg: parameter allowlist only accepted plain names for video, audio and hardware, so the documented HomeKit source with #audio=opus/16000 was rejected by PUT /go2rtc/streams, config/save and config/set. Names may now carry a numeric /rate suffix. A rate without a name is still rejected. All ffmpeg: sources in the docs pass the check. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Restore the environment after go2rtc exec override tests Tests that removed or set GO2RTC_ALLOW_ARBITRARY_EXEC did not restore a value that was already present, so later tests could run with a changed security setting. They now use patch.dict on os.environ. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Revert the go2rtc startup script to str.format substitution Restores create_config.py to its state on master. The config API check rejects any stream source whose {FRIGATE_*} placeholders it cannot resolve with substitute_frigate_vars, and for every source it accepts, str.format() at startup either produces the same string or fails and stops go2rtc startup. Changing the startup script is therefore not needed for the fix, and leaving it alone avoids a conflict with dev, which already moved this script to the new env helpers. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * 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> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-15 01:53:02 +03:00 · 2026-10-14 15:10:05 -04:00
parent d3b2af932c
commit a3ca8742b0
13 changed files with 825 additions and 24 deletions
@@ -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."""
@@ -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()
@@ -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"
@@ -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()
@@ -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()