Coverage for paperless/network.py: 21%
93 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 09:07 +0000
1import ipaddress
2import socket
3from collections.abc import Collection
4from urllib.parse import ParseResult
5from urllib.parse import urlparse
7import httpx
9# Ranges ipaddress does not report as private, but which routinely front
10# internal infrastructure.
11_NON_PUBLIC_NETWORKS = (
12 # RFC 6598 shared address space: ISP CGNAT, and the default pod/service
13 # CIDR on several managed Kubernetes offerings.
14 ipaddress.ip_network("100.64.0.0/10"),
15 # RFC 6052 NAT64 well-known prefix: 64:ff9b::7f00:1 is 127.0.0.1 wherever
16 # a NAT64 gateway exists.
17 ipaddress.ip_network("64:ff9b::/96"),
18)
21def is_public_ip(ip: str | int) -> bool:
22 try:
23 obj = ipaddress.ip_address(ip)
24 return not (
25 obj.is_private
26 or obj.is_loopback
27 or obj.is_link_local
28 or obj.is_multicast
29 or obj.is_unspecified
30 or any(obj in network for network in _NON_PUBLIC_NETWORKS)
31 )
32 except ValueError: # pragma: no cover
33 return False
36def resolve_hostname_ips(hostname: str) -> list[str]:
37 try:
38 addr_info = socket.getaddrinfo(hostname, None)
39 except socket.gaierror as e:
40 raise ValueError(f"Could not resolve hostname: {hostname}") from e
42 ips = [info[4][0] for info in addr_info if info and info[4]]
43 if not ips:
44 raise ValueError(f"Could not resolve hostname: {hostname}")
45 return ips
48def format_host_for_url(host: str) -> str:
49 """
50 Format IP address for URL use (wrap IPv6 in brackets).
51 """
52 try:
53 ip_obj = ipaddress.ip_address(host)
54 if ip_obj.version == 6:
55 return f"[{host}]"
56 return host
57 except ValueError:
58 return host
61def validate_outbound_http_url(
62 url: str,
63 *,
64 allowed_schemes: Collection[str] = ("http", "https"),
65 allowed_ports: Collection[int] | None = None,
66 allow_internal: bool = False,
67) -> ParseResult:
68 parsed = urlparse(url)
69 scheme = parsed.scheme.lower()
71 if scheme not in allowed_schemes or not parsed.hostname: 71 ↛ 74line 71 didn't jump to line 74 because the condition on line 71 was always true
72 raise ValueError("Invalid URL scheme or hostname.")
74 default_port = 443 if scheme == "https" else 80
75 try:
76 port = parsed.port or default_port
77 except ValueError as e:
78 raise ValueError("Invalid URL scheme or hostname.") from e
80 if allowed_ports and port not in allowed_ports:
81 raise ValueError("Destination port not permitted.")
83 if not allow_internal:
84 for ip_str in resolve_hostname_ips(parsed.hostname):
85 if not is_public_ip(ip_str):
86 raise ValueError(
87 f"Connection blocked: {parsed.hostname} resolves to a non-public address",
88 )
90 return parsed
93def _rewrite_request_to_pinned_ip(
94 request: httpx.Request,
95 *,
96 allow_internal: bool,
97) -> httpx.Request:
98 hostname = request.url.host
100 if not hostname:
101 raise httpx.ConnectError("No hostname in request URL")
103 try:
104 ips = resolve_hostname_ips(hostname)
105 except ValueError as e:
106 raise httpx.ConnectError(str(e)) from e
108 if not allow_internal:
109 for ip_str in ips:
110 if not is_public_ip(ip_str):
111 raise httpx.ConnectError(
112 f"Connection blocked: {hostname} resolves to a non-public address",
113 )
115 ip_str = ips[0]
116 formatted_ip = format_host_for_url(ip_str)
118 new_headers = httpx.Headers(request.headers)
119 if "host" in new_headers:
120 del new_headers["host"]
121 host_header = format_host_for_url(hostname)
122 default_port = 443 if request.url.scheme == "https" else 80
123 if request.url.port and request.url.port != default_port:
124 host_header = f"{host_header}:{request.url.port}"
125 new_headers["Host"] = host_header
126 new_url = request.url.copy_with(host=formatted_ip)
128 rewritten_request = httpx.Request(
129 method=request.method,
130 url=new_url,
131 headers=new_headers,
132 stream=request.stream,
133 extensions=request.extensions,
134 )
135 rewritten_request.extensions["sni_hostname"] = hostname
137 return rewritten_request
140class PinnedHostHTTPTransport(httpx.HTTPTransport):
141 """
142 HTTP transport that resolves/validates hostnames per request and connects to
143 a vetted IP while preserving the original Host header and TLS SNI hostname.
144 """
146 def __init__(
147 self,
148 *args,
149 allow_internal: bool = False,
150 **kwargs,
151 ) -> None:
152 super().__init__(*args, **kwargs)
153 self.allow_internal = allow_internal
155 def handle_request(self, request: httpx.Request) -> httpx.Response:
156 request = _rewrite_request_to_pinned_ip(
157 request,
158 allow_internal=self.allow_internal,
159 )
160 return super().handle_request(request)
163class PinnedHostAsyncHTTPTransport(httpx.AsyncHTTPTransport):
164 """
165 Async variant of PinnedHostHTTPTransport.
166 """
168 def __init__(
169 self,
170 *args,
171 allow_internal: bool = False,
172 **kwargs,
173 ) -> None:
174 super().__init__(*args, **kwargs)
175 self.allow_internal = allow_internal
177 async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
178 request = _rewrite_request_to_pinned_ip(
179 request,
180 allow_internal=self.allow_internal,
181 )
182 return await super().handle_async_request(request)
185def create_pinned_httpx_client(
186 url: str,
187 *,
188 allow_internal: bool = False,
189 **kwargs,
190) -> httpx.Client:
191 validate_outbound_http_url(url, allow_internal=allow_internal)
192 return httpx.Client(
193 transport=PinnedHostHTTPTransport(allow_internal=allow_internal),
194 **kwargs,
195 )
198def create_pinned_async_httpx_client(
199 url: str,
200 *,
201 allow_internal: bool = False,
202 **kwargs,
203) -> httpx.AsyncClient:
204 validate_outbound_http_url(url, allow_internal=allow_internal)
205 return httpx.AsyncClient(
206 transport=PinnedHostAsyncHTTPTransport(allow_internal=allow_internal),
207 **kwargs,
208 )