From 9e1f938d5629d523e0855c3b8bca7ab64647620d Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 22 Sep 2026 08:30:14 -0700 Subject: [PATCH] Guard outbound connections in the httpcore network backend With internal addresses disallowed, the backend resolves the origin host, rejects it if any address is non-public, and dials the validated literals in family-interleaved order under the caller's connect timeout. Resolver failures surface as connect errors; unix sockets are always refused. Co-Authored-By: Claude Opus 5 --- src/paperless/network.py | 236 +++++++++- src/paperless/tests/test_network.py | 648 ++++++++++++++++++++++++++++ 2 files changed, 882 insertions(+), 2 deletions(-) diff --git a/src/paperless/network.py b/src/paperless/network.py index 83d48e12d..7448a4a00 100644 --- a/src/paperless/network.py +++ b/src/paperless/network.py @@ -1,20 +1,27 @@ import functools import ipaddress +import logging +import math import re import socket +import time from collections.abc import Callable from collections.abc import Collection from collections.abc import Iterable from enum import StrEnum from typing import Any +from typing import Final from typing import Self from typing import TypeAlias from urllib.parse import ParseResult from urllib.parse import urlparse import anyio +import httpcore import httpx +logger = logging.getLogger("paperless.network") + # requires-python is >=3.11, so no PEP 695 `type` statement. IPAddress: TypeAlias = ipaddress.IPv4Address | ipaddress.IPv6Address @@ -109,10 +116,12 @@ def is_public_ip(ip: IPAddress) -> bool: ) -# Resolver indirection so tests can fake DNS for this module without changing -# how the stock httpcore backends resolve the literals the guard dials. +# Resolver and clock indirection so tests can fake DNS and time for this module +# without changing how the stock httpcore backends resolve the literals the +# guard dials. _getaddrinfo = socket.getaddrinfo _agetaddrinfo = anyio.getaddrinfo +_monotonic = time.monotonic def _parse_ip_literal(host: str) -> IPAddress | None: @@ -183,6 +192,229 @@ async def aresolve_public_addresses( return _require_public(host, port, _collect_addresses(host, infos)) +MAX_ADDRESSES_TRIED: Final = 8 +MIN_ATTEMPT_TIMEOUT: Final = 2.0 +MAX_ATTEMPT_TIMEOUT: Final = 10.0 + + +def _require_positive_timeout(host: str, timeout: float | None) -> None: + # A zero timeout makes the socket non-blocking and a negative one is + # rejected by settimeout; neither can produce a useful connection attempt. + if timeout is not None and timeout <= 0: + raise httpcore.ConnectTimeout( + f"Connect timeout for {host} must be positive, got {timeout}", + ) + + +def _deadline(timeout: float | None) -> float: + return math.inf if timeout is None else _monotonic() + timeout + + +def _attempt_order(addresses: tuple[IPAddress, ...]) -> list[IPAddress]: + # Alternate address families, starting with the resolver's first family + # (RFC 8305 section 4), so one unreachable family cannot delay the other. + first_version = addresses[0].version + primary = [a for a in addresses if a.version == first_version] + secondary = [a for a in addresses if a.version != first_version] + ordered: list[IPAddress] = [] + for index in range(max(len(primary), len(secondary))): + ordered.extend(primary[index : index + 1]) + ordered.extend(secondary[index : index + 1]) + return ordered[:MAX_ADDRESSES_TRIED] + + +def _attempt_timeout(remaining: float, attempts_left: int) -> float: + """ + Budget for the next attempt. Once the budget is too small to split, or on + the last address, the attempt gets everything left. Otherwise it gets an + equal share clamped to [MIN, MAX], always leaving MIN for a later attempt. + The floor survives one lost SYN; the ceiling bounds how long a black-holed + address delays the next one. + """ + if attempts_left == 1 or remaining < 2 * MIN_ATTEMPT_TIMEOUT: + return remaining + share = remaining / attempts_left + return min( + MAX_ATTEMPT_TIMEOUT, + max(MIN_ATTEMPT_TIMEOUT, share), + remaining - MIN_ATTEMPT_TIMEOUT, + ) + + +def _as_httpcore_timeout(seconds: float) -> float | None: + return None if math.isinf(seconds) else seconds + + +def _log_block(error: OutboundRequestBlockedError) -> None: + logger.warning("Blocked outbound connection: %s", error) + + +def _budget_exhausted(host: str, tried: int, total: int) -> httpcore.ConnectTimeout: + return httpcore.ConnectTimeout( + f"Timed out connecting to {host} after trying {tried} of {total} addresses", + ) + + +def _resolve_for_connect(host: str, port: int) -> tuple[IPAddress, ...]: + try: + return resolve_public_addresses(host, port) + except OutboundRequestBlockedError as e: + _log_block(e) + raise + except HostResolutionError as e: + raise httpcore.ConnectError(str(e)) from e + + +class _GuardedSyncBackend(httpcore.NetworkBackend): + """ + Wraps httpcore's sync backend. With internal addresses disallowed, it + resolves the origin host itself, rejects the name if any address is + non-public, and dials the validated literals so the checked address is + the connected one. TLS still verifies against the origin hostname. + """ + + def __init__(self, inner: httpcore.NetworkBackend, *, allow_internal: bool) -> None: + self._inner = inner + self._allow_internal = allow_internal + + def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.NetworkStream: + if self._allow_internal: + return self._inner.connect_tcp( + host, + port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + _require_positive_timeout(host, timeout) + # Resolution is not charged to the budget, matching the stock backend. + candidates = _attempt_order(_resolve_for_connect(host, port)) + deadline = _deadline(timeout) + last_error: httpcore.ConnectError | httpcore.ConnectTimeout | None = None + for index, address in enumerate(candidates): + remaining = deadline - _monotonic() + if remaining <= 0: + raise _budget_exhausted(host, index, len(candidates)) + budget = _attempt_timeout(remaining, len(candidates) - index) + try: + return self._inner.connect_tcp( + str(address), + port, + timeout=_as_httpcore_timeout(budget), + local_address=local_address, + socket_options=socket_options, + ) + except (httpcore.ConnectError, httpcore.ConnectTimeout) as e: + logger.debug("Connecting to %s via %s failed: %s", host, address, e) + last_error = e + # candidates is never empty, so every address was tried and failed + raise last_error or _budget_exhausted(host, len(candidates), len(candidates)) + + def connect_unix_socket( + self, + path: str, + timeout: float | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.NetworkStream: + error = OutboundRequestBlockedError( + host=path, + port=None, + reason=BlockReason.UNIX_SOCKET, + ) + _log_block(error) + raise error + + def sleep(self, seconds: float) -> None: + self._inner.sleep(seconds) + + +class _GuardedAsyncBackend(httpcore.AsyncNetworkBackend): + """Async twin of _GuardedSyncBackend.""" + + def __init__( + self, + inner: httpcore.AsyncNetworkBackend, + *, + allow_internal: bool, + ) -> None: + self._inner = inner + self._allow_internal = allow_internal + + async def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + if self._allow_internal: + return await self._inner.connect_tcp( + host, + port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + _require_positive_timeout(host, timeout) + # Resolution counts against the budget, matching the stock backend. + # This scope closes before dialling; attempts are not nested inside it. + deadline = _deadline(timeout) + try: + with anyio.fail_after(timeout): + addresses = await aresolve_public_addresses(host, port) + except TimeoutError as e: + raise httpcore.ConnectTimeout(f"Timed out resolving {host}") from e + except OutboundRequestBlockedError as e: + _log_block(e) + raise + except HostResolutionError as e: + raise httpcore.ConnectError(str(e)) from e + candidates = _attempt_order(addresses) + last_error: httpcore.ConnectError | httpcore.ConnectTimeout | None = None + for index, address in enumerate(candidates): + remaining = deadline - _monotonic() + if remaining <= 0: + raise _budget_exhausted(host, index, len(candidates)) + budget = _attempt_timeout(remaining, len(candidates) - index) + try: + return await self._inner.connect_tcp( + str(address), + port, + timeout=_as_httpcore_timeout(budget), + local_address=local_address, + socket_options=socket_options, + ) + except (httpcore.ConnectError, httpcore.ConnectTimeout) as e: + logger.debug("Connecting to %s via %s failed: %s", host, address, e) + last_error = e + raise last_error or _budget_exhausted(host, len(candidates), len(candidates)) + + async def connect_unix_socket( + self, + path: str, + timeout: float | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + error = OutboundRequestBlockedError( + host=path, + port=None, + reason=BlockReason.UNIX_SOCKET, + ) + _log_block(error) + raise error + + async def sleep(self, seconds: float) -> None: + await self._inner.sleep(seconds) + + def resolve_hostname_ips(hostname: str) -> list[str]: try: addr_info = socket.getaddrinfo(hostname, None) diff --git a/src/paperless/tests/test_network.py b/src/paperless/tests/test_network.py index 23d9fb3ba..3bfa82233 100644 --- a/src/paperless/tests/test_network.py +++ b/src/paperless/tests/test_network.py @@ -1,19 +1,26 @@ import ipaddress +import logging import pickle import socket +from collections.abc import Iterable from typing import Any from unittest import mock from unittest.mock import MagicMock +import anyio +import httpcore import httpx import pytest from celery.utils.serialization import get_pickleable_exception from pytest_mock import MockerFixture +from paperless.network import MAX_ADDRESSES_TRIED from paperless.network import BlockReason from paperless.network import HostResolutionError from paperless.network import OutboundRequestBlockedError from paperless.network import PinnedHostHTTPTransport +from paperless.network import _GuardedAsyncBackend +from paperless.network import _GuardedSyncBackend from paperless.network import aresolve_public_addresses from paperless.network import blocked_message from paperless.network import is_public_ip @@ -639,3 +646,644 @@ class TestValidateOutboundHttpUrl: r"http://127.0.0.1\@evil.example/", allow_internal=True, ) + + +class FakeClock: + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +@pytest.fixture +def clock(mocker: MockerFixture) -> FakeClock: + fake = FakeClock() + mocker.patch("paperless.network._monotonic", new=fake) + return fake + + +class ScriptedBackend(httpcore.NetworkBackend): + """ + Inner backend double. Each host either connects (default), refuses + immediately, or black-holes (uses its whole timeout, then times out). + """ + + def __init__(self, clock: FakeClock, outcomes: dict[str, str]) -> None: + self.clock = clock + self.outcomes = outcomes + self.calls: list[tuple[str, float | None]] = [] + self.slept: list[float] = [] + + def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.NetworkStream: + self.calls.append((host, timeout)) + outcome = self.outcomes.get(host, "connect") + if outcome == "refuse": + raise httpcore.ConnectError(f"refused {host}") + if outcome == "blackhole": + self.clock.advance(timeout or 0.0) + raise httpcore.ConnectTimeout(f"timed out {host}") + return httpcore.MockStream([]) + + def sleep(self, seconds: float) -> None: + self.slept.append(seconds) + + +class AsyncScriptedBackend(httpcore.AsyncNetworkBackend): + """Async twin of ScriptedBackend.""" + + def __init__(self, clock: FakeClock, outcomes: dict[str, str]) -> None: + self.clock = clock + self.outcomes = outcomes + self.calls: list[tuple[str, float | None]] = [] + self.slept: list[float] = [] + + async def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + self.calls.append((host, timeout)) + outcome = self.outcomes.get(host, "connect") + if outcome == "refuse": + raise httpcore.ConnectError(f"refused {host}") + if outcome == "blackhole": + self.clock.advance(timeout or 0.0) + raise httpcore.ConnectTimeout(f"timed out {host}") + return httpcore.AsyncMockStream([]) + + async def sleep(self, seconds: float) -> None: + self.slept.append(seconds) + + +def _public_ipv4(count: int) -> list[str]: + return [f"93.184.216.{index + 1}" for index in range(count)] + + +_ATTEMPT_BUDGETS = [ + pytest.param(5.0, 2, [2.5, 2.5], id="5s-2-addresses"), + pytest.param(5.0, 3, [2.0, 3.0], id="5s-3-addresses"), + pytest.param(5.0, 8, [2.0, 3.0], id="5s-8-addresses"), + pytest.param(1.0, 2, [1.0], id="1s-2-addresses"), + pytest.param(120.0, 8, [10.0] * 7 + [50.0], id="120s-8-addresses"), + pytest.param(None, 3, [10.0, 10.0, None], id="no-timeout-3-addresses"), +] + + +class TestGuardedSyncBackend: + def test_dials_validated_literals_in_order( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A hostname whose first address refuses and second accepts + WHEN: + - The guard connects + THEN: + - It dials the IP literals in turn, never the hostname + """ + _answer(mocker, "93.184.216.1", "93.184.216.2") + inner = ScriptedBackend(clock, {"93.184.216.1": "refuse"}) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert [host for host, _ in inner.calls] == ["93.184.216.1", "93.184.216.2"] + + @pytest.mark.parametrize( + ("answers", "expected"), + [ + pytest.param( + ["2606:4700::1", "2606:4700::2", "93.184.216.1", "93.184.216.2"], + ["2606:4700::1", "93.184.216.1", "2606:4700::2", "93.184.216.2"], + id="balanced", + ), + pytest.param( + ["2606:4700::1", "2606:4700::2", "2606:4700::3", "93.184.216.1"], + ["2606:4700::1", "93.184.216.1", "2606:4700::2", "2606:4700::3"], + id="one-family-runs-out", + ), + pytest.param( + ["93.184.216.1", "93.184.216.2"], + ["93.184.216.1", "93.184.216.2"], + id="single-family", + ), + ], + ) + def test_interleaves_address_families( + self, + mocker: MockerFixture, + clock: FakeClock, + answers: list[str], + expected: list[str], + ) -> None: + """ + GIVEN: + - Resolver results grouped by address family + WHEN: + - Every address refuses + THEN: + - Families are alternated, starting with the first result's family + """ + _answer(mocker, *answers) + inner = ScriptedBackend(clock, dict.fromkeys(answers, "refuse")) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectError): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert [host for host, _ in inner.calls] == expected + + @pytest.mark.parametrize(("timeout", "count", "expected"), _ATTEMPT_BUDGETS) + def test_attempt_budgets( + self, + mocker: MockerFixture, + clock: FakeClock, + timeout: float | None, + count: int, + expected: list[float | None], + ) -> None: + """ + GIVEN: + - Every resolved address black-holed + WHEN: + - The guard connects with a given timeout + THEN: + - Each attempt gets the budget the connect-time policy prescribes + """ + addresses = _public_ipv4(count) + _answer(mocker, *addresses) + inner = ScriptedBackend(clock, dict.fromkeys(addresses, "blackhole")) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectTimeout): + guard.connect_tcp("example.com", 443, timeout=timeout) + + assert [budget for _, budget in inner.calls] == expected + + def test_quick_refusal_leaves_budget_for_next_address( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A 1 second budget, a refusing first address and a working second + WHEN: + - The guard connects + THEN: + - Both addresses are tried and the connection succeeds + """ + _answer(mocker, "93.184.216.1", "93.184.216.2") + inner = ScriptedBackend(clock, {"93.184.216.1": "refuse"}) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + guard.connect_tcp("example.com", 443, timeout=1.0) + + assert inner.calls == [("93.184.216.1", 1.0), ("93.184.216.2", 1.0)] + + def test_resolution_time_is_not_charged( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A resolver that takes 4 seconds against a 5 second timeout + WHEN: + - The guard connects to two black-holed addresses + THEN: + - The attempts share the full 5 seconds + """ + infos = _addrinfo("93.184.216.1", "93.184.216.2") + + def slow_resolver(*_args: object, **_kwargs: object) -> list[tuple[Any, ...]]: + clock.advance(4.0) + return infos + + mocker.patch("paperless.network._getaddrinfo", side_effect=slow_resolver) + inner = ScriptedBackend( + clock, + {"93.184.216.1": "blackhole", "93.184.216.2": "blackhole"}, + ) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectTimeout): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert [budget for _, budget in inner.calls] == [2.5, 2.5] + + @pytest.mark.parametrize( + "timeout", + [pytest.param(0.0, id="zero"), pytest.param(-1.0, id="negative")], + ) + def test_non_positive_timeout( + self, + mocker: MockerFixture, + clock: FakeClock, + timeout: float, + ) -> None: + """ + GIVEN: + - A zero or negative connect timeout + WHEN: + - The guard connects + THEN: + - It times out without resolving or dialling + """ + resolver = _answer(mocker, "93.184.216.1") + inner = ScriptedBackend(clock, {}) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectTimeout): + guard.connect_tcp("example.com", 443, timeout=timeout) + + resolver.assert_not_called() + assert inner.calls == [] + + def test_caps_attempts(self, mocker: MockerFixture, clock: FakeClock) -> None: + """ + GIVEN: + - More public addresses than the attempt cap, all refusing + WHEN: + - The guard connects + THEN: + - Only the capped number of addresses is dialled + """ + addresses = _public_ipv4(MAX_ADDRESSES_TRIED + 2) + _answer(mocker, *addresses) + inner = ScriptedBackend(clock, dict.fromkeys(addresses, "refuse")) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectError): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert len(inner.calls) == MAX_ADDRESSES_TRIED + + def test_private_address_beyond_cap_still_blocks( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A private address after more public addresses than the cap + WHEN: + - The guard connects + THEN: + - The name is blocked and nothing is dialled + """ + _answer(mocker, *_public_ipv4(MAX_ADDRESSES_TRIED), "10.0.0.1") + inner = ScriptedBackend(clock, {}) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + with pytest.raises(OutboundRequestBlockedError): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert inner.calls == [] + + def test_block_is_logged( + self, + mocker: MockerFixture, + clock: FakeClock, + caplog: pytest.LogCaptureFixture, + ) -> None: + """ + GIVEN: + - A hostname resolving to a private address + WHEN: + - The guard connects + THEN: + - A warning names the destination and the reason + - The resolved internal address is not logged + """ + _answer(mocker, "10.0.0.1") + guard = _GuardedSyncBackend(ScriptedBackend(clock, {}), allow_internal=False) + + with ( + caplog.at_level(logging.DEBUG, logger="paperless.network"), + pytest.raises(OutboundRequestBlockedError), + ): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert "example.com:443" in caplog.text + assert "non_public_address" in caplog.text + assert "10.0.0.1" not in caplog.text + + def test_resolves_again_for_every_connection( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A hostname that resolves to a public address, then to a private + one (DNS rebinding) + WHEN: + - The guard opens two connections to it + THEN: + - The first connects to the public address + - The second is blocked and nothing more is dialled + """ + mocker.patch( + "paperless.network._getaddrinfo", + side_effect=[_addrinfo("93.184.216.34"), _addrinfo("10.0.0.1")], + ) + inner = ScriptedBackend(clock, {}) + guard = _GuardedSyncBackend(inner, allow_internal=False) + + guard.connect_tcp("example.com", 443, timeout=5.0) + with pytest.raises(OutboundRequestBlockedError): + guard.connect_tcp("example.com", 443, timeout=5.0) + + assert [host for host, _ in inner.calls] == ["93.184.216.34"] + + def test_resolution_failure_is_a_connect_error( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A resolver that fails + WHEN: + - The guard connects + THEN: + - A connect error is raised, not a policy block + """ + mocker.patch( + "paperless.network._getaddrinfo", + side_effect=socket.gaierror(-2, "Name or service not known"), + ) + guard = _GuardedSyncBackend(ScriptedBackend(clock, {}), allow_internal=False) + + with pytest.raises( + httpcore.ConnectError, + match=r"Could not resolve example\.com", + ): + guard.connect_tcp("example.com", 443, timeout=5.0) + + def test_allow_internal_passes_hostname_through( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - Internal addresses allowed + WHEN: + - The guard connects + THEN: + - The inner backend gets the hostname and no resolution happens + """ + resolver = _answer(mocker, "10.0.0.1") + inner = ScriptedBackend(clock, {}) + guard = _GuardedSyncBackend(inner, allow_internal=True) + + guard.connect_tcp("localhost", 8080, timeout=5.0) + + assert inner.calls == [("localhost", 5.0)] + resolver.assert_not_called() + + def test_unix_sockets_are_refused(self, clock: FakeClock) -> None: + """ + GIVEN: + - A guard + WHEN: + - A unix socket connection is requested + THEN: + - It is blocked with the unix socket reason + """ + guard = _GuardedSyncBackend(ScriptedBackend(clock, {}), allow_internal=True) + + with pytest.raises(OutboundRequestBlockedError) as exc_info: + guard.connect_unix_socket("/run/app.sock") + + assert exc_info.value.reason is BlockReason.UNIX_SOCKET + assert exc_info.value.host == "/run/app.sock" + + def test_sleep_delegates(self, clock: FakeClock) -> None: + """ + GIVEN: + - A guard + WHEN: + - httpcore asks it to sleep between retries + THEN: + - The inner backend sleeps + """ + inner = ScriptedBackend(clock, {}) + + _GuardedSyncBackend(inner, allow_internal=False).sleep(0.5) + + assert inner.slept == [0.5] + + +class TestGuardedAsyncBackend: + @pytest.fixture(autouse=True) + def anyio_backend(self) -> str: + return "asyncio" + + @pytest.mark.anyio + async def test_falls_back_to_next_address( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A refusing first address and a working second + WHEN: + - The async guard connects + THEN: + - Both literals are dialled in order + """ + _answer(mocker, "2606:4700::1", "93.184.216.1") + inner = AsyncScriptedBackend(clock, {"2606:4700::1": "refuse"}) + guard = _GuardedAsyncBackend(inner, allow_internal=False) + + await guard.connect_tcp("example.com", 443, timeout=5.0) + + assert [host for host, _ in inner.calls] == ["2606:4700::1", "93.184.216.1"] + + @pytest.mark.anyio + @pytest.mark.parametrize(("timeout", "count", "expected"), _ATTEMPT_BUDGETS) + async def test_attempt_budgets( + self, + mocker: MockerFixture, + clock: FakeClock, + timeout: float | None, + count: int, + expected: list[float | None], + ) -> None: + """ + GIVEN: + - Every resolved address black-holed + WHEN: + - The async guard connects with a given timeout + THEN: + - Each attempt gets the prescribed budget + """ + addresses = _public_ipv4(count) + _answer(mocker, *addresses) + inner = AsyncScriptedBackend(clock, dict.fromkeys(addresses, "blackhole")) + guard = _GuardedAsyncBackend(inner, allow_internal=False) + + with pytest.raises(httpcore.ConnectTimeout): + await guard.connect_tcp("example.com", 443, timeout=timeout) + + assert [budget for _, budget in inner.calls] == expected + + @pytest.mark.anyio + async def test_blocks_non_public( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - A hostname resolving to a private address + WHEN: + - The async guard connects + THEN: + - It is blocked and nothing is dialled + """ + _answer(mocker, "10.0.0.1") + inner = AsyncScriptedBackend(clock, {}) + guard = _GuardedAsyncBackend(inner, allow_internal=False) + + with pytest.raises(OutboundRequestBlockedError): + await guard.connect_tcp("example.com", 443, timeout=5.0) + + assert inner.calls == [] + + @pytest.mark.anyio + async def test_slow_resolution_times_out(self, mocker: MockerFixture) -> None: + """ + GIVEN: + - A resolver slower than the connect timeout + WHEN: + - The async guard connects + THEN: + - It raises a connect timeout (real time: fail_after follows the + event loop clock) + """ + + async def slow_resolver( + *_args: object, + **_kwargs: object, + ) -> list[tuple[Any, ...]]: + await anyio.sleep(0.2) + return _addrinfo("93.184.216.1") + + mocker.patch("paperless.network._agetaddrinfo", new=slow_resolver) + guard = _GuardedAsyncBackend( + AsyncScriptedBackend(FakeClock(), {}), + allow_internal=False, + ) + + with pytest.raises(httpcore.ConnectTimeout, match="Timed out resolving"): + await guard.connect_tcp("example.com", 443, timeout=0.05) + + @pytest.mark.anyio + async def test_resolution_failure_is_a_connect_error( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - An async resolver that fails + WHEN: + - The async guard connects + THEN: + - A connect error is raised, not a policy block + """ + mocker.patch( + "paperless.network._agetaddrinfo", + new=mocker.AsyncMock( + side_effect=socket.gaierror(-2, "Name or service not known"), + ), + ) + guard = _GuardedAsyncBackend( + AsyncScriptedBackend(clock, {}), + allow_internal=False, + ) + + with pytest.raises( + httpcore.ConnectError, + match=r"Could not resolve example\.com", + ): + await guard.connect_tcp("example.com", 443, timeout=5.0) + + @pytest.mark.anyio + async def test_allow_internal_passes_hostname_through( + self, + mocker: MockerFixture, + clock: FakeClock, + ) -> None: + """ + GIVEN: + - Internal addresses allowed + WHEN: + - The async guard connects + THEN: + - The inner backend gets the hostname and no resolution happens + """ + _answer(mocker, "10.0.0.1") + inner = AsyncScriptedBackend(clock, {}) + guard = _GuardedAsyncBackend(inner, allow_internal=True) + + await guard.connect_tcp("localhost", 8080, timeout=5.0) + + assert inner.calls == [("localhost", 5.0)] + + @pytest.mark.anyio + async def test_unix_sockets_are_refused(self, clock: FakeClock) -> None: + """ + GIVEN: + - An async guard + WHEN: + - A unix socket connection is requested + THEN: + - It is blocked with the unix socket reason + """ + guard = _GuardedAsyncBackend( + AsyncScriptedBackend(clock, {}), + allow_internal=True, + ) + + with pytest.raises(OutboundRequestBlockedError) as exc_info: + await guard.connect_unix_socket("/run/app.sock") + + assert exc_info.value.reason is BlockReason.UNIX_SOCKET + + @pytest.mark.anyio + async def test_sleep_delegates(self, clock: FakeClock) -> None: + """ + GIVEN: + - An async guard + WHEN: + - httpcore asks it to sleep between retries + THEN: + - The inner backend sleeps + """ + inner = AsyncScriptedBackend(clock, {}) + + await _GuardedAsyncBackend(inner, allow_internal=False).sleep(0.5) + + assert inner.slept == [0.5]