From caf79dc66844058c3d073c223354b283bbbfe396 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Sun, 18 Oct 2026 12:08:55 -0500 Subject: [PATCH] reject push endpoints that crash the vapid audience parse The validator accepted hostnames like "." so https://./x passed, and check_registrations then raised ValueError slicing the endpoint, killing the dispatcher's IPC thread. The validator now rejects empty hostname labels and lives in webpush.py, the audience comes from urlparse, and _build_web_pushers skips stored rows whose endpoint fails validation. --- frigate/api/notification.py | 68 +---------------- frigate/comms/webpush.py | 89 ++++++++++++++++++++++- frigate/test/test_webpush_registration.py | 77 +++++++++++++------- 3 files changed, 138 insertions(+), 96 deletions(-) diff --git a/frigate/api/notification.py b/frigate/api/notification.py index 42efc94c47..e35430cd63 100644 --- a/frigate/api/notification.py +++ b/frigate/api/notification.py @@ -1,10 +1,8 @@ """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 @@ -13,6 +11,7 @@ from py_vapid import Vapid01, utils from frigate.api.auth import allow_any_authenticated from frigate.api.defs.tags import Tags +from frigate.comms.webpush import validate_push_endpoint from frigate.const import CONFIG_DIR from frigate.models import User @@ -20,76 +19,13 @@ 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")) + reason = validate_push_endpoint(sub.get("endpoint")) if reason: return reason diff --git a/frigate/comms/webpush.py b/frigate/comms/webpush.py index 5f3fb945da..36d41ce1d4 100644 --- a/frigate/comms/webpush.py +++ b/frigate/comms/webpush.py @@ -1,6 +1,7 @@ """Handle sending notifications for Frigate via Firebase.""" import datetime +import ipaddress import json import logging import os @@ -10,6 +11,7 @@ from collections.abc import Callable, Iterator from dataclasses import dataclass from multiprocessing.synchronize import Event as MpEvent from typing import Any +from urllib.parse import urlparse from py_vapid import Vapid01 from pywebpush import WebPusher @@ -41,11 +43,92 @@ class PushNotification: ttl: int = 0 +# 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 "" in hostname.split("."): + return "endpoint hostname must not contain empty labels" + elif hostname.endswith(INTERNAL_HOST_SUFFIXES): + return "endpoint must not use an internal hostname" + + # The subscription token lives in the path + if len(parsed.path) <= 1: + return "endpoint must include a subscription path" + + return None + + +def _vapid_audience(endpoint: str) -> str: + parsed = urlparse(endpoint) + return f"{parsed.scheme}://{parsed.netloc}" + + def _build_web_pushers(user: str, subs: list[dict[str, Any]]) -> list[WebPusher]: """Build pushers for a user's stored subscriptions, skipping unusable ones.""" pushers: list[WebPusher] = [] for sub in subs: + # Rows stored before registration validation can hold any endpoint + reason = validate_push_endpoint( + sub.get("endpoint") if isinstance(sub, dict) else None + ) + + if reason: + logger.warning( + "Skipping notification subscription for %s: %s", user, reason + ) + continue + # WebPusher decodes the stored keys and raises on malformed ones try: pushers.append(WebPusher(sub)) @@ -131,7 +214,7 @@ class WebPushClient(Communicator): for pushers in self.web_pushers.values(): for push in pushers: endpoint: str = push.subscription_info["endpoint"] - endpoints.add(endpoint[0 : endpoint.index("/", 10)]) + endpoints.add(_vapid_audience(endpoint)) # create new claim for endpoint in endpoints: @@ -320,9 +403,7 @@ class WebPushClient(Communicator): for pusher in self.web_pushers[notification.user]: endpoint = pusher.subscription_info["endpoint"] - headers = self.claim_headers[ - endpoint[: endpoint.index("/", 10)] - ].copy() + headers = self.claim_headers[_vapid_audience(endpoint)].copy() headers["urgency"] = "high" resp = pusher.send( diff --git a/frigate/test/test_webpush_registration.py b/frigate/test/test_webpush_registration.py index 733659489a..daff9e5a62 100644 --- a/frigate/test/test_webpush_registration.py +++ b/frigate/test/test_webpush_registration.py @@ -3,8 +3,12 @@ import unittest from base64 import urlsafe_b64encode -from frigate.api.notification import _validate_push_endpoint, _validate_subscription -from frigate.comms.webpush import _build_web_pushers +from frigate.api.notification import _validate_subscription +from frigate.comms.webpush import ( + _build_web_pushers, + _vapid_audience, + validate_push_endpoint, +) VALID_ENDPOINTS = [ "https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu", @@ -26,11 +30,11 @@ 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)) + self.assertIsNone(validate_push_endpoint(endpoint)) def test_rejects_http(self): self.assertIsNotNone( - _validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token") + validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token") ) def test_rejects_non_http_schemes(self): @@ -40,7 +44,7 @@ class TestValidatePushEndpoint(unittest.TestCase): "//example.com/token", ): with self.subTest(endpoint=endpoint): - self.assertIsNotNone(_validate_push_endpoint(endpoint)) + self.assertIsNotNone(validate_push_endpoint(endpoint)) def test_rejects_localhost(self): for endpoint in ( @@ -50,7 +54,7 @@ class TestValidatePushEndpoint(unittest.TestCase): "https://[::1]/token", ): with self.subTest(endpoint=endpoint): - self.assertIsNotNone(_validate_push_endpoint(endpoint)) + self.assertIsNotNone(validate_push_endpoint(endpoint)) def test_rejects_private_addresses(self): for endpoint in ( @@ -61,7 +65,7 @@ class TestValidatePushEndpoint(unittest.TestCase): "https://0.0.0.0/token", ): with self.subTest(endpoint=endpoint): - self.assertIsNotNone(_validate_push_endpoint(endpoint)) + self.assertIsNotNone(validate_push_endpoint(endpoint)) def test_rejects_internal_hostnames(self): for endpoint in ( @@ -71,16 +75,16 @@ class TestValidatePushEndpoint(unittest.TestCase): "https://host.home.arpa/token", ): with self.subTest(endpoint=endpoint): - self.assertIsNotNone(_validate_push_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") + validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token") ) def test_rejects_embedded_credentials(self): self.assertIsNotNone( - _validate_push_endpoint( + validate_push_endpoint( "https://user:pass@fcm.googleapis.com/fcm/send/token" ) ) @@ -88,26 +92,26 @@ class TestValidatePushEndpoint(unittest.TestCase): 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)) + 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_empty_hostname_labels(self): + for endpoint in ( + "https://./x", + "https://fcm..googleapis.com/token", + "https://.fcm.googleapis.com/token", + "https://fcm.googleapis.com./token", + ): + with self.subTest(endpoint=endpoint): + 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)) + self.assertIsNotNone(validate_push_endpoint(endpoint)) def test_rejects_overlong_endpoint(self): self.assertIsNotNone( - _validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}") + validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}") ) @@ -148,10 +152,6 @@ class TestValidateSubscription(unittest.TestCase): self.assertIsNotNone(_validate_subscription(sub)) -if __name__ == "__main__": - unittest.main() - - class TestBuildWebPushers(unittest.TestCase): def test_skips_subscriptions_with_malformed_keys(self): bad = _subscription(VALID_ENDPOINTS[0]) @@ -163,3 +163,28 @@ class TestBuildWebPushers(unittest.TestCase): self.assertEqual(len(pushers), 1) self.assertEqual(pushers[0].receiver_key, b"\x04" + bytes(64)) + + def test_skips_subscriptions_with_invalid_endpoints(self): + good = _subscription(VALID_ENDPOINTS[0]) + good["keys"]["p256dh"] = urlsafe_b64encode(b"\x04" + bytes(64)).decode() + bad = dict(good, endpoint="https://./x") + + with self.assertLogs("frigate.comms.webpush", level="WARNING"): + pushers = _build_web_pushers("viewer", [bad, "not-a-dict", good]) + + self.assertEqual(len(pushers), 1) + self.assertEqual(pushers[0].subscription_info["endpoint"], VALID_ENDPOINTS[0]) + + +class TestVapidAudience(unittest.TestCase): + def test_uses_scheme_and_host(self): + self.assertEqual( + _vapid_audience(VALID_ENDPOINTS[0]), "https://fcm.googleapis.com" + ) + self.assertEqual( + _vapid_audience(VALID_ENDPOINTS[4]), "https://fcm.googleapis.com:443" + ) + + +if __name__ == "__main__": + unittest.main()