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: