Type outbound block errors and classify addresses with is_global

is_public_ip now takes an ipaddress object and relies on is_global, keeping
multicast and the NAT64 well-known prefix as explicit extra exclusions.
Adds OutboundRequestBlockedError and HostResolutionError, both picklable so
Celery keeps them intact on task failure.

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

src/paperless/network.py

@@ -1,36 +1,108 @@
+import functools
 import ipaddress
 import socket
+from collections.abc import Callable
 from collections.abc import Collection
+from enum import StrEnum
+from typing import Self
+from typing import TypeAlias
 from urllib.parse import ParseResult
 from urllib.parse import urlparse
 
 import httpx
 
-# Ranges ipaddress does not report as private, but which routinely front
-# internal infrastructure.
+# 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
+
+
+class HostResolutionError(Exception):
+    """The resolver returned no usable addresses for a host."""
+
+    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 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:
+    """
+    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)
+    )
 
 
 def resolve_hostname_ips(hostname: str) -> list[str]:
@@ -82,7 +154,7 @@ def validate_outbound_http_url(
 
     if not allow_internal:
         for ip_str in resolve_hostname_ips(parsed.hostname):
-            if not is_public_ip(ip_str):
+            if not is_public_ip(ipaddress.ip_address(ip_str)):
                 raise ValueError(
                     f"Connection blocked: {parsed.hostname} resolves to a non-public address",
                 )
@@ -107,7 +179,7 @@ def _rewrite_request_to_pinned_ip(
 
     if not allow_internal:
         for ip_str in ips:
-            if not is_public_ip(ip_str):
+            if not is_public_ip(ipaddress.ip_address(ip_str)):
                 raise httpx.ConnectError(
                     f"Connection blocked: {hostname} resolves to a non-public address",
                 )

src/paperless/tests/test_network.py

@@ -1,9 +1,16 @@
+import ipaddress
+import pickle
 from unittest import mock
 
 import httpx
 import pytest
+from celery.utils.serialization import get_pickleable_exception
 
+from paperless.network import BlockReason
+from paperless.network import HostResolutionError
+from paperless.network import OutboundRequestBlockedError
 from paperless.network import PinnedHostHTTPTransport
+from paperless.network import blocked_message
 from paperless.network import is_public_ip
 
 
@@ -51,40 +58,187 @@ def test_pinned_host_transport_rewrites_to_vetted_ip():
     assert response.status_code == 200
 
 
-@pytest.mark.parametrize(
-    "address",
-    [
-        "127.0.0.1",
-        "10.0.0.5",
-        "169.254.169.254",
-        "::1",
-        "fc00::1",
-        "fe80::1",
-        # RFC 6598 shared address space, incl. both edges of the /10
-        "100.64.0.0",
-        "100.64.0.1",
-        "100.127.255.255",
-        # RFC 6052 NAT64 well-known prefix, embedding 127.0.0.1 and 10.0.0.5
-        "64:ff9b::7f00:1",
-        "64:ff9b::a00:5",
-    ],
-)
-def test_is_public_ip_blocks_non_public_addresses(address):
-    assert not is_public_ip(address)
+class TestIsPublicIp:
+    @pytest.mark.parametrize(
+        "address",
+        [
+            pytest.param("127.0.0.1", id="ipv4-loopback"),
+            pytest.param("10.0.0.5", id="ipv4-private"),
+            pytest.param("169.254.169.254", id="ipv4-link-local-metadata"),
+            pytest.param("::1", id="ipv6-loopback"),
+            pytest.param("fc00::1", id="ipv6-unique-local"),
+            pytest.param("fe80::1", id="ipv6-link-local"),
+            pytest.param("100.64.0.0", id="cgnat-first"),
+            pytest.param("100.127.255.255", id="cgnat-last"),
+            pytest.param("64:ff9b::", id="nat64-well-known-first"),
+            pytest.param("64:ff9b::7f00:1", id="nat64-wrapping-loopback"),
+            pytest.param("64:ff9b::ffff:ffff", id="nat64-well-known-last"),
+            pytest.param("224.0.0.1", id="ipv4-multicast"),
+            pytest.param("ff02::1", id="ipv6-multicast"),
+            pytest.param("0.0.0.0", id="ipv4-unspecified"),
+            pytest.param("::", id="ipv6-unspecified"),
+            pytest.param("::ffff:127.0.0.1", id="ipv4-mapped-loopback"),
+            pytest.param("::ffff:169.254.169.254", id="ipv4-mapped-metadata"),
+            pytest.param("::ffff:10.0.0.1", id="ipv4-mapped-private"),
+            pytest.param("::ffff:100.64.0.1", id="ipv4-mapped-cgnat"),
+        ],
+    )
+    def test_rejects_non_public_addresses(self, address: str) -> None:
+        """
+        GIVEN:
+            - An address that is internal, multicast, or reaches internal hosts
+        WHEN:
+            - It is classified
+        THEN:
+            - It is not public
+        """
+        assert not is_public_ip(ipaddress.ip_address(address))
+
+    @pytest.mark.parametrize(
+        "address",
+        [
+            pytest.param("8.8.8.8", id="ipv4-public"),
+            pytest.param("2606:4700:4700::1111", id="ipv6-public"),
+            pytest.param("100.63.255.255", id="below-cgnat"),
+            pytest.param("100.128.0.0", id="above-cgnat"),
+            pytest.param("64:ff9a:ffff:ffff:ffff:ffff:ffff:ffff", id="below-nat64"),
+            pytest.param("64:ff9b::1:0:0", id="above-nat64-well-known"),
+        ],
+    )
+    def test_accepts_public_addresses(self, address: str) -> None:
+        """
+        GIVEN:
+            - A globally routable unicast address outside our extra ranges
+        WHEN:
+            - It is classified
+        THEN:
+            - It is public
+        """
+        assert is_public_ip(ipaddress.ip_address(address))
 
 
-@pytest.mark.parametrize(
-    "address",
-    [
-        "8.8.8.8",
-        "142.250.185.196",
-        "2606:4700:4700::1111",
-        # just outside the ranges above, must stay reachable
-        "100.63.255.255",
-        "100.128.0.0",
-        "64:ff9a::1",
-        "64:ff9c::1",
-    ],
-)
-def test_is_public_ip_allows_public_addresses(address):
-    assert is_public_ip(address)
+def _sample_errors() -> list[object]:
+    return [
+        pytest.param(
+            OutboundRequestBlockedError(
+                host="example.com",
+                port=443,
+                reason=BlockReason.NON_PUBLIC_ADDRESS,
+                address=ipaddress.ip_address("10.0.0.1"),
+            ),
+            id="non-public-address",
+        ),
+        pytest.param(
+            OutboundRequestBlockedError(
+                host="/run/app.sock",
+                port=None,
+                reason=BlockReason.UNIX_SOCKET,
+            ),
+            id="unix-socket",
+        ),
+        pytest.param(
+            HostResolutionError(
+                host="missing.example",
+                detail="Name or service not known",
+            ),
+            id="unresolvable",
+        ),
+    ]
+
+
+class TestOutboundErrors:
+    @pytest.mark.parametrize("error", _sample_errors())
+    def test_round_trips_through_pickle(self, error: Exception) -> None:
+        """
+        GIVEN:
+            - An outbound error with keyword-only fields
+        WHEN:
+            - It is pickled and unpickled
+        THEN:
+            - Type, fields and message are preserved
+        """
+        restored = pickle.loads(pickle.dumps(error))
+
+        assert type(restored) is type(error)
+        assert vars(restored) == vars(error)
+        assert str(restored) == str(error)
+
+    @pytest.mark.parametrize("error", _sample_errors())
+    def test_celery_keeps_the_original_exception(self, error: Exception) -> None:
+        """
+        GIVEN:
+            - An outbound error raised from a Celery task
+        WHEN:
+            - Celery prepares it for failure handling
+        THEN:
+            - The exception itself is kept, not an unpickleable wrapper
+        """
+        assert get_pickleable_exception(error) is error
+
+    def test_message_names_the_destination_but_not_the_address(self) -> None:
+        """
+        GIVEN:
+            - A block for a non-public address
+        WHEN:
+            - The error is rendered
+        THEN:
+            - Host, port and reason are in the message
+            - The resolved internal address is not, so it never reaches logs or
+              stored task results
+        """
+        error = OutboundRequestBlockedError(
+            host="example.com",
+            port=443,
+            reason=BlockReason.NON_PUBLIC_ADDRESS,
+            address=ipaddress.ip_address("10.0.0.1"),
+        )
+
+        message = str(error)
+
+        assert "example.com:443" in message
+        assert "non_public_address" in message
+        assert "10.0.0.1" not in message
+
+    @pytest.mark.parametrize(
+        ("error", "expected"),
+        [
+            pytest.param(
+                OutboundRequestBlockedError(
+                    host="internal.example",
+                    port=443,
+                    reason=BlockReason.NON_PUBLIC_ADDRESS,
+                    address=ipaddress.ip_address("10.0.0.1"),
+                ),
+                "Connection blocked: internal.example resolves to a non-public address",
+                id="non-public-address",
+            ),
+            pytest.param(
+                HostResolutionError(host="missing.example", detail="NXDOMAIN"),
+                "Could not resolve hostname: missing.example",
+                id="unresolvable",
+            ),
+            pytest.param(
+                OutboundRequestBlockedError(
+                    host="/run/app.sock",
+                    port=None,
+                    reason=BlockReason.UNIX_SOCKET,
+                ),
+                "Connection blocked: unix sockets are not permitted",
+                id="unix-socket",
+            ),
+        ],
+    )
+    def test_blocked_message(
+        self,
+        error: OutboundRequestBlockedError | HostResolutionError,
+        expected: str,
+    ) -> None:
+        """
+        GIVEN:
+            - An outbound error
+        WHEN:
+            - The user-facing message is requested
+        THEN:
+            - It matches the established wording
+        """
+        assert blocked_message(error) == expected

src/paperless_mail/mail.py

@@ -1,5 +1,6 @@
 import datetime
 import imaplib
+import ipaddress
 import itertools
 import logging
 import socket
@@ -527,7 +528,7 @@ def get_mailbox(server, port, security) -> MailBox:
             raise MailError(str(e)) from e
 
         for ip_str in pinned_ips:
-            if not is_public_ip(ip_str):
+            if not is_public_ip(ipaddress.ip_address(ip_str)):
                 raise MailError(
                     f"Connection blocked: {server} resolves to a non-public address",
                 )