From d414297c0571b8d0283fdac7455e7885d6ddba46 Mon Sep 17 00:00:00 2001 From: stumpylog <797416+stumpylog@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:01:01 -0700 Subject: [PATCH] Fix: Validate IP literals from the resolver's answer only Outbound host checks parsed IP literals themselves and skipped the resolver for them. That second parser is what let a host such as 8.8.8.8%2eexample pass as the public address 8.8.8.8, and even with zone ids limited to IPv6 it remains one more place where the checked host can be read differently from the connected one. The separate literal parsing is removed. Every host now goes to getaddrinfo, which answers numeric literals itself without a lookup, and only the addresses it returns are classified. Zone ids are still dropped from the resolver's answers, where a scoped IPv6 literal comes back as fe80::1%1. Co-Authored-By: Claude Opus 5 --- src/paperless/network.py | 24 +++--------- src/paperless/tests/test_network.py | 58 +++++++++++------------------ 2 files changed, 28 insertions(+), 54 deletions(-) diff --git a/src/paperless/network.py b/src/paperless/network.py index 072ee1c4b..541c086be 100644 --- a/src/paperless/network.py +++ b/src/paperless/network.py @@ -127,22 +127,12 @@ _agetaddrinfo = anyio.getaddrinfo _monotonic = time.monotonic -def _parse_ip_literal(host: str) -> IPAddress | None: - address, zone_sep, _zone = host.partition("%") - try: - parsed = ipaddress.ip_address(address) - except ValueError: - return None - # Zone ids exist only on IPv6; anything else after "%" is not a literal. - if zone_sep and parsed.version != 6: - return None - return parsed - - 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: @@ -174,10 +164,11 @@ 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. """ - literal = _parse_ip_literal(host) - if literal is not None: - return _require_public(host, port, (literal,)) try: infos = _getaddrinfo(host, port, type=socket.SOCK_STREAM) except (OSError, UnicodeError) as e: @@ -190,9 +181,6 @@ async def aresolve_public_addresses( port: int | None, ) -> tuple[IPAddress, ...]: """Async variant of resolve_public_addresses.""" - literal = _parse_ip_literal(host) - if literal is not None: - return _require_public(host, port, (literal,)) try: infos = await _agetaddrinfo(host, port, type=socket.SOCK_STREAM) except (OSError, UnicodeError) as e: diff --git a/src/paperless/tests/test_network.py b/src/paperless/tests/test_network.py index e5d6cb300..31e30c583 100644 --- a/src/paperless/tests/test_network.py +++ b/src/paperless/tests/test_network.py @@ -240,38 +240,46 @@ def _answer(mocker: MockerFixture, *addresses: str) -> MagicMock: class TestResolvePublicAddresses: - def test_ip_literal_skips_dns(self, mocker: MockerFixture) -> None: + def test_ip_literal_is_validated_from_resolver_answer( + self, + mocker: MockerFixture, + ) -> None: """ GIVEN: - - A public IP literal + - A public IP literal, which the resolver answers with itself WHEN: - It is resolved THEN: - - It is returned without a resolver call + - The literal is passed to the resolver and its answer returned """ - resolver = _answer(mocker) + resolver = _answer(mocker, "93.184.216.34") assert resolve_public_addresses("93.184.216.34", 443) == ( ipaddress.ip_address("93.184.216.34"), ) - resolver.assert_not_called() + resolver.assert_called_once_with("93.184.216.34", 443, type=socket.SOCK_STREAM) - def test_private_ip_literal_is_blocked(self, mocker: MockerFixture) -> None: + def test_private_ip_literal_is_blocked_from_resolver_answer( + self, + mocker: MockerFixture, + ) -> None: """ GIVEN: - - A private IP literal + - A private IP literal, which the resolver answers with itself WHEN: - It is resolved THEN: - - It is blocked as a non-public address + - The literal is passed to the resolver and blocked as a + non-public address """ - _answer(mocker) + resolver = _answer(mocker, "10.0.0.1") with pytest.raises(OutboundRequestBlockedError) as exc_info: resolve_public_addresses("10.0.0.1", 443) assert exc_info.value.reason is BlockReason.NON_PUBLIC_ADDRESS assert exc_info.value.address == ipaddress.ip_address("10.0.0.1") + resolver.assert_called_once_with("10.0.0.1", 443, type=socket.SOCK_STREAM) def test_asks_for_stream_sockets_on_the_port(self, mocker: MockerFixture) -> None: """ @@ -383,13 +391,12 @@ class TestResolvePublicAddresses: ) -> None: """ GIVEN: - - A dotted quad followed by "%" and more text, which is not an IP - literal since zone ids exist only on IPv6 + - A dotted quad followed by "%" and more text - A resolver answering with a public address WHEN: - It is resolved THEN: - - The whole host is looked up as a name and its answer returned + - The whole host is passed to the resolver and its answer returned """ resolver = _answer(mocker, "93.184.216.34") @@ -413,7 +420,7 @@ class TestResolvePublicAddresses: WHEN: - It is resolved THEN: - - The name is blocked instead of passing as the public literal + - The name is blocked, not taken as the public address before "%" """ resolver = _answer(mocker, "169.254.169.254") @@ -423,27 +430,6 @@ class TestResolvePublicAddresses: assert exc_info.value.address == ipaddress.ip_address("169.254.169.254") resolver.assert_called_once() - def test_scoped_ipv6_literal_is_blocked_without_dns( - self, - mocker: MockerFixture, - ) -> None: - """ - GIVEN: - - A link-local IPv6 literal with a zone id - WHEN: - - It is resolved - THEN: - - It is parsed as the IPv6 literal and blocked without a resolver - call - """ - resolver = _answer(mocker, "93.184.216.34") - - with pytest.raises(OutboundRequestBlockedError) as exc_info: - resolve_public_addresses("fe80::1%eth0", 443) - - assert exc_info.value.address == ipaddress.ip_address("fe80::1") - resolver.assert_not_called() - class TestAsyncResolvePublicAddresses: @pytest.fixture(autouse=True) @@ -493,7 +479,7 @@ class TestAsyncResolvePublicAddresses: WHEN: - It is resolved asynchronously THEN: - - The whole host is looked up as a name and its answer returned + - The whole host is passed to the resolver and its answer returned """ resolver = mocker.patch( "paperless.network._agetaddrinfo", @@ -521,7 +507,7 @@ class TestAsyncResolvePublicAddresses: WHEN: - It is resolved asynchronously THEN: - - The name is blocked instead of passing as the public literal + - The name is blocked, not taken as the public address before "%" """ resolver = mocker.patch( "paperless.network._agetaddrinfo",