From 39d522eadb3998a5f0a90f07d01420b1fc4ca4ee Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Mon, 11 May 2026 11:19:21 -0600 Subject: [PATCH] Implement cross-camera safety for indexed media folders --- frigate/api/auth.py | 34 ++-- frigate/api/media_auth.py | 229 +++++++++++++++++++++++++ frigate/test/test_media_auth.py | 290 ++++++++++++++++++++++++++++++++ 3 files changed, 543 insertions(+), 10 deletions(-) create mode 100644 frigate/api/media_auth.py create mode 100644 frigate/test/test_media_auth.py diff --git a/frigate/api/auth.py b/frigate/api/auth.py index 6411312087..eca51df1a4 100644 --- a/frigate/api/auth.py +++ b/frigate/api/auth.py @@ -26,6 +26,7 @@ from frigate.api.defs.request.app_body import ( AppPutRoleBody, ) from frigate.api.defs.tags import Tags +from frigate.api.media_auth import check_camera_access, deny_response_for_media_uri from frigate.config import AuthConfig, NetworkingConfig, ProxyConfig from frigate.const import CONFIG_DIR, JWT_SECRET_ENV_VAR, PASSWORD_HASH_ALGORITHM from frigate.models import User @@ -633,6 +634,9 @@ def auth(request: Request): logger.debug("X-Proxy-Secret header does not match configured secret value") return fail_response + original_url = request.headers.get("x-original-url") + frigate_config = request.app.frigate_config + # if auth is disabled, just apply the proxy header map and return success if not auth_config.enabled: # pass the user header value from the upstream proxy if a mapping is specified @@ -649,6 +653,11 @@ def auth(request: Request): role = resolve_role(request.headers, proxy_config, config_roles_set) success_response.headers["remote-role"] = role + + deny_status = deny_response_for_media_uri(original_url, role, frigate_config) + if deny_status is not None: + return Response("", status_code=deny_status) + return success_response # now apply authentication @@ -743,6 +752,11 @@ def auth(request: Request): success_response.headers["remote-user"] = user success_response.headers["remote-role"] = role + + deny_status = deny_response_for_media_uri(original_url, role, frigate_config) + if deny_status is not None: + return Response("", status_code=deny_status) + return success_response except Exception as e: logger.error(f"Error parsing jwt: {e}") @@ -1069,19 +1083,19 @@ async def require_camera_access( raise HTTPException(status_code=current_user.status_code, detail=detail) role = current_user["role"] - all_camera_names = set(request.app.frigate_config.cameras.keys()) - roles_dict = request.app.frigate_config.auth.roles - allowed_cameras = User.get_allowed_cameras(role, roles_dict, all_camera_names) + frigate_config = request.app.frigate_config - # Admin or full access bypasses - if role == "admin" or not roles_dict.get(role): + if check_camera_access(role, camera_name, frigate_config): return - if camera_name not in allowed_cameras: - raise HTTPException( - status_code=403, - detail=f"Access denied to camera '{camera_name}'. Allowed: {allowed_cameras}", - ) + all_camera_names = set(frigate_config.cameras.keys()) + allowed_cameras = User.get_allowed_cameras( + role, frigate_config.auth.roles, all_camera_names + ) + raise HTTPException( + status_code=403, + detail=f"Access denied to camera '{camera_name}'. Allowed: {allowed_cameras}", + ) def _get_stream_owner_cameras(request: Request, stream_name: str) -> set[str]: diff --git a/frigate/api/media_auth.py b/frigate/api/media_auth.py new file mode 100644 index 0000000000..1c060a4631 --- /dev/null +++ b/frigate/api/media_auth.py @@ -0,0 +1,229 @@ +"""URI-aware authorization for nginx-served static media. + +The `/auth` endpoint (used as nginx `auth_request` target) calls into this +module to classify the requested URI from the `X-Original-URL` header and, for +camera-scoped resources, decide whether the current role may access them. + +Without this, `auth_request` only verifies the JWT — every authenticated user +could read clips, recordings, and exports for *any* camera, bypassing the +per-camera authorization the regular API enforces via `require_camera_access`. +""" + +from __future__ import annotations + +import logging +from enum import Enum +from typing import Optional +from urllib.parse import unquote, urlparse + +from peewee import DoesNotExist + +from frigate.config import FrigateConfig +from frigate.models import Export, User + +logger = logging.getLogger(__name__) + + +class MediaAuthResolution(str, Enum): + """Classification of an `X-Original-URL` path for media-auth purposes.""" + + CAMERA = "camera" + ADMIN_ONLY = "admin_only" + LISTING_MULTI_CAMERA = "listing_multi_camera" + LISTING_NEUTRAL = "listing_neutral" + UNKNOWN = "unknown" + + +def extract_path(original_url: Optional[str]) -> Optional[str]: + """Return just the path component of nginx's `X-Original-URL` header. + + nginx sets the value to `{scheme}://{host}{request_uri}`; we strip + scheme/host and query string and percent-decode. + """ + if not original_url: + return None + + parsed = urlparse(original_url) + path = parsed.path or original_url + return unquote(path) or None + + +def resolve_media_uri( + uri: str, frigate_config: Optional[FrigateConfig] = None +) -> tuple[MediaAuthResolution, Optional[str]]: + """Classify a URI and return the owning camera if applicable. + + `frigate_config` is used to disambiguate clip filenames whose camera name + contains hyphens by matching against the longest configured camera-name + prefix. + """ + if not uri: + return MediaAuthResolution.UNKNOWN, None + + parts = [p for p in uri.split("/") if p] + if not parts: + return MediaAuthResolution.UNKNOWN, None + + root = parts[0] + if root == "recordings": + return _resolve_recording(parts) + if root == "clips": + return _resolve_clip(parts, frigate_config) + if root == "exports": + return _resolve_export(parts) + + return MediaAuthResolution.UNKNOWN, None + + +def _resolve_recording( + parts: list[str], +) -> tuple[MediaAuthResolution, Optional[str]]: + # /recordings → neutral + # /recordings/{date} → neutral + # /recordings/{date}/{hour} → multi-camera listing + # /recordings/{date}/{hour}/{cam}/... → camera + if len(parts) <= 2: + return MediaAuthResolution.LISTING_NEUTRAL, None + if len(parts) == 3: + return MediaAuthResolution.LISTING_MULTI_CAMERA, None + return MediaAuthResolution.CAMERA, parts[3] + + +def _resolve_clip( + parts: list[str], frigate_config: Optional[FrigateConfig] +) -> tuple[MediaAuthResolution, Optional[str]]: + # /clips → multi-camera listing + # /clips/thumbs → multi-camera listing + # /clips/thumbs/{cam}/... → camera + # /clips/faces/... → admin-only + # /clips/{model}/train|dataset/... → admin-only + # /clips/{cam}-{event_id}[-clean].{ext} → camera (resolved via config) + # other /clips/{subdir}/... → admin-only (conservative) + if len(parts) == 1: + return MediaAuthResolution.LISTING_MULTI_CAMERA, None + + second = parts[1] + if second == "thumbs": + if len(parts) == 2: + return MediaAuthResolution.LISTING_MULTI_CAMERA, None + return MediaAuthResolution.CAMERA, parts[2] + + if second == "faces": + return MediaAuthResolution.ADMIN_ONLY, None + + if len(parts) >= 3 and parts[2] in ("train", "dataset"): + return MediaAuthResolution.ADMIN_ONLY, None + + if len(parts) == 2: + camera = _camera_from_clip_filename(second, frigate_config) + if camera: + return MediaAuthResolution.CAMERA, camera + return MediaAuthResolution.UNKNOWN, None + + return MediaAuthResolution.ADMIN_ONLY, None + + +def _camera_from_clip_filename( + filename: str, frigate_config: Optional[FrigateConfig] +) -> Optional[str]: + """Match a flat clip filename `{camera}-{event_id}[-clean].{ext}` against + configured camera names. Longest-prefix wins so camera names containing + hyphens (e.g. `front-door`) resolve correctly. + """ + if frigate_config is None: + return None + + dot = filename.rfind(".") + stem = filename[:dot] if dot > 0 else filename + + for cam in sorted(frigate_config.cameras.keys(), key=len, reverse=True): + if stem.startswith(cam + "-"): + return cam + return None + + +def _resolve_export( + parts: list[str], +) -> tuple[MediaAuthResolution, Optional[str]]: + # /exports → multi-camera listing + # /exports/{filename}.mp4 → camera (DB lookup) + if len(parts) == 1: + return MediaAuthResolution.LISTING_MULTI_CAMERA, None + if len(parts) != 2: + return MediaAuthResolution.UNKNOWN, None + + filename = parts[1] + try: + export = Export.get(Export.video_path.endswith(filename)) + return MediaAuthResolution.CAMERA, export.camera + except DoesNotExist: + return MediaAuthResolution.UNKNOWN, None + except Exception as e: + logger.debug("Export DB lookup failed for %s: %s", filename, e) + return MediaAuthResolution.UNKNOWN, None + + +def check_camera_access( + role: str, camera: str, frigate_config: FrigateConfig +) -> bool: + """Return True iff `role` may access `camera`. + + Mirrors the gating logic in `require_camera_access`: admin and any role + without a non-empty allow-list bypass the check. + """ + if role == "admin": + return True + + roles_dict = frigate_config.auth.roles + if not roles_dict.get(role): + return True + + all_camera_names = set(frigate_config.cameras.keys()) + allowed = User.get_allowed_cameras(role, roles_dict, all_camera_names) + return camera in allowed + + +def is_role_restricted(role: str, frigate_config: FrigateConfig) -> bool: + """True if `role` has a non-empty allow-list (i.e. not full-access).""" + if role == "admin": + return False + return bool(frigate_config.auth.roles.get(role)) + + +def deny_response_for_media_uri( + original_url: Optional[str], role: Optional[str], frigate_config: FrigateConfig +) -> Optional[int]: + """Decide whether the current role should be blocked from `original_url`. + + Returns an HTTP status code (403) when access should be denied, or `None` + when the request is allowed or the URI is not a recognized media path. + """ + path = extract_path(original_url) + if not path: + return None + + resolution, camera = resolve_media_uri(path, frigate_config) + if resolution == MediaAuthResolution.UNKNOWN: + return None + + if not role or role == "admin": + return None + + if not is_role_restricted(role, frigate_config): + return None + + if resolution == MediaAuthResolution.LISTING_NEUTRAL: + return None + + if resolution in ( + MediaAuthResolution.LISTING_MULTI_CAMERA, + MediaAuthResolution.ADMIN_ONLY, + ): + return 403 + + if resolution == MediaAuthResolution.CAMERA: + if camera and check_camera_access(role, camera, frigate_config): + return None + return 403 + + return 403 diff --git a/frigate/test/test_media_auth.py b/frigate/test/test_media_auth.py new file mode 100644 index 0000000000..bfebbb9bcb --- /dev/null +++ b/frigate/test/test_media_auth.py @@ -0,0 +1,290 @@ +"""Unit tests for `frigate.api.media_auth`. + +Covers URI classification, the role-vs-camera decision matrix, and the export +DB-lookup path. These are pure functions/DB lookups — no HTTP stack involved. +""" + +import datetime +import logging +import os +import unittest + +from peewee_migrate import Router +from playhouse.sqlite_ext import SqliteExtDatabase +from playhouse.sqliteq import SqliteQueueDatabase + +from frigate.api.media_auth import ( + MediaAuthResolution, + deny_response_for_media_uri, + extract_path, + resolve_media_uri, +) +from frigate.config import FrigateConfig +from frigate.models import Event, Export, Recordings, ReviewSegment +from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS + +_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}, + }, + # Camera name with a hyphen — exercises longest-prefix match. + "back-yard": { + "ffmpeg": { + "inputs": [{"path": "rtsp://10.0.0.3:554/video", "roles": ["detect"]}] + }, + "detect": {"height": 1080, "width": 1920, "fps": 5}, + }, + }, +} + + +class TestExtractPath(unittest.TestCase): + def test_full_url(self): + self.assertEqual( + extract_path("http://host:8971/clips/front_door-1.jpg"), + "/clips/front_door-1.jpg", + ) + + def test_strips_query_string(self): + self.assertEqual( + extract_path("http://h/recordings/2026-05-11/14/front_door/00.00.mp4?t=1"), + "/recordings/2026-05-11/14/front_door/00.00.mp4", + ) + + def test_path_only(self): + self.assertEqual(extract_path("/exports/x.mp4"), "/exports/x.mp4") + + def test_percent_decoded(self): + self.assertEqual( + extract_path("http://h/clips/front%20door-1.jpg"), + "/clips/front door-1.jpg", + ) + + def test_empty(self): + self.assertIsNone(extract_path(None)) + self.assertIsNone(extract_path("")) + + +class TestResolveMediaUri(unittest.TestCase): + def setUp(self): + self.config = FrigateConfig(**_CONFIG) + + def _assert(self, uri, resolution, camera=None): + got_resolution, got_camera = resolve_media_uri(uri, self.config) + self.assertEqual(got_resolution, resolution, uri) + self.assertEqual(got_camera, camera, uri) + + def test_unknown_paths(self): + self._assert("/api/events", MediaAuthResolution.UNKNOWN) + self._assert("/", MediaAuthResolution.UNKNOWN) + self._assert("", MediaAuthResolution.UNKNOWN) + + def test_recordings(self): + self._assert("/recordings/", MediaAuthResolution.LISTING_NEUTRAL) + self._assert("/recordings/2026-05-11/", MediaAuthResolution.LISTING_NEUTRAL) + self._assert( + "/recordings/2026-05-11/14/", MediaAuthResolution.LISTING_MULTI_CAMERA + ) + self._assert( + "/recordings/2026-05-11/14/front_door/", + MediaAuthResolution.CAMERA, + camera="front_door", + ) + self._assert( + "/recordings/2026-05-11/14/back_door/00.00.mp4", + MediaAuthResolution.CAMERA, + camera="back_door", + ) + + def test_clip_flat_filename_resolves_camera(self): + self._assert( + "/clips/front_door-1234.jpg", + MediaAuthResolution.CAMERA, + camera="front_door", + ) + self._assert( + "/clips/back_door-1234-clean.webp", + MediaAuthResolution.CAMERA, + camera="back_door", + ) + + def test_clip_filename_with_hyphenated_camera_name(self): + # Camera name "back-yard" itself contains a hyphen; longest-prefix + # match must pick `back-yard`, not the bogus `back` prefix. + self._assert( + "/clips/back-yard-1234.jpg", + MediaAuthResolution.CAMERA, + camera="back-yard", + ) + + def test_clip_filename_no_matching_camera(self): + self._assert( + "/clips/nonexistent-1234.jpg", MediaAuthResolution.UNKNOWN + ) + + def test_clip_thumbs(self): + self._assert("/clips/thumbs/", MediaAuthResolution.LISTING_MULTI_CAMERA) + self._assert( + "/clips/thumbs/front_door/", + MediaAuthResolution.CAMERA, + camera="front_door", + ) + self._assert( + "/clips/thumbs/back_door/abc.webp", + MediaAuthResolution.CAMERA, + camera="back_door", + ) + + def test_clip_admin_only_subtrees(self): + self._assert("/clips/faces/train/foo.webp", MediaAuthResolution.ADMIN_ONLY) + self._assert("/clips/faces/", MediaAuthResolution.ADMIN_ONLY) + self._assert("/clips/some_model/train/x.jpg", MediaAuthResolution.ADMIN_ONLY) + self._assert("/clips/some_model/dataset/x.jpg", MediaAuthResolution.ADMIN_ONLY) + + def test_clip_top_level_listing(self): + self._assert("/clips/", MediaAuthResolution.LISTING_MULTI_CAMERA) + + def test_exports_listing(self): + self._assert("/exports/", MediaAuthResolution.LISTING_MULTI_CAMERA) + + +class TestExportResolution(unittest.TestCase): + """Export resolution requires a DB lookup.""" + + def setUp(self): + migrate_db = SqliteExtDatabase("test.db") + del logging.getLogger("peewee_migrate").handlers[:] + Router(migrate_db).run() + migrate_db.close() + self.db = SqliteQueueDatabase(TEST_DB) + self.db.bind([Event, ReviewSegment, Recordings, Export]) + self.config = FrigateConfig(**_CONFIG) + + def tearDown(self): + if not self.db.is_closed(): + self.db.close() + for f in TEST_DB_CLEANUPS: + try: + os.remove(f) + except OSError: + pass + + def _insert_export(self, export_id, camera, filename): + Export.insert( + id=export_id, + camera=camera, + name=f"export-{export_id}", + date=datetime.datetime.now(), + video_path=f"/media/frigate/exports/{filename}", + thumb_path=f"/media/frigate/exports/{filename}.jpg", + in_progress=False, + ).execute() + + def test_export_resolves_camera(self): + self._insert_export( + "exp1", "back_door", "back_door_20260511_140000-20260511_150000_abc123.mp4" + ) + resolution, camera = resolve_media_uri( + "/exports/back_door_20260511_140000-20260511_150000_abc123.mp4", + self.config, + ) + self.assertEqual(resolution, MediaAuthResolution.CAMERA) + self.assertEqual(camera, "back_door") + + def test_unknown_export(self): + resolution, camera = resolve_media_uri( + "/exports/does_not_exist.mp4", self.config + ) + self.assertEqual(resolution, MediaAuthResolution.UNKNOWN) + self.assertIsNone(camera) + + +class TestDenyResponseForMediaUri(unittest.TestCase): + """End-to-end decision check used by /auth.""" + + def setUp(self): + self.config = FrigateConfig(**_CONFIG) + + def _deny(self, url, role): + return deny_response_for_media_uri(url, role, self.config) + + def test_admin_always_allowed(self): + self.assertIsNone(self._deny("/clips/back_door-1.jpg", "admin")) + self.assertIsNone(self._deny("/clips/", "admin")) + self.assertIsNone(self._deny("/clips/faces/x.webp", "admin")) + self.assertIsNone( + self._deny("/recordings/2026-05-11/14/back_door/00.00.mp4", "admin") + ) + + def test_unrestricted_role_allowed(self): + # "viewer" role has no entry in roles_dict → full access (matches the + # behavior of require_camera_access). + self.assertIsNone(self._deny("/clips/back_door-1.jpg", "viewer")) + self.assertIsNone(self._deny("/clips/", "viewer")) + + def test_restricted_role_allowed_camera(self): + self.assertIsNone( + self._deny("/clips/front_door-1.jpg", "limited_user") + ) + self.assertIsNone( + self._deny( + "/recordings/2026-05-11/14/front_door/00.00.mp4", "limited_user" + ) + ) + self.assertIsNone( + self._deny("/clips/thumbs/front_door/abc.webp", "limited_user") + ) + + def test_restricted_role_blocked_other_camera(self): + self.assertEqual( + self._deny("/clips/back_door-1.jpg", "limited_user"), 403 + ) + self.assertEqual( + self._deny( + "/recordings/2026-05-11/14/back_door/00.00.mp4", "limited_user" + ), + 403, + ) + self.assertEqual( + self._deny("/clips/thumbs/back_door/abc.webp", "limited_user"), 403 + ) + + def test_restricted_role_blocked_admin_only(self): + self.assertEqual( + self._deny("/clips/faces/train/foo.webp", "limited_user"), 403 + ) + + def test_restricted_role_blocked_multi_camera_listing(self): + self.assertEqual(self._deny("/clips/", "limited_user"), 403) + self.assertEqual(self._deny("/exports/", "limited_user"), 403) + self.assertEqual( + self._deny("/recordings/2026-05-11/14/", "limited_user"), 403 + ) + + def test_restricted_role_allowed_neutral_listing(self): + self.assertIsNone(self._deny("/recordings/", "limited_user")) + self.assertIsNone(self._deny("/recordings/2026-05-11/", "limited_user")) + + def test_non_media_uri_passes_through(self): + self.assertIsNone(self._deny("/api/events", "limited_user")) + self.assertIsNone(self._deny("http://h/login", "limited_user")) + + def test_missing_header(self): + self.assertIsNone(self._deny(None, "limited_user")) + self.assertIsNone(self._deny("", "limited_user")) + + +if __name__ == "__main__": + unittest.main()