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.

frigate/comms/webpush.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import queue
+import socket
 import threading
 from collections.abc import Callable, Iterator
 from dataclasses import dataclass
@@ -13,6 +14,7 @@ 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
@@ -108,6 +110,34 @@ def validate_push_endpoint(endpoint: Any) -> str | None:
     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 _resolves_to_public_address(endpoint: str) -> bool:
+    hostname = urlparse(endpoint).hostname
+
+    if not hostname:
+        return False
+
+    try:
+        infos = socket.getaddrinfo(hostname, 443, proto=socket.IPPROTO_TCP)
+    except (socket.gaierror, UnicodeError):
+        return False
+
+    return all(ipaddress.ip_address(info[4][0]).is_global for info in infos)
+
+
 def _vapid_audience(endpoint: str) -> str:
     parsed = urlparse(endpoint)
     return f"{parsed.scheme}://{parsed.netloc}"
@@ -131,7 +161,7 @@ def _build_web_pushers(user: str, subs: list[dict[str, Any]]) -> list[WebPusher]
 
         # 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)
 
@@ -403,6 +433,16 @@ class WebPushClient(Communicator):
 
                 for pusher in self.web_pushers[notification.user]:
                     endpoint = pusher.subscription_info["endpoint"]
+
+                    # Resolve-then-send leaves a DNS rebinding window. Closing it
+                    # needs the connection pinned to the checked address.
+                    if not _resolves_to_public_address(endpoint):
+                        logger.warning(
+                            "Skipping notification to %s: endpoint does not resolve to a public address",
+                            notification.user,
+                        )
+                        continue
+
                     headers = self.claim_headers[_vapid_audience(endpoint)].copy()
                     headers["urgency"] = "high"
 

frigate/test/test_webpush_registration.py

@@ -6,6 +6,7 @@ from base64 import urlsafe_b64encode
 from frigate.api.notification import _validate_subscription
 from frigate.comms.webpush import (
     _build_web_pushers,
+    _NoRedirectSession,
     _vapid_audience,
     validate_push_endpoint,
 )
@@ -174,6 +175,7 @@ class TestBuildWebPushers(unittest.TestCase):
 
         self.assertEqual(len(pushers), 1)
         self.assertEqual(pushers[0].subscription_info["endpoint"], VALID_ENDPOINTS[0])
+        self.assertIsInstance(pushers[0].mod_or_session, _NoRedirectSession)
 
 
 class TestVapidAudience(unittest.TestCase):

frigate/test/test_webpush_send_guard.py

@@ -0,0 +1,121 @@
+"""Tests that push sends cannot be aimed at internal addresses."""
+
+import queue
+import socket
+import unittest
+from unittest.mock import MagicMock, patch
+
+import requests
+from requests.adapters import BaseAdapter
+
+from frigate.comms.webpush import (
+    PushNotification,
+    WebPushClient,
+    _NoRedirectSession,
+    _resolves_to_public_address,
+)
+
+ENDPOINT = "https://push.example.com/token"
+
+
+def _addrinfo(address: str) -> list:
+    return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, 443))]
+
+
+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)
+
+
+class TestResolvesToPublicAddress(unittest.TestCase):
+    def test_accepts_global_address(self):
+        with patch("socket.getaddrinfo", return_value=_addrinfo("8.8.8.8")):
+            self.assertTrue(_resolves_to_public_address(ENDPOINT))
+
+    def test_rejects_non_global_addresses(self):
+        for address in ("127.0.0.1", "10.0.0.5", "169.254.169.254", "::1"):
+            with (
+                self.subTest(address=address),
+                patch("socket.getaddrinfo", return_value=_addrinfo(address)),
+            ):
+                self.assertFalse(_resolves_to_public_address(ENDPOINT))
+
+    def test_rejects_when_any_address_is_non_global(self):
+        infos = _addrinfo("8.8.8.8") + _addrinfo("127.0.0.1")
+
+        with patch("socket.getaddrinfo", return_value=infos):
+            self.assertFalse(_resolves_to_public_address(ENDPOINT))
+
+    def test_rejects_unresolvable_host(self):
+        with patch("socket.getaddrinfo", side_effect=socket.gaierror):
+            self.assertFalse(_resolves_to_public_address(ENDPOINT))
+
+
+class TestProcessNotifications(unittest.TestCase):
+    def _client(self, pusher: MagicMock) -> WebPushClient:
+        client = WebPushClient.__new__(WebPushClient)
+        client.stop_event = MagicMock()
+        client.stop_event.is_set.side_effect = [False, True]
+        client.notification_queue = queue.Queue()
+        client.notification_queue.put(
+            PushNotification(user="viewer", payload={}, title="t", message="m")
+        )
+        client.check_registrations = MagicMock()
+        client.claim_headers = {"https://push.example.com": {}}
+        client.web_pushers = {"viewer": [pusher]}
+        client.expired_subs = {}
+        return client
+
+    def test_skips_endpoint_resolving_to_internal_address(self):
+        pusher = MagicMock()
+        pusher.subscription_info = {"endpoint": ENDPOINT}
+        client = self._client(pusher)
+
+        with (
+            patch("socket.getaddrinfo", return_value=_addrinfo("127.0.0.1")),
+            self.assertLogs("frigate.comms.webpush", level="WARNING"),
+        ):
+            client._process_notifications()
+
+        pusher.send.assert_not_called()
+
+    def test_sends_to_public_endpoint(self):
+        pusher = MagicMock()
+        pusher.subscription_info = {"endpoint": ENDPOINT}
+        pusher.send.return_value.status_code = 201
+        client = self._client(pusher)
+
+        with patch("socket.getaddrinfo", return_value=_addrinfo("8.8.8.8")):
+            client._process_notifications()
+
+        pusher.send.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()