mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-15 11:12:49 +03:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e23b6b9a6c | ||
|
|
b99977192a | ||
|
|
a3ca8742b0 |
+5
-34
@@ -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
|
||||
|
||||
|
||||
+42
-10
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user