diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dfaecb..fcaef68 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/research/dns-resolution-resource-bounds.md b/docs/research/dns-resolution-resource-bounds.md new file mode 100644 index 0000000..891b715 --- /dev/null +++ b/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 diff --git a/src/egressweave/validation.py b/src/egressweave/validation.py index 9542ab9..4e3fcf2 100644 --- a/src/egressweave/validation.py +++ b/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,10 +234,8 @@ 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: @@ -233,71 +243,121 @@ def _resolve_all_global_addresses_blocking( 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 \ No newline at end of file diff --git a/tests/test_dns_resolution_singleflight.py b/tests/test_dns_resolution_singleflight.py new file mode 100644 index 0000000..8f3cd19 --- /dev/null +++ b/tests/test_dns_resolution_singleflight.py @@ -0,0 +1,322 @@ +"""Regression contracts for bounded in-flight DNS resolution work.""" + +from __future__ import annotations + +import asyncio +import queue +import threading +import time + +import pytest + +from egressweave import ( + EGRESS_NOT_ALLOWED, + EgressNotAllowedError, + EgressPolicy, + validate_egress_url_details, + validate_egress_url_details_async, + validation, +) + +PUBLIC_ADDRESS = "93.184.216.34" +SECOND_PUBLIC_ADDRESS = "93.184.216.35" +CALLER_COUNT = 4 +DNS_TIMEOUT_SECONDS = 0.75 +AUTHORITY_KEY = ("api.example.com", 443) + + +class _CountingResolutionSlots: + """Allow resolver work while recording acquired and released worker slots.""" + + def __init__(self) -> None: + self.acquire_count = 0 + self.release_count = 0 + self._condition = threading.Condition() + + def acquire(self, *, timeout: float) -> bool: + """Record one successful acquisition without imposing a test-only ceiling.""" + assert timeout > 0 + with self._condition: + self.acquire_count += 1 + self._condition.notify_all() + return True + + def release(self) -> None: + """Record one worker-slot release.""" + with self._condition: + self.release_count += 1 + self._condition.notify_all() + + def wait_until_balanced(self, timeout: float) -> bool: + """Wait until every slot acquired by this test has been released.""" + deadline = time.monotonic() + timeout + with self._condition: + while self.release_count < self.acquire_count: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._condition.wait(remaining) + return True + + +def _install_blocking_resolver(monkeypatch): + """Install one deterministic resolver that stays live past caller deadlines.""" + release_resolver = threading.Event() + call_lock = threading.Lock() + call_count = 0 + + def slow_getaddrinfo(hostname, port, *, type): + nonlocal call_count + assert hostname == AUTHORITY_KEY[0] + assert port == AUTHORITY_KEY[1] + assert type == validation.socket.SOCK_STREAM + with call_lock: + call_count += 1 + if not release_resolver.wait(timeout=5.0): + raise RuntimeError("test resolver was not released") + return [(2, 1, 6, "", (PUBLIC_ADDRESS, port))] + + monkeypatch.setattr(validation.socket, "getaddrinfo", slow_getaddrinfo) + + def observed_calls() -> int: + with call_lock: + return call_count + + return release_resolver, observed_calls + + +def _active_authority_flight(): + """Return the currently registered test-authority flight, if one is live.""" + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + return validation._DNS_RESOLUTION_FLIGHTS.get(AUTHORITY_KEY) + + +def _release_and_wait_for_flight( + release_resolver: threading.Event, + slots: _CountingResolutionSlots, +) -> None: + """Release a controlled worker and wait for slot plus registry cleanup.""" + flight = _active_authority_flight() + release_resolver.set() + if flight is not None: + assert flight.completed.wait(timeout=3.0) + assert slots.wait_until_balanced(timeout=3.0) + assert _active_authority_flight() is None + + +def _policy(*, max_resolved_addresses: int = 16) -> EgressPolicy: + """Return the short-deadline policy used by the concurrency regressions.""" + return EgressPolicy.from_hosts( + AUTHORITY_KEY[0], + dns_timeout_seconds=DNS_TIMEOUT_SECONDS, + max_resolved_addresses=max_resolved_addresses, + ) + + +def test_sync_same_authority_timeouts_share_one_live_resolver(monkeypatch) -> None: + """Repeated sync timeouts for one authority must share one live DNS worker.""" + slots = _CountingResolutionSlots() + release_resolver, observed_calls = _install_blocking_resolver(monkeypatch) + monkeypatch.setattr(validation, "_DNS_RESOLUTION_SLOTS", slots) + caller_barrier = threading.Barrier(CALLER_COUNT + 1) + outcomes: queue.Queue[str] = queue.Queue() + + def validate_from_caller() -> None: + try: + caller_barrier.wait(timeout=5.0) + validate_egress_url_details( + "https://api.example.com", + policy=_policy(), + ) + except EgressNotAllowedError as exc: + outcomes.put(str(exc)) + except Exception as exc: # noqa: BLE001 - preserve thread diagnostics + outcomes.put(f"unexpected {type(exc).__name__}: {exc}") + else: + outcomes.put("unexpected success") + + callers = [ + threading.Thread(target=validate_from_caller, name=f"dns-caller-{index}") + for index in range(CALLER_COUNT) + ] + for caller in callers: + caller.start() + + try: + caller_barrier.wait(timeout=5.0) + for caller in callers: + caller.join(timeout=DNS_TIMEOUT_SECONDS + 2.0) + + assert not any(caller.is_alive() for caller in callers) + assert sorted(outcomes.get_nowait() for _ in range(CALLER_COUNT)) == [ + EGRESS_NOT_ALLOWED + ] * CALLER_COUNT + assert observed_calls() == 1 + assert slots.acquire_count == 1 + finally: + _release_and_wait_for_flight(release_resolver, slots) + + +async def test_async_same_authority_timeouts_share_one_live_resolver(monkeypatch) -> None: + """Repeated async timeouts for one authority must share the same live worker.""" + slots = _CountingResolutionSlots() + release_resolver, observed_calls = _install_blocking_resolver(monkeypatch) + monkeypatch.setattr(validation, "_DNS_RESOLUTION_SLOTS", slots) + start_callers = asyncio.Event() + + async def validate_from_caller() -> None: + await start_callers.wait() + with pytest.raises( + EgressNotAllowedError, + match=f"^{EGRESS_NOT_ALLOWED}$", + ): + await validate_egress_url_details_async( + "https://api.example.com", + policy=_policy(), + ) + + callers = [asyncio.create_task(validate_from_caller()) for _ in range(CALLER_COUNT)] + await asyncio.sleep(0) + + try: + start_callers.set() + await asyncio.wait_for( + asyncio.gather(*callers), + timeout=DNS_TIMEOUT_SECONDS + 3.0, + ) + assert observed_calls() == 1 + assert slots.acquire_count == 1 + finally: + _release_and_wait_for_flight(release_resolver, slots) + + +def test_empty_shared_resolver_result_remains_generic(monkeypatch) -> None: + """Normalize one shared worker's empty DNS result to the public denial.""" + monkeypatch.setattr(validation.socket, "getaddrinfo", lambda *args, **kwargs: []) + + with pytest.raises( + EgressNotAllowedError, + match=f"^{EGRESS_NOT_ALLOWED}$", + ): + validate_egress_url_details("https://api.example.com", policy=_policy()) + + +def test_completed_flight_without_outcome_fails_closed() -> None: + """Reject an internally incomplete completed flight instead of trusting it.""" + flight = validation._DNSResolutionFlight() + flight.completed.set() + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + validation._DNS_RESOLUTION_FLIGHTS[AUTHORITY_KEY] = flight + + try: + with pytest.raises( + EgressNotAllowedError, + match=f"^{EGRESS_NOT_ALLOWED}$", + ): + validate_egress_url_details("https://api.example.com", policy=_policy()) + finally: + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + validation._DNS_RESOLUTION_FLIGHTS.pop(AUTHORITY_KEY, None) + + +def test_completed_resolution_is_not_cached(monkeypatch) -> None: + """Perform a fresh DNS lookup after each completed same-authority flight.""" + call_count = 0 + + def counting_getaddrinfo(hostname, port, *, type): + nonlocal call_count + call_count += 1 + return [(2, 1, 6, "", (PUBLIC_ADDRESS, port))] + + monkeypatch.setattr(validation.socket, "getaddrinfo", counting_getaddrinfo) + + first = validate_egress_url_details("https://api.example.com", policy=_policy()) + second = validate_egress_url_details("https://api.example.com", policy=_policy()) + + assert first is not None + assert second is not None + assert first.addresses == (PUBLIC_ADDRESS,) + assert second.addresses == (PUBLIC_ADDRESS,) + assert call_count == 2 + + +def test_shared_raw_result_is_validated_under_each_callers_policy(monkeypatch) -> None: + """Keep policy-specific address cardinality outside the shared DNS worker.""" + release_resolver = threading.Event() + resolver_started = threading.Event() + joiner_arrived = threading.Event() + call_count = 0 + original_join = validation._join_dns_resolution_flight + + def observing_join(hostname: str, port: int): + result = original_join(hostname, port) + if not result[2]: + joiner_arrived.set() + return result + + def two_address_getaddrinfo(hostname, port, *, type): + nonlocal call_count + call_count += 1 + resolver_started.set() + assert release_resolver.wait(timeout=5.0) + return [ + (2, 1, 6, "", (PUBLIC_ADDRESS, port)), + (2, 1, 6, "", (SECOND_PUBLIC_ADDRESS, port)), + ] + + monkeypatch.setattr(validation, "_join_dns_resolution_flight", observing_join) + monkeypatch.setattr(validation.socket, "getaddrinfo", two_address_getaddrinfo) + outcomes: queue.Queue[tuple[str, tuple[str, ...] | None]] = queue.Queue() + + def validate_with_limit(limit: int) -> None: + try: + result = validate_egress_url_details( + "https://api.example.com", + policy=_policy(max_resolved_addresses=limit), + ) + except EgressNotAllowedError: + outcomes.put(("denied", None)) + else: + assert result is not None + outcomes.put(("allowed", result.addresses)) + + strict_caller = threading.Thread(target=validate_with_limit, args=(1,)) + broad_caller = threading.Thread(target=validate_with_limit, args=(2,)) + strict_caller.start() + assert resolver_started.wait(timeout=2.0) + broad_caller.start() + assert joiner_arrived.wait(timeout=2.0) + release_resolver.set() + strict_caller.join(timeout=3.0) + broad_caller.join(timeout=3.0) + + assert not strict_caller.is_alive() + assert not broad_caller.is_alive() + assert call_count == 1 + assert sorted(outcomes.get_nowait() for _ in range(2)) == [ + ("allowed", (PUBLIC_ADDRESS, SECOND_PUBLIC_ADDRESS)), + ("denied", None), + ] + assert _active_authority_flight() is None + + +def test_unexpected_shared_resolver_failure_has_no_private_provenance(monkeypatch) -> None: + """Normalize dependency-controlled resolver failures without retaining a cause.""" + + def failing_getaddrinfo(hostname, port, *, type): + assert hostname == AUTHORITY_KEY[0] + assert port == AUTHORITY_KEY[1] + assert type == validation.socket.SOCK_STREAM + raise RuntimeError("private resolver detail") + + monkeypatch.setattr(validation.socket, "getaddrinfo", failing_getaddrinfo) + + with pytest.raises( + EgressNotAllowedError, + match=f"^{EGRESS_NOT_ALLOWED}$", + ) as exc_info: + validate_egress_url_details("https://api.example.com", policy=_policy()) + + assert exc_info.value.__cause__ is None + assert exc_info.value.__context__ is None + assert _active_authority_flight() is None diff --git a/tests/test_dns_resolution_singleflight_documentation.py b/tests/test_dns_resolution_singleflight_documentation.py new file mode 100644 index 0000000..7e4a33e --- /dev/null +++ b/tests/test_dns_resolution_singleflight_documentation.py @@ -0,0 +1,70 @@ +"""Documentation contracts for bounded same-authority DNS single-flight work.""" + +from __future__ import annotations + +import re +from pathlib import Path + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +RESEARCH_PATH = REPOSITORY_ROOT / "docs" / "research" / "dns-resolution-resource-bounds.md" +CHANGELOG_PATH = REPOSITORY_ROOT / "CHANGELOG.md" + + +def _normalized_section(path: Path, heading: str) -> str: + """Return one named Markdown section with whitespace normalized.""" + text = path.read_text(encoding="utf-8") + marker = f"{heading}\n" + section = text.split(marker, 1)[1] + level = len(heading) - len(heading.lstrip("#")) + next_heading = re.search(rf"\n#{{1,{level}}} ", section) + if next_heading is not None: + section = section[: next_heading.start()] + return " ".join(section.split()) + + +def test_dns_singleflight_operator_guidance_records_the_exact_boundary() -> None: + """Require operator guidance for live-only sharing and per-caller validation.""" + guidance = _normalized_section(RESEARCH_PATH, "## Implemented boundary") + + required = ( + "in-flight", + "(hostname, port)", + "never caches a completed DNS result", + "each caller retains its own", + "max_resolved_addresses", + "before executor scheduling", + ) + missing = [fragment for fragment in required if fragment not in guidance] + assert not missing, f"DNS single-flight implemented boundary is missing: {missing}" + + +def test_dns_singleflight_operator_guidance_records_residual_platform_limits() -> None: + """Keep non-cancellable platform-resolver limits in their canonical section.""" + guidance = _normalized_section(RESEARCH_PATH, "## Residual platform limitation") + + required = ("socket.getaddrinfo", "cannot safely cancel", "DNS rebinding") + missing = [fragment for fragment in required if fragment not in guidance] + assert not missing, f"DNS residual platform guidance is missing: {missing}" + + +def test_changelog_records_same_authority_dns_worker_deduplication() -> None: + """Keep the buyer-facing release history aligned with the resolver repair.""" + changelog = _normalized_section(CHANGELOG_PATH, "### Security") + + assert "same-authority DNS" in changelog + assert "in-flight" in changelog + assert "Completed DNS results" in changelog + + +def test_operator_guidance_counts_async_executor_scheduling_inside_deadline() -> None: + """State that async timeout accounting starts before executor scheduling.""" + guidance = _normalized_section(RESEARCH_PATH, "## Implemented boundary") + + assert "before executor scheduling" in guidance + + +def test_changelog_records_async_executor_scheduling_deadline() -> None: + """Keep the async DNS deadline tightening visible in release history.""" + changelog = _normalized_section(CHANGELOG_PATH, "### Security") + + assert "executor scheduling" in changelog diff --git a/tests/test_dns_singleflight_docstrings.py b/tests/test_dns_singleflight_docstrings.py new file mode 100644 index 0000000..a1af9f8 --- /dev/null +++ b/tests/test_dns_singleflight_docstrings.py @@ -0,0 +1,25 @@ +"""Documentation contracts for DNS single-flight production helpers.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +VALIDATION_SOURCE = Path("src/egressweave/validation.py") + + +def test_dns_singleflight_worker_has_beginner_readable_docstring() -> None: + """Keep the resolver worker helper understandable at the production boundary.""" + module = ast.parse(VALIDATION_SOURCE.read_text(encoding="utf-8")) + worker = next( + node + for node in ast.walk(module) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "resolve_on_worker" + ) + + docstring = ast.get_docstring(worker) + assert docstring is not None + normalized = docstring.lower() + assert "resolver" in normalized or "address" in normalized + assert "shared" in normalized diff --git a/tests/test_dns_timeout.py b/tests/test_dns_timeout.py index 30bd89a..01bd158 100644 --- a/tests/test_dns_timeout.py +++ b/tests/test_dns_timeout.py @@ -1,3 +1,4 @@ +import asyncio import math import threading import time @@ -12,6 +13,8 @@ ) from egressweave import validation as v +_AUTHORITY_KEY = ("api.openai.com", 443) + @pytest.mark.parametrize( "timeout", @@ -48,6 +51,18 @@ def fake_getaddrinfo(host, port, type=None): return started, release +def _release_and_wait_for_resolver(release: threading.Event) -> None: + with v._DNS_RESOLUTION_FLIGHTS_LOCK: + flight = v._DNS_RESOLUTION_FLIGHTS.get(_AUTHORITY_KEY) + + assert flight is not None + release.set() + assert flight.completed.wait(timeout=1.0) + + with v._DNS_RESOLUTION_FLIGHTS_LOCK: + assert _AUTHORITY_KEY not in v._DNS_RESOLUTION_FLIGHTS + + def test_sync_validation_enforces_dns_timeout(monkeypatch): started, release = _install_blocking_resolver(monkeypatch) policy = EgressPolicy.from_hosts( @@ -67,7 +82,7 @@ def test_sync_validation_enforces_dns_timeout(monkeypatch): assert started.wait(timeout=0.2) assert time.monotonic() - started_at < 0.5 finally: - release.set() + _release_and_wait_for_resolver(release) async def test_async_validation_uses_same_bounded_dns_timeout(monkeypatch): @@ -89,7 +104,34 @@ async def test_async_validation_uses_same_bounded_dns_timeout(monkeypatch): assert started.wait(timeout=0.2) assert time.monotonic() - started_at < 0.5 finally: - release.set() + _release_and_wait_for_resolver(release) + + +async def test_async_dns_timeout_includes_to_thread_scheduling_delay(monkeypatch): + """Count executor scheduling delay inside the public async DNS deadline.""" + policy = EgressPolicy.from_hosts( + "api.openai.com", + dns_timeout_seconds=0.05, + ) + + async def delayed_to_thread(function, *args): + assert function is v._resolve_all_global_addresses + assert args == ("api.openai.com", 443, policy) + await asyncio.sleep(0.25) + return ("93.184.216.34",) + + monkeypatch.setattr(v.asyncio, "to_thread", delayed_to_thread) + started_at = time.monotonic() + + with pytest.raises( + EgressNotAllowedError, match="^egress URL is not allowed$" + ): + await validate_egress_url_details_async( + "https://api.openai.com/v1", + policy=policy, + ) + + assert time.monotonic() - started_at < 0.2 def test_resolver_failure_remains_generic(monkeypatch): diff --git a/tests/test_dns_worker_start_failure.py b/tests/test_dns_worker_start_failure.py new file mode 100644 index 0000000..a2e5bb7 --- /dev/null +++ b/tests/test_dns_worker_start_failure.py @@ -0,0 +1,142 @@ +"""Regression tests for DNS resolver worker-start failure containment.""" + +from __future__ import annotations + +import ast +import inspect +import textwrap + +import pytest + +from egressweave import ( + EGRESS_NOT_ALLOWED, + EgressNotAllowedError, + EgressPolicy, + validation, +) + +_HOSTNAME = "worker-start.example.com" +_PORT = 443 +_AUTHORITY_KEY = (_HOSTNAME, _PORT) +_POLICY = EgressPolicy.from_hosts(_HOSTNAME) + + +class _CountingResolutionSlots: + """Record resolver-slot acquisitions and releases for start-failure tests.""" + + def __init__(self) -> None: + self.acquire_count = 0 + self.release_count = 0 + + def acquire(self, *, timeout: float) -> bool: + """Record one successful slot acquisition.""" + assert timeout > 0 + self.acquire_count += 1 + return True + + def release(self) -> None: + """Record one resolver-slot release.""" + self.release_count += 1 + + +class _SyntheticThreadStartFailure(Exception): + """Model a non-RuntimeError ordinary failure before the resolver worker starts.""" + + +def _install_failing_resolver_thread(monkeypatch, failure: BaseException) -> None: + """Fail only EgressWeave's resolver thread while preserving other threads.""" + original_thread = validation.threading.Thread + + class _BrokenThread: + """Synthetic resolver thread that fails before any worker can run.""" + + def start(self) -> None: + """Raise the configured platform startup failure.""" + raise failure + + def selective_thread(*args, **kwargs): + if kwargs.get("name") == "egressweave-dns-resolver": + return _BrokenThread() + return original_thread(*args, **kwargs) + + monkeypatch.setattr(validation.threading, "Thread", selective_thread) + + +def _assert_generic_start_failure(monkeypatch, failure: Exception) -> None: + """Require one failed worker start to release its slot and flight.""" + slots = _CountingResolutionSlots() + monkeypatch.setattr(validation, "_DNS_RESOLUTION_SLOTS", slots) + _install_failing_resolver_thread(monkeypatch, failure) + + try: + with pytest.raises( + EgressNotAllowedError, + match=f"^{EGRESS_NOT_ALLOWED}$", + ) as exc_info: + validation._resolve_all_global_addresses(_HOSTNAME, _PORT, _POLICY) + finally: + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + validation._DNS_RESOLUTION_FLIGHTS.pop(_AUTHORITY_KEY, None) + + assert exc_info.value.__cause__ is None + assert exc_info.value.__context__ is None + assert slots.acquire_count == 1 + assert slots.release_count == slots.acquire_count + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + assert _AUTHORITY_KEY not in validation._DNS_RESOLUTION_FLIGHTS + + +def test_dns_worker_start_runtime_failure_erases_private_provenance(monkeypatch) -> None: + """Normalize the documented RuntimeError startup failure and clean resources.""" + _assert_generic_start_failure( + monkeypatch, + RuntimeError("private thread-start detail"), + ) + + +def test_dns_worker_start_non_runtime_failure_fails_closed(monkeypatch) -> None: + """Clean resources when an ordinary non-RuntimeError failure prevents startup.""" + _assert_generic_start_failure( + monkeypatch, + _SyntheticThreadStartFailure("private thread-start detail"), + ) + + +def test_dns_worker_start_avoids_direct_base_exception_handler() -> None: + """Keep worker-start cleanup explicit without a catch-all exception handler.""" + source = textwrap.dedent(inspect.getsource(validation._resolve_all_global_addresses)) + syntax_tree = ast.parse(source) + direct_base_exception_handlers = [ + handler + for node in ast.walk(syntax_tree) + if isinstance(node, ast.Try) + for handler in node.handlers + if handler.type is None + or ( + isinstance(handler.type, ast.Name) + and handler.type.id == "BaseException" + ) + ] + + assert direct_base_exception_handlers == [] + + +def test_dns_worker_start_interrupt_cleans_before_propagating(monkeypatch) -> None: + """Release owned resolver state before propagating a real process-control interrupt.""" + slots = _CountingResolutionSlots() + failure = KeyboardInterrupt("synthetic process-control interruption") + monkeypatch.setattr(validation, "_DNS_RESOLUTION_SLOTS", slots) + _install_failing_resolver_thread(monkeypatch, failure) + + try: + with pytest.raises(KeyboardInterrupt) as exc_info: + validation._resolve_all_global_addresses(_HOSTNAME, _PORT, _POLICY) + + assert exc_info.value is failure + assert slots.acquire_count == 1 + assert slots.release_count == slots.acquire_count + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + assert _AUTHORITY_KEY not in validation._DNS_RESOLUTION_FLIGHTS + finally: + with validation._DNS_RESOLUTION_FLIGHTS_LOCK: + validation._DNS_RESOLUTION_FLIGHTS.pop(_AUTHORITY_KEY, None)