Add real-socket tests for the outbound connection guard

A local HTTP server, a per-hostname resolver fake and dial spies exercise
the guarded transports end to end: address fallback, blocking before any
connection, the Host header, TLS server name, per-host connection pooling,
numeric host spellings, environment proxies and redirects to blocked hosts.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/conftest.py

@@ -17,10 +17,14 @@ if TYPE_CHECKING:
 
     from django.contrib.auth.models import User
     from pytest_django.fixtures import Settings
+    from pytest_mock import MockerFixture
     from rest_framework.test import APIClient
 
     from paperless_testing.dirs import PaperlessDirs
     from paperless_testing.fakes.progress import FakeProgressManager
+    from paperless_testing.outbound import DialRecorder
+    from paperless_testing.outbound import FakeDNS
+    from paperless_testing.outbound import LocalHTTPServer
 
 
 @pytest.fixture(scope="session", autouse=True)
@@ -149,3 +153,28 @@ def fake_progress_manager(
 
     monkeypatch.setattr("documents.tasks.ProgressManager", FakeProgressManager)
     return FakeProgressManager
+
+
+@pytest.fixture
+def local_http_server() -> Generator[LocalHTTPServer, None, None]:
+    """A recording HTTP server on 127.0.0.1, for outbound connection tests."""
+    from paperless_testing.outbound import running_http_server
+
+    with running_http_server() as server:
+        yield server
+
+
+@pytest.fixture
+def fake_dns(mocker: MockerFixture) -> FakeDNS:
+    """Per-hostname answers for the outbound guard's resolver hooks."""
+    from paperless_testing.outbound import install_fake_dns
+
+    return install_fake_dns(mocker)
+
+
+@pytest.fixture
+def dial_recorder(mocker: MockerFixture) -> DialRecorder:
+    """Records which addresses the outbound guard actually dialled."""
+    from paperless_testing.outbound import install_dial_recorder
+
+    return install_dial_recorder(mocker)

src/paperless/tests/test_network_integration.py

@@ -0,0 +1,382 @@
+import ipaddress
+import os
+
+import httpcore
+import httpx
+import pytest
+from pytest_mock import MockerFixture
+
+from paperless.network import GuardedAsyncHTTPTransport
+from paperless.network import GuardedHTTPTransport
+from paperless.network import OutboundRequestBlockedError
+from paperless_testing.outbound import DialRecorder
+from paperless_testing.outbound import FakeDNS
+from paperless_testing.outbound import LocalHTTPServer
+
+
+class TestGuardedTransportSync:
+    def test_pinned_connection_falls_back_to_next_address(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - A hostname resolving to ::1 then 127.0.0.1
+            - A server listening on 127.0.0.1 only
+            - Internal addresses disallowed, with loopback treated as public
+        WHEN:
+            - A request is made
+        THEN:
+            - ::1 fails, 127.0.0.1 is dialled next and the request succeeds
+        """
+        fake_dns.add("dual-stack.test", "::1", "127.0.0.1")
+        mocker.patch("paperless.network.is_public_ip", return_value=True)
+
+        with httpx.Client(
+            transport=GuardedHTTPTransport(allow_internal=False),
+            timeout=5.0,
+        ) as client:
+            response = client.get(f"http://dual-stack.test:{local_http_server.port}/")
+
+        assert response.status_code == 200
+        assert dial_recorder.hosts() == ["::1", "127.0.0.1"]
+
+    def test_allow_internal_uses_stock_resolution(
+        self,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses allowed
+        WHEN:
+            - A request is made to localhost
+        THEN:
+            - It succeeds without the guard resolving anything
+        """
+        with httpx.Client(
+            transport=GuardedHTTPTransport(allow_internal=True),
+            timeout=5.0,
+        ) as client:
+            response = client.get(f"http://localhost:{local_http_server.port}/")
+
+        assert response.status_code == 200
+        assert fake_dns.lookups == []
+
+    def test_blocks_internal_host_without_connecting(
+        self,
+        local_http_server: LocalHTTPServer,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses disallowed
+        WHEN:
+            - A request is made to localhost through the transport
+        THEN:
+            - It is blocked and the server never sees a connection
+        """
+        with (
+            httpx.Client(
+                transport=GuardedHTTPTransport(allow_internal=False),
+                timeout=5.0,
+            ) as client,
+            pytest.raises(OutboundRequestBlockedError),
+        ):
+            client.get(f"http://localhost:{local_http_server.port}/")
+
+        assert local_http_server.connections == 0
+
+    def test_host_header_is_the_hostname(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+    ) -> None:
+        """
+        GIVEN:
+            - A pinned connection to a named host
+        WHEN:
+            - A request is made
+        THEN:
+            - The server receives the hostname in Host, not the dialled IP
+        """
+        fake_dns.add("pinned.test", "127.0.0.1")
+        mocker.patch("paperless.network.is_public_ip", return_value=True)
+
+        with httpx.Client(
+            transport=GuardedHTTPTransport(allow_internal=False),
+            timeout=5.0,
+        ) as client:
+            client.get(f"http://pinned.test:{local_http_server.port}/")
+
+        assert local_http_server.requests[0].headers["host"] == (
+            f"pinned.test:{local_http_server.port}"
+        )
+
+    def test_redirect_to_internal_host_is_blocked(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - An allowed origin that redirects to a host resolving to a blocked
+              address, and a client that follows redirects
+        WHEN:
+            - The origin is requested
+        THEN:
+            - The redirect hop is blocked without dialling the blocked address
+        """
+        allowed = ipaddress.ip_address("127.0.0.1")
+        mocker.patch(
+            "paperless.network.is_public_ip",
+            side_effect=lambda address: address == allowed,
+        )
+        fake_dns.add("origin.test", "127.0.0.1")
+        fake_dns.add("internal.test", "127.0.0.2")
+        local_http_server.redirect_to = (
+            f"http://internal.test:{local_http_server.port}/"
+        )
+
+        with (
+            httpx.Client(
+                transport=GuardedHTTPTransport(allow_internal=False),
+                timeout=5.0,
+                follow_redirects=True,
+            ) as client,
+            pytest.raises(OutboundRequestBlockedError) as exc_info,
+        ):
+            client.get(f"http://origin.test:{local_http_server.port}/")
+
+        assert exc_info.value.address == ipaddress.ip_address("127.0.0.2")
+        assert dial_recorder.hosts() == ["127.0.0.1"]
+        assert len(local_http_server.requests) == 1
+
+    def test_connections_are_not_shared_between_hosts_on_one_address(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - Two hostnames resolving to the same address
+            - Internal addresses disallowed, with loopback treated as public
+        WHEN:
+            - One client requests the first host twice, then the second host
+        THEN:
+            - The first host's connection is reused for its second request
+            - The second host gets its own connection, so its certificate would
+              be checked rather than inheriting the first host's session
+        """
+        fake_dns.add("first.test", "127.0.0.1")
+        fake_dns.add("second.test", "127.0.0.1")
+        mocker.patch("paperless.network.is_public_ip", return_value=True)
+
+        with httpx.Client(
+            transport=GuardedHTTPTransport(allow_internal=False),
+            timeout=5.0,
+        ) as client:
+            client.get(f"http://first.test:{local_http_server.port}/")
+            client.get(f"http://first.test:{local_http_server.port}/")
+            client.get(f"http://second.test:{local_http_server.port}/")
+
+        assert dial_recorder.hosts() == ["127.0.0.1", "127.0.0.1"]
+        assert local_http_server.connections == 2
+        assert [request.headers["host"] for request in local_http_server.requests] == [
+            f"first.test:{local_http_server.port}",
+            f"first.test:{local_http_server.port}",
+            f"second.test:{local_http_server.port}",
+        ]
+
+    def test_tls_uses_the_hostname_not_the_dialled_address(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - A pinned HTTPS connection to a named host
+            - A plain HTTP server, so the handshake itself fails
+        WHEN:
+            - A request is made
+        THEN:
+            - The validated address is dialled
+            - TLS is started with the hostname for SNI and certificate checks
+        """
+        fake_dns.add("pinned.test", "127.0.0.1")
+        mocker.patch("paperless.network.is_public_ip", return_value=True)
+        start_tls = mocker.spy(httpcore._backends.sync.SyncStream, "start_tls")
+
+        with (
+            httpx.Client(
+                transport=GuardedHTTPTransport(allow_internal=False),
+                timeout=5.0,
+            ) as client,
+            pytest.raises(httpx.ConnectError),
+        ):
+            client.get(f"https://pinned.test:{local_http_server.port}/")
+
+        assert dial_recorder.hosts() == ["127.0.0.1"]
+        start_tls.assert_called_once()
+        assert start_tls.call_args.kwargs["server_hostname"] == "pinned.test"
+
+    @pytest.mark.parametrize(
+        "host",
+        [
+            pytest.param("2130706433", id="decimal"),
+            pytest.param("0x7f.1", id="hex-short"),
+            pytest.param("127.1", id="short-dotted"),
+        ],
+    )
+    def test_numeric_host_forms_are_blocked(
+        self,
+        local_http_server: LocalHTTPServer,
+        host: str,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses disallowed
+            - A URL whose host is a non-canonical spelling of 127.0.0.1
+        WHEN:
+            - A request is made through the transport
+        THEN:
+            - The resolved address is checked, the request is blocked and the
+              server never sees a connection
+        """
+        with (
+            httpx.Client(
+                transport=GuardedHTTPTransport(allow_internal=False),
+                timeout=5.0,
+            ) as client,
+            pytest.raises(OutboundRequestBlockedError),
+        ):
+            client.get(f"http://{host}:{local_http_server.port}/")
+
+        assert local_http_server.connections == 0
+
+    def test_environment_proxy_is_not_used(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+    ) -> None:
+        """
+        GIVEN:
+            - Proxy variables in the environment pointing at an unreachable proxy
+            - Internal addresses disallowed
+        WHEN:
+            - A request is made to localhost
+        THEN:
+            - The guard blocks it, rather than the request going to the proxy
+        """
+        unreachable = "http://127.0.0.1:9"
+        mocker.patch.dict(
+            os.environ,
+            {
+                "HTTP_PROXY": unreachable,
+                "HTTPS_PROXY": unreachable,
+                "ALL_PROXY": unreachable,
+            },
+        )
+
+        with (
+            httpx.Client(
+                transport=GuardedHTTPTransport(allow_internal=False),
+                timeout=5.0,
+            ) as client,
+            pytest.raises(OutboundRequestBlockedError),
+        ):
+            client.get(f"http://localhost:{local_http_server.port}/")
+
+        assert local_http_server.connections == 0
+
+
+class TestGuardedTransportAsync:
+    @pytest.fixture(autouse=True)
+    def anyio_backend(self) -> str:
+        return "asyncio"
+
+    @pytest.mark.anyio
+    async def test_pinned_connection_falls_back_to_next_address(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - A hostname resolving to ::1 then 127.0.0.1
+            - A server listening on 127.0.0.1 only
+            - Internal addresses disallowed, with loopback treated as public
+        WHEN:
+            - An async request is made
+        THEN:
+            - ::1 fails, 127.0.0.1 is dialled next and the request succeeds
+        """
+        fake_dns.add("dual-stack.test", "::1", "127.0.0.1")
+        mocker.patch("paperless.network.is_public_ip", return_value=True)
+
+        async with httpx.AsyncClient(
+            transport=GuardedAsyncHTTPTransport(allow_internal=False),
+            timeout=5.0,
+        ) as client:
+            response = await client.get(
+                f"http://dual-stack.test:{local_http_server.port}/",
+            )
+
+        assert response.status_code == 200
+        assert dial_recorder.hosts() == ["::1", "127.0.0.1"]
+
+    @pytest.mark.anyio
+    async def test_allow_internal_uses_stock_resolution(
+        self,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses allowed
+        WHEN:
+            - An async request is made to localhost
+        THEN:
+            - It succeeds without the guard resolving anything
+        """
+        async with httpx.AsyncClient(
+            transport=GuardedAsyncHTTPTransport(allow_internal=True),
+            timeout=5.0,
+        ) as client:
+            response = await client.get(f"http://localhost:{local_http_server.port}/")
+
+        assert response.status_code == 200
+        assert fake_dns.lookups == []
+
+    @pytest.mark.anyio
+    async def test_blocks_internal_host_without_connecting(
+        self,
+        local_http_server: LocalHTTPServer,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses disallowed
+        WHEN:
+            - An async request is made to localhost through the transport
+        THEN:
+            - It is blocked and the server never sees a connection
+        """
+        async with httpx.AsyncClient(
+            transport=GuardedAsyncHTTPTransport(allow_internal=False),
+            timeout=5.0,
+        ) as client:
+            with pytest.raises(OutboundRequestBlockedError):
+                await client.get(f"http://localhost:{local_http_server.port}/")
+
+        assert local_http_server.connections == 0

src/paperless_testing/outbound.py

@@ -0,0 +1,209 @@
+"""
+Real-socket helpers for tests of the outbound connection guard in
+paperless.network: a local HTTP server, a per-hostname resolver fake and
+spies recording which addresses were actually dialled.
+
+The fixtures wrapping these live in the root conftest.
+"""
+
+from __future__ import annotations
+
+import http.server
+import socket
+import threading
+from contextlib import contextmanager
+from dataclasses import dataclass
+from dataclasses import field
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import cast
+
+import anyio
+import httpcore
+
+from paperless.network import GuardedAsyncHTTPTransport
+from paperless.network import GuardedHTTPTransport
+from paperless.network import _GuardedAsyncBackend
+from paperless.network import _GuardedSyncBackend
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from unittest.mock import MagicMock
+    from unittest.mock import _Call
+
+    import httpx
+    from pytest_mock import MockerFixture
+
+_REAL_GETADDRINFO = socket.getaddrinfo
+_REAL_AGETADDRINFO = anyio.getaddrinfo
+
+
+@dataclass
+class ReceivedRequest:
+    method: str
+    path: str
+    headers: dict[str, str]
+    body: bytes
+
+
+@dataclass
+class LocalHTTPServer:
+    """State of a threaded HTTP server bound to 127.0.0.1 on an ephemeral port."""
+
+    port: int
+    requests: list[ReceivedRequest] = field(default_factory=list)
+    connections: int = 0
+    redirect_to: str | None = None
+
+
+class _Handler(http.server.BaseHTTPRequestHandler):
+    # HTTP/1.1 keeps connections open, so tests can observe connection reuse.
+    # Every response sets Content-Length, which keep-alive requires.
+    protocol_version = "HTTP/1.1"
+
+    def _handle(self) -> None:
+        length = int(self.headers.get("Content-Length") or 0)
+        body = self.rfile.read(length) if length else b""
+        # BaseHTTPRequestHandler types server as the base socketserver.BaseServer;
+        # narrowing the attribute's declared type is a variance error, so the
+        # subclass is recovered here instead of on the class body.
+        server = cast("_RecordingHTTPServer", self.server)
+        state = server.state
+        state.requests.append(
+            ReceivedRequest(
+                method=self.command,
+                path=self.path,
+                headers={key.lower(): value for key, value in self.headers.items()},
+                body=body,
+            ),
+        )
+        if state.redirect_to is not None:
+            self.send_response(302)
+            self.send_header("Location", state.redirect_to)
+            self.send_header("Content-Length", "0")
+            self.end_headers()
+            return
+        self.send_response(200)
+        self.send_header("Content-Length", "2")
+        self.end_headers()
+        self.wfile.write(b"ok")
+
+    do_GET = _handle
+    do_POST = _handle
+
+    def log_message(self, format: str, *args: Any) -> None:
+        return None
+
+
+class _RecordingHTTPServer(http.server.ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self) -> None:
+        super().__init__(("127.0.0.1", 0), _Handler)
+        self.state = LocalHTTPServer(port=self.socket.getsockname()[1])
+
+    def verify_request(self, request: Any, client_address: Any) -> bool:
+        self.state.connections += 1
+        return True
+
+
+@contextmanager
+def running_http_server() -> Iterator[LocalHTTPServer]:
+    """Serve on 127.0.0.1 in a background thread until the block exits."""
+    server = _RecordingHTTPServer()
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield server.state
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def _addrinfo(address: str, port: int | None) -> tuple[Any, ...]:
+    if ":" in address:
+        return (socket.AF_INET6, socket.SOCK_STREAM, 6, "", (address, port or 0, 0, 0))
+    return (socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, port or 0))
+
+
+class FakeDNS:
+    """
+    Answers the guard's resolver hooks for registered names and delegates
+    every other name to the real resolver. The stock httpcore backends keep
+    using the unpatched socket.getaddrinfo.
+    """
+
+    def __init__(self) -> None:
+        self._answers: dict[str, list[str]] = {}
+        self.lookups: list[str] = []
+
+    def add(self, hostname: str, *addresses: str) -> None:
+        self._answers[hostname] = list(addresses)
+
+    def getaddrinfo(
+        self,
+        host: str,
+        port: int | None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> list[tuple[Any, ...]]:
+        self.lookups.append(host)
+        if host in self._answers:
+            return [_addrinfo(address, port) for address in self._answers[host]]
+        return list(_REAL_GETADDRINFO(host, port, *args, **kwargs))
+
+    async def agetaddrinfo(
+        self,
+        host: str,
+        port: int | None,
+        **kwargs: Any,
+    ) -> list[tuple[Any, ...]]:
+        self.lookups.append(host)
+        if host in self._answers:
+            return [_addrinfo(address, port) for address in self._answers[host]]
+        return list(await _REAL_AGETADDRINFO(host, port, **kwargs))
+
+
+def install_fake_dns(mocker: MockerFixture) -> FakeDNS:
+    """Patch the guard's resolver hooks with a FakeDNS for the current test."""
+    dns = FakeDNS()
+    mocker.patch("paperless.network._getaddrinfo", new=dns.getaddrinfo)
+    mocker.patch("paperless.network._agetaddrinfo", new=dns.agetaddrinfo)
+    return dns
+
+
+def _dialled_host(call: _Call) -> str:
+    # The spy sits on the class, so args[0] is the backend instance.
+    if "host" in call.kwargs:
+        return str(call.kwargs["host"])
+    return str(call.args[1])
+
+
+@dataclass
+class DialRecorder:
+    sync_spy: MagicMock
+    async_spy: MagicMock
+
+    def hosts(self) -> list[str]:
+        calls = [*self.sync_spy.call_args_list, *self.async_spy.call_args_list]
+        return [_dialled_host(call) for call in calls]
+
+
+def install_dial_recorder(mocker: MockerFixture) -> DialRecorder:
+    """Spy on the stock backends' connect_tcp for the current test."""
+    return DialRecorder(
+        sync_spy=mocker.spy(httpcore.SyncBackend, "connect_tcp"),
+        async_spy=mocker.spy(httpcore.AnyIOBackend, "connect_tcp"),
+    )
+
+
+def guard_of(
+    client: httpx.Client | httpx.AsyncClient,
+) -> _GuardedSyncBackend | _GuardedAsyncBackend:
+    """Return the guard installed on a client's transport."""
+    transport = client._transport
+    assert isinstance(transport, GuardedHTTPTransport | GuardedAsyncHTTPTransport)
+    backend = transport._pool._network_backend
+    assert isinstance(backend, _GuardedSyncBackend | _GuardedAsyncBackend)
+    return backend