fix: restrict Web Push endpoint egress (Closes #653)

src/main.py

@@ -48,6 +48,7 @@ from src.live_snapshot_store import (
 from src.models import Issue, Milestone, PullRequest, Repo, User
 from src.passkey_store import PasskeyStore
 from src.push_notifications import PushConfiguration, dispatch_unread_updates
+from src.push_endpoint_policy import UnsafePushEndpoint, validate_public_push_endpoint
 from src.push_subscription_store import PushSubscriptionStore
 from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
 from src.security_event_store import SecurityEventStore, SecurityEventStoreError
@@ -1808,6 +1809,10 @@ async def push_status(request: Request):
 async def subscribe_push(payload: PushSubscriptionPayload, request: Request):
     if not _push_configuration().enabled:
         raise HTTPException(status_code=503, detail="Push notifications are not configured")
+    try:
+        await validate_public_push_endpoint(payload.endpoint)
+    except UnsafePushEndpoint as error:
+        raise HTTPException(status_code=422, detail=str(error)) from error
     device_id = await dashboard_auth.session_management_id(
         request.state.dashboard_session
     )

src/push_endpoint_policy.py

@@ -0,0 +1,62 @@
+import asyncio
+import ipaddress
+import socket
+from collections.abc import Awaitable, Callable
+from urllib.parse import urlsplit
+
+
+class UnsafePushEndpoint(ValueError):
+    """Raised when a Web Push endpoint could reach a non-public service."""
+
+
+Resolver = Callable[[str, int], Awaitable[list[str]]]
+
+
+async def _resolve(host: str, port: int) -> list[str]:
+    loop = asyncio.get_running_loop()
+    records = await loop.getaddrinfo(
+        host,
+        port,
+        family=socket.AF_UNSPEC,
+        type=socket.SOCK_STREAM,
+    )
+    return sorted({record[4][0] for record in records})
+
+
+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."""
+    parsed = urlsplit(endpoint)
+    try:
+        port = parsed.port
+    except ValueError as error:
+        raise UnsafePushEndpoint(
+            "Endpoint must be a canonical public Web Push service URL"
+        ) from error
+    if (
+        parsed.scheme != "https"
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.fragment
+        or port not in {None, 443}
+    ):
+        raise UnsafePushEndpoint("Endpoint must be a canonical public Web Push service URL")
+    try:
+        addresses = await asyncio.wait_for(
+            resolver(parsed.hostname, port or 443),
+            timeout=max(0.01, timeout_seconds),
+        )
+    except (OSError, asyncio.TimeoutError) as error:
+        raise UnsafePushEndpoint("Endpoint must resolve to a public Web Push service") from error
+    try:
+        public = addresses and all(ipaddress.ip_address(address).is_global for address in addresses)
+    except ValueError as error:
+        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

src/push_notifications.py

@@ -6,6 +6,7 @@ from dataclasses import dataclass
 from typing import Awaitable, Callable
 
 from src.push_subscription_store import PushSubscriptionStore
+from src.push_endpoint_policy import UnsafePushEndpoint, validate_public_push_endpoint
 
 
 @dataclass(frozen=True)
@@ -22,8 +23,12 @@ class PushConfiguration:
 async def send_web_push(
     subscription: dict, payload: str, configuration: PushConfiguration
 ) -> None:
+    import requests
     from pywebpush import webpush
 
+    session = requests.Session()
+    session.max_redirects = 0
+
     await asyncio.to_thread(
         webpush,
         subscription_info=subscription,
@@ -31,6 +36,8 @@ async def send_web_push(
         vapid_private_key=configuration.private_key,
         vapid_claims={"sub": configuration.subject},
         ttl=300,
+        timeout=10,
+        requests_session=session,
     )
 
 
@@ -45,6 +52,7 @@ async def dispatch_unread_updates(
     send_timeout_seconds: float = 10.0,
     max_concurrency: int = 8,
     max_individual_notifications: int = 3,
+    endpoint_validator: Callable[[str], Awaitable[str]] | None = None,
 ) -> int:
     if not configuration.enabled:
         return 0
@@ -90,6 +98,15 @@ async def dispatch_unread_updates(
             async with semaphore:
                 if ownership_lost.is_set():
                     return 0
+                validate_endpoint = endpoint_validator
+                if validate_endpoint is None and send is None:
+                    validate_endpoint = validate_public_push_endpoint
+                try:
+                    if validate_endpoint is not None:
+                        await validate_endpoint(delivery.subscription["endpoint"])
+                except UnsafePushEndpoint:
+                    await asyncio.to_thread(store.delete_session, delivery.session_id)
+                    return 0
                 count = 0
                 digest_pending = set(delivery.digest_revisions)
                 new_revisions = tuple(