Improve robustness

frigate/api/media_auth.py

@@ -12,6 +12,7 @@ per-camera authorization the regular API enforces via `require_camera_access`.
 from __future__ import annotations
 
 import logging
+import os
 from enum import Enum
 from typing import Optional
 from urllib.parse import unquote, urlparse
@@ -19,6 +20,7 @@ from urllib.parse import unquote, urlparse
 from peewee import DoesNotExist
 
 from frigate.config import FrigateConfig
+from frigate.const import EXPORT_DIR
 from frigate.models import Export, User
 
 logger = logging.getLogger(__name__)
@@ -31,21 +33,42 @@ class MediaAuthResolution(str, Enum):
     ADMIN_ONLY = "admin_only"
     LISTING_MULTI_CAMERA = "listing_multi_camera"
     LISTING_NEUTRAL = "listing_neutral"
+    # Under a recognized media root (/clips, /recordings, /exports) but
+    # unclassifiable (unknown subtree, no matching DB row, DB error).
+    # Restricted users are denied; admins/full-access roles are allowed
+    # (nginx will likely return 404 if the file genuinely doesn't exist).
+    UNRESOLVED_MEDIA = "unresolved_media"
+    # Not a media URI at all (e.g. /api/events, /login).
     UNKNOWN = "unknown"
 
 
 def extract_path(original_url: Optional[str]) -> Optional[str]:
-    """Return just the path component of nginx's `X-Original-URL` header.
+    """Return the decoded path component of nginx's `X-Original-URL` header.
 
-    nginx sets the value to `{scheme}://{host}{request_uri}`; we strip
+    nginx forwards the *raw* request URI (with `..` segments intact) via
-    scheme/host and query string and percent-decode.
+    `$request_uri`. nginx normalizes the path before serving the file, so a
+    request like `/recordings/.../allowed_cam/../forbidden_cam/file.mp4`
+    would (1) parse as the allowed camera in our auth check, (2) be served
+    as the forbidden camera by nginx. To close the bypass we reject any URI
+    whose path contains `.` or `..` segments outright.
     """
     if not original_url:
         return None
 
     parsed = urlparse(original_url)
-    path = parsed.path or original_url
+    raw_path = parsed.path or original_url
-    return unquote(path) or None
+    decoded = unquote(raw_path)
+    if not decoded:
+        return None
+
+    if not decoded.startswith("/"):
+        decoded = "/" + decoded
+
+    segments = decoded.split("/")
+    if ".." in segments or "." in segments:
+        return None
+
+    return decoded
 
 
 def resolve_media_uri(
@@ -53,9 +76,9 @@ def resolve_media_uri(
 ) -> 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
+    `frigate_config` is used to disambiguate clip/review filenames whose
-    contains hyphens by matching against the longest configured camera-name
+    camera name contains hyphens by matching against the longest configured
-    prefix.
+    camera-name prefix.
     """
     if not uri:
         return MediaAuthResolution.UNKNOWN, None
@@ -93,22 +116,34 @@ 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/previews/{cam}/...                       → camera
+    # /clips/review/thumb-{cam}-{review_id}.webp      → camera (parsed)
     # /clips/faces/...                                → admin-only
+    # /clips/genai-requests/...                       → admin-only
+    # /clips/preview_restart_cache/...                → admin-only
     # /clips/{model}/train|dataset/...                → admin-only
-    # /clips/{cam}-{event_id}[-clean].{ext}      → camera (resolved via config)
+    # /clips/{cam}-{event_id}[-clean].{ext}           → camera (parsed)
-    # other /clips/{subdir}/...                  → admin-only (conservative)
+    # other /clips/{subdir}/...                       → unresolved (deny restricted)
     if len(parts) == 1:
         return MediaAuthResolution.LISTING_MULTI_CAMERA, None
 
     second = parts[1]
-    if second == "thumbs":
+
+    if second in ("thumbs", "previews"):
         if len(parts) == 2:
             return MediaAuthResolution.LISTING_MULTI_CAMERA, None
         return MediaAuthResolution.CAMERA, parts[2]
 
