check go2rtc stream sources on the internal port

auth() returned admin for internal port requests before the go2rtc src check ran, so anyone who could reach port 5000 could send an unknown src to /api/ws, and go2rtc created a stream from it and started its producer, including ffmpeg sources with #raw= arguments. The internal port now rejects any src that isn't an existing stream, the same as the external port, and still skips the per-camera role checks.

frigate/api/auth.py

@@ -739,6 +739,13 @@ def auth(request: Request):
         int(request.headers.get("x-server-port", default=0))
         == request.app.auth_internal_port
     ):
+        # go2rtc runs the producer an unknown src describes for any caller
+        deny_status = deny_response_for_go2rtc_stream(
+            request.headers.get("x-original-url"), None, request
+        )
+        if deny_status is not None:
+            return Response("", status_code=deny_status)
+
         success_response.headers["remote-user"] = "anonymous"
         success_response.headers["remote-role"] = "admin"
         return success_response

frigate/test/http_api/test_http_auth_internal_port.py

@@ -5,6 +5,7 @@ import tempfile
 import unittest
 from unittest.mock import MagicMock, Mock, patch
 
+import requests
 import ruamel.yaml
 from fastapi import Request
 
@@ -234,6 +235,51 @@ class TestAuthInternalPort(BaseTestHttp):
         finally:
             os.unlink(config_path)
 
+    @patch(
+        "frigate.api.auth.requests.get",
+        side_effect=requests.ConnectionError("go2rtc unavailable"),
+    )
+    def test_internal_port_rejects_unknown_go2rtc_src(self, _mock_get):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            for url in (
+                "/live/mse/api/ws?src=exec:id",
+                "/live/webrtc/api/ws?src=ffmpeg:x%23raw=-f",
+                "/api/go2rtc/webrtc?src=front_door&src=unknown",
+            ):
+                resp = client.get(
+                    "/auth",
+                    headers={"x-server-port": "5000", "x-original-url": url},
+                )
+                self.assertEqual(resp.status_code, 403, url)
+
+    def test_internal_port_allows_existing_go2rtc_stream(self):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            resp = client.get(
+                "/auth",
+                headers={
+                    "x-server-port": "5000",
+                    "x-original-url": "/live/mse/api/ws?src=front_door",
+                },
+            )
+
+            self.assertEqual(resp.status_code, 202)
+            self.assertEqual(resp.headers["remote-role"], "admin")
+
+    def test_internal_port_skips_go2rtc_check_for_other_paths(self):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            for url in ("/api/stats", "/api/go2rtc/streams?src=unknown"):
+                resp = client.get(
+                    "/auth",
+                    headers={"x-server-port": "5000", "x-original-url": url},
+                )
+                self.assertEqual(resp.status_code, 202, url)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)