diff --git a/frigate/comms/webpush.py b/frigate/comms/webpush.py index 36d41ce1d4..987463b5ab 100644 --- a/frigate/comms/webpush.py +++ b/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" diff --git a/frigate/test/test_webpush_registration.py b/frigate/test/test_webpush_registration.py index daff9e5a62..72d7a270aa 100644 --- a/frigate/test/test_webpush_registration.py +++ b/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): diff --git a/frigate/test/test_webpush_send_guard.py b/frigate/test/test_webpush_send_guard.py new file mode 100644 index 0000000000..aac3746fff --- /dev/null +++ b/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()