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

@@ -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

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:
+            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
 
+            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:

frigate/comms/inter_process.py

@@ -50,7 +50,13 @@ class InterProcessCommunicator(Communicator):
 
                     if isinstance(raw, list):
                         (topic, value) = raw
+
+                        # 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)}"

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(

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

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

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

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: