reliability: reconstruct DNS single-flight on protected main

CHANGELOG.md

@@ -91,6 +91,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - Harden release publication evidence with validated integrating-PR identity,
   cross-repository required-workflow source checks, and Strix check-run
   annotations without adding an elevated release credential.
+- Deduplicate overlapping same-authority DNS resolution through one live
+  in-flight `(hostname, port)` worker while preserving the finite global
+  resolver ceiling, per-caller deadlines, and caller-specific address policy.
+  Completed DNS results are never cached, later validations perform a fresh
+  lookup, and unexpected resolver failures remain behind the generic denial
+  boundary without private exception provenance. Asynchronous validation starts
+  its caller-owned deadline before executor scheduling, so `asyncio.to_thread`
+  queue delay cannot extend the configured public DNS wait budget.
 - Require the exact `TLSConfiguration` type before TLS context creation. A
   subclass can no longer override `create_ssl_context()` to replace the reviewed
   immutable policy with a context that disables hostname or certificate

docs/research/dns-resolution-resource-bounds.md

@@ -0,0 +1,133 @@
+# DNS resolution resource bounds
+
+## Scope
+
+EgressWeave resolves an already-authorized canonical `(hostname, port)` before a
+pinned transport can connect. DNS remains an untrusted and potentially slow
+platform dependency: callers need a finite wait, the process needs a finite
+resolver-worker budget, and a slow authority must not multiply background work
+merely because several requests arrive together.
+
+This note records the exact runtime boundary implemented by the DNS validation
+layer. It does not make DNS an authorization source and does not replace the
+existing all-address scope checks, exact authority policy, transport pinning, or
+per-connect address revalidation.
+
+## Implemented boundary
+
+The process retains one finite global resolver-worker pool. When overlapping
+validations target the same exact authority, they join one live **in-flight**
+resolver operation keyed by canonical `(hostname, port)` instead of creating a
+new platform resolver thread for every caller.
+
+The sharing boundary is intentionally live-only:
+
+- the registry **never caches a completed DNS result**;
+- the entry is removed as soon as the worker completes and current waiters are
+  released;
+- a later validation therefore starts a fresh lookup, preserving the product's
+  DNS rebinding freshness model instead of converting single-flight into a DNS
+  cache;
+- the worker publishes only raw address strings; **each caller retains its own**
+  finite `dns_timeout_seconds` deadline and independently applies all-address
+  classification, local-development rules, deduplication, and its own
+  `max_resolved_addresses` ceiling;
+- for asynchronous validation, that caller-owned deadline starts **before executor scheduling** of the synchronous resolver wrapper, so time spent waiting for `asyncio.to_thread(...)` executor capacity counts against `dns_timeout_seconds` instead of extending the public wait budget;
+- empty answers, capacity exhaustion, worker-start failure, resolver failure,
+  caller deadline exhaustion, and incomplete shared outcomes all fail closed;
+- unexpected resolver exceptions are normalized to the public generic denial
+  without retaining dependency-controlled `__cause__` or `__context__` text.
+
+This arrangement reduces same-authority amplification without weakening policy
+isolation. A permissive caller cannot donate its larger address cardinality or
+local-address scope to a stricter caller that happened to share the same raw DNS
+lookup.
+
+## Residual platform limitation
+
+The standard-library resolver path ultimately calls `socket.getaddrinfo`, whose
+work can block in the operating-system or C-library resolver. EgressWeave runs
+that call on a daemon thread so the caller can stop waiting at its configured
+deadline, but Python does not provide a safe general operation for forcibly
+terminating arbitrary already-running thread work. EgressWeave therefore
+**cannot safely cancel** an already-running platform `socket.getaddrinfo` call.
+
+For the asynchronous API, `asyncio.wait_for(...)` bounds the caller-visible
+operation beginning before executor scheduling. Cancellation of that await does
+not imply that a resolver wrapper or platform resolver already running on a
+thread has been force-stopped. The same live-only single-flight and finite global
+resolver-slot rules remain the resource backstop after a caller stops waiting.
+
+A timed-out resolver worker may remain alive and keep one global resolver slot
+until the platform call returns. Single-flight prevents repeated callers for one
+slow authority from consuming additional slots, but distinct stalled
+authorities can still occupy the finite global pool. Once capacity is exhausted,
+new validations fail closed rather than creating unbounded resolver work. The
+caller timeout consequently bounds caller waiting time; it does not claim to
+bound the operating-system resolver's lifetime.
+
+This residual is deliberate. The package does not interrupt Python threads,
+ship a recursive DNS resolver, silently detach unbounded work, or persist a
+completed result merely to avoid another lookup. Those alternatives would
+introduce larger correctness, portability, security, or DNS rebinding risks.
+
+## Security interpretation
+
+The original failure mode is an asymmetric resource-consumption problem: a
+small number of repeated validations could leave disproportionate live resolver
+work after the initiating callers had already failed. CWE-405 describes this
+amplification family, while CWE-410 describes exhaustion of a finite resource
+pool and CWE-400 is the broader uncontrolled-resource-consumption class. The
+runtime fix both meters the global resource and collapses duplicate
+same-authority live work.
+
+These CWE entries are diagnostic taxonomy, not proof of vulnerability severity.
+CWE-400 is a high-level class and MITRE discourages mapping to overly broad
+entries when a more specific weakness is available.
+
+## Operational expectations
+
+Host applications and operators should preserve these assumptions:
+
+1. A DNS timeout is a caller-wait bound, not a guarantee that the platform
+   resolver thread has terminated. For async validation the bound includes
+   executor scheduling delay before the synchronous wrapper starts.
+2. Repeated generic DNS denials or resolver-capacity pressure should be measured
+   outside the core library using purpose-limited counters; raw candidate URLs,
+   credentials, resolved IP addresses, and resolver exception text should not be
+   added to default telemetry.
+3. Resolver availability, operating-system DNS configuration, recursive-server
+   behavior, and network reachability remain host/platform responsibilities.
+4. EgressWeave must continue to re-resolve after a completed flight. Persisting
+   completed DNS results would change the reviewed DNS rebinding and freshness
+   boundary and requires a separate architecture decision.
+5. A future resolver implementation must preserve exact authority identity,
+   finite concurrency, per-caller policy validation, generic failure behavior,
+   and fresh post-completion resolution before it can replace this boundary.
+
+RFC 8305 treats hostname resolution and subsequent connection concurrency as
+separate stages and explicitly accounts for DNS answers changing during
+connection setup. EgressWeave does not claim RFC 8305 defines its single-flight
+mechanism; the RFC is relevant because the live-only design avoids turning a
+resource-control optimization into a persistent answer cache that would erase
+fresh DNS observations.
+
+## References
+
+MITRE. (2026). *CWE-400: Uncontrolled resource consumption (Version 4.20)*.
+https://cwe.mitre.org/data/definitions/400.html
+
+MITRE. (2026). *CWE-405: Asymmetric resource consumption (amplification)
+(Version 4.20)*. https://cwe.mitre.org/data/definitions/405.html
+
+MITRE. (2026). *CWE-410: Insufficient resource pool (Version 4.20)*.
+https://cwe.mitre.org/data/definitions/410.html
+
+Python Software Foundation. (2026). *socket — Low-level networking interface
+(Python 3.13.14 documentation)*. https://docs.python.org/3.13/library/socket.html
+
+Python Software Foundation. (2026). *threading — Thread-based parallelism
+(Python 3.13.14 documentation)*. https://docs.python.org/3.13/library/threading.html
+
+Schinazi, D., & Pauly, T. (2017). *Happy Eyeballs Version 2: Better connectivity
+using concurrency* (RFC 8305). RFC Editor. https://doi.org/10.17487/RFC8305

src/egressweave/validation.py

@@ -18,7 +18,6 @@
 import hashlib
 import hmac
 import ipaddress
-import queue
 import secrets
 import socket
 import threading
@@ -34,6 +33,19 @@
 _MAX_CONCURRENT_DNS_RESOLUTIONS = 32
 _DNS_RESOLUTION_SLOTS = threading.BoundedSemaphore(_MAX_CONCURRENT_DNS_RESOLUTIONS)
 
+
+@dataclass
+class _DNSResolutionFlight:
+    """Share one live resolver result among callers of the same authority."""
+
+    completed: threading.Event = field(default_factory=threading.Event)
+    raw_addresses: tuple[str, ...] | None = None
+    error: Exception | None = None
+
+
+_DNS_RESOLUTION_FLIGHTS_LOCK = threading.Lock()
+_DNS_RESOLUTION_FLIGHTS: dict[tuple[str, int], _DNSResolutionFlight] = {}
+
 _LOCAL_DEV_HOSTNAMES = frozenset({"localhost", "localhost.localdomain"})
 _PRIVATE_LOCAL_NETWORKS = (
     ipaddress.ip_network("10.0.0.0/8"),
@@ -222,82 +234,130 @@ def _validate_bounded_unique_addresses(
     return tuple(validated_addresses)
 
 
-def _resolve_all_global_addresses_blocking(
-    hostname: str, port: int, policy: EgressPolicy
-) -> tuple[str, ...]:
-    """Resolve and validate a finite unique address set on the worker thread."""
+def _resolve_raw_addresses_blocking(hostname: str, port: int) -> tuple[str, ...]:
+    """Resolve raw address strings without applying any caller-specific policy."""
     try:
         address_infos = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM)
     except (OSError, UnicodeError) as exc:
         raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from exc
 
     if not address_infos:
         raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
+    return tuple(str(address_info[4][0]) for address_info in address_infos)
+
+
+def _resolve_all_global_addresses_blocking(
+    hostname: str, port: int, policy: EgressPolicy
+) -> tuple[str, ...]:
+    """Resolve and validate a finite unique address set on the worker thread."""
     return _validate_bounded_unique_addresses(
-        (str(address_info[4][0]) for address_info in address_infos),
+        _resolve_raw_addresses_blocking(hostname, port),
         policy,
         hostname=hostname,
     )
 
 
+def _join_dns_resolution_flight(
+    hostname: str,
+    port: int,
+) -> tuple[tuple[str, int], _DNSResolutionFlight, bool]:
+    """Return the live authority flight, creating it for exactly one owner."""
+    key = (hostname, port)
+    with _DNS_RESOLUTION_FLIGHTS_LOCK:
+        flight = _DNS_RESOLUTION_FLIGHTS.get(key)
+        if flight is not None:
+            return key, flight, False
+        flight = _DNSResolutionFlight()
+        _DNS_RESOLUTION_FLIGHTS[key] = flight
+        return key, flight, True
+
+
+def _complete_dns_resolution_flight(
+    key: tuple[str, int],
+    flight: _DNSResolutionFlight,
+) -> None:
+    """Remove one completed live flight before waking its current waiters."""
+    with _DNS_RESOLUTION_FLIGHTS_LOCK:
+        _DNS_RESOLUTION_FLIGHTS.pop(key, None)
+    flight.completed.set()
+
+
 def _resolve_all_global_addresses(
     hostname: str, port: int, policy: EgressPolicy
 ) -> tuple[str, ...]:
     """Resolve addresses within a bounded, fail-closed synchronous deadline."""
     deadline = time.monotonic() + policy.dns_timeout_seconds
-    if not _DNS_RESOLUTION_SLOTS.acquire(timeout=policy.dns_timeout_seconds):
-        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
+    key, flight, owns_flight = _join_dns_resolution_flight(hostname, port)
 
-    remaining_timeout = deadline - time.monotonic()
-    if remaining_timeout <= 0:
-        _DNS_RESOLUTION_SLOTS.release()
-        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
+    if owns_flight:
+        if not _DNS_RESOLUTION_SLOTS.acquire(timeout=policy.dns_timeout_seconds):
+            flight.error = EgressNotAllowedError(EGRESS_NOT_ALLOWED)
+            _complete_dns_resolution_flight(key, flight)
+            raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
 
-    resolution_result: queue.Queue[
-        tuple[tuple[str, ...] | None, Exception | None]
-    ] = queue.Queue(maxsize=1)
+        remaining_timeout = deadline - time.monotonic()
+        if remaining_timeout <= 0:
+            _DNS_RESOLUTION_SLOTS.release()
+            flight.error = EgressNotAllowedError(EGRESS_NOT_ALLOWED)
+            _complete_dns_resolution_flight(key, flight)
+            raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
 
-    def resolve_on_worker() -> None:
+        def resolve_on_worker() -> None:
+            """Publish one shared resolver address flight and release its worker slot."""
+            try:
+                flight.raw_addresses = _resolve_raw_addresses_blocking(hostname, port)
+            except Exception as exc:  # noqa: BLE001
+                flight.error = exc
+            finally:
+                _DNS_RESOLUTION_SLOTS.release()
+                _complete_dns_resolution_flight(key, flight)
+
+        worker = threading.Thread(
+            target=resolve_on_worker,
+            name="egressweave-dns-resolver",
+            daemon=True,
+        )
+        worker_start_failed = False
         try:
-            addresses = _resolve_all_global_addresses_blocking(hostname, port, policy)
-            resolution_result.put((addresses, None))
+            worker.start()
         except Exception as exc:  # noqa: BLE001
-            resolution_result.put((None, exc))
-        finally:
             _DNS_RESOLUTION_SLOTS.release()
+            flight.error = exc
+            worker_start_failed = True
+            _complete_dns_resolution_flight(key, flight)
+        except (KeyboardInterrupt, SystemExit, GeneratorExit):
+            _DNS_RESOLUTION_SLOTS.release()
+            _complete_dns_resolution_flight(key, flight)
+            raise
+        if worker_start_failed:
+            raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
 
-    worker = threading.Thread(
-        target=resolve_on_worker,
-        name="egressweave-dns-resolver",
-        daemon=True,
-    )
-    try:
-        worker.start()
-    except RuntimeError as exc:
-        _DNS_RESOLUTION_SLOTS.release()
-        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from exc
-
-    try:
-        addresses, error = resolution_result.get(
-            timeout=max(0.0, deadline - time.monotonic())
-        )
-    except queue.Empty as exc:
-        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from exc
+    remaining_timeout = deadline - time.monotonic()
+    if remaining_timeout <= 0 or not flight.completed.wait(timeout=remaining_timeout):
+        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
 
-    if error is not None:
-        if isinstance(error, EgressNotAllowedError):
-            raise error
-        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from error
-    if addresses is None:
+    if flight.error is not None:
+        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
+    if flight.raw_addresses is None:
         raise EgressNotAllowedError(EGRESS_NOT_ALLOWED)
-    return addresses
+    return _validate_bounded_unique_addresses(
+        flight.raw_addresses,
+        policy,
+        hostname=hostname,
+    )
 
 
 async def _resolve_all_global_addresses_async(
     hostname: str, port: int, policy: EgressPolicy
 ) -> tuple[str, ...]:
-    """Run the same bounded resolver without blocking the event loop."""
-    return await asyncio.to_thread(_resolve_all_global_addresses, hostname, port, policy)
+    """Run the bounded resolver under one caller-owned asynchronous deadline."""
+    try:
+        return await asyncio.wait_for(
+            asyncio.to_thread(_resolve_all_global_addresses, hostname, port, policy),
+            timeout=policy.dns_timeout_seconds,
+        )
+    except asyncio.TimeoutError:
+        raise EgressNotAllowedError(EGRESS_NOT_ALLOWED) from None
 
 
 def _parse_and_validate_candidate_url(
@@ -479,4 +539,4 @@ async def validate_egress_url_async(
     validated = await validate_egress_url_details_async(value, policy=policy)
     if validated is None:
         return None
-    return validated.normalized_url
+    return validated.normalized_url