From c28adcb2cc82d4cdd7b69c94ee04c28fc0768d4c Mon Sep 17 00:00:00 2001 From: Nicolas Mowen Date: Mon, 11 May 2026 12:26:01 -0600 Subject: [PATCH] Improve robustness --- frigate/api/media_auth.py | 132 ++++++++++++++++++++++++-------- frigate/test/test_media_auth.py | 111 ++++++++++++++++++++++++++- 2 files changed, 206 insertions(+), 37 deletions(-) diff --git a/frigate/api/media_auth.py b/frigate/api/media_auth.py index 449743d312..cc06eb75cb 100644 --- a/frigate/api/media_auth.py +++ b/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 - scheme/host and query string and percent-decode. + nginx forwards the *raw* request URI (with `..` segments intact) via + `$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 - return unquote(path) or None + raw_path = parsed.path or original_url + 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 - contains hyphens by matching against the longest configured camera-name - prefix. + `frigate_config` is used to disambiguate clip/review filenames whose + camera name contains hyphens by matching against the longest configured + camera-name prefix. """ if not uri: return MediaAuthResolution.UNKNOWN, None @@ -92,23 +115,35 @@ def _resolve_recording( 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) + # /clips → 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 (parsed) + # 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 + "-"): - return cam - return None + +def _camera_from_thumb_filename( + 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) - return MediaAuthResolution.UNKNOWN, None + logger.warning("Export DB lookup failed for %s: %s", filename, e) + 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 diff --git a/frigate/test/test_media_auth.py b/frigate/test/test_media_auth.py index 66689a1331..80dc6fb8b0 100644 --- a/frigate/test/test_media_auth.py +++ b/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()