diff --git a/frigate/api/auth.py b/frigate/api/auth.py index 76d330548e..1a842377f2 100644 --- a/frigate/api/auth.py +++ b/frigate/api/auth.py @@ -11,7 +11,7 @@ import secrets import time from datetime import datetime from pathlib import Path -from urllib.parse import parse_qs, unquote +from urllib.parse import parse_qs import requests from fastapi import APIRouter, Depends, HTTPException, Request, Response @@ -31,6 +31,8 @@ from frigate.api.media_auth import ( check_camera_access, deny_response_for_media_uri, is_role_restricted, + normalize_request_path, + split_request_uri, ) from frigate.config import AuthConfig, ProxyConfig from frigate.const import CONFIG_DIR, JWT_SECRET_ENV_VAR, PASSWORD_HASH_ALGORITHM @@ -1185,37 +1187,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. - - nginx percent-decodes the path, merges repeated slashes and resolves dot - segments, while `X-Original-URL` carries the raw `$request_uri`. - """ - segments: list[str] = [] - for segment in unquote(path).split("/"): - if segment in ("", "."): - continue - if segment == "..": - if segments: - segments.pop() - continue - segments.append(segment) - - return "/" + "/".join(segments) - def _is_go2rtc_stream_proxy_request(path: str, request: Request) -> bool: # nginx marks requests routed through the go2rtc stream locations, and the @@ -1223,7 +1194,7 @@ def _is_go2rtc_stream_proxy_request(path: str, request: Request) -> bool: if request.headers.get("x-go2rtc-stream-proxy") == "1": return True - normalized = _normalize_request_path(path) + normalized = normalize_request_path(path) return any(normalized.startswith(p) for p in GO2RTC_STREAM_PROXY_PATHS) @@ -1242,7 +1213,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..802077188a 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,36 @@ 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. + nginx ends the URI at a literal `#`, so anything after it is dropped. + """ + uri = _URI_SCHEME_AND_AUTHORITY.sub("", request_uri, count=1) + path, _, query = uri.partition("#")[0].partition("?") + return path, query + + +def normalize_request_path(path: str) -> str: + """Normalize a raw request path the way nginx does before matching locations. + + nginx percent-decodes the path, merges repeated slashes and resolves dot + segments, while `X-Original-URL` carries the raw `$request_uri`. + """ + segments: list[str] = [] + for segment in unquote(path).split("/"): + if segment in ("", "."): + continue + if segment == "..": + if segments: + segments.pop() + continue + segments.append(segment) + + return "/" + "/".join(segments) + + def extract_path(original_url: str | None) -> str | None: """Return the decoded path component of nginx's `X-Original-URL` header. @@ -49,14 +82,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 @@ -252,12 +285,11 @@ def deny_response_for_media_uri( path = extract_path(original_url) # `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). + # Deny them when nginx would resolve them under a media root, and pass + # the rest through to nginx and the backend. if path is None: - raw = urlparse(original_url).path or original_url - decoded = unquote(raw) - first = decoded.lstrip("/").split("/", 1)[0] if decoded else "" + normalized = normalize_request_path(split_request_uri(original_url)[0]) + first = normalized.lstrip("/").split("/", 1)[0] if first in ("clips", "recordings", "exports"): return 403 return None diff --git a/frigate/test/test_media_auth.py b/frigate/test/test_media_auth.py index d025fea614..f8d6289ed7 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,34 @@ 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 + ) + + def test_traversal_from_non_media_prefix_denied(self): + # nginx resolves these into a media root before serving + for url in ( + "/x/../clips/back_door-1.jpg", + "/x/%2E%2E/clips/back_door-1.jpg", + "/./clips/front_door/../back_door-1.jpg", + "/api/../recordings/2026-05-11/14/back_door/00.00.mp4", + "/clips/back_door-1.jpg#/../../x", + ): + self.assertEqual(self._deny(url, "limited_user"), 403, url) + self.assertIsNone(self._deny("/x/../api/events", "limited_user")) + if __name__ == "__main__": unittest.main()