Merge pull request 'Pin Web Push connections to validated public addresses' (#1097) from timmy/1096-pin-web-push-connections into main

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:
+) -> ResolvedPushEndpoint:
-    """Return a canonical public HTTPS push endpoint or fail closed."""
+    """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

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

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