From 8c10717103d3ac2d8be7a66458afc4791b7e96ac Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Mon, 23 Feb 2026 09:57:45 -0600 Subject: [PATCH] add tests --- .../test/http_api/test_http_camera_access.py | 204 ++++++++++++++++++ 1 file changed, 204 insertions(+) diff --git a/frigate/test/http_api/test_http_camera_access.py b/frigate/test/http_api/test_http_camera_access.py index 5cd115417..211c84bb4 100644 --- a/frigate/test/http_api/test_http_camera_access.py +++ b/frigate/test/http_api/test_http_camera_access.py @@ -1,6 +1,7 @@ from unittest.mock import patch from fastapi import HTTPException, Request +from fastapi.testclient import TestClient from frigate.api.auth import ( get_allowed_cameras_for_filter, @@ -9,6 +10,33 @@ from frigate.api.auth import ( from frigate.models import Event, Recordings, ReviewSegment from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp +# Minimal multi-camera config used by go2rtc stream access tests. +# front_door has a stream alias "front_door_main"; back_door uses its own name. +# The "limited_user" role is restricted to front_door only. +_MULTI_CAMERA_CONFIG = { + "mqtt": {"host": "mqtt"}, + "auth": { + "roles": { + "limited_user": ["front_door"], + } + }, + "cameras": { + "front_door": { + "ffmpeg": { + "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + "live": {"streams": {"default": "front_door_main"}}, + }, + "back_door": { + "ffmpeg": { + "inputs": [{"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]}] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + }, + }, +} + class TestCameraAccessEventReview(BaseTestHttp): def setUp(self): @@ -190,3 +218,179 @@ class TestCameraAccessEventReview(BaseTestHttp): resp = client.get("/events/summary") summary_list = resp.json() assert len(summary_list) == 2 + + +class TestGo2rtcStreamAccess(BaseTestHttp): + """Tests for require_go2rtc_stream_access — the auth dependency on + GET /go2rtc/streams/{stream_name}. + + go2rtc is not running in unit tests, so an authorized request returns + 500 (the proxy call fails), while an unauthorized request returns 401/403 + before the proxy is ever reached. + """ + + def _make_app(self, config_override: dict | None = None): + """Build a test app, optionally replacing self.minimal_config.""" + if config_override is not None: + self.minimal_config = config_override + app = super().create_app() + + # Allow tests to control the current user via request headers. + async def mock_get_current_user(request: Request): + username = request.headers.get("remote-user") + role = request.headers.get("remote-role") + if not username or not role: + from fastapi.responses import JSONResponse + + return JSONResponse( + content={"message": "No authorization headers."}, + status_code=401, + ) + return {"username": username, "role": role} + + app.dependency_overrides[get_current_user] = mock_get_current_user + return app + + def setUp(self): + super().setUp([Event, ReviewSegment, Recordings]) + + def tearDown(self): + super().tearDown() + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _get_stream( + self, app, stream_name: str, role: str = "admin", user: str = "test" + ): + """Issue GET /go2rtc/streams/{stream_name} with the given role.""" + with AuthTestClient(app) as client: + return client.get( + f"/go2rtc/streams/{stream_name}", + headers={"remote-user": user, "remote-role": role}, + ) + + # ------------------------------------------------------------------ + # Tests + # ------------------------------------------------------------------ + + def test_admin_can_access_any_stream(self): + """Admin role bypasses camera restrictions.""" + app = self._make_app(_MULTI_CAMERA_CONFIG) + # front_door stream — go2rtc is not running so expect 500, not 401/403 + resp = self._get_stream(app, "front_door", role="admin") + assert resp.status_code not in (401, 403), ( + f"Admin should not be blocked; got {resp.status_code}" + ) + + # back_door stream + resp = self._get_stream(app, "back_door", role="admin") + assert resp.status_code not in (401, 403) + + def test_missing_auth_headers_returns_401(self): + """Requests without auth headers must be rejected with 401.""" + app = self._make_app(_MULTI_CAMERA_CONFIG) + # Use plain TestClient (not AuthTestClient) so no headers are injected. + with TestClient(app, raise_server_exceptions=False) as client: + resp = client.get("/go2rtc/streams/front_door") + assert resp.status_code == 401, f"Expected 401, got {resp.status_code}" + + def test_unconfigured_role_can_access_any_stream(self): + """When no camera restrictions are configured for a role the user + should have access to all streams (no roles_dict entry ⇒ no restriction).""" + no_roles_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}, + }, + "back_door": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]} + ] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + }, + }, + } + app = self._make_app(no_roles_config) + + # "myuser" role is not listed in roles_dict — should be allowed everywhere + for stream in ("front_door", "back_door"): + resp = self._get_stream(app, stream, role="myuser") + assert resp.status_code not in (401, 403), ( + f"Unconfigured role should not be blocked on '{stream}'; " + f"got {resp.status_code}" + ) + + def test_restricted_role_can_access_allowed_camera(self): + """limited_user role (restricted to front_door) can access front_door stream.""" + app = self._make_app(_MULTI_CAMERA_CONFIG) + resp = self._get_stream(app, "front_door", role="limited_user") + assert resp.status_code not in (401, 403), ( + f"limited_user should be allowed on front_door; got {resp.status_code}" + ) + + def test_restricted_role_blocked_from_disallowed_camera(self): + """limited_user role (restricted to front_door) cannot access back_door stream.""" + app = self._make_app(_MULTI_CAMERA_CONFIG) + resp = self._get_stream(app, "back_door", role="limited_user") + assert resp.status_code == 403, ( + f"limited_user should be denied on back_door; got {resp.status_code}" + ) + + def test_stream_alias_allowed_for_owning_camera(self): + """Stream alias 'front_door_main' is owned by front_door; limited_user (who + is allowed front_door) should be permitted.""" + app = self._make_app(_MULTI_CAMERA_CONFIG) + # front_door_main is the alias defined in live.streams for front_door + resp = self._get_stream(app, "front_door_main", role="limited_user") + assert resp.status_code not in (401, 403), ( + f"limited_user should be allowed on alias front_door_main; " + f"got {resp.status_code}" + ) + + 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.""" + # Give back_door a stream alias and restrict limited_user to front_door only + config = { + "mqtt": {"host": "mqtt"}, + "auth": { + "roles": { + "limited_user": ["front_door"], + } + }, + "cameras": { + "front_door": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]} + ] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + }, + "back_door": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]} + ] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + "live": {"streams": {"default": "back_door_main"}}, + }, + }, + } + app = self._make_app(config) + resp = self._get_stream(app, "back_door_main", role="limited_user") + assert resp.status_code == 403, ( + f"limited_user should be denied on alias back_door_main; " + f"got {resp.status_code}" + )