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,8 +2308,8 @@ 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:

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

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

@@ -710,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),
 ):
@@ -728,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,

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()