fix: pin Web Push connections to validated addresses (Closes #1096)

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

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