administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-11-13 17:13:01 +03:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Harden web push notifications (#24715)

Browse Source 
* 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.
This commit is contained in:
Josh Hawkins
2026-10-19 04:55:59 -10:00
committed by GitHub
parent 53ca9c929b
commit 3d3f6a0678
8 changed files with 341 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/test/test_ipc_handler_errors.py
+76
View File
@@ -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
+76 -26
View File
@@ -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
+45
View File
@@ -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()