diff --git a/docs/static/frigate-api.yaml b/docs/static/frigate-api.yaml index 247ab02031..ff34021df2 100644 --- a/docs/static/frigate-api.yaml +++ b/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: diff --git a/frigate/api/auth.py b/frigate/api/auth.py index beff30b836..1e4524e262 100644 --- a/frigate/api/auth.py +++ b/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 diff --git a/frigate/api/notification.py b/frigate/api/notification.py index 502e76dbd2..04749fc260 100644 --- a/frigate/api/notification.py +++ b/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 diff --git a/frigate/api/review.py b/frigate/api/review.py index d10c72c49d..2194c7c2fb 100644 --- a/frigate/api/review.py +++ b/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, diff --git a/frigate/test/http_api/test_http_password.py b/frigate/test/http_api/test_http_password.py new file mode 100644 index 0000000000..9305aa2c05 --- /dev/null +++ b/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 diff --git a/frigate/test/test_webpush_registration.py b/frigate/test/test_webpush_registration.py new file mode 100644 index 0000000000..16e0129152 --- /dev/null +++ b/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()