diff --git a/src/push_endpoint_policy.py b/src/push_endpoint_policy.py index a9cca10..8bb463a 100644 --- a/src/push_endpoint_policy.py +++ b/src/push_endpoint_policy.py @@ -2,6 +2,7 @@ import asyncio import ipaddress import socket from collections.abc import Awaitable, Callable +from dataclasses import dataclass from urllib.parse import urlsplit @@ -12,6 +13,14 @@ class UnsafePushEndpoint(ValueError): Resolver = Callable[[str, int], Awaitable[list[str]]] +@dataclass(frozen=True) +class ResolvedPushEndpoint: + endpoint: str + hostname: str + port: int + addresses: tuple[str, ...] + + async def _resolve(host: str, port: int) -> list[str]: loop = asyncio.get_running_loop() records = await loop.getaddrinfo( @@ -23,13 +32,13 @@ async def _resolve(host: str, port: int) -> list[str]: return sorted({record[4][0] for record in records}) -async def validate_public_push_endpoint( +async def resolve_public_push_endpoint( endpoint: str, *, resolver: Resolver = _resolve, timeout_seconds: float = 2.0, -) -> str: - """Return a canonical public HTTPS push endpoint or fail closed.""" +) -> ResolvedPushEndpoint: + """Resolve a canonical endpoint to public addresses for a pinned connection.""" parsed = urlsplit(endpoint) try: port = parsed.port @@ -59,4 +68,24 @@ async def validate_public_push_endpoint( raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") from error if not public: raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") - return endpoint + return ResolvedPushEndpoint( + endpoint=endpoint, + hostname=parsed.hostname, + port=port or 443, + addresses=tuple(sorted(set(addresses))), + ) + + +async def validate_public_push_endpoint( + endpoint: str, + *, + resolver: Resolver = _resolve, + timeout_seconds: float = 2.0, +) -> str: + """Return a canonical public HTTPS push endpoint or fail closed.""" + resolved = await resolve_public_push_endpoint( + endpoint, + resolver=resolver, + timeout_seconds=timeout_seconds, + ) + return resolved.endpoint diff --git a/src/push_notifications.py b/src/push_notifications.py index 1dcb7f4..ecc7bfc 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -5,10 +5,18 @@ import time from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Awaitable, Callable +from urllib.parse import urlsplit from zoneinfo import ZoneInfo, ZoneInfoNotFoundError +import requests + from src.push_subscription_store import PushSubscriptionStore -from src.push_endpoint_policy import UnsafePushEndpoint, validate_public_push_endpoint +from src.push_endpoint_policy import ( + ResolvedPushEndpoint, + UnsafePushEndpoint, + resolve_public_push_endpoint, + validate_public_push_endpoint, +) @dataclass(frozen=True) @@ -22,17 +30,59 @@ class PushConfiguration: return bool(self.public_key and self.private_key and self.subject) +class _PinnedHTTPSAdapter(requests.adapters.HTTPAdapter): + """Dial one approved IP while authenticating the endpoint's original host.""" + + def __init__(self, resolved: ResolvedPushEndpoint): + self.resolved = resolved + super().__init__() + + def add_headers(self, request, **kwargs): + super().add_headers(request, **kwargs) + request.headers["Host"] = self.resolved.hostname + + def build_connection_pool_key_attributes(self, request, verify, cert=None): + parsed = urlsplit(request.url) + if parsed.scheme != "https" or parsed.hostname != self.resolved.hostname: + raise UnsafePushEndpoint("Push transport attempted an unvalidated destination") + host, tls = super().build_connection_pool_key_attributes(request, verify, cert) + host.update( + host=self.resolved.addresses[0], + port=self.resolved.port, + ) + tls.update( + assert_hostname=self.resolved.hostname, + server_hostname=self.resolved.hostname, + ) + return host, tls + + async def send_web_push( - subscription: dict, payload: str, configuration: PushConfiguration + subscription: dict, + payload: str, + configuration: PushConfiguration, + *, + endpoint_resolver: Callable[[str], Awaitable[ResolvedPushEndpoint]] | None = None, + webpush_sender: Callable[..., object] | None = None, ) -> None: - import requests - from pywebpush import webpush + if endpoint_resolver is None: + endpoint_resolver = resolve_public_push_endpoint + resolved = await endpoint_resolver(subscription["endpoint"]) + if not resolved.addresses: + raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") + if webpush_sender is None: + from pywebpush import webpush + + webpush_sender = webpush session = requests.Session() + session.trust_env = False session.max_redirects = 0 + origin = f"https://{resolved.hostname}" + session.mount(origin, _PinnedHTTPSAdapter(resolved)) await asyncio.to_thread( - webpush, + webpush_sender, subscription_info=subscription, data=payload, vapid_private_key=configuration.private_key, @@ -160,7 +210,7 @@ async def dispatch_unread_updates( status = getattr( getattr(error, "response", None), "status_code", None ) - if status in {404, 410}: + if isinstance(error, UnsafePushEndpoint) or status in {404, 410}: await asyncio.to_thread( store.delete_session, delivery.session_id ) @@ -207,7 +257,7 @@ async def dispatch_unread_updates( status = getattr( getattr(error, "response", None), "status_code", None ) - if status in {404, 410}: + if isinstance(error, UnsafePushEndpoint) or status in {404, 410}: await asyncio.to_thread( store.delete_session, delivery.session_id ) @@ -388,7 +438,9 @@ async def _dispatch_deadline_reminders_unlocked( else send_web_push(device.subscription, payload, configuration) ) await asyncio.wait_for(operation, timeout=send_timeout_seconds) - except Exception: + except Exception as error: + if isinstance(error, UnsafePushEndpoint): + await asyncio.to_thread(store.delete_session, device.session_id) return 0 await asyncio.to_thread( store.mark_deadline_reminder_delivered, device.session_id, local_day diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 5952b72..08856fd 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -7,9 +7,14 @@ from types import SimpleNamespace import httpx import pytest +import requests from src import dashboard_auth, gitea_proxy, main -from src.push_notifications import PushConfiguration, dispatch_unread_updates +from src.push_notifications import ( + PushConfiguration, + dispatch_unread_updates, + send_web_push, +) from src.push_subscription_store import PushSubscriptionStore from src.push_endpoint_policy import UnsafePushEndpoint @@ -692,6 +697,75 @@ async def test_dispatch_removes_an_endpoint_that_rebinds_private_without_contact assert store.is_subscribed("unsafe-device") is False +@pytest.mark.anyio +async def test_web_push_transport_dials_the_address_validated_for_the_original_tls_host(): + endpoint = "https://push.example/device-a" + resolved = SimpleNamespace( + endpoint=endpoint, + hostname="push.example", + port=443, + addresses=("93.184.216.34",), + ) + resolutions = [] + connection = {} + + async def resolve_once(value): + resolutions.append(value) + return resolved + + def inspect_transport(**kwargs): + session = kwargs["requests_session"] + assert session.trust_env is False + request = session.prepare_request(requests.Request("POST", endpoint)) + adapter = session.get_adapter(endpoint) + adapter.add_headers(request) + host, tls = adapter.build_connection_pool_key_attributes(request, True) + connection.update(host=host, tls=tls, headers=dict(request.headers)) + + await send_web_push( + {"endpoint": endpoint, "keys": {"p256dh": "key", "auth": "secret"}}, + "{}", + PushConfiguration("public", "private", "mailto:ops@example.com"), + endpoint_resolver=resolve_once, + webpush_sender=inspect_transport, + ) + + assert resolutions == [endpoint] + assert connection["host"] == { + "scheme": "https", + "host": "93.184.216.34", + "port": 443, + } + assert connection["tls"]["assert_hostname"] == "push.example" + assert connection["tls"]["server_hostname"] == "push.example" + assert connection["headers"]["Host"] == "push.example" + + +@pytest.mark.anyio +async def test_connect_time_rebinding_rejection_removes_only_the_unsafe_device(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + for session_id in ("unsafe-device", "healthy-device"): + store.upsert(session_id, { + "endpoint": f"https://push.example/{session_id}", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + sent = [] + + async def unread(): + return {"items": [{"id": 8}]} + + async def send(subscription, _payload): + if subscription["endpoint"].endswith("unsafe-device"): + raise UnsafePushEndpoint("Endpoint rebound before connection") + sent.append(subscription["endpoint"]) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates(store, config, unread, send) == 1 + assert store.is_subscribed("unsafe-device") is False + assert store.is_subscribed("healthy-device") is True + assert sent == ["https://push.example/healthy-device"] + + @pytest.mark.anyio async def test_transient_endpoint_failure_does_not_block_healthy_devices(tmp_path): store = PushSubscriptionStore(tmp_path / "push.sqlite3")