diff --git a/frigate/api/auth.py b/frigate/api/auth.py index 76d330548e..2446589b15 100644 --- a/frigate/api/auth.py +++ b/frigate/api/auth.py @@ -31,6 +31,7 @@ from frigate.api.media_auth import ( check_camera_access, deny_response_for_media_uri, is_role_restricted, + split_request_uri, ) from frigate.config import AuthConfig, ProxyConfig from frigate.const import CONFIG_DIR, JWT_SECRET_ENV_VAR, PASSWORD_HASH_ALGORITHM @@ -1185,18 +1186,6 @@ GO2RTC_STREAM_PROXY_PATHS = frozenset( } ) -_URI_SCHEME_AND_AUTHORITY = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://[^/?]*") - - -def _split_request_uri(request_uri: str) -> tuple[str, str]: - """Split a raw request URI into its path and query string. - - urlparse is not used since it reads a path starting with `//` as a host. - """ - uri = _URI_SCHEME_AND_AUTHORITY.sub("", request_uri, count=1) - path, _, query = uri.partition("?") - return path, query - def _normalize_request_path(path: str) -> str: """Normalize a raw request path the way nginx does before matching locations. @@ -1242,7 +1231,7 @@ def deny_response_for_go2rtc_stream( if not original_url: return None - path, query = _split_request_uri(original_url) + path, query = split_request_uri(original_url) if not _is_go2rtc_stream_proxy_request(path, request): return None diff --git a/frigate/api/media_auth.py b/frigate/api/media_auth.py index 0630e3604f..a94e720196 100644 --- a/frigate/api/media_auth.py +++ b/frigate/api/media_auth.py @@ -13,8 +13,9 @@ from __future__ import annotations import logging import os +import re from enum import Enum -from urllib.parse import unquote, urlparse +from urllib.parse import unquote from peewee import DoesNotExist @@ -24,6 +25,8 @@ from frigate.models import Export, User logger = logging.getLogger(__name__) +_URI_SCHEME_AND_AUTHORITY = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://[^/?]*") + class MediaAuthResolution(str, Enum): """Classification of an `X-Original-URL` path for media-auth purposes.""" @@ -41,6 +44,16 @@ class MediaAuthResolution(str, Enum): UNKNOWN = "unknown" +def split_request_uri(request_uri: str) -> tuple[str, str]: + """Split a raw request URI into its path and query string. + + urlparse is not used since it reads a path starting with `//` as a host. + """ + uri = _URI_SCHEME_AND_AUTHORITY.sub("", request_uri, count=1) + path, _, query = uri.partition("?") + return path, query + + def extract_path(original_url: str | None) -> str | None: """Return the decoded path component of nginx's `X-Original-URL` header. @@ -49,14 +62,14 @@ def extract_path(original_url: str | None) -> str | None: 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. + whose path contains `.` or `..` segments outright. Repeated slashes are + merged, as nginx does before matching locations. """ if not original_url: return None - parsed = urlparse(original_url) - raw_path = parsed.path or original_url - decoded = unquote(raw_path) + path, _ = split_request_uri(original_url) + decoded = re.sub(r"/{2,}", "/", unquote(path)) if not decoded: return None @@ -255,8 +268,7 @@ def deny_response_for_media_uri( # 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) + decoded = unquote(split_request_uri(original_url)[0]) first = decoded.lstrip("/").split("/", 1)[0] if decoded else "" if first in ("clips", "recordings", "exports"): return 403 diff --git a/frigate/test/test_media_auth.py b/frigate/test/test_media_auth.py index d025fea614..ea0e6164ac 100644 --- a/frigate/test/test_media_auth.py +++ b/frigate/test/test_media_auth.py @@ -76,6 +76,20 @@ class TestExtractPath(unittest.TestCase): self.assertIsNone(extract_path(None)) self.assertIsNone(extract_path("")) + def test_leading_double_slash_is_path(self): + self.assertEqual( + extract_path("//clips/front_door-1.jpg"), "/clips/front_door-1.jpg" + ) + + def test_repeated_slashes_merged(self): + self.assertEqual( + extract_path("/recordings//2026-05-11/14//front_door/00.00.mp4"), + "/recordings/2026-05-11/14/front_door/00.00.mp4", + ) + self.assertEqual( + extract_path("/%2Fclips/front_door-1.jpg"), "/clips/front_door-1.jpg" + ) + class TestResolveMediaUri(unittest.TestCase): def setUp(self): @@ -376,6 +390,22 @@ class TestDenyResponseForMediaUri(unittest.TestCase): self.assertIsNone(self._deny("/clips/nonexistent-1.jpg", "admin")) self.assertIsNone(self._deny("/clips/nonexistent-1.jpg", "viewer")) + def test_repeated_slashes_resolve_to_owning_camera(self): + # nginx merges repeated slashes before matching the media locations + for url in ( + "//clips/back_door-1.jpg", + "/clips//back_door-1.jpg", + "//recordings/2026-05-11/14/back_door/00.00.mp4", + "/recordings//2026-05-11/14/back_door/00.00.mp4", + ): + self.assertEqual(self._deny(url, "limited_user"), 403, url) + self.assertIsNone(self._deny("//clips/front_door-1.jpg", "limited_user")) + + def test_traversal_with_leading_double_slash_denied(self): + self.assertEqual( + self._deny("//clips/front_door/../back_door-1.jpg", "admin"), 403 + ) + if __name__ == "__main__": unittest.main()