-    if second == "faces":
+    if second == "review":
+        if len(parts) == 2:
+            return MediaAuthResolution.LISTING_MULTI_CAMERA, None
+        camera = _camera_from_thumb_filename(parts[2], frigate_config)
+        if camera:
+            return MediaAuthResolution.CAMERA, camera
+        return MediaAuthResolution.UNRESOLVED_MEDIA, None
+
+    if second in ("faces", "genai-requests", "preview_restart_cache"):
         return MediaAuthResolution.ADMIN_ONLY, None
 
     if len(parts) >= 3 and parts[2] in ("train", "dataset"):
@@ -118,9 +153,20 @@ def _resolve_clip(
         camera = _camera_from_clip_filename(second, frigate_config)
         if camera:
             return MediaAuthResolution.CAMERA, camera
-        return MediaAuthResolution.UNKNOWN, None
+        return MediaAuthResolution.UNRESOLVED_MEDIA, None
 
-    return MediaAuthResolution.ADMIN_ONLY, None
+    return MediaAuthResolution.UNRESOLVED_MEDIA, None
+
+
+def _longest_prefix_camera(
+    stem: str, frigate_config: Optional[FrigateConfig]
+) -> Optional[str]:
+    if frigate_config is None:
+        return None
+    for cam in sorted(frigate_config.cameras.keys(), key=len, reverse=True):
+        if stem.startswith(cam + "-"):
+            return cam
+    return None
 
 
 def _camera_from_clip_filename(
@@ -130,37 +176,42 @@ def _camera_from_clip_filename(
     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
+    return _longest_prefix_camera(stem, frigate_config)
 
-    for cam in sorted(frigate_config.cameras.keys(), key=len, reverse=True):
+
-        if stem.startswith(cam + "-"):
+def _camera_from_thumb_filename(
-            return cam
+    filename: str, frigate_config: Optional[FrigateConfig]
+) -> Optional[str]:
+    """Match a review thumbnail filename `thumb-{camera}-{review_id}.webp`."""
+    if not filename.startswith("thumb-"):
         return None
+    dot = filename.rfind(".")
+    stem = filename[len("thumb-") : dot] if dot > 0 else filename[len("thumb-") :]
+    return _longest_prefix_camera(stem, frigate_config)
 
 
 def _resolve_export(
     parts: list[str],
 ) -> tuple[MediaAuthResolution, Optional[str]]:
     # /exports                  → multi-camera listing
-    # /exports/{filename}.mp4   → camera (DB lookup)
+    # /exports/{filename}.mp4   → camera (DB lookup by exact path)
     if len(parts) == 1:
         return MediaAuthResolution.LISTING_MULTI_CAMERA, None
     if len(parts) != 2:
-        return MediaAuthResolution.UNKNOWN, None
+        return MediaAuthResolution.UNRESOLVED_MEDIA, None
 
     filename = parts[1]
+    full_path = os.path.join(EXPORT_DIR, filename)
     try:
-        export = Export.get(Export.video_path.endswith(filename))
+        export = Export.get(Export.video_path == full_path)
         return MediaAuthResolution.CAMERA, export.camera
     except DoesNotExist:
-        return MediaAuthResolution.UNKNOWN, None
+        return MediaAuthResolution.UNRESOLVED_MEDIA, None
     except Exception as e:
-        logger.debug("Export DB lookup failed for %s: %s", filename, e)
+        logger.warning("Export DB lookup failed for %s: %s", filename, e)
-        return MediaAuthResolution.UNKNOWN, None
+        return MediaAuthResolution.UNRESOLVED_MEDIA, None
 
 
 def check_camera_access(role: str, camera: str, frigate_config: FrigateConfig) -> bool:
@@ -194,10 +245,22 @@ def deny_response_for_media_uri(
     """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.
+    when the request is allowed.
     """
+    if not original_url:
+        return None
+
     path = extract_path(original_url)
-    if not path:
+
+    # `extract_path` returns None for URIs containing `.` or `..` segments.
+    # For media-root URIs that's a traversal attempt — deny outright. For
+    # non-media URIs, pass through (nginx / the backend handle them).
+    if path is None:
+        raw = urlparse(original_url).path or original_url
+        decoded = unquote(raw)
+        first = decoded.lstrip("/").split("/", 1)[0] if decoded else ""
+        if first in ("clips", "recordings", "exports"):
+            return 403
         return None
 
     resolution, camera = resolve_media_uri(path, frigate_config)
@@ -216,6 +279,7 @@ def deny_response_for_media_uri(
     if resolution in (
         MediaAuthResolution.LISTING_MULTI_CAMERA,
         MediaAuthResolution.ADMIN_ONLY,
+        MediaAuthResolution.UNRESOLVED_MEDIA,
     ):
         return 403
 

frigate/test/test_media_auth.py

@@ -130,7 +130,11 @@ class TestResolveMediaUri(unittest.TestCase):
         )
 
     def test_clip_filename_no_matching_camera(self):
-        self._assert("/clips/nonexistent-1234.jpg", MediaAuthResolution.UNKNOWN)
+        # Looks like a media path but couldn't classify — fail closed for
+        # restricted users (UNRESOLVED_MEDIA), not pass-through.
+        self._assert(
+            "/clips/nonexistent-1234.jpg", MediaAuthResolution.UNRESOLVED_MEDIA
+        )
 
     def test_clip_thumbs(self):
         self._assert("/clips/thumbs/", MediaAuthResolution.LISTING_MULTI_CAMERA)
@@ -145,12 +149,54 @@ class TestResolveMediaUri(unittest.TestCase):
             camera="back_door",
         )
 
+    def test_clip_previews(self):
+        self._assert("/clips/previews/", MediaAuthResolution.LISTING_MULTI_CAMERA)
+        self._assert(
+            "/clips/previews/front_door/",
+            MediaAuthResolution.CAMERA,
+            camera="front_door",
+        )
+        self._assert(
+            "/clips/previews/back_door/segment.mp4",
+            MediaAuthResolution.CAMERA,
+            camera="back_door",
+        )
+
+    def test_clip_review_thumbs(self):
+        # Format: /clips/review/thumb-{camera}-{review_id}.webp (frigate/review/maintainer.py).
+        self._assert(
+            "/clips/review/thumb-front_door-abc123.webp",
+            MediaAuthResolution.CAMERA,
+            camera="front_door",
+        )
+        # Hyphenated camera name — longest-prefix match.
+        self._assert(
+            "/clips/review/thumb-back-yard-abc123.webp",
+            MediaAuthResolution.CAMERA,
+            camera="back-yard",
+        )
+        # Unknown camera prefix → unresolved, not allowed for restricted users.
+        self._assert(
+            "/clips/review/thumb-unknown-cam-abc123.webp",
+            MediaAuthResolution.UNRESOLVED_MEDIA,
+        )
+
     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/genai-requests/x/0.webp", MediaAuthResolution.ADMIN_ONLY)
+        self._assert(
+            "/clips/preview_restart_cache/x.mp4", 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_unknown_subtree_is_unresolved(self):
+        # Unknown /clips/{x}/{y}/... subtree falls through as unresolved (not
+        # admin-only) so restricted users get 403 without admins being denied
+        # access to legitimate but unrecognized resources.
+        self._assert("/clips/random_dir/foo.jpg", MediaAuthResolution.UNRESOLVED_MEDIA)
+
     def test_clip_top_level_listing(self):
         self._assert("/clips/", MediaAuthResolution.LISTING_MULTI_CAMERA)
 
@@ -201,13 +247,24 @@ class TestExportResolution(unittest.TestCase):
         self.assertEqual(resolution, MediaAuthResolution.CAMERA)
         self.assertEqual(camera, "back_door")
 
-    def test_unknown_export(self):
+    def test_unknown_export_is_unresolved(self):
+        # No matching row → UNRESOLVED_MEDIA (fail closed for restricted users),
+        # not UNKNOWN (which would pass-through).
         resolution, camera = resolve_media_uri(
             "/exports/does_not_exist.mp4", self.config
         )
-        self.assertEqual(resolution, MediaAuthResolution.UNKNOWN)
+        self.assertEqual(resolution, MediaAuthResolution.UNRESOLVED_MEDIA)
         self.assertIsNone(camera)
 
+    def test_export_anchored_match_not_endswith(self):
+        # Anchored exact-path equality must NOT match by filename suffix.
+        # A request like /exports/clip.mp4 must not authorize against a row at
+        # /media/frigate/exports/back_door_clip.mp4 just because the suffix matches.
+        self._insert_export("exp_bd", "back_door", "back_door_clip.mp4")
+        self._insert_export("exp_fd", "front_door", "front_door_clip.mp4")
+        resolution, _ = resolve_media_uri("/exports/clip.mp4", self.config)
+        self.assertEqual(resolution, MediaAuthResolution.UNRESOLVED_MEDIA)
+
 
 class TestDenyResponseForMediaUri(unittest.TestCase):
     """End-to-end decision check used by /auth."""
@@ -271,6 +328,54 @@ class TestDenyResponseForMediaUri(unittest.TestCase):
         self.assertIsNone(self._deny(None, "limited_user"))
         self.assertIsNone(self._deny("", "limited_user"))
 
+    def test_traversal_in_media_uri_denied_for_all_roles(self):
+        # Bypass attempt: parts[3] looks like an allowed camera, but the
+        # normalized path nginx would serve points at a forbidden camera.
+        # Both restricted and admin should be denied — the URI is malformed
+        # and we refuse to make an auth decision against it.
+        traversal_uris = [
+            "/recordings/2026-05-11/14/front_door/../back_door/00.00.mp4",
+            "/clips/front_door-1.jpg/../back_door-1.jpg",
+            "/exports/../recordings/2026-05-11/14/back_door/00.00.mp4",
+            "/clips/./back_door-1.jpg",
+        ]
+        for uri in traversal_uris:
+            self.assertEqual(self._deny(uri, "limited_user"), 403, uri)
+            self.assertEqual(self._deny(uri, "admin"), 403, uri)
+            self.assertEqual(self._deny(uri, "viewer"), 403, uri)
+
+    def test_traversal_outside_media_passes_through(self):
+        # `..` in non-media URIs is not our problem; the backend handles it.
+        self.assertIsNone(self._deny("/api/foo/../bar", "limited_user"))
+
+    def test_percent_encoded_traversal_denied(self):
+        # nginx may decode percent-encoded `%2E%2E` to `..` before serving;
+        # we must apply the same denial after percent-decoding.
+        self.assertEqual(
+            self._deny(
+                "/recordings/2026-05-11/14/front_door/%2E%2E/back_door/00.mp4",
+                "limited_user",
+            ),
+            403,
+        )
+
+    def test_unresolved_media_fails_closed_for_restricted(self):
+        # Restricted user requesting a media URI we can't classify (no DB row,
+        # unknown clip prefix, unknown clip subtree) must be denied.
+        self.assertEqual(self._deny("/clips/nonexistent-1.jpg", "limited_user"), 403)
+        self.assertEqual(self._deny("/clips/random_dir/foo.jpg", "limited_user"), 403)
+        self.assertEqual(
+            self._deny("/clips/review/thumb-unknown_cam-1.webp", "limited_user"),
+            403,
+        )
+
+    def test_unresolved_media_allowed_for_admin(self):
+        # Admin and full-access roles are *not* denied on UNRESOLVED_MEDIA —
+        # nginx returns 404 if the file doesn't exist on disk anyway, and we
+        # don't want a stale DB to lock out admins.
+        self.assertIsNone(self._deny("/clips/nonexistent-1.jpg", "admin"))
+        self.assertIsNone(self._deny("/clips/nonexistent-1.jpg", "viewer"))
+
 
 if __name__ == "__main__":
     unittest.main()