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>

frigate/api/auth.py

@@ -13,6 +13,7 @@ from datetime import datetime
 from pathlib import Path
 from urllib.parse import parse_qs, urlparse
 
+import requests
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from fastapi.responses import JSONResponse, RedirectResponse
 from joserfc import jwt
@@ -1138,6 +1139,38 @@ 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 —
@@ -1155,11 +1188,12 @@ GO2RTC_STREAM_PROXY_PATHS = frozenset(
 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
+    Returns 403 when any `src` named in `original_url` is not an existing go2rtc
-    camera outside the role's allow-list (or when no `src` is provided on a
+    stream (for every role), or when it resolves to a camera outside a
-    stream-proxy path), otherwise None. Mirrors the resolution logic in
+    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.
     """
@@ -1171,12 +1205,18 @@ def deny_response_for_go2rtc_stream(
         return None
 
     frigate_config = request.app.frigate_config
+    sources = parse_qs(parsed.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/test/http_api/test_http_go2rtc_proxy_auth.py

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

frigate/test/test_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": {
@@ -54,6 +58,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 +188,77 @@ 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
+        )
+
 
 if __name__ == "__main__":
     unittest.main()