From 3d3f6a067871897864c7d5a044e6f6564b6d2d22 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Mon, 19 Oct 2026 09:55:59 -0500 Subject: [PATCH] Harden web push notifications (#24715) * 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. * keep the ipc thread alive when a handler raises An exception from any dispatcher handler escaped InterProcessCommunicator.read, which only caught ZMQError. The REP thread died owing a reply, so every InterProcessRequestor blocked forever. read now logs the error and replies with an empty response, and Dispatcher.publish and publish_local catch per communicator so one failing communicator can't take down the rest. * stop push sends from reaching internal addresses pywebpush posts through requests with redirects enabled, so an endpoint could answer 307 to http://127.0.0.1:5000/api/restart and the sender would follow it into the internal admin port on every alert. Pushers now use a session that never follows redirects, and each send first resolves the endpoint host and skips it unless every address is global. * keep the previous region grid when the refresh request fails The IPC reader now replies [] when a handler raises, and process_frames assigned that reply to region_grid unconditionally. The next region lookup then indexed into an empty list and killed the camera's detect process. An empty reply now keeps the current grid until the next daily refresh. * drop the send-time public address check Refusing non-global resolved addresses broke Tailscale (100.64.0.0/10), split-horizon DNS and HTTPS_PROXY-only setups, and added little once redirects are no longer followed and TLS verification stays on. * only skip stored push rows that break the vapid audience Running the full registration validator over stored rows dropped self-hosted endpoints registered before 0.18 with a non-443 port or a bare / path. _build_web_pushers now skips only non-dict rows, rows with no endpoint, unparsable endpoints and hosts with empty labels, and the warning names the host so the row can be found. Registration still uses the full validator. --- frigate/api/notification.py | 68 +----------- frigate/comms/dispatcher.py | 14 ++- frigate/comms/inter_process.py | 8 +- frigate/comms/webpush.py | 122 +++++++++++++++++++++- frigate/test/test_ipc_handler_errors.py | 76 ++++++++++++++ frigate/test/test_webpush_registration.py | 102 +++++++++++++----- frigate/test/test_webpush_send_guard.py | 45 ++++++++ frigate/video/detect.py | 7 +- 8 files changed, 341 insertions(+), 101 deletions(-) create mode 100644 frigate/test/test_ipc_handler_errors.py create mode 100644 frigate/test/test_webpush_send_guard.py 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/dispatcher.py b/frigate/comms/dispatcher.py index f9373c30ee..f0988b5710 100644 --- a/frigate/comms/dispatcher.py +++ b/frigate/comms/dispatcher.py @@ -490,7 +490,12 @@ class Dispatcher: def publish(self, topic: str, payload: Any, retain: bool = False) -> None: """Handle publishing to communicators.""" for comm in self.comms: - comm.publish(topic, payload, retain) + try: + comm.publish(topic, payload, retain) + except Exception: + logger.exception( + "Error publishing %s to %s", topic, type(comm).__name__ + ) def publish_local(self, topic: str, payload: Any) -> None: """Publish to every communicator except MQTT. @@ -501,7 +506,12 @@ class Dispatcher: if isinstance(comm, MqttClient): continue - comm.publish(topic, payload, False) + try: + comm.publish(topic, payload, False) + except Exception: + logger.exception( + "Error publishing %s to %s", topic, type(comm).__name__ + ) def _publish_notices(self) -> None: if self.notice_registry is None: diff --git a/frigate/comms/inter_process.py b/frigate/comms/inter_process.py index c40e53ddb7..c45f94ffc6 100644 --- a/frigate/comms/inter_process.py +++ b/frigate/comms/inter_process.py @@ -50,7 +50,13 @@ class InterProcessCommunicator(Communicator): if isinstance(raw, list): (topic, value) = raw - response = self._dispatcher(topic, value) + + # REP owes a reply, so a handler error must not end the thread + try: + response = self._dispatcher(topic, value) + except Exception: + logger.exception("Error handling IPC topic %s", topic) + response = None else: logging.warning( f"Received unexpected data type in ZMQ recv_json: {type(raw)}" diff --git a/frigate/comms/webpush.py b/frigate/comms/webpush.py index 5f3fb945da..f7267896ed 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,7 +11,9 @@ 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 +import requests from py_vapid import Vapid01 from pywebpush import WebPusher from titlecase import titlecase @@ -41,14 +44,124 @@ 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 + + +class _NoRedirectSession(requests.Session): + """Session that never follows redirects. + + A push endpoint could otherwise redirect the send to an internal URL. + """ + + def request(self, *args: Any, **kwargs: Any) -> requests.Response: + kwargs["allow_redirects"] = False + return super().request(*args, **kwargs) + + +_push_session = _NoRedirectSession() + + +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: + # Stored rows may predate registration validation, so only skip the + # endpoints that would break building the VAPID audience + endpoint = sub.get("endpoint") if isinstance(sub, dict) else None + + if not isinstance(endpoint, str): + logger.warning( + "Skipping notification subscription for %s: no endpoint", user + ) + continue + + try: + hostname = urlparse(endpoint).hostname or "" + except ValueError: + logger.warning( + "Skipping notification subscription for %s: unparsable endpoint", user + ) + continue + + if "" in hostname.split("."): + logger.warning( + "Skipping notification subscription for %s: invalid endpoint host %r", + user, + hostname, + ) + continue + # WebPusher decodes the stored keys and raises on malformed ones try: - pushers.append(WebPusher(sub)) + pushers.append(WebPusher(sub, requests_session=_push_session)) except Exception: logger.warning("Skipping invalid notification subscription for %s", user) @@ -131,7 +244,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 +433,8 @@ 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_ipc_handler_errors.py b/frigate/test/test_ipc_handler_errors.py new file mode 100644 index 0000000000..d78d09cd4b --- /dev/null +++ b/frigate/test/test_ipc_handler_errors.py @@ -0,0 +1,76 @@ +"""Tests that IPC handler errors cannot stall inter-process communication.""" + +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +import zmq + +from frigate.comms.dispatcher import Dispatcher +from frigate.comms.inter_process import InterProcessCommunicator + + +class TestInterProcessHandlerErrors(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.address = f"ipc://{os.path.join(self.tmp.name, 'comms')}" + + with patch("frigate.comms.inter_process.SOCKET_REP_REQ", self.address): + self.communicator = InterProcessCommunicator() + + self.context = zmq.Context() + self.requestor = self.context.socket(zmq.REQ) + self.requestor.setsockopt(zmq.RCVTIMEO, 5000) + self.requestor.connect(self.address) + + def tearDown(self): + self.requestor.close(linger=0) + self.context.destroy(linger=0) + self.communicator.stop() + self.tmp.cleanup() + + def test_handler_exception_still_replies_and_keeps_reading(self): + def dispatcher(topic, value): + if topic == "boom": + raise ValueError("handler failed") + return value + + self.communicator.subscribe(dispatcher) + self.communicator.start() + + with self.assertLogs("frigate.comms.inter_process", level="ERROR"): + self.requestor.send_json(("boom", None)) + self.assertEqual(self.requestor.recv_json(), []) + + self.requestor.send_json(("echo", "ok")) + self.assertEqual(self.requestor.recv_json(), "ok") + self.assertTrue(self.communicator.reader_thread.is_alive()) + + +class TestDispatcherPublishErrors(unittest.TestCase): + def test_failing_communicator_does_not_block_the_others(self): + failing = MagicMock() + failing.publish.side_effect = ValueError("publish failed") + healthy = MagicMock() + + with ( + patch("frigate.comms.dispatcher.CameraActivityManager"), + patch("frigate.comms.dispatcher.AudioActivityManager"), + ): + dispatcher = Dispatcher( + MagicMock(), MagicMock(), MagicMock(), {}, [failing, healthy] + ) + + healthy.publish.reset_mock() + + with self.assertLogs("frigate.comms.dispatcher", level="ERROR"): + dispatcher.publish("reviews", "{}") + dispatcher.publish_local("notices", "[]") + + healthy.publish.assert_any_call("reviews", "{}", False) + healthy.publish.assert_any_call("notices", "[]", False) + + +if __name__ == "__main__": + unittest.main() diff --git a/frigate/test/test_webpush_registration.py b/frigate/test/test_webpush_registration.py index 733659489a..d1787a133a 100644 --- a/frigate/test/test_webpush_registration.py +++ b/frigate/test/test_webpush_registration.py @@ -3,8 +3,13 @@ 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, + _NoRedirectSession, + _vapid_audience, + validate_push_endpoint, +) VALID_ENDPOINTS = [ "https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu", @@ -26,11 +31,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 +45,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 +55,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 +66,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 +76,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 +93,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 +153,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 +164,52 @@ class TestBuildWebPushers(unittest.TestCase): self.assertEqual(len(pushers), 1) self.assertEqual(pushers[0].receiver_key, b"\x04" + bytes(64)) + + def _valid_keys_subscription(self, endpoint: str) -> dict: + sub = _subscription(endpoint) + sub["keys"]["p256dh"] = urlsafe_b64encode(b"\x04" + bytes(64)).decode() + return sub + + def test_skips_stored_rows_that_break_the_audience(self): + good = self._valid_keys_subscription(VALID_ENDPOINTS[0]) + rows = [ + self._valid_keys_subscription("https://./secret-token"), + self._valid_keys_subscription("https://[::1/secret-token"), + {"keys": good["keys"]}, + "not-a-dict", + good, + ] + + with self.assertLogs("frigate.comms.webpush", level="WARNING") as logs: + pushers = _build_web_pushers("viewer", rows) + + self.assertEqual(len(logs.output), 4) + self.assertIn("'.'", logs.output[0]) + self.assertNotIn("secret-token", "".join(logs.output)) + self.assertEqual(len(pushers), 1) + self.assertEqual(pushers[0].subscription_info["endpoint"], VALID_ENDPOINTS[0]) + self.assertIsInstance(pushers[0].mod_or_session, _NoRedirectSession) + + def test_keeps_stored_self_hosted_endpoints(self): + rows = [ + self._valid_keys_subscription("https://ntfy.example.com:8443/up123"), + self._valid_keys_subscription("https://ntfy.example.com/"), + ] + + pushers = _build_web_pushers("viewer", rows) + + self.assertEqual(len(pushers), 2) + + +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() diff --git a/frigate/test/test_webpush_send_guard.py b/frigate/test/test_webpush_send_guard.py new file mode 100644 index 0000000000..ebcd9a3fc5 --- /dev/null +++ b/frigate/test/test_webpush_send_guard.py @@ -0,0 +1,45 @@ +"""Tests that push sends never follow redirects.""" + +import unittest + +import requests +from requests.adapters import BaseAdapter + +from frigate.comms.webpush import _NoRedirectSession + +ENDPOINT = "https://push.example.com/token" + + +class RedirectAdapter(BaseAdapter): + def __init__(self): + super().__init__() + self.calls = 0 + + def send(self, request, **kwargs): + self.calls += 1 + response = requests.Response() + response.status_code = 307 + response.headers["Location"] = "http://127.0.0.1:5000/api/restart" + response.url = request.url + response.request = request + return response + + def close(self): + pass + + +class TestNoRedirectSession(unittest.TestCase): + def test_does_not_follow_redirects(self): + adapter = RedirectAdapter() + session = _NoRedirectSession() + session.mount("https://", adapter) + session.mount("http://", adapter) + + response = session.post(ENDPOINT, data=b"payload") + + self.assertEqual(response.status_code, 307) + self.assertEqual(adapter.calls, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/frigate/video/detect.py b/frigate/video/detect.py index a2877f83c0..591b57f97e 100644 --- a/frigate/video/detect.py +++ b/frigate/video/detect.py @@ -338,7 +338,12 @@ def process_frames( continue if datetime.now().astimezone(UTC) > next_region_update: - region_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name) + updated_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name) + + # A failed request replies empty, so keep the current grid + if updated_grid: + region_grid = updated_grid + next_region_update = get_tomorrow_at_time(2) try: