Implement cross-camera safety for indexed media folders

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())
+    frigate_config = request.app.frigate_config
-    roles_dict = request.app.frigate_config.auth.roles
-    allowed_cameras = User.get_allowed_cameras(role, roles_dict, all_camera_names)
 
-    # Admin or full access bypasses
+    if check_camera_access(role, camera_name, frigate_config):
-    if role == "admin" or not roles_dict.get(role):
         return
 
-    if camera_name not in allowed_cameras:
+    all_camera_names = set(frigate_config.cameras.keys())
-        raise HTTPException(
+    allowed_cameras = User.get_allowed_cameras(
-            status_code=403,
+        role, frigate_config.auth.roles, all_camera_names
-            detail=f"Access denied to camera '{camera_name}'. Allowed: {allowed_cameras}",
+    )
-        )
+    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]:

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

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()