Reject cross-origin websocket connections (#24714)

* reject cross-origin websocket handshakes

Neither /ws nor the jsmpeg websockets checked Origin, and the internal port grants admin to everyone, so any page a LAN user visited could open ws://frigate:5000/ws and send commands or read streams. Both servers now reject a browser handshake whose Origin host matches neither Host nor X-Forwarded-Host, and allow handshakes without an Origin for non-browser clients. nginx now forwards an outer proxy's X-Forwarded-Host instead of overwriting it, so proxies that rewrite Host and the Home Assistant integration's proxy still match.

* apply runtime role changes to jsmpeg streams

The output process never subscribed to config/auth, so role edits from the Roles UI (requires_restart 0) did not reach it and jsmpeg kept streaming cameras that had been removed from a role until restart. The output process now replaces its auth config on config/auth updates, which the camera and birdseye broadcast threads read through the shared config.

* tidy origin check docstring and auth update comment

* reject cross-origin go2rtc stream requests in /auth

The go2rtc websockets and the webrtc offer POST still accepted any Origin, since go2rtc runs with origin "*" and /auth grants admin on the internal port, so a page on another site could pull live video through a LAN user's browser. nginx now passes Origin, Host, X-Forwarded-Host and Upgrade to /auth, which rejects a websocket upgrade or go2rtc stream request with a foreign Origin before the internal-port early return. The hostname rule moves to frigate/util/origin.py so /auth and the ws4py servers share it.

* set Host in the nginx proxy doc examples

* drain all queued auth updates in the output process

* note that cards must stream through the integration proxy

* strip the query before matching go2rtc paths in the origin check

frigate/test/http_api/test_http_auth_internal_port.py

@@ -101,6 +101,71 @@ class TestAuthInternalPort(BaseTestHttp):
             self.assertEqual(resp.headers["remote-user"], "anonymous")
             self.assertEqual(resp.headers["remote-role"], "admin")
 
+    def _internal_auth(self, original_url: str, **headers: str):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            return client.get(
+                "/auth",
+                headers={
+                    "x-server-port": "5000",
+                    "x-original-url": original_url,
+                    "x-original-host": "frigate.lan:5000",
+                    **{k.replace("_", "-"): v for k, v in headers.items()},
+                },
+            )
+
+    def test_cross_origin_websocket_on_internal_port_denied(self):
+        resp = self._internal_auth(
+            "/ws", origin="https://evil.example", x_original_upgrade="websocket"
+        )
+        self.assertEqual(resp.status_code, 403)
+
+    def test_cross_origin_go2rtc_stream_denied(self):
+        resp = self._internal_auth(
+            "/live/mse/api/ws?src=front_door",
+            origin="https://evil.example",
+            x_original_upgrade="websocket",
+            x_go2rtc_stream_proxy="1",
+        )
+        self.assertEqual(resp.status_code, 403)
+
+    def test_cross_origin_webrtc_post_denied(self):
+        # the offer POST is not an upgrade, so the stream marker alone gates it
+        resp = self._internal_auth(
+            "/api/go2rtc/webrtc?src=front_door",
+            origin="https://evil.example",
+            x_go2rtc_stream_proxy="1",
+        )
+        self.assertEqual(resp.status_code, 403)
+
+    def test_same_origin_websocket_allowed(self):
+        resp = self._internal_auth(
+            "/live/mse/api/ws?src=front_door",
+            origin="http://frigate.lan:5000",
+            x_original_upgrade="websocket",
+            x_go2rtc_stream_proxy="1",
+        )
+        self.assertEqual(resp.status_code, 202)
+
+    def test_websocket_without_origin_allowed(self):
+        resp = self._internal_auth("/ws", x_original_upgrade="websocket")
+        self.assertEqual(resp.status_code, 202)
+
+    def test_forwarded_host_origin_allowed(self):
+        # Home Assistant integration proxy: Host is Frigate, Origin is HA
+        resp = self._internal_auth(
+            "/api/go2rtc/webrtc?src=front_door",
+            origin="https://ha.example.com:8123",
+            x_forwarded_host="ha.example.com:8123",
+            x_go2rtc_stream_proxy="1",
+        )
+        self.assertEqual(resp.status_code, 202)
+
+    def test_cross_origin_plain_http_not_checked(self):
+        resp = self._internal_auth("/api/events", origin="https://evil.example")
+        self.assertEqual(resp.status_code, 202)
+
     def test_external_port_requires_auth(self):
         app = self._create_app()
 

frigate/test/test_output_ws_auth.py

@@ -3,7 +3,7 @@
 import unittest
 from types import SimpleNamespace
 
-from frigate.config import FrigateConfig
+from frigate.config import AuthConfig, FrigateConfig
 from frigate.output.ws_auth import ws_has_camera_access
 
 
@@ -55,3 +55,11 @@ class TestWsHasCameraAccess(unittest.TestCase):
         self.assertFalse(
             ws_has_camera_access(self._make_ws("limited_user"), "birdseye", self.config)
         )
+
+    def test_replaced_auth_config_revokes_camera(self):
+        ws = self._make_ws("limited_user")
+        self.assertTrue(ws_has_camera_access(ws, "front_door", self.config))
+
+        self.config.auth = AuthConfig(roles={"limited_user": ["back_door"]})
+        self.assertFalse(ws_has_camera_access(ws, "front_door", self.config))
+        self.assertTrue(ws_has_camera_access(ws, "back_door", self.config))

frigate/test/test_ws_origin.py

@@ -0,0 +1,126 @@
+"""Tests for the same-origin check on WebSocket handshakes."""
+
+import unittest
+from unittest.mock import MagicMock
+
+from frigate.comms.ws import OriginCheckedWebSocketApplication, WebSocket
+from frigate.util.origin import origin_allowed
+
+
+def ws_origin_allowed(environ: dict[str, str]) -> bool:
+    return origin_allowed(
+        environ.get("HTTP_ORIGIN"),
+        environ.get("HTTP_HOST"),
+        environ.get("HTTP_X_FORWARDED_HOST"),
+    )
+
+
+class TestOriginAllowed(unittest.TestCase):
+    def test_missing_origin_allowed(self):
+        self.assertTrue(ws_origin_allowed({"HTTP_HOST": "frigate.lan"}))
+
+    def test_direct_access_same_origin(self):
+        # nginx forwards Host without the port and X-Forwarded-Host with it
+        environ = {
+            "HTTP_ORIGIN": "http://192.168.1.10:5000",
+            "HTTP_HOST": "192.168.1.10",
+            "HTTP_X_FORWARDED_HOST": "192.168.1.10:5000",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_default_port_origin(self):
+        environ = {
+            "HTTP_ORIGIN": "https://frigate.example.com",
+            "HTTP_HOST": "frigate.example.com",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_host_comparison_is_case_insensitive(self):
+        environ = {
+            "HTTP_ORIGIN": "https://Frigate.Example.com",
+            "HTTP_HOST": "frigate.example.COM",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_ipv6_host(self):
+        environ = {"HTTP_ORIGIN": "http://[::1]:8971", "HTTP_HOST": "[::1]"}
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_cross_origin_rejected(self):
+        environ = {
+            "HTTP_ORIGIN": "https://evil.example",
+            "HTTP_HOST": "frigate.lan",
+            "HTTP_X_FORWARDED_HOST": "frigate.lan:5000",
+        }
+        self.assertFalse(ws_origin_allowed(environ))
+
+    def test_null_origin_rejected(self):
+        self.assertFalse(
+            ws_origin_allowed({"HTTP_ORIGIN": "null", "HTTP_HOST": "frigate.lan"})
+        )
+
+    def test_proxy_rewriting_host_with_forwarded_host(self):
+        environ = {
+            "HTTP_ORIGIN": "https://frigate.example.com",
+            "HTTP_HOST": "10.0.0.5",
+            "HTTP_X_FORWARDED_HOST": "frigate.example.com",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_forwarded_host_list_uses_first_entry(self):
+        environ = {
+            "HTTP_ORIGIN": "https://frigate.example.com",
+            "HTTP_HOST": "10.0.0.5",
+            "HTTP_X_FORWARDED_HOST": "frigate.example.com, proxy.internal",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+        environ["HTTP_ORIGIN"] = "https://proxy.internal"
+        self.assertFalse(ws_origin_allowed(environ))
+
+    def test_home_assistant_ingress(self):
+        # HA core copies Host and sets X-Forwarded-Host to its own host
+        environ = {
+            "HTTP_ORIGIN": "https://ha.example.com:8123",
+            "HTTP_HOST": "ha.example.com",
+            "HTTP_X_FORWARDED_HOST": "ha.example.com:8123",
+            "HTTP_X_INGRESS_PATH": "/api/hassio_ingress/abc123",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+    def test_ingress_path_does_not_bypass_origin_check(self):
+        environ = {
+            "HTTP_ORIGIN": "https://evil.example",
+            "HTTP_HOST": "frigate.example.com",
+            "HTTP_X_INGRESS_PATH": "/frigate",
+        }
+        self.assertFalse(ws_origin_allowed(environ))
+
+    def test_home_assistant_integration_proxy(self):
+        # hass-web-proxy-lib drops Host, so Host is the integration's Frigate URL
+        environ = {
+            "HTTP_ORIGIN": "https://ha.example.com:8123",
+            "HTTP_HOST": "frigate.lan",
+            "HTTP_X_FORWARDED_HOST": "ha.example.com:8123",
+        }
+        self.assertTrue(ws_origin_allowed(environ))
+
+
+class TestOriginCheckedWebSocketApplication(unittest.TestCase):
+    def test_cross_origin_handshake_gets_403(self):
+        app = OriginCheckedWebSocketApplication(handler_cls=WebSocket)
+        start_response = MagicMock()
+        environ = {
+            "REQUEST_METHOD": "GET",
+            "HTTP_ORIGIN": "https://evil.example",
+            "HTTP_HOST": "frigate.lan",
+        }
+
+        self.assertEqual(app(environ, start_response), [])
+        start_response.assert_called_once()
+        self.assertEqual(start_response.call_args.args[0], "403 Forbidden")
+        self.assertNotIn("ws4py.websocket", environ)
+
+
+if __name__ == "__main__":
+    unittest.main()