API Consistency / Security Fixes (#24057)

* Make review user read status consistent with other APIs

* Validate URLs for web push endpoint

* Validate the role for a custom viewer, rate limit password changing

* Cleanup

docs/static/frigate-api.yaml

@@ -2308,15 +2308,15 @@ paths:
                 $ref: '#/components/schemas/HTTPValidationError'
       security:
         - frigateUserAuth: []
-      x-required-role: any
-      description: '**Access:** Any authenticated user.'
+      x-required-role: camera
+      description: '**Access:** Authenticated user with access to the referenced camera.'
   /review/summarize/start/{start_ts}/end/{end_ts}:
     post:
       tags:
         - Review
       summary: Generate Review Summary
       description: |-
-        **Access:** Admin role required.
+        **Access:** Authenticated user with access to all cameras.
 
         Use GenAI to summarize review items over a period of time.
       operationId: 
@@ -2347,8 +2347,8 @@ paths:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
       security:
-        - frigateAdminAuth: []
-      x-required-role: admin
+        - frigateUserAuth: []
+      x-required-role: all_cameras
   /:
     get:
       tags:

frigate/api/auth.py

@@ -971,6 +971,7 @@ def delete_user(request: Request, username: str):
     summary="Update user password",
     description="Updates a user's password. Users can only change their own password unless they have admin role. Requires the current password to verify identity for non-admin users. Password must be at least 12 characters long. If user changes their own password, a new JWT cookie is automatically issued.",
 )
+@limiter.limit(limit_value=rateLimiter.get_limit)
 async def update_password(
     request: Request,
     username: str,
@@ -984,10 +985,11 @@ async def update_password(
     current_username = current_user.get("username")
     current_role = current_user.get("role")
 
-    # viewers can only change their own password
-    if current_role == "viewer" and current_username != username:
+    # Only admins may target another account. This has to cover every non-admin
+    # role rather than just viewer, since custom roles are arbitrary names
+    if current_role != "admin" and current_username != username:
         raise HTTPException(
-            status_code=403, detail="Viewers can only update their own password"
+            status_code=403, detail="Users can only update their own password"
         )
 
     HASH_ITERATIONS = request.app.frigate_config.auth.hash_iterations
@@ -1251,3 +1253,23 @@ async def get_allowed_cameras_for_filter(request: Request):
     all_camera_names = set(request.app.frigate_config.cameras.keys())
     roles_dict = request.app.frigate_config.auth.roles
     return User.get_allowed_cameras(role, roles_dict, all_camera_names)
+
+
+async def require_full_camera_access(
+    request: Request,
+    allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
+):
+    """Dependency for endpoints returning data that spans every camera.
+
+    Some responses cannot be meaningfully scoped to a subset of cameras, so
+    rather than filter them the endpoint is limited to callers who can already
+    see every camera. Admin and viewer always qualify; a custom role qualifies
+    only when its camera list covers all configured cameras.
+    """
+    all_camera_names = set(request.app.frigate_config.cameras.keys())
+
+    if not all_camera_names.issubset(allowed_cameras):
+        raise HTTPException(
+            status_code=403,
+            detail="Access to all cameras is required for this endpoint",
+        )

frigate/api/notification.py

@@ -1,8 +1,10 @@
 """Notification apis."""
 
+import ipaddress
 import logging
 import os
 from typing import Any
+from urllib.parse import urlparse
 
 from cryptography.hazmat.primitives import serialization
 from fastapi import APIRouter, Depends, Request
@@ -19,6 +21,95 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=[Tags.notifications])
 
+# Push endpoints are opaque URLs but stay well under this in practice
+MAX_ENDPOINT_LENGTH = 2048
+
+# Suffixes that only ever resolve on the local network
+INTERNAL_HOST_SUFFIXES = (".local", ".localdomain", ".internal", ".home.arpa")
+
+
+def _validate_push_endpoint(endpoint: Any) -> str | None:
+    """Return a reason the endpoint is unusable, or None when it is valid.
+
+    Subscriptions are issued by the browser vendor's push service, so a valid
+    endpoint is always a public https URL. Anything else is either a broken
+    registration or an attempt to aim the notification sender somewhere it
+    should not reach.
+    """
+    if not isinstance(endpoint, str) or not endpoint:
+        return "endpoint must be a url"
+
+    if len(endpoint) > MAX_ENDPOINT_LENGTH:
+        return "endpoint is too long"
+
+    try:
+        parsed = urlparse(endpoint)
+        port = parsed.port
+    except ValueError:
+        return "endpoint is not a valid url"
+
+    if parsed.scheme != "https":
+        return "endpoint must use https"
+
+    if parsed.username or parsed.password:
+        return "endpoint must not include credentials"
+
+    if port is not None and port != 443:
+        return "endpoint must use the default https port"
+
+    hostname = parsed.hostname
+
+    if not hostname:
+        return "endpoint must include a hostname"
+
+    try:
+        address = ipaddress.ip_address(hostname)
+    except ValueError:
+        address = None
+
+    if address is not None:
+        # A push service is never reachable at an address only this network can
+        # route, so anything non-global is a misconfiguration at best
+        if not address.is_global:
+            return "endpoint must not use a private address"
+    elif hostname == "localhost" or "." not in hostname:
+        return "endpoint must use a fully qualified hostname"
+    elif hostname.endswith(INTERNAL_HOST_SUFFIXES):
+        return "endpoint must not use an internal hostname"
+
+    # The subscription token lives in the path, and webpush.py assumes there is
+    # a separator after the host when it builds the VAPID audience
+    if len(parsed.path) <= 1:
+        return "endpoint must include a subscription path"
+
+    return None
+
+
+def _validate_subscription(sub: Any) -> str | None:
+    """Return a reason the subscription is unusable, or None when it is valid."""
+    if not isinstance(sub, dict):
+        return "subscription must be an object"
+
+    reason = _validate_push_endpoint(sub.get("endpoint"))
+
+    if reason:
+        return reason
+
+    keys = sub.get("keys")
+
+    if not isinstance(keys, dict):
+        return "subscription must include keys"
+
+    # WebPusher raises on a missing key, which would break every send for the
+    # user rather than just this registration
+    for name in ("p256dh", "auth"):
+        value = keys.get(name)
+
+        if not isinstance(value, str) or not value:
+            return f"subscription keys must include {name}"
+
+    return None
+
 
 @router.get(
     "/notifications/pubkey",
@@ -71,6 +162,17 @@ def register_notifications(request: Request, body: dict = None):
             status_code=400,
         )
 
+    reason = _validate_subscription(sub)
+
+    if reason:
+        logger.warning(
+            "Rejected notification registration for %s: %s", username, reason
+        )
+        return JSONResponse(
+            content={"success": False, "message": f"Invalid subscription: {reason}"},
+            status_code=400,
+        )
+
     try:
         User.update(notification_tokens=User.notification_tokens.append(sub)).where(
             User.username == username

frigate/api/review.py

@@ -17,6 +17,7 @@ from frigate.api.auth import (
     get_allowed_cameras_for_filter,
     get_current_user,
     require_camera_access,
+    require_full_camera_access,
     require_role,
 )
 from frigate.api.defs.query.review_query_parameters import (
@@ -709,6 +710,7 @@ async def get_review(request: Request, review_id: str):
     dependencies=[Depends(allow_any_authenticated())],
 )
 async def set_not_reviewed(
+    request: Request,
     review_id: str,
     current_user: dict = Depends(get_current_user),
 ):
@@ -727,6 +729,8 @@ async def set_not_reviewed(
             status_code=404,
         )
 
+    await require_camera_access(review.camera, request=request)
+
     try:
         user_review = UserReviewStatus.get(
             UserReviewStatus.user_id == user_id,
@@ -743,9 +747,12 @@ async def set_not_reviewed(
     )
 
 
+# Intentionally not camera scoped, as the summary correlates each flagged event
+# with overlapping activity on other cameras. Restricted to callers who can
+# already see every camera, so the unscoped query discloses nothing.
 @router.post(
     "/review/summarize/start/{start_ts}/end/{end_ts}",
-    dependencies=[Depends(require_role(["admin"]))],
+    dependencies=[Depends(require_full_camera_access)],
     description="Use GenAI to summarize review items over a period of time.",
 )
 def generate_review_summary(request: Request, start_ts: float, end_ts: float):

frigate/test/http_api/test_http_camera_access.py

@@ -440,3 +440,68 @@ class TestGo2rtcStreamAccess(BaseTestHttp):
             f"limited_user should be denied on alias back_door_main; "
             f"got {resp.status_code}"
         )
+
+
+class TestReviewSummaryAccess(BaseTestHttp):
+    """Tests for POST /review/summarize/start/{start_ts}/end/{end_ts}.
+
+    The summary correlates each flagged event with overlapping activity on
+    other cameras, so it is gated on full camera access rather than scoped to
+    the caller's cameras. These tests pin that decision so the dependency is
+    not loosened without first scoping the query.
+
+    GenAI is not configured in unit tests, so an authorized request returns 400
+    while an unauthorized one is rejected with 403 before the handler runs.
+    """
+
+    def setUp(self):
+        super().setUp([Event, ReviewSegment, Recordings])
+        self.minimal_config = _MULTI_CAMERA_CONFIG
+        self.app = super().create_app()
+
+    def tearDown(self):
+        self.app.dependency_overrides.clear()
+        super().tearDown()
+
+    def _summarize(self, allowed_cameras: list[str]):
+        async def mock_cameras(request: Request):
+            return allowed_cameras
+
+        self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
+        with AuthTestClient(self.app) as client:
+            return client.post("/review/summarize/start/0/end/9999999999")
+
+    def _assert_allowed(self, resp):
+        assert resp.status_code not in (401, 403), (
+            f"Caller should not be blocked; got {resp.status_code}"
+        )
+
+    def test_partial_camera_access_blocked(self):
+        assert self._summarize(["front_door"]).status_code == 403
+
+    def test_no_camera_access_blocked(self):
+        assert self._summarize([]).status_code == 403
+
+    def test_full_camera_access_allowed(self):
+        # Covers admin and viewer, which always resolve to every camera, and a
+        # custom role whose list happens to name them all.
+        self._assert_allowed(self._summarize(["front_door", "back_door"]))
+
+    def _summarize_as_role(self, role: str):
+        """Summarize using the real role to allowed-cameras resolution."""
+        self.app.dependency_overrides.pop(get_allowed_cameras_for_filter, None)
+        with AuthTestClient(self.app) as client:
+            return client.post(
+                "/review/summarize/start/0/end/9999999999",
+                headers={"remote-user": "test", "remote-role": role},
+            )
+
+    def test_viewer_role_allowed(self):
+        # viewer is never camera restricted, so it resolves to every camera.
+        self._assert_allowed(self._summarize_as_role("viewer"))
+
+    def test_admin_role_allowed(self):
+        self._assert_allowed(self._summarize_as_role("admin"))
+
+    def test_restricted_role_blocked(self):
+        assert self._summarize_as_role("limited_user").status_code == 403

frigate/test/http_api/test_http_password.py

@@ -0,0 +1,113 @@
+"""Tests for password change authorization."""
+
+from fastapi import Request
+
+from frigate.api.auth import get_current_user, hash_password, verify_password
+from frigate.models import Event, Recordings, ReviewSegment, User
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+# Config carrying a custom role, which is the class of user the literal
+# "viewer" check used to let through.
+_CUSTOM_ROLE_CONFIG = {
+    "mqtt": {"host": "mqtt"},
+    "auth": {"roles": {"neighbor": ["front_door"]}, "hash_iterations": 10},
+    "cameras": {
+        "front_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+    },
+}
+
+ADMIN_PASSWORD = "admin-real-password"
+NEW_PASSWORD = "AttackerChosenPassword123!"
+
+
+class TestUpdatePasswordAccess(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event, ReviewSegment, Recordings, User])
+        self.minimal_config = _CUSTOM_ROLE_CONFIG
+        self.app = super().create_app()
+        User.insert(
+            username="admin",
+            password_hash=hash_password(ADMIN_PASSWORD, iterations=10),
+            role="admin",
+            notification_tokens=[],
+        ).execute()
+
+        async def mock_get_current_user(request: Request):
+            return {
+                "username": request.headers.get("remote-user"),
+                "role": request.headers.get("remote-role"),
+            }
+
+        self.app.dependency_overrides[get_current_user] = mock_get_current_user
+
+    def tearDown(self):
+        self.app.dependency_overrides.clear()
+        super().tearDown()
+
+    def _change_password(self, actor: str, role: str, target: str, old_password: str):
+        with AuthTestClient(self.app) as client:
+            return client.put(
+                f"/users/{target}/password",
+                json={"password": NEW_PASSWORD, "old_password": old_password},
+                headers={"remote-user": actor, "remote-role": role},
+            )
+
+    def _admin_password_unchanged(self) -> bool:
+        return verify_password(ADMIN_PASSWORD, User.get_by_id("admin").password_hash)
+
+    def test_custom_role_cannot_target_another_account(self):
+        resp = self._change_password("neighbor", "neighbor", "admin", "wrong-guess")
+        assert resp.status_code == 403
+        assert self._admin_password_unchanged()
+
+    def test_custom_role_cannot_target_another_account_with_correct_password(self):
+        # The 403 must land before old_password is checked, so knowing the
+        # target's password is not a way through
+        resp = self._change_password("neighbor", "neighbor", "admin", ADMIN_PASSWORD)
+        assert resp.status_code == 403
+        assert self._admin_password_unchanged()
+
+    def test_viewer_cannot_target_another_account(self):
+        resp = self._change_password("viewer_user", "viewer", "admin", ADMIN_PASSWORD)
+        assert resp.status_code == 403
+        assert self._admin_password_unchanged()
+
+    def test_admin_can_target_another_account(self):
+        User.insert(
+            username="neighbor",
+            password_hash=hash_password("neighbor-password", iterations=10),
+            role="neighbor",
+            notification_tokens=[],
+        ).execute()
+
+        resp = self._change_password("admin", "admin", "neighbor", "")
+        assert resp.status_code == 200
+
+    def test_non_admin_can_change_own_password(self):
+        User.insert(
+            username="neighbor",
+            password_hash=hash_password("neighbor-password", iterations=10),
+            role="neighbor",
+            notification_tokens=[],
+        ).execute()
+
+        resp = self._change_password(
+            "neighbor", "neighbor", "neighbor", "neighbor-password"
+        )
+        assert resp.status_code == 200
+
+    def test_non_admin_own_password_still_requires_old_password(self):
+        User.insert(
+            username="neighbor",
+            password_hash=hash_password("neighbor-password", iterations=10),
+            role="neighbor",
+            notification_tokens=[],
+        ).execute()
+
+        resp = self._change_password("neighbor", "neighbor", "neighbor", "wrong-guess")
+        assert resp.status_code == 401

frigate/test/test_webpush_registration.py

@@ -0,0 +1,150 @@
+"""Tests for push notification subscription validation."""
+
+import unittest
+
+from frigate.api.notification import _validate_push_endpoint, _validate_subscription
+
+VALID_ENDPOINTS = [
+    "https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu",
+    "https://updates.push.services.mozilla.com/wpush/v2/dGhpcy1pcy1hLXRva2Vu",
+    "https://web.push.apple.com/dGhpcy1pcy1hLXRva2Vu",
+    "https://wns2-by3p.notify.windows.com/w/?token=dGhpcy1pcy1hLXRva2Vu",
+    "https://fcm.googleapis.com:443/fcm/send/dGhpcy1pcy1hLXRva2Vu",
+]
+
+
+def _subscription(endpoint: str) -> dict:
+    return {
+        "endpoint": endpoint,
+        "keys": {"p256dh": "cHVibGljLWtleQ", "auth": "YXV0aC1zZWNyZXQ"},
+    }
+
+
+class TestValidatePushEndpoint(unittest.TestCase):
+    def test_accepts_real_push_service_endpoints(self):
+        for endpoint in VALID_ENDPOINTS:
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_http(self):
+        self.assertIsNotNone(
+            _validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token")
+        )
+
+    def test_rejects_non_http_schemes(self):
+        for endpoint in (
+            "file:///etc/passwd",
+            "ftp://example.com/token",
+            "//example.com/token",
+        ):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_localhost(self):
+        for endpoint in (
+            "https://localhost/token",
+            "https://localhost:443/token",
+            "https://127.0.0.1/token",
+            "https://[::1]/token",
+        ):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_private_addresses(self):
+        for endpoint in (
+            "https://192.168.1.10/token",
+            "https://10.0.0.5/token",
+            "https://172.16.0.1/token",
+            "https://169.254.169.254/token",
+            "https://0.0.0.0/token",
+        ):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_internal_hostnames(self):
+        for endpoint in (
+            "https://frigate/token",
+            "https://nas.local/token",
+            "https://push.internal/token",
+            "https://host.home.arpa/token",
+        ):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_non_default_port(self):
+        self.assertIsNotNone(
+            _validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token")
+        )
+
+    def test_rejects_embedded_credentials(self):
+        self.assertIsNotNone(
+            _validate_push_endpoint(
+                "https://user:pass@fcm.googleapis.com/fcm/send/token"
+            )
+        )
+
+    def test_rejects_endpoint_without_path(self):
+        for endpoint in ("https://fcm.googleapis.com", "https://fcm.googleapis.com/"):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_endpoint_that_breaks_audience_parsing(self):
+        # webpush.py locates the host by searching for a separator after index
+        # 10, which raises ValueError when the url has no path at all
+        endpoint = "https://fcm.googleapis.com"
+
+        with self.assertRaises(ValueError):
+            endpoint.index("/", 10)
+
+        self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_missing_or_non_string_endpoint(self):
+        for endpoint in (None, "", 5, {"url": "https://example.com/token"}):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+
+    def test_rejects_overlong_endpoint(self):
+        self.assertIsNotNone(
+            _validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}")
+        )
+
+
+class TestValidateSubscription(unittest.TestCase):
+    def test_accepts_valid_subscription(self):
+        self.assertIsNone(_validate_subscription(_subscription(VALID_ENDPOINTS[0])))
+
+    def test_accepts_extra_fields_sent_by_the_browser(self):
+        sub = _subscription(VALID_ENDPOINTS[0])
+        sub["expirationTime"] = None
+        self.assertIsNone(_validate_subscription(sub))
+
+    def test_rejects_non_object(self):
+        for sub in ("https://fcm.googleapis.com/fcm/send/token", ["endpoint"], 5):
+            with self.subTest(sub=sub):
+                self.assertIsNotNone(_validate_subscription(sub))
+
+    def test_rejects_bad_endpoint(self):
+        self.assertIsNotNone(
+            _validate_subscription(_subscription("https://localhost/t"))
+        )
+
+    def test_rejects_missing_keys(self):
+        sub = _subscription(VALID_ENDPOINTS[0])
+        del sub["keys"]
+        self.assertIsNotNone(_validate_subscription(sub))
+
+    def test_rejects_incomplete_keys(self):
+        for keys in (
+            {"p256dh": "cHVibGljLWtleQ"},
+            {"auth": "YXV0aC1zZWNyZXQ"},
+            {"p256dh": "cHVibGljLWtleQ", "auth": ""},
+            {"p256dh": None, "auth": "YXV0aC1zZWNyZXQ"},
+        ):
+            with self.subTest(keys=keys):
+                sub = _subscription(VALID_ENDPOINTS[0])
+                sub["keys"] = keys
+                self.assertIsNotNone(_validate_subscription(sub))
+
+
+if __name__ == "__main__":
+    unittest.main()

generate_api_auth_spec.py

@@ -94,6 +94,7 @@ SPEC_SERVERS = [
 PUBLIC = "public"
 AUTHENTICATED = "any"
 CAMERA = "camera"
+ALL_CAMERAS = "all_cameras"
 ADMIN = "admin"
 
 ADMIN_SCHEME = "frigateAdminAuth"
@@ -128,6 +129,7 @@ ACCESS_NOTES = {
     PUBLIC: "**Access:** Public — no authentication required.",
     AUTHENTICATED: "**Access:** Any authenticated user.",
     CAMERA: "**Access:** Authenticated user with access to the referenced camera.",
+    ALL_CAMERAS: "**Access:** Authenticated user with access to all cameras.",
     ADMIN: "**Access:** Admin role required.",
 }
 
@@ -197,6 +199,8 @@ def _route_markers(route: APIRoute) -> tuple[set[str], list[str] | None]:
                 pass
         elif name in ("require_camera_access", "require_go2rtc_stream_access"):
             markers.add(CAMERA)
+        elif name == "require_full_camera_access":
+            markers.add(ALL_CAMERAS)
         elif "auth_checker" in qualname:
             markers.add(AUTHENTICATED)
         elif "public_checker" in qualname:
@@ -254,6 +258,8 @@ def _classify_base(
     # Explicit route-level markers win, in order of specificity.
     if ADMIN in markers:
         return ADMIN, admin_roles or ["admin"], None
+    if ALL_CAMERAS in markers:
+        return ALL_CAMERAS, None, None
     if CAMERA in markers:
         return CAMERA, None, None
     if AUTHENTICATED in markers:
@@ -337,8 +343,8 @@ def security_for(level: str) -> list:
         return []
     if level == ADMIN:
         return [{ADMIN_SCHEME: []}]
-    # AUTHENTICATED and CAMERA both require any authenticated session; the
-    # camera-specific scoping is conveyed in the note and x-required-role.
+    # AUTHENTICATED, CAMERA and ALL_CAMERAS all require any authenticated
+    # session; the camera scoping is conveyed in the note and x-required-role.
     return [{USER_SCHEME: []}]