mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-26 00:14:41 +03:00
Josh HawkinsandGitHub
5bb2cde8b3
CI / AMD64 Build (push) Canceled after 0s
CI / AMD64 Smoke Test (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
* 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
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""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()
|