Collapse near-duplicate outbound guard tests into parametrized cases

Several tests in test_network.py and test_network_integration.py had
grown into near-copies of each other, differing only in the URL or
host they fed the guard while asserting the same thing. That made the
suite noisier to read and to extend than the behaviour it covers
warrants.

Merge the invalid-host rejection tests into one parametrized test over
all seven unusual host forms, and likewise for their allow_internal
counterparts that assert nothing is rejected. Merge the
resolve_public_addresses tests that only check the host reaches the
resolver unchanged, and the ones that check a private answer blocks
regardless of how the host was spelled, keeping the stricter resolver
call assertion on both merged cases. Add localhost as a fourth case to
the numeric-host-forms test in the transport integration tests, so the
name and the non-canonical numeric spellings of loopback are covered
by one test.

No behaviour or assertion strength changes; this only reduces
duplicated test bodies while keeping every original case addressable
by its own parametrize id.

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

docs/configuration.md

@@ -1576,6 +1576,9 @@ ports.
 #### [`PAPERLESS_WEBHOOKS_ALLOW_INTERNAL_REQUESTS=<bool>`](#PAPERLESS_WEBHOOKS_ALLOW_INTERNAL_REQUESTS) {#PAPERLESS_WEBHOOKS_ALLOW_INTERNAL_REQUESTS}
 
 : If set to false, webhooks cannot be sent to internal URLs (e.g., localhost).
+A hostname is blocked if any of the addresses it resolves to is non-public.
+Webhook requests connect directly, without using the `HTTP_PROXY` or
+`HTTPS_PROXY` environment variables, and never follow redirects.
 
     Defaults to true, which allows internal requests.
 
@@ -1584,7 +1587,7 @@ ports.
 #### [`PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS=<bool>`](#PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS) {#PAPERLESS_EMAIL_ALLOW_INTERNAL_HOSTS}
 
 : If set to false, incoming mail account connections are blocked when the
-configured IMAP hostname resolves to a non-public address (for example,
+configured IMAP hostname resolves to any non-public address (for example,
 localhost, link-local, or RFC1918 private ranges).
 
     Defaults to true, which allows internal hosts.
@@ -2214,6 +2217,8 @@ used with the OpenAI-compatible backend to target a custom provider or local gat
 #### [`PAPERLESS_AI_LLM_ALLOW_INTERNAL_ENDPOINTS=<bool>`](#PAPERLESS_AI_LLM_ALLOW_INTERNAL_ENDPOINTS) {#PAPERLESS_AI_LLM_ALLOW_INTERNAL_ENDPOINTS}
 
 : If set to false, Paperless blocks AI endpoint URLs that resolve to non-public addresses (e.g., localhost, etc).
+A hostname is blocked if any of the addresses it resolves to is non-public, and redirects are checked the same way.
+Requests to a configured AI endpoint connect directly, without using the `HTTP_PROXY` or `HTTPS_PROXY` environment variables.
 
     Defaults to true, which allows internal endpoints.
 

docs/usage.md

@@ -613,7 +613,7 @@ The following workflow action types are available:
 - The request headers as key-value pairs
 
 For security reasons, webhooks can be limited to specific ports and disallowed from connecting to local URLs. See the relevant
-[configuration settings](configuration.md#workflow-webhooks) to change this behavior. If you are allowing non-admins to create workflows,
+[configuration settings](configuration.md#workflow-webhooks) to change this behavior. Webhook requests connect directly (proxy environment variables are not used) and do not follow redirects. If you are allowing non-admins to create workflows,
 you may want to adjust these settings to prevent abuse.
 
 ##### Move to Trash {#workflow-action-move-to-trash}

pyproject.toml

@@ -17,6 +17,7 @@ classifiers = [
 # TODO: Move certain things to groups and then utilize that further
 # This will allow testing to not install a webserver, mysql, etc
 dependencies = [
+  "anyio>=4.12",
   "azure-ai-documentintelligence>=1.0.2",
   "babel>=2.17",
   "bleach~=6.4.0",
@@ -47,6 +48,8 @@ dependencies = [
   "filelock~=3.32.0",
   "flower>=2.0.1,<2.2",
   "gotenberg-client[httpx]~=1.0",
+  "httpcore~=1.0.9",
+  "httpx~=0.28.1",
   "httpx-oauth~=0.17",
   "ijson>=3.5.1",
   "imap-tools>=1.14,<1.16",

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,41 @@ 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)
+
+
+@pytest.fixture
+def every_address_is_public(mocker: MockerFixture) -> None:
+    """Disable the outbound guard's address policy: every address passes.
+
+    For tests that are not themselves exercising which addresses the guard
+    accepts, so loopback and other private addresses dial just like a
+    public one.
+    """
+    from paperless_testing.outbound import allow_all_addresses
+
+    allow_all_addresses(mocker)

src/documents/tests/test_tasks.py

@@ -18,6 +18,7 @@ from documents.models import WorkflowAction
 from documents.sanity_checker import SanityCheckFailedException
 from documents.sanity_checker import SanityCheckMessages
 from documents.tests.helpers import dummy_preprocess
+from paperless_ai.exceptions import LLMBlockedError
 from paperless_testing.assertions import FileSystemAssertsMixin
 from paperless_testing.dirs import DirectoriesMixin
 
@@ -555,3 +556,37 @@ class TestApplyAISuggestionsTask(DirectoriesMixin, TestCase):
 
         apply_suggestions.assert_not_called()
         self.assertIn("no longer exists", "".join(cm.output))
+
+    @override_settings(AI_ENABLED=True)
+    def test_blocked_request_fails_without_retry(self) -> None:
+        """
+        GIVEN:
+            - AI enabled and a document with content
+            - The AI classification call blocked by the outbound request policy
+        WHEN:
+            - The task runs through Celery
+        THEN:
+            - The workflow code does not swallow the block
+            - The task fails with LLMBlockedError and is never retried
+        """
+        with (
+            mock.patch(
+                "documents.workflows.ai.get_ai_document_classification",
+                side_effect=LLMBlockedError(
+                    "AI backend request was blocked by the outbound request "
+                    "policy: detail",
+                ),
+            ),
+            mock.patch.object(
+                tasks.apply_ai_suggestions,
+                "retry",
+                wraps=tasks.apply_ai_suggestions.retry,
+            ) as retry,
+        ):
+            result = tasks.apply_ai_suggestions.apply(
+                args=(self.action.pk, self.doc.pk),
+            )
+
+        self.assertTrue(result.failed())
+        self.assertIsInstance(result.result, LLMBlockedError)
+        retry.assert_not_called()

src/documents/tests/test_views.py

@@ -29,6 +29,7 @@ from documents.models import Tag
 from documents.models import UiSettings
 from documents.signals.handlers import update_llm_suggestions_cache
 from paperless.models import ApplicationConfiguration
+from paperless_ai.exceptions import LLMBlockedError
 from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 from paperless_testing.dirs import DirectoriesMixin
@@ -770,6 +771,48 @@ class TestAISuggestions(DirectoriesMixin, TestCase):
             get_llm_suggestion_cache(self.document.pk, backend="openai-like"),
         )
 
+    @patch("documents.views.get_ai_document_classification")
+    @override_settings(
+        AI_ENABLED=True,
+        LLM_BACKEND="openai-like",
+    )
+    def test_ai_suggestions_with_blocked_llm_request(
+        self,
+        mock_get_ai_classification,
+    ) -> None:
+        """
+        GIVEN:
+            - An AI backend request blocked by the outbound request policy
+        WHEN:
+            - AI suggestions are requested
+        THEN:
+            - 502 is returned with a generic message and nothing is cached
+        """
+        mock_get_ai_classification.side_effect = LLMBlockedError(
+            "AI backend request was blocked by the outbound request policy: detail",
+        )
+
+        self.client.force_login(user=self.user)
+        response = self.client.get(
+            f"/api/documents/{self.document.pk}/ai_suggestions/",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
+        self.assertEqual(
+            response.json(),
+            {
+                "ai": [
+                    (
+                        "AI backend request was blocked by the outbound request "
+                        "policy. Check logs for details."
+                    ),
+                ],
+            },
+        )
+        self.assertIsNone(
+            get_llm_suggestion_cache(self.document.pk, backend="openai-like"),
+        )
+
     @patch("documents.views.get_ai_document_classification")
     @override_settings(
         AI_ENABLED=True,

src/documents/tests/test_workflows.py

@@ -1,9 +1,7 @@
 import datetime
 import json
 import shutil
-import socket
 import tempfile
-from collections.abc import Callable
 from datetime import timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -19,11 +17,11 @@ from django.test import override_settings
 from django.utils import timezone
 from guardian.shortcuts import get_groups_with_perms
 from guardian.shortcuts import get_users_with_perms
-from httpx import ConnectError
 from httpx import HTTPError
 from httpx import HTTPStatusError
 from pytest_django.fixtures import Settings
 from pytest_httpx import HTTPXMock
+from pytest_mock import MockerFixture
 from rest_framework.test import APIClient
 from rest_framework.test import APITestCase
 
@@ -33,8 +31,12 @@ from documents.file_handling import generate_unique_filename
 from documents.signals.handlers import run_workflows
 from documents.workflows.ai import apply_ai_suggestions_to_document
 from documents.workflows.webhooks import send_webhook
+from paperless.network import OutboundRequestBlockedError
 from paperless_ai.base_model import ClassificationSuggestions
 from paperless_ai.exceptions import LLMTimeoutError
+from paperless_testing.outbound import DialRecorder
+from paperless_testing.outbound import FakeDNS
+from paperless_testing.outbound import LocalHTTPServer
 
 if TYPE_CHECKING:
     from django.db.models import QuerySet
@@ -5069,25 +5071,6 @@ class TestWebhookSend:
         assert httpx_mock.get_request().headers["Content-Type"] == "application/json"
 
 
-@pytest.fixture
-def resolve_to(monkeypatch: pytest.MonkeyPatch) -> Callable[[str], None]:
-    """
-    Force DNS resolution to a specific IP for any hostname.
-    """
-
-    def _set(ip: str) -> None:
-        def fake_getaddrinfo(
-            host: str,
-            *_args: object,
-            **_kwargs: object,
-        ) -> list[tuple[Any, ...]]:
-            return [(socket.AF_INET, None, None, "", (ip, 0))]
-
-        monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo)
-
-    return _set
-
-
 class TestWebhookSecurity:
     def test_blocks_invalid_scheme_or_hostname(self, httpx_mock: HTTPXMock) -> None:
         """
@@ -5137,60 +5120,145 @@ class TestWebhookSecurity:
 
         assert httpx_mock.get_request() is None
 
+    @pytest.mark.parametrize(
+        "address",
+        [
+            pytest.param("127.0.0.1", id="loopback"),
+            pytest.param("10.0.0.1", id="private"),
+            pytest.param("169.254.169.254", id="link-local-metadata"),
+            pytest.param("::ffff:127.0.0.1", id="ipv4-mapped-loopback"),
+            pytest.param("64:ff9b::7f00:1", id="nat64-wrapping-loopback"),
+        ],
+    )
     @override_settings(WEBHOOKS_ALLOW_INTERNAL_REQUESTS=False)
     def test_blocks_private_loopback_linklocal(
         self,
-        httpx_mock: HTTPXMock,
-        resolve_to,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+        address: str,
     ) -> None:
         """
         GIVEN:
-            - URL with a private, loopback, or link-local IP address
+            - A webhook host resolving to a non-public address
             - WEBHOOKS_ALLOW_INTERNAL_REQUESTS is False
         WHEN:
-            - send_webhook is called with such URL
+            - send_webhook is called
         THEN:
-            - ValueError is raised
+            - The request is blocked before any connection is opened
         """
-        resolve_to("127.0.0.1")
-        with pytest.raises(ConnectError):
+        fake_dns.add("webhook.test", address)
+
+        with pytest.raises(OutboundRequestBlockedError):
             send_webhook(
-                "http://paperless-ngx.com",
+                f"http://webhook.test:{local_http_server.port}",
                 data="",
                 headers={},
                 files=None,
                 as_json=False,
             )
 
-    def test_allows_public_ip_and_sends(
+        assert local_http_server.connections == 0
+        assert dial_recorder.hosts() == []
+
+    @override_settings(WEBHOOKS_ALLOW_INTERNAL_REQUESTS=False)
+    @pytest.mark.usefixtures("every_address_is_public")
+    def test_sends_to_validated_address(
         self,
-        httpx_mock: HTTPXMock,
-        resolve_to,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
     ) -> None:
         """
         GIVEN:
-            - URL with a public IP address
+            - A webhook host resolving to an address the policy accepts
+            - WEBHOOKS_ALLOW_INTERNAL_REQUESTS is False
         WHEN:
-            - send_webhook is called with such URL
+            - send_webhook is called
         THEN:
-            - Request is sent successfully
+            - The payload arrives with the webhook hostname in the Host header
         """
-        resolve_to("52.207.186.75")
-        httpx_mock.add_response(content=b"ok")
+        fake_dns.add("webhook.test", "127.0.0.1")
 
         send_webhook(
-            url="http://paperless-ngx.com",
+            url=f"http://webhook.test:{local_http_server.port}",
             data="hi",
             headers={},
             files=None,
             as_json=False,
         )
 
-        req = httpx_mock.get_request()
-        assert req.url.host == "52.207.186.75"
-        assert req.headers["host"] == "paperless-ngx.com"
+        received = local_http_server.requests[0]
+        assert received.body == b"hi"
+        assert received.headers["host"] == f"webhook.test:{local_http_server.port}"
 
-    def test_follow_redirects_disabled(self, httpx_mock: HTTPXMock, resolve_to) -> None:
+    @override_settings(WEBHOOKS_ALLOW_INTERNAL_REQUESTS=True)
+    def test_allow_internal_sends_to_internal_address(
+        self,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+    ) -> None:
+        """
+        GIVEN:
+            - A webhook to localhost
+            - WEBHOOKS_ALLOW_INTERNAL_REQUESTS is True
+        WHEN:
+            - send_webhook is called
+        THEN:
+            - The payload arrives at the internal address
+            - The guard does not resolve the host, leaving it to the stock
+              connection path
+        """
+        send_webhook(
+            url=f"http://localhost:{local_http_server.port}",
+            data="hi",
+            headers={},
+            files=None,
+            as_json=False,
+        )
+
+        received = local_http_server.requests[0]
+        assert received.body == b"hi"
+        assert fake_dns.lookups == []
+
+    @override_settings(WEBHOOKS_ALLOW_INTERNAL_REQUESTS=False)
+    def test_block_is_an_expected_task_failure(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - A webhook host resolving to a loopback address
+            - WEBHOOKS_ALLOW_INTERNAL_REQUESTS is False
+        WHEN:
+            - The webhook task runs through Celery
+        THEN:
+            - The task fails with the original block error, not a wrapper,
+              so it matches the task's expected errors, and is not retried
+        """
+        fake_dns.add("webhook.test", "127.0.0.1")
+        retry = mocker.spy(send_webhook, "retry")
+
+        result = send_webhook.apply(
+            kwargs={
+                "url": f"http://webhook.test:{local_http_server.port}",
+                "data": "",
+                "headers": {},
+                "files": None,
+                "as_json": False,
+            },
+        )
+
+        assert result.failed()
+        assert isinstance(result.result, OutboundRequestBlockedError)
+        assert isinstance(result.result, send_webhook.throws)
+        retry.assert_not_called()
+        assert local_http_server.connections == 0
+        assert dial_recorder.hosts() == []
+
+    def test_follow_redirects_disabled(self, httpx_mock: HTTPXMock) -> None:
         """
         GIVEN:
             - A URL that redirects
@@ -5199,7 +5267,6 @@ class TestWebhookSecurity:
         THEN:
             - Request is made to the original URL and does not follow the redirect
         """
-        resolve_to("52.207.186.75")
         # Return a redirect and ensure we don't follow it (only one request recorded)
         httpx_mock.add_response(
             status_code=302,
@@ -5221,7 +5288,6 @@ class TestWebhookSecurity:
     def test_strips_user_supplied_host_header(
         self,
         httpx_mock: HTTPXMock,
-        resolve_to: Callable[[str], None],
     ) -> None:
         """
         GIVEN:
@@ -5229,9 +5295,8 @@ class TestWebhookSecurity:
         WHEN:
             - send_webhook is called with a malicious Host header
         THEN:
-            - The Host header is stripped and replaced with the resolved hostname
+            - The Host header is stripped and set from the URL hostname
         """
-        resolve_to("52.207.186.75")
         httpx_mock.add_response(content=b"ok")
 
         send_webhook(

src/documents/views.py

@@ -256,6 +256,7 @@ from paperless.views import StandardPagination
 from paperless_ai.ai_classifier import get_ai_document_classification
 from paperless_ai.ai_classifier import get_llm_output_language
 from paperless_ai.chat import stream_chat_with_documents
+from paperless_ai.exceptions import LLMBlockedError
 from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 from paperless_ai.matching import extract_unmatched_names
@@ -1697,6 +1698,23 @@ class DocumentViewSet(
                     },
                     status=status.HTTP_502_BAD_GATEWAY,
                 )
+            except LLMBlockedError as exc:
+                logger.warning(
+                    "AI backend request for document %s was blocked: %s",
+                    doc.pk,
+                    exc,
+                )
+                return Response(
+                    {
+                        "ai": [
+                            _(
+                                "AI backend request was blocked by the outbound "
+                                "request policy. Check logs for details.",
+                            ),
+                        ],
+                    },
+                    status=status.HTTP_502_BAD_GATEWAY,
+                )
             set_llm_suggestions_cache(
                 doc.pk,
                 llm_suggestions,

src/documents/workflows/webhooks.py

@@ -4,7 +4,8 @@ import httpx
 from celery import shared_task
 from django.conf import settings
 
-from paperless.network import PinnedHostHTTPTransport
+from paperless.network import GuardedHTTPTransport
+from paperless.network import OutboundRequestBlockedError
 from paperless.network import validate_outbound_http_url
 
 logger = logging.getLogger("paperless.workflows.webhooks")
@@ -14,7 +15,7 @@ logger = logging.getLogger("paperless.workflows.webhooks")
     retry_backoff=True,
     autoretry_for=(httpx.HTTPStatusError,),
     max_retries=3,
-    throws=(httpx.HTTPError,),
+    throws=(httpx.HTTPError, OutboundRequestBlockedError),
 )
 def send_webhook(
     url: str,
@@ -29,14 +30,15 @@ def send_webhook(
             url,
             allowed_schemes=settings.WEBHOOKS_ALLOWED_SCHEMES,
             allowed_ports=settings.WEBHOOKS_ALLOWED_PORTS,
-            # Internal-address checks happen in transport to preserve ConnectError behavior.
+            # Scheme and port only; the transport enforces the internal-address
+            # policy at connect time, on the address actually dialled.
             allow_internal=True,
         )
     except ValueError as e:
         logger.warning("Webhook blocked: %s", e)
         raise
 
-    transport = PinnedHostHTTPTransport(
+    transport = GuardedHTTPTransport(
         allow_internal=settings.WEBHOOKS_ALLOW_INTERNAL_REQUESTS,
     )
 

src/paperless/network.py

@@ -1,61 +1,533 @@
+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
 
-# Ranges ipaddress does not report as private, but which routinely front
-# internal infrastructure.
+# Not exported by httpcore; the guard asserts it is still the async default.
+from httpcore._backends.auto import AutoBackend
+
+logger = logging.getLogger("paperless.network")
+
+# requires-python is >=3.11, so no PEP 695 `type` statement.
+IPAddress: TypeAlias = ipaddress.IPv4Address | ipaddress.IPv6Address
+
+# Ranges that ipaddress reports as global but which still reach internal hosts.
 _NON_PUBLIC_NETWORKS = (
-    # RFC 6598 shared address space: ISP CGNAT, and the default pod/service
-    # CIDR on several managed Kubernetes offerings.
-    ipaddress.ip_network("100.64.0.0/10"),
     # RFC 6052 NAT64 well-known prefix: 64:ff9b::7f00:1 is 127.0.0.1 wherever
-    # a NAT64 gateway exists.
+    # a NAT64 gateway exists, yet ipaddress classifies the prefix as global.
     ipaddress.ip_network("64:ff9b::/96"),
 )
 
 
-def is_public_ip(ip: str | int) -> bool:
-    try:
-        obj = ipaddress.ip_address(ip)
-        return not (
-            obj.is_private
-            or obj.is_loopback
-            or obj.is_link_local
-            or obj.is_multicast
-            or obj.is_unspecified
-            or any(obj in network for network in _NON_PUBLIC_NETWORKS)
+class BlockReason(StrEnum):
+    NON_PUBLIC_ADDRESS = "non_public_address"
+    UNIX_SOCKET = "unix_socket"
+
+
+class OutboundRequestBlockedError(Exception):
+    """
+    An outbound connection was refused by policy before any socket was opened.
+
+    For NON_PUBLIC_ADDRESS, ``host`` is the name or literal being connected to
+    and ``address`` the first offending address. For UNIX_SOCKET, ``host`` is
+    the socket path and ``port`` and ``address`` are None.
+
+    ``address`` is deliberately left out of the message: the message is logged
+    and stored on failed tasks, and must not disclose internal addresses.
+    """
+
+    def __init__(
+        self,
+        *,
+        host: str,
+        port: int | None,
+        reason: BlockReason,
+        address: IPAddress | None = None,
+    ) -> None:
+        self.host = host
+        self.port = port
+        self.reason = reason
+        self.address = address
+        target = host if port is None else f"{host}:{port}"
+        super().__init__(f"Outbound connection to {target} blocked ({reason})")
+
+    def __reduce__(self) -> tuple[Callable[..., Self], tuple[object, ...]]:
+        # Celery rebuilds failed-task exceptions by pickling; keyword-only
+        # fields cannot be recovered from ``args`` alone.
+        return (
+            functools.partial(
+                type(self),
+                host=self.host,
+                port=self.port,
+                reason=self.reason,
+                address=self.address,
+            ),
+            (),
         )
-    except ValueError:  # pragma: no cover
-        return False
 
 
-def resolve_hostname_ips(hostname: str) -> list[str]:
-    try:
-        addr_info = socket.getaddrinfo(hostname, None)
-    except socket.gaierror as e:
-        raise ValueError(f"Could not resolve hostname: {hostname}") from e
+class HostResolutionError(Exception):
+    """The resolver returned no usable addresses for a host."""
 
-    ips = [info[4][0] for info in addr_info if info and info[4]]
-    if not ips:
-        raise ValueError(f"Could not resolve hostname: {hostname}")
-    return ips
+    def __init__(self, *, host: str, detail: str) -> None:
+        self.host = host
+        self.detail = detail
+        super().__init__(f"Could not resolve {host}: {detail}")
+
+    def __reduce__(self) -> tuple[Callable[..., Self], tuple[object, ...]]:
+        return (
+            functools.partial(type(self), host=self.host, detail=self.detail),
+            (),
+        )
 
 
-def format_host_for_url(host: str) -> str:
+def blocked_message(exc: OutboundRequestBlockedError | HostResolutionError) -> str:
+    """User-facing text for validation errors, kept stable for existing callers."""
+    if isinstance(exc, HostResolutionError):
+        return f"Could not resolve hostname: {exc.host}"
+    if exc.reason is BlockReason.UNIX_SOCKET:
+        return "Connection blocked: unix sockets are not permitted"
+    return f"Connection blocked: {exc.host} resolves to a non-public address"
+
+
+def is_public_ip(ip: IPAddress) -> bool:
     """
-    Format IP address for URL use (wrap IPv6 in brackets).
+    True when ``ip`` is globally routable unicast and not in a range that
+    ipaddress reports as global but which still reaches internal hosts.
+    """
+    return (
+        ip.is_global
+        and not ip.is_multicast
+        and not any(ip in network for network in _NON_PUBLIC_NETWORKS)
+    )
+
+
+# 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
+# The clock is a seam because time-machine does not mock monotonic clocks, and
+# patching time.monotonic globally would also replace the asyncio event loop's
+# own clock, hanging or misfiring its timers for the rest of the test.
+_monotonic = time.monotonic
+
+
+def _collect_addresses(
+    host: str,
+    infos: Iterable[tuple[Any, ...]],
+) -> tuple[IPAddress, ...]:
+    # Resolver output is always an address, but a scoped IPv6 answer carries a
+    # zone id ("fe80::1%1"), which is dropped before classification.
+    # dict keys keep the first occurrence and resolver order
+    addresses: dict[IPAddress, None] = {}
+    for info in infos:
+        address = ipaddress.ip_address(str(info[4][0]).split("%", 1)[0])
+        addresses.setdefault(address, None)
+    if not addresses:
+        raise HostResolutionError(host=host, detail="no addresses returned")
+    return tuple(addresses)
+
+
+def _require_public(
+    host: str,
+    port: int | None,
+    addresses: tuple[IPAddress, ...],
+) -> tuple[IPAddress, ...]:
+    for address in addresses:
+        if not is_public_ip(address):
+            raise OutboundRequestBlockedError(
+                host=host,
+                port=port,
+                reason=BlockReason.NON_PUBLIC_ADDRESS,
+                address=address,
+            )
+    return addresses
+
+
+def resolve_public_addresses(host: str, port: int | None) -> tuple[IPAddress, ...]:
+    """
+    Resolve ``host`` and return its addresses in resolver order, or raise if
+    any of them is non-public. A name is rejected as a whole; offending
+    addresses are never filtered out.
+
+    IP literals go through the resolver too: getaddrinfo answers them without
+    a lookup, and validating only its answer means no second parser can read
+    the host differently from the one that connects.
     """
     try:
-        ip_obj = ipaddress.ip_address(host)
-        if ip_obj.version == 6:
-            return f"[{host}]"
-        return host
-    except ValueError:
-        return host
+        infos = _getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except (OSError, UnicodeError) as e:
+        raise HostResolutionError(host=host, detail=str(e)) from e
+    return _require_public(host, port, _collect_addresses(host, infos))
+
+
+async def aresolve_public_addresses(
+    host: str,
+    port: int | None,
+) -> tuple[IPAddress, ...]:
+    """Async variant of resolve_public_addresses."""
+    try:
+        infos = await _agetaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except (OSError, UnicodeError) as e:
+        raise HostResolutionError(host=host, detail=str(e)) from e
+    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 _next_attempt_budget(
+    host: str,
+    deadline: float,
+    candidates: list[IPAddress],
+    index: int,
+) -> float:
+    """Budget for the attempt at index, or a timeout if none is left."""
+    remaining = deadline - _monotonic()
+    if remaining <= 0:
+        raise _budget_exhausted(host, index, len(candidates))
+    return _attempt_timeout(remaining, len(candidates) - index)
+
+
+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
+
+
+async def _aresolve_for_connect(
+    host: str,
+    port: int,
+    timeout: float | None,
+) -> tuple[IPAddress, ...]:
+    # The scope closes before dialling; attempts are not nested inside it.
+    try:
+        with anyio.fail_after(timeout):
+            return 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
+
+
+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):
+            budget = _next_attempt_budget(host, deadline, 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.
+        deadline = _deadline(timeout)
+        candidates = _attempt_order(await _aresolve_for_connect(host, port, timeout))
+        last_error: httpcore.ConnectError | httpcore.ConnectTimeout | None = None
+        for index, address in enumerate(candidates):
+            budget = _next_attempt_budget(host, deadline, 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)
+
+
+_LAYOUT_ERROR = (
+    "Unexpected httpx transport layout; refusing to create a transport "
+    "without the outbound connection guard"
+)
+
+
+class GuardedHTTPTransport(httpx.HTTPTransport):
+    """
+    httpx transport whose connections pass through the outbound guard.
+
+    Deliberately accepts no proxy, uds or retries options: a proxy would be
+    dialled instead of the destination, and a unix socket bypasses TCP
+    entirely. Adding an option here is a reviewed change, not a pass-through.
+    """
+
+    def __init__(self, *, allow_internal: bool) -> None:
+        super().__init__()
+        # httpx has no public hook for the network backend. Check the exact
+        # layout before swapping so an httpx or httpcore change fails loudly.
+        pool = self._pool
+        if (
+            type(pool) is not httpcore.ConnectionPool
+            or type(pool._network_backend) is not httpcore.SyncBackend
+        ):
+            raise RuntimeError(_LAYOUT_ERROR)
+        pool._network_backend = _GuardedSyncBackend(
+            pool._network_backend,
+            allow_internal=allow_internal,
+        )
+
+
+class GuardedAsyncHTTPTransport(httpx.AsyncHTTPTransport):
+    """Async twin of GuardedHTTPTransport."""
+
+    def __init__(self, *, allow_internal: bool) -> None:
+        super().__init__()
+        pool = self._pool
+        if (
+            type(pool) is not httpcore.AsyncConnectionPool
+            or type(pool._network_backend) is not AutoBackend
+        ):
+            raise RuntimeError(_LAYOUT_ERROR)
+        pool._network_backend = _GuardedAsyncBackend(
+            pool._network_backend,
+            allow_internal=allow_internal,
+        )
+
+
+def create_guarded_httpx_client(
+    url: str,
+    *,
+    allow_internal: bool,
+    timeout: float,
+) -> httpx.Client:
+    """
+    Validate ``url`` up front, then build a client that re-checks at connect
+    time. The up-front check turns static misconfiguration into a ValueError
+    before any retry layer sees it.
+    """
+    validate_outbound_http_url(url, allow_internal=allow_internal)
+    return httpx.Client(
+        transport=GuardedHTTPTransport(allow_internal=allow_internal),
+        timeout=timeout,
+    )
+
+
+def create_guarded_async_httpx_client(
+    url: str,
+    *,
+    allow_internal: bool,
+    timeout: float,
+) -> httpx.AsyncClient:
+    """Async twin of create_guarded_httpx_client."""
+    validate_outbound_http_url(url, allow_internal=allow_internal)
+    return httpx.AsyncClient(
+        transport=GuardedAsyncHTTPTransport(allow_internal=allow_internal),
+        timeout=timeout,
+    )
+
+
+# urllib3 treats a backslash as ending the authority while urlparse and httpx do
+# not, so the host checked here could differ from the one that is dialled.
+# Control and whitespace characters are refused for the same reason.
+_UNSAFE_URL_CHARS = re.compile(r"[\\\x00-\x1f\x7f\s]")
+
+
+def _dns_name(url: str) -> str:
+    """
+    The ASCII hostname that httpx and urllib3 look up for ``url``.
+
+    urlparse keeps a non-ASCII hostname as typed, and getaddrinfo would then
+    encode it with the stdlib IDNA 2003 codec. That maps some characters
+    differently from the IDNA 2008 encoding the HTTP clients use ("faß"
+    becomes "fass" instead of "xn--fa-hia"), so the check would resolve a
+    different name from the one that is connected to.
+    """
+    try:
+        return httpx.URL(url).raw_host.decode("ascii")
+    except (httpx.InvalidURL, UnicodeError) as e:
+        raise ValueError("Invalid URL scheme or hostname.") from e
 
 
 def validate_outbound_http_url(
@@ -81,128 +553,17 @@ def validate_outbound_http_url(
         raise ValueError("Destination port not permitted.")
 
     if not allow_internal:
-        for ip_str in resolve_hostname_ips(parsed.hostname):
-            if not is_public_ip(ip_str):
-                raise ValueError(
-                    f"Connection blocked: {parsed.hostname} resolves to a non-public address",
-                )
+        if _UNSAFE_URL_CHARS.search(url):
+            raise ValueError("Invalid URL scheme or hostname.")
+        host = _dns_name(url)
+        # HTTP clients may percent-decode the host before resolving it, so the
+        # checked name could differ from the dialled one. An IPv6 zone id is the
+        # only legitimate use, and link-local addresses are non-public anyway.
+        if "%" in host:
+            raise ValueError("Invalid URL scheme or hostname.")
+        try:
+            resolve_public_addresses(host, port)
+        except (OutboundRequestBlockedError, HostResolutionError) as e:
+            raise ValueError(blocked_message(e)) from e
 
     return parsed
-
-
-def _rewrite_request_to_pinned_ip(
-    request: httpx.Request,
-    *,
-    allow_internal: bool,
-) -> httpx.Request:
-    hostname = request.url.host
-
-    if not hostname:
-        raise httpx.ConnectError("No hostname in request URL")
-
-    try:
-        ips = resolve_hostname_ips(hostname)
-    except ValueError as e:
-        raise httpx.ConnectError(str(e)) from e
-
-    if not allow_internal:
-        for ip_str in ips:
-            if not is_public_ip(ip_str):
-                raise httpx.ConnectError(
-                    f"Connection blocked: {hostname} resolves to a non-public address",
-                )
-
-    ip_str = ips[0]
-    formatted_ip = format_host_for_url(ip_str)
-
-    new_headers = httpx.Headers(request.headers)
-    if "host" in new_headers:
-        del new_headers["host"]
-    host_header = format_host_for_url(hostname)
-    default_port = 443 if request.url.scheme == "https" else 80
-    if request.url.port and request.url.port != default_port:
-        host_header = f"{host_header}:{request.url.port}"
-    new_headers["Host"] = host_header
-    new_url = request.url.copy_with(host=formatted_ip)
-
-    rewritten_request = httpx.Request(
-        method=request.method,
-        url=new_url,
-        headers=new_headers,
-        stream=request.stream,
-        extensions=request.extensions,
-    )
-    rewritten_request.extensions["sni_hostname"] = hostname
-
-    return rewritten_request
-
-
-class PinnedHostHTTPTransport(httpx.HTTPTransport):
-    """
-    HTTP transport that resolves/validates hostnames per request and connects to
-    a vetted IP while preserving the original Host header and TLS SNI hostname.
-    """
-
-    def __init__(
-        self,
-        *args,
-        allow_internal: bool = False,
-        **kwargs,
-    ) -> None:
-        super().__init__(*args, **kwargs)
-        self.allow_internal = allow_internal
-
-    def handle_request(self, request: httpx.Request) -> httpx.Response:
-        request = _rewrite_request_to_pinned_ip(
-            request,
-            allow_internal=self.allow_internal,
-        )
-        return super().handle_request(request)
-
-
-class PinnedHostAsyncHTTPTransport(httpx.AsyncHTTPTransport):
-    """
-    Async variant of PinnedHostHTTPTransport.
-    """
-
-    def __init__(
-        self,
-        *args,
-        allow_internal: bool = False,
-        **kwargs,
-    ) -> None:
-        super().__init__(*args, **kwargs)
-        self.allow_internal = allow_internal
-
-    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
-        request = _rewrite_request_to_pinned_ip(
-            request,
-            allow_internal=self.allow_internal,
-        )
-        return await super().handle_async_request(request)
-
-
-def create_pinned_httpx_client(
-    url: str,
-    *,
-    allow_internal: bool = False,
-    **kwargs,
-) -> httpx.Client:
-    validate_outbound_http_url(url, allow_internal=allow_internal)
-    return httpx.Client(
-        transport=PinnedHostHTTPTransport(allow_internal=allow_internal),
-        **kwargs,
-    )
-
-
-def create_pinned_async_httpx_client(
-    url: str,
-    *,
-    allow_internal: bool = False,
-    **kwargs,
-) -> httpx.AsyncClient:
-    validate_outbound_http_url(url, allow_internal=allow_internal)
-    return httpx.AsyncClient(
-        transport=PinnedHostAsyncHTTPTransport(allow_internal=allow_internal),
-        **kwargs,
-    )

src/paperless/tests/test_network_integration.py

@@ -0,0 +1,375 @@
+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.network import create_guarded_httpx_client
+from paperless_testing.outbound import DialRecorder
+from paperless_testing.outbound import FakeDNS
+from paperless_testing.outbound import LocalHTTPServer
+from paperless_testing.outbound import running_http_server
+
+
+class TestGuardedTransportSync:
+    @pytest.mark.usefixtures("every_address_is_public")
+    def test_pinned_connection_falls_back_to_next_address(
+        self,
+        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")
+
+        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 == []
+
+    @pytest.mark.usefixtures("every_address_is_public")
+    def test_host_header_is_the_hostname(
+        self,
+        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")
+
+        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
+
+    @pytest.mark.usefixtures("every_address_is_public")
+    def test_connections_are_not_shared_between_hosts_on_one_address(
+        self,
+        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")
+
+        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}",
+        ]
+
+    @pytest.mark.usefixtures("every_address_is_public")
+    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")
+        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("localhost", id="name"),
+            pytest.param("2130706433", id="decimal"),
+            pytest.param("0x7f.1", id="hex-short"),
+            pytest.param("127.1", id="short-dotted"),
+        ],
+    )
+    def test_blocks_internal_host_without_connecting(
+        self,
+        local_http_server: LocalHTTPServer,
+        dial_recorder: DialRecorder,
+        host: str,
+    ) -> None:
+        """
+        GIVEN:
+            - Internal addresses disallowed
+            - A URL whose host reaches loopback, by name or by 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
+        assert dial_recorder.hosts() == []
+
+    @pytest.mark.usefixtures("every_address_is_public")
+    def test_environment_proxy_is_not_used(
+        self,
+        mocker: MockerFixture,
+        local_http_server: LocalHTTPServer,
+        fake_dns: FakeDNS,
+        dial_recorder: DialRecorder,
+    ) -> None:
+        """
+        GIVEN:
+            - Proxy variables in the environment pointing at a second local server
+            - Internal addresses disallowed
+        WHEN:
+            - A request is made through the production client factory to an
+              allowed origin
+        THEN:
+            - The origin server receives the request directly and the proxy
+              server never sees a connection
+        """
+        with running_http_server() as proxy_server:
+            mocker.patch.dict(
+                os.environ,
+                {
+                    "HTTP_PROXY": f"http://127.0.0.1:{proxy_server.port}",
+                    "HTTPS_PROXY": f"http://127.0.0.1:{proxy_server.port}",
+                    "ALL_PROXY": f"http://127.0.0.1:{proxy_server.port}",
+                },
+            )
+            fake_dns.add("origin.test", "127.0.0.1")
+
+            url = f"http://origin.test:{local_http_server.port}/"
+            with create_guarded_httpx_client(
+                url,
+                allow_internal=False,
+                timeout=5.0,
+            ) as client:
+                response = client.get(url)
+
+            assert response.status_code == 200
+            assert len(local_http_server.requests) == 1
+            assert local_http_server.requests[0].headers["host"] == (
+                f"origin.test:{local_http_server.port}"
+            )
+            assert proxy_server.connections == 0
+            assert proxy_server.requests == []
+            assert dial_recorder.hosts() == ["127.0.0.1"]
+
+
+class TestGuardedTransportAsync:
+    @pytest.fixture(autouse=True)
+    def anyio_backend(self) -> str:
+        return "asyncio"
+
+    @pytest.mark.anyio
+    @pytest.mark.usefixtures("every_address_is_public")
+    async def test_pinned_connection_falls_back_to_next_address(
+        self,
+        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")
+
+        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,
+        dial_recorder: DialRecorder,
+    ) -> 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
+        assert dial_recorder.hosts() == []

src/paperless_ai/client.py

@@ -14,14 +14,16 @@ if TYPE_CHECKING:
     from llama_index.llms.openai_like import OpenAILike
 
 from paperless.config import AIConfig
-from paperless.network import PinnedHostAsyncHTTPTransport
-from paperless.network import PinnedHostHTTPTransport
-from paperless.network import create_pinned_async_httpx_client
-from paperless.network import create_pinned_httpx_client
+from paperless.network import GuardedAsyncHTTPTransport
+from paperless.network import GuardedHTTPTransport
+from paperless.network import OutboundRequestBlockedError
+from paperless.network import create_guarded_async_httpx_client
+from paperless.network import create_guarded_httpx_client
 from paperless.network import validate_outbound_http_url
 from paperless_ai.base_model import ClassificationSuggestions
 from paperless_ai.base_model import DocumentClassifierSchema
 from paperless_ai.base_model import model_to_classification_suggestions
+from paperless_ai.exceptions import LLMBlockedError
 from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
 
@@ -43,6 +45,19 @@ LLM_SYSTEM_PROMPT = (
 PLACEHOLDER_API_KEY: Final = "fake"
 
 
+def _find_blocked_cause(exc: BaseException) -> OutboundRequestBlockedError | None:
+    # The openai SDK wraps transport errors in APIConnectionError, so the
+    # block can sit anywhere in the __cause__ chain.
+    current: BaseException | None = exc
+    seen: set[int] = set()
+    while current is not None and id(current) not in seen:
+        if isinstance(current, OutboundRequestBlockedError):
+            return current
+        seen.add(id(current))
+        current = current.__cause__
+    return None
+
+
 class AIClient:
     """
     A client for interacting with an LLM backend.
@@ -63,10 +78,10 @@ class AIClient:
                 endpoint,
                 allow_internal=self.settings.llm_allow_internal_endpoints,
             )
-            transport = PinnedHostHTTPTransport(
+            transport = GuardedHTTPTransport(
                 allow_internal=self.settings.llm_allow_internal_endpoints,
             )
-            async_transport = PinnedHostAsyncHTTPTransport(
+            async_transport = GuardedAsyncHTTPTransport(
                 allow_internal=self.settings.llm_allow_internal_endpoints,
             )
             return Ollama(
@@ -93,12 +108,12 @@ class AIClient:
             http_client = None
             async_http_client = None
             if endpoint:
-                http_client = create_pinned_httpx_client(
+                http_client = create_guarded_httpx_client(
                     endpoint,
                     allow_internal=self.settings.llm_allow_internal_endpoints,
                     timeout=self.settings.llm_request_timeout,
                 )
-                async_http_client = create_pinned_async_httpx_client(
+                async_http_client = create_guarded_async_httpx_client(
                     endpoint,
                     allow_internal=self.settings.llm_allow_internal_endpoints,
                     timeout=self.settings.llm_request_timeout,
@@ -179,6 +194,12 @@ class AIClient:
         except httpx.TimeoutException as exc:
             raise LLMTimeoutError from exc
         except Exception as exc:
+            blocked = _find_blocked_cause(exc)
+            if blocked is not None:
+                raise LLMBlockedError(
+                    "AI backend request was blocked by the outbound request "
+                    f"policy: {blocked}",
+                ) from exc
             if self._is_openai_timeout(exc):
                 raise LLMTimeoutError from exc
             if self._is_provider_error(exc):

src/paperless_ai/embedding.py

@@ -9,10 +9,10 @@ if TYPE_CHECKING:
 from documents.models import Document
 from paperless.config import AIConfig
 from paperless.models import LLMEmbeddingBackend
-from paperless.network import PinnedHostAsyncHTTPTransport
-from paperless.network import PinnedHostHTTPTransport
-from paperless.network import create_pinned_async_httpx_client
-from paperless.network import create_pinned_httpx_client
+from paperless.network import GuardedAsyncHTTPTransport
+from paperless.network import GuardedHTTPTransport
+from paperless.network import create_guarded_async_httpx_client
+from paperless.network import create_guarded_httpx_client
 from paperless.network import validate_outbound_http_url
 from paperless_ai.client import PLACEHOLDER_API_KEY
 
@@ -29,12 +29,12 @@ def get_embedding_model(config: AIConfig) -> "BaseEmbedding":
             http_client = None
             async_http_client = None
             if endpoint:
-                http_client = create_pinned_httpx_client(
+                http_client = create_guarded_httpx_client(
                     endpoint,
                     allow_internal=config.llm_allow_internal_endpoints,
                     timeout=config.llm_request_timeout,
                 )
-                async_http_client = create_pinned_async_httpx_client(
+                async_http_client = create_guarded_async_httpx_client(
                     endpoint,
                     allow_internal=config.llm_allow_internal_endpoints,
                     timeout=config.llm_request_timeout,
@@ -77,14 +77,14 @@ def get_embedding_model(config: AIConfig) -> "BaseEmbedding":
             embedding._client = Client(
                 host=endpoint,
                 timeout=config.llm_request_timeout,
-                transport=PinnedHostHTTPTransport(
+                transport=GuardedHTTPTransport(
                     allow_internal=config.llm_allow_internal_endpoints,
                 ),
             )
             embedding._async_client = AsyncClient(
                 host=endpoint,
                 timeout=config.llm_request_timeout,
-                transport=PinnedHostAsyncHTTPTransport(
+                transport=GuardedAsyncHTTPTransport(
                     allow_internal=config.llm_allow_internal_endpoints,
                 ),
             )

src/paperless_ai/exceptions.py

@@ -4,3 +4,7 @@ class LLMTimeoutError(Exception):
 
 class LLMProviderError(Exception):
     """The LLM backend rejected the request."""
+
+
+class LLMBlockedError(Exception):
+    """The outbound request policy refused the connection to the LLM backend."""

src/paperless_ai/tests/test_client.py

@@ -1,3 +1,4 @@
+import ipaddress
 import json
 from unittest.mock import ANY
 from unittest.mock import MagicMock
@@ -9,11 +10,15 @@ import openai
 import pytest
 from llama_index.core.llms.llm import ToolSelection
 
+from paperless.network import BlockReason
+from paperless.network import OutboundRequestBlockedError
 from paperless_ai.client import LLM_SYSTEM_PROMPT
 from paperless_ai.client import PLACEHOLDER_API_KEY
 from paperless_ai.client import AIClient
+from paperless_ai.exceptions import LLMBlockedError
 from paperless_ai.exceptions import LLMProviderError
 from paperless_ai.exceptions import LLMTimeoutError
+from paperless_testing.outbound import guard_of
 
 
 @pytest.fixture
@@ -277,3 +282,142 @@ def test_run_llm_query_httpx_timeout_raises_local_error(
 
     with pytest.raises(LLMTimeoutError):
         client.run_llm_query("test_prompt")
+
+
+class TestGuardedLLMClients:
+    @pytest.mark.parametrize(
+        ("endpoint", "allow_internal"),
+        [
+            pytest.param("http://test-url", True, id="internal-allowed"),
+            pytest.param("http://93.184.216.34:11434", False, id="internal-blocked"),
+        ],
+    )
+    def test_ollama_clients_are_guarded(
+        self,
+        mock_ai_config: MagicMock,
+        mock_ollama_llm: MagicMock,
+        endpoint: str,
+        *,
+        allow_internal: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - The Ollama backend
+        WHEN:
+            - The LLM is built
+        THEN:
+            - Its sync and async clients use guarded transports with the setting
+        """
+        mock_ai_config.llm_backend = "ollama"
+        mock_ai_config.llm_model = "test_model"
+        mock_ai_config.llm_endpoint = endpoint
+        mock_ai_config.llm_allow_internal_endpoints = allow_internal
+
+        AIClient()
+
+        kwargs = mock_ollama_llm.call_args.kwargs
+        assert guard_of(kwargs["client"]._client)._allow_internal is allow_internal
+        assert (
+            guard_of(kwargs["async_client"]._client)._allow_internal is allow_internal
+        )
+
+    @pytest.mark.parametrize(
+        ("endpoint", "allow_internal"),
+        [
+            pytest.param("http://test-url", True, id="internal-allowed"),
+            pytest.param("http://93.184.216.34:8080", False, id="internal-blocked"),
+        ],
+    )
+    def test_openai_like_clients_are_guarded(
+        self,
+        mock_ai_config: MagicMock,
+        mock_openai_llm: MagicMock,
+        endpoint: str,
+        *,
+        allow_internal: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - The OpenAI-like backend with an endpoint
+        WHEN:
+            - The LLM is built
+        THEN:
+            - Its sync and async http clients use guarded transports
+        """
+        mock_ai_config.llm_backend = "openai-like"
+        mock_ai_config.llm_model = "test_model"
+        mock_ai_config.llm_api_key = "key"
+        mock_ai_config.llm_endpoint = endpoint
+        mock_ai_config.llm_allow_internal_endpoints = allow_internal
+
+        AIClient()
+
+        kwargs = mock_openai_llm.call_args.kwargs
+        assert guard_of(kwargs["http_client"])._allow_internal is allow_internal
+        assert guard_of(kwargs["async_http_client"])._allow_internal is allow_internal
+
+
+def _block() -> OutboundRequestBlockedError:
+    return OutboundRequestBlockedError(
+        host="llm.example",
+        port=443,
+        reason=BlockReason.NON_PUBLIC_ADDRESS,
+        address=ipaddress.ip_address("10.0.0.1"),
+    )
+
+
+class TestBlockedLLMRequests:
+    def test_ollama_block_becomes_llm_blocked_error(
+        self,
+        mock_ai_config: MagicMock,
+        mock_ollama_llm: MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - The Ollama backend and a connection blocked by policy
+        WHEN:
+            - An LLM query runs
+        THEN:
+            - LLMBlockedError is raised with a message, chained to the block
+            - The message, which tracked tasks store, names the destination but
+              not the resolved internal address
+        """
+        mock_ai_config.llm_backend = "ollama"
+        mock_ai_config.llm_model = "test_model"
+        mock_ai_config.llm_endpoint = "http://test-url"
+        block = _block()
+        mock_ollama_llm.return_value.chat.side_effect = block
+
+        with pytest.raises(LLMBlockedError) as exc_info:
+            AIClient().run_llm_query("test_prompt")
+
+        assert exc_info.value.__cause__ is block
+        assert "llm.example:443" in str(exc_info.value)
+        assert "10.0.0.1" not in str(exc_info.value)
+
+    def test_openai_wrapped_block_becomes_llm_blocked_error(
+        self,
+        mock_ai_config: MagicMock,
+        mock_openai_llm: MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - The OpenAI-like backend, whose SDK wraps the block in
+              APIConnectionError
+        WHEN:
+            - An LLM query runs
+        THEN:
+            - LLMBlockedError is raised
+        """
+        mock_ai_config.llm_backend = "openai-like"
+        mock_ai_config.llm_model = "test_model"
+        mock_ai_config.llm_api_key = "key"
+        mock_ai_config.llm_endpoint = "http://test-url"
+        wrapped = openai.APIConnectionError(
+            request=httpx.Request("POST", "http://test-url/v1/chat/completions"),
+        )
+        wrapped.__cause__ = _block()
+        mock_openai_llm.return_value.chat_with_tools.side_effect = wrapped
+
+        with pytest.raises(LLMBlockedError):
+            AIClient().run_llm_query("test_prompt")

src/paperless_ai/tests/test_embedding.py

@@ -1,9 +1,12 @@
+from typing import TYPE_CHECKING
+from typing import cast
 from unittest.mock import ANY
 from unittest.mock import MagicMock
 from unittest.mock import patch
 
 import pytest
 from django.conf import settings
+from pytest_mock import MockerFixture
 
 from documents.models import Document
 from paperless.models import LLMEmbeddingBackend
@@ -12,6 +15,10 @@ from paperless_ai.embedding import _normalize_llm_index_text
 from paperless_ai.embedding import build_llm_index_text
 from paperless_ai.embedding import get_configured_model_name
 from paperless_ai.embedding import get_embedding_model
+from paperless_testing.outbound import guard_of
+
+if TYPE_CHECKING:
+    from llama_index.embeddings.ollama import OllamaEmbedding
 
 
 @pytest.fixture
@@ -283,3 +290,61 @@ def test_normalize_llm_index_text_collapses_ocr_leaders_without_joining_lines():
 
 def test_normalize_llm_index_text_collapses_non_breaking_spaces():
     assert _normalize_llm_index_text("A\u00a0........\u00a0B") == "A B"
+
+
+class TestGuardedEmbeddingClients:
+    def test_ollama_embedding_clients_are_guarded(
+        self,
+        mocker: MockerFixture,
+        mock_ai_config: MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - The Ollama embedding backend
+        WHEN:
+            - The embedding model is built
+        THEN:
+            - The clients swapped onto it use guarded transports
+        """
+        config = mock_ai_config.return_value
+        config.llm_embedding_backend = LLMEmbeddingBackend.OLLAMA
+        config.llm_embedding_model = "embeddinggemma"
+        config.llm_endpoint = "http://93.184.216.34:11434"
+        config.llm_allow_internal_endpoints = False
+
+        mocker.patch("llama_index.embeddings.ollama.OllamaEmbedding")
+
+        model = cast("OllamaEmbedding", get_embedding_model(config))
+
+        assert guard_of(model._client._client)._allow_internal is False
+        assert guard_of(model._async_client._client)._allow_internal is False
+
+    def test_openai_like_embedding_clients_are_guarded(
+        self,
+        mocker: MockerFixture,
+        mock_ai_config: MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - The OpenAI-like embedding backend with an endpoint
+        WHEN:
+            - The embedding model is built
+        THEN:
+            - Its http clients use guarded transports
+        """
+        config = mock_ai_config.return_value
+        config.llm_embedding_backend = LLMEmbeddingBackend.OPENAI_LIKE
+        config.llm_embedding_model = "text-embedding-3-small"
+        config.llm_api_key = "key"
+        config.llm_endpoint = "http://93.184.216.34:8080"
+        config.llm_allow_internal_endpoints = False
+
+        embedding_class = mocker.patch(
+            "llama_index.embeddings.openai_like.OpenAILikeEmbedding",
+        )
+
+        get_embedding_model(config)
+
+        kwargs = embedding_class.call_args.kwargs
+        assert guard_of(kwargs["http_client"])._allow_internal is False
+        assert guard_of(kwargs["async_http_client"])._allow_internal is False

src/paperless_mail/mail.py

@@ -45,8 +45,11 @@ from documents.models import Correspondent
 from documents.models import PaperlessTask
 from documents.parsers import is_mime_type_supported
 from documents.tasks import consume_file
-from paperless.network import is_public_ip
-from paperless.network import resolve_hostname_ips
+from paperless.network import HostResolutionError
+from paperless.network import IPAddress
+from paperless.network import OutboundRequestBlockedError
+from paperless.network import blocked_message
+from paperless.network import resolve_public_addresses
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
 from paperless_mail.models import ProcessedMail
@@ -445,18 +448,34 @@ class PinnedIMAP4(imaplib.IMAP4):
 
     Without pinned addresses, and with the ssl_context of the matching imaplib
     class, this behaves exactly like imaplib.IMAP4 / imaplib.IMAP4_SSL.
+
+    ``pinned_ips`` of ``None`` means no pinning was requested and the stock
+    imaplib connection path is used. An empty tuple means pinning was requested
+    and yielded nothing, and the connection fails without opening a socket
+    rather than falling back to a hostname lookup.
     """
 
-    def __init__(self, host, port, pinned_ips, ssl_context=None, timeout=None) -> None:
+    def __init__(
+        self,
+        host: str,
+        port: int | None,
+        pinned_ips: tuple[IPAddress, ...] | None,
+        ssl_context: ssl.SSLContext | None = None,
+        timeout: float | None = None,
+    ) -> None:
         self._pinned_ips = pinned_ips
         self.ssl_context = ssl_context
         super().__init__(host, port, timeout=timeout)
 
-    def _connect_pinned(self, timeout):
+    def _connect_pinned(
+        self,
+        pinned_ips: tuple[IPAddress, ...],
+        timeout: float | None,
+    ) -> socket.socket:
         last_error: OSError | None = None
-        for ip_str in self._pinned_ips:
+        for ip in pinned_ips:
             try:
-                address = (ip_str, self.port)
+                address = (str(ip), self.port)
                 if timeout is not None:
                     return socket.create_connection(address, timeout)
                 return socket.create_connection(address)
@@ -464,9 +483,9 @@ class PinnedIMAP4(imaplib.IMAP4):
                 last_error = e
         raise last_error or OSError(f"Could not connect to {self.host}")
 
-    def _create_socket(self, timeout):
-        if self._pinned_ips:
-            sock = self._connect_pinned(timeout)
+    def _create_socket(self, timeout: float | None) -> socket.socket:
+        if self._pinned_ips is not None:
+            sock = self._connect_pinned(self._pinned_ips, timeout)
         else:
             sock = super()._create_socket(timeout)
         if self.ssl_context is None:
@@ -477,7 +496,12 @@ class PinnedIMAP4(imaplib.IMAP4):
 class PinnedClientMixin:
     """Builds the imaplib client against the pre-resolved addresses, if any."""
 
-    def __init__(self, *args, pinned_ips: list[str] | None, **kwargs) -> None:
+    def __init__(
+        self,
+        *args,
+        pinned_ips: tuple[IPAddress, ...] | None,
+        **kwargs,
+    ) -> None:
         self._pinned_ips = pinned_ips
         super().__init__(*args, **kwargs)
 
@@ -515,22 +539,20 @@ class PinnedMailBoxStartTls(PinnedClientMixin, MailBoxStartTls):
         return client
 
 
-def get_mailbox(server, port, security) -> MailBox:
+def get_mailbox(
+    server: str,
+    port: int | None,
+    security: int,
+) -> MailBox:
     """
     Returns the correct MailBox instance for the given configuration.
     """
-    pinned_ips: list[str] | None = None
+    pinned_ips: tuple[IPAddress, ...] | None = None
     if not settings.EMAIL_ALLOW_INTERNAL_HOSTS:
         try:
-            pinned_ips = resolve_hostname_ips(server)
-        except ValueError as e:
-            raise MailError(str(e)) from e
-
-        for ip_str in pinned_ips:
-            if not is_public_ip(ip_str):
-                raise MailError(
-                    f"Connection blocked: {server} resolves to a non-public address",
-                )
+            pinned_ips = resolve_public_addresses(server, port)
+        except (OutboundRequestBlockedError, HostResolutionError) as e:
+            raise MailError(blocked_message(e)) from e
 
     ssl_context = ssl.create_default_context()
     if settings.EMAIL_CERTIFICATE_FILE is not None:  # pragma: no cover

src/paperless_mail/tests/test_mail.py

@@ -1,9 +1,12 @@
 import dataclasses
+import ipaddress
+import socket
 import time
 import uuid
 from collections import namedtuple
 from datetime import timedelta
 from unittest import mock
+from unittest.mock import MagicMock
 
 import pytest
 from django.contrib.auth.models import Permission
@@ -25,6 +28,7 @@ from documents.models import MatchingModel
 from paperless_mail import tasks
 from paperless_mail.mail import MailAccountHandler
 from paperless_mail.mail import MailError
+from paperless_mail.mail import PinnedIMAP4
 from paperless_mail.mail import TagMailAction
 from paperless_mail.mail import apply_mail_action
 from paperless_mail.mail import error_callback
@@ -2045,10 +2049,13 @@ class TestMailAccountTestView(APITestCase):
         self.assertEqual(response.content.decode(), "Unable to connect to server")
 
     @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False)
-    @mock.patch("paperless_mail.mail.resolve_hostname_ips", return_value=["127.0.0.1"])
+    @mock.patch(
+        "paperless.network._getaddrinfo",
+        return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 993))],
+    )
     def test_mail_account_test_view_blocks_internal_host_when_disabled(
         self,
-        _mock_resolve_hostname_ips,
+        _mock_getaddrinfo: MagicMock,
     ) -> None:
         data = {
             "imap_server": "internal.example",
@@ -2205,10 +2212,10 @@ class TestGetMailboxHostPinning(TestCase):
 
     @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False)
     @mock.patch(
-        "paperless_mail.mail.resolve_hostname_ips",
-        return_value=["93.184.216.34"],
+        "paperless_mail.mail.resolve_public_addresses",
+        return_value=(ipaddress.ip_address("93.184.216.34"),),
     )
-    def test_connects_to_validated_ip(self, _mock_resolve) -> None:
+    def test_connects_to_validated_ip(self, _mock_resolve: MagicMock) -> None:
         with mock.patch(
             "paperless_mail.mail.socket.create_connection",
             side_effect=OSError("no connection in tests"),
@@ -2225,10 +2232,13 @@ class TestGetMailboxHostPinning(TestCase):
 
     @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False)
     @mock.patch(
-        "paperless_mail.mail.resolve_hostname_ips",
-        return_value=["93.184.216.34"],
+        "paperless_mail.mail.resolve_public_addresses",
+        return_value=(ipaddress.ip_address("93.184.216.34"),),
     )
-    def test_ssl_pins_ip_but_keeps_hostname_for_sni(self, _mock_resolve) -> None:
+    def test_ssl_pins_ip_but_keeps_hostname_for_sni(
+        self,
+        _mock_resolve: MagicMock,
+    ) -> None:
         ssl_context = mock.MagicMock()
         ssl_context.wrap_socket.return_value.makefile.side_effect = OSError(
             "no connection in tests",
@@ -2259,13 +2269,51 @@ class TestGetMailboxHostPinning(TestCase):
 
     @override_settings(EMAIL_ALLOW_INTERNAL_HOSTS=False)
     @mock.patch(
-        "paperless_mail.mail.resolve_hostname_ips",
-        return_value=["93.184.216.34", "127.0.0.1"],
+        "paperless.network._getaddrinfo",
+        return_value=[
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 993)),
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 993)),
+        ],
     )
-    def test_blocks_when_any_resolved_address_is_internal(self, _mock_resolve) -> None:
-        with self.assertRaises(MailError):
+    def test_blocks_when_any_resolved_address_is_internal(
+        self,
+        _mock_resolve: MagicMock,
+    ) -> None:
+        """
+        GIVEN:
+            - A mail host resolving to one public and one loopback address
+            - EMAIL_ALLOW_INTERNAL_HOSTS is False
+        WHEN:
+            - A mailbox is requested
+        THEN:
+            - The whole host is blocked with the existing message
+        """
+        with self.assertRaisesMessage(
+            MailError,
+            "Connection blocked: mail.example.com resolves to a non-public address",
+        ):
             get_mailbox("mail.example.com", 993, MailAccount.ImapSecurity.SSL)
 
+    def test_empty_pin_list_never_falls_back_to_hostname_lookup(self) -> None:
+        """
+        GIVEN:
+            - A pinned IMAP client given an empty tuple of addresses
+        WHEN:
+            - It connects
+        THEN:
+            - It fails without opening any socket, rather than resolving the
+              hostname itself
+        """
+        with (
+            mock.patch("paperless_mail.mail.socket.create_connection") as pinned,
+            mock.patch("imaplib.IMAP4._create_socket") as unpinned,
+            self.assertRaises(OSError),
+        ):
+            PinnedIMAP4("mail.example.com", 143, ())
+
+        pinned.assert_not_called()
+        unpinned.assert_not_called()
+
 
 class TestMailAccountProcess(APITestCase):
     def setUp(self) -> None:

src/paperless_testing/outbound.py

@@ -0,0 +1,218 @@
+"""
+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 allow_all_addresses(mocker: MockerFixture) -> None:
+    """Patch the guard's public-address check to accept every address.
+
+    Loopback and other private addresses pass just like a public one, for
+    tests that exercise something other than the address policy itself.
+    """
+    mocker.patch("paperless.network.is_public_ip", return_value=True)
+
+
+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

uv.lock

@@ -2874,6 +2874,7 @@ name = "paperless-ngx"
 version = "3.2.1"
 source = { virtual = "." }
 dependencies = [
+    { name = "anyio" },
     { name = "azure-ai-documentintelligence" },
     { name = "babel" },
     { name = "bleach" },
@@ -2902,6 +2903,8 @@ dependencies = [
     { name = "filelock" },
     { name = "flower" },
     { name = "gotenberg-client", extra = ["httpx"] },
+    { name = "httpcore" },
+    { name = "httpx" },
     { name = "httpx-oauth" },
     { name = "ijson" },
     { name = "imap-tools" },
@@ -3026,6 +3029,7 @@ typing = [
 
 [package.metadata]
 requires-dist = [
+    { name = "anyio", specifier = ">=4.12" },
     { name = "azure-ai-documentintelligence", specifier = ">=1.0.2" },
     { name = "babel", specifier = ">=2.17" },
     { name = "bleach", specifier = "~=6.4.0" },
@@ -3055,6 +3059,8 @@ requires-dist = [
     { name = "flower", specifier = ">=2.0.1,<2.2" },
     { name = "gotenberg-client", extras = ["httpx"], specifier = "~=1.0" },
     { name = "granian", extras = ["uvloop"], marker = "extra == 'webserver'", specifier = ">=2.7,<2.9" },
+    { name = "httpcore", specifier = "~=1.0.9" },
+    { name = "httpx", specifier = "~=0.28.1" },
     { name = "httpx-oauth", specifier = "~=0.17" },
     { name = "ijson", specifier = ">=3.5.1" },
     { name = "imap-tools", specifier = ">=1.14,<1.16" },