Coverage for netbox_data_import/inference_trust.py: 96.99%
166 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The `api_root` trust boundary (specification 8.3).
5`api_root` names a destination the NetBox server itself calls, so the same rules apply at the form
6boundary and again at request time. Resolution is a separate step: the allowlist approves a name,
7and `assert_resolved_address_allowed` decides whether the address that name answered with is one
8NetBox may reach.
10Specification 8.3 asks for a recheck after resolution. The inference and Vault request paths hand
11that result to the address-pinned transport, which connects without resolving the hostname again.
12The transport keeps the original HTTP Host header and TLS hostname verification.
14A bearer credential may travel in cleartext over HTTP to an approved local endpoint.
15`is_local_endpoint` covers loopback, private and link-local addresses, so this includes a private
16network, not only loopback. This residual risk is an accepted deployment choice.
17"""
19import ipaddress
20import json
21import os
22import signal
23import socket
24import subprocess
25import sys
27from collections.abc import Iterable, Sequence
28from contextlib import suppress
29from pathlib import Path
30from urllib.parse import urlsplit, urlunsplit
32from .inference_transport import WallClockDeadline, WallClockDeadlineExceeded
34SUPPORTED_SCHEMES = ("https", "http")
35DNS_WORKER_COMMAND = ("-I", "-S", str(Path(__file__).with_name("_dns_worker.py")))
36# The deployment owns the token; the plugin never stores one.
37VAULT_TOKEN_ENVIRONMENT_VARIABLE = "VAULT_TOKEN" # noqa: S105 - This is an environment variable name.
39# Link-local already covers 169.254.0.0/16, so these name the destinations worth their own message.
40CLOUD_METADATA_ADDRESSES = frozenset(
41 {
42 "169.254.169.254", # AWS, Azure, GCP, OpenStack
43 "169.254.170.2", # AWS ECS task metadata
44 "100.100.100.200", # Alibaba Cloud
45 "fd00:ec2::254", # AWS IMDSv2 over IPv6
46 }
47)
49# Parsed from the exported strings, so an alternate spelling of the same address cannot slip past.
50_CLOUD_METADATA_IPS = frozenset(ipaddress.ip_address(address) for address in CLOUD_METADATA_ADDRESSES)
52LOCAL_HOST_NAMES = frozenset({"localhost", "localhost.localdomain"})
55class InvalidInferenceConfiguration(ValueError):
56 """An Inference Backend configuration value the deployment may not use."""
59def split_url(value: str, setting: str, *, quote_value: bool = True):
60 """Return the parsed URL, rejecting a value that is not a usable absolute URL.
62 `quote_value=False` keeps the rejected value out of the message, for a setting that can carry
63 a secret in its userinfo and whose failures are persisted.
64 """
65 if not isinstance(value, str):
66 raise InvalidInferenceConfiguration(f"'{setting}' must be a string, got {type(value).__name__}.")
67 got = f" Got '{value}'." if quote_value else ""
68 try:
69 # urlsplit raises a bare ValueError on malformed bracket syntax, which callers do not catch.
70 parts = urlsplit(value.strip())
71 except ValueError as exc:
72 raise InvalidInferenceConfiguration(f"'{setting}' is not a usable URL.{got}") from exc
73 if not parts.scheme:
74 raise InvalidInferenceConfiguration(f"'{setting}' must name a scheme, for example https://host:443.{got}")
75 if parts.scheme.lower() not in SUPPORTED_SCHEMES:
76 scheme = f" Got '{parts.scheme}'." if quote_value else ""
77 raise InvalidInferenceConfiguration(
78 f"'{setting}' must use the scheme {' or '.join(SUPPORTED_SCHEMES)}.{scheme}"
79 )
80 if parts.username or parts.password:
81 raise InvalidInferenceConfiguration(f"'{setting}' must not carry a credential in its userinfo component.")
82 if not parts.hostname:
83 raise InvalidInferenceConfiguration(f"'{setting}' must name a host.{got}")
84 if "*" in parts.netloc:
85 raise InvalidInferenceConfiguration(f"'{setting}' must not use a wildcard.{got}")
86 return parts
89def origin_of(value: str, setting: str) -> str:
90 """Return the scheme, host and port of one absolute URL, lower-cased."""
91 parts = split_url(value, setting)
92 unusable = f"'{setting}' must name a port between 1 and 65535. Got '{value}'."
93 try:
94 # urlsplit defers the cast, so a non-numeric or out-of-range port raises only here.
95 port = parts.port
96 except ValueError as exc:
97 raise InvalidInferenceConfiguration(unusable) from exc
98 if port is None:
99 raise InvalidInferenceConfiguration(
100 f"'{setting}' must name an explicit port, for example https://host:443. Got '{value}'."
101 )
102 if port < 1:
103 # urlsplit returns zero rather than raising, and no connection can use it.
104 raise InvalidInferenceConfiguration(unusable)
105 host = parts.hostname.lower()
106 # urlsplit strips the brackets, and only an IPv6 literal can leave a colon in a hostname.
107 if ":" in host:
108 host = f"[{host}]"
109 return urlunsplit((parts.scheme.lower(), f"{host}:{port}", "", "", ""))
112def validate_origin(value: str, setting: str) -> str:
113 """Return one exact allowlist origin, rejecting a path, query, fragment or wildcard."""
114 parts = split_url(value, setting)
115 if parts.path or parts.query or parts.fragment:
116 raise InvalidInferenceConfiguration(
117 f"'{setting}' entries carry no path, query or fragment component. Got '{value}'."
118 )
119 return origin_of(value, setting)
122def _allowlist_entry_for(origin: str, allowlist: Iterable[str]) -> str | None:
123 """Return the allowlist entry that approves *origin*, or None."""
124 for entry in allowlist:
125 if origin_of(entry, setting="allowlist") == origin:
126 return entry
127 return None
130def is_local_endpoint(origin: str) -> bool:
131 """Return whether an origin literally names a local address, which is the approval to reach one."""
132 host = urlsplit(origin).hostname or ""
133 if host.lower() in LOCAL_HOST_NAMES:
134 return True
135 try:
136 address = ipaddress.ip_address(host)
137 except ValueError:
138 return False
139 return address.is_loopback or address.is_private or address.is_link_local
142def _assert_origin_approved(url: str, allowlist: Sequence[str], authentication: str, setting: str) -> str:
143 """Reject a URL whose origin the deployment has not approved, or whose scheme it may not use."""
144 parts = split_url(url, setting)
145 origin = origin_of(url, setting)
146 if _allowlist_entry_for(origin, allowlist) is None:
147 raise InvalidInferenceConfiguration(
148 f"'{setting}' origin '{origin}' is not on the inference_backend_origin_allowlist."
149 )
150 if parts.scheme.lower() != "https" and authentication == "bearer" and not is_local_endpoint(origin):
151 raise InvalidInferenceConfiguration(
152 f"'{setting}' must use https when authentication is bearer, unless the allowlist approves it as a "
153 f"local endpoint. Got '{url}'."
154 )
155 return origin
158def validate_api_root(
159 api_root: str,
160 allowlist: Sequence[str],
161 authentication: str = "bearer",
162 setting: str = "api_root",
163) -> str:
164 """Return the validated API root, rejecting an origin the deployment has not approved."""
165 api_root = api_root.strip() if isinstance(api_root, str) else api_root
166 parts = split_url(api_root, setting)
167 if parts.path.endswith("/"):
168 raise InvalidInferenceConfiguration(
169 f"'{setting}' must have no trailing slash. The client appends /chat/completions. Got '{api_root}'."
170 )
171 # A bare `?` or `#` splits into an empty component, so the raw value is what shows it.
172 if parts.query or parts.fragment or "?" in api_root or "#" in api_root:
173 raise InvalidInferenceConfiguration(
174 f"'{setting}' carries no query or fragment component. The client appends /chat/completions to the "
175 f"path, which either one would swallow. Got '{api_root}'."
176 )
177 _assert_origin_approved(api_root, allowlist, authentication, setting)
178 return api_root
181def assert_resolved_address_allowed(
182 api_root: str,
183 allowlist: Sequence[str],
184 addresses: Iterable[str],
185 setting: str = "api_root",
186) -> None:
187 """Reject a destination whose resolved address NetBox must not reach."""
188 origin = origin_of(api_root, setting)
189 approved_local = is_local_endpoint(origin) and _allowlist_entry_for(origin, allowlist) is not None
190 resolved = tuple(addresses)
191 if not resolved:
192 raise InvalidInferenceConfiguration(f"'{setting}' host '{origin}' resolved to no address.")
193 for candidate in resolved:
194 address = ipaddress.ip_address(candidate)
195 if address in _CLOUD_METADATA_IPS:
196 raise InvalidInferenceConfiguration(
197 f"'{setting}' resolves to the cloud metadata address {candidate}, which NetBox must not call."
198 )
199 if approved_local:
200 continue
201 category = next(
202 (
203 label
204 for matches, label in (
205 (address.is_loopback, "loopback"),
206 (address.is_link_local, "link-local"),
207 (address.is_reserved, "reserved"),
208 (address.is_private, "private"),
209 )
210 if matches
211 ),
212 None,
213 )
214 if category:
215 raise InvalidInferenceConfiguration(
216 f"'{setting}' resolves to the {category} address {candidate}, which the allowlist does not approve "
217 f"as a local endpoint."
218 )
221def _resolution_failure(setting: str, host: str) -> InvalidInferenceConfiguration:
222 """Return the public failure for a host that did not produce usable addresses."""
223 return InvalidInferenceConfiguration(f"'{setting}' host '{host}' could not be resolved.")
226def _worker_addresses(stdout: str) -> tuple[str, ...] | None:
227 """Return validated worker output, or None when the child did not produce addresses."""
228 try:
229 payload = json.loads(stdout)
230 except (TypeError, ValueError):
231 return None
232 if not isinstance(payload, list) or not all(isinstance(address, str) for address in payload):
233 return None
234 try:
235 for address in payload:
236 ipaddress.ip_address(address)
237 except ValueError:
238 return None
239 return tuple(payload)
242def _kill_and_reap(process: subprocess.Popen[str]) -> None:
243 """Stop the worker and anything it started, then reap it, without replacing the caller's failure."""
244 with suppress(OSError):
245 # The worker owns its process group, so this also reaches a helper the resolver started.
246 os.killpg(process.pid, signal.SIGKILL)
247 with suppress(OSError):
248 process.kill()
249 with suppress(OSError, subprocess.SubprocessError):
250 process.wait()
253def _communicate_before_deadline(
254 process: subprocess.Popen[str],
255 request: str,
256 deadline: WallClockDeadline,
257 setting: str,
258 host: str,
259) -> str:
260 """Exchange one worker request within the deadline and keep failures typed."""
261 timeout = deadline.remaining()
262 try:
263 stdout, _stderr = process.communicate(input=request, timeout=timeout)
264 except subprocess.TimeoutExpired:
265 raise WallClockDeadlineExceeded("The operation exceeded its overall time limit.") from None
266 except Exception as exc:
267 raise _resolution_failure(setting, host) from exc
268 return stdout
271def _resolve_addresses_before_deadline(
272 host: str,
273 port: int,
274 setting: str,
275 deadline: WallClockDeadline,
276) -> tuple[str, ...]:
277 """Resolve one host in a child that the parent can terminate and reap."""
278 child_timeout = deadline.remaining()
279 if not sys.executable:
280 raise InvalidInferenceConfiguration(
281 f"'{setting}' cannot be resolved before its deadline because the Python executable is unavailable."
282 )
283 environment = os.environ.copy()
284 environment.pop(VAULT_TOKEN_ENVIRONMENT_VARIABLE, None)
285 try:
286 process = subprocess.Popen( # noqa: S603 - the executable and arguments are module-owned
287 (sys.executable, *DNS_WORKER_COMMAND),
288 stdin=subprocess.PIPE,
289 stdout=subprocess.PIPE,
290 stderr=subprocess.DEVNULL,
291 text=True,
292 env=environment,
293 # Its own process group, so the deadline can stop descendants and not just the worker.
294 start_new_session=True,
295 )
296 except Exception as exc:
297 raise _resolution_failure(setting, host) from exc
298 try:
299 request = json.dumps((host, port, child_timeout))
300 stdout = _communicate_before_deadline(process, request, deadline, setting, host)
301 addresses = _worker_addresses(stdout)
302 deadline.remaining()
303 finally:
304 _kill_and_reap(process)
305 if process.returncode != 0 or addresses is None:
306 raise _resolution_failure(setting, host)
307 return addresses
310def resolve_addresses(
311 api_root: str,
312 setting: str = "api_root",
313 deadline: WallClockDeadline | None = None,
314) -> tuple[str, ...]:
315 """Return every address the API root's host answers with."""
316 parts = split_url(api_root, setting)
317 port = parts.port or (443 if parts.scheme.lower() == "https" else 80)
318 if deadline is not None:
319 return _resolve_addresses_before_deadline(parts.hostname, port, setting, deadline)
320 try:
321 answers = socket.getaddrinfo(parts.hostname, port, proto=socket.IPPROTO_TCP)
322 except OSError as exc:
323 raise _resolution_failure(setting, parts.hostname) from exc
324 return tuple(dict.fromkeys(str(answer[4][0]) for answer in answers))
327__all__ = (
328 "CLOUD_METADATA_ADDRESSES",
329 "InvalidInferenceConfiguration",
330 "assert_resolved_address_allowed",
331 "is_local_endpoint",
332 "origin_of",
333 "resolve_addresses",
334 "split_url",
335 "validate_api_root",
336 "validate_origin",
337)