amorfo77/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2026-09-25 02:40:32 +00:00
Code Issues Projects Releases Wiki Activity

Fix: Validate IP literals from the resolver's answer only

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
stumpylog
2026-09-22 12:49:06 -07:00
co-authored by Claude Opus 5
parent d53a930aed
commit d414297c05
2 changed files with 28 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/paperless/network.py
+6 -18
View File
@@ -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:
src/paperless/tests/test_network.py
+22 -36
View File
@@ -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",