Coverage for netbox_data_import/inference_trust.py: 98%
104 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +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, which is what this module performs. It does
11not pin the socket to the address it checked, so a name that answers differently between the check
12and the connect is a residual window. Closing it needs an address-pinned transport with its own TLS
13hostname handling, which is tracked in issue #147.
15A bearer credential may travel in cleartext over HTTP to an approved local endpoint.
16`is_local_endpoint` covers loopback, private and link-local addresses, so this includes a private
17network, not only loopback. This residual risk is an accepted deployment choice.
18"""
20import ipaddress
21import socket
23from collections.abc import Iterable, Sequence
24from urllib.parse import urlsplit, urlunsplit
26SUPPORTED_SCHEMES = ("https", "http")
28# Link-local already covers 169.254.0.0/16, so these name the destinations worth their own message.
29CLOUD_METADATA_ADDRESSES = frozenset(
30 {
31 "169.254.169.254", # AWS, Azure, GCP, OpenStack
32 "169.254.170.2", # AWS ECS task metadata
33 "100.100.100.200", # Alibaba Cloud
34 "fd00:ec2::254", # AWS IMDSv2 over IPv6
35 }
36)
38# Parsed from the exported strings, so an alternate spelling of the same address cannot slip past.
39_CLOUD_METADATA_IPS = frozenset(ipaddress.ip_address(address) for address in CLOUD_METADATA_ADDRESSES)
41LOCAL_HOST_NAMES = frozenset({"localhost", "localhost.localdomain"})
44class InvalidInferenceConfiguration(ValueError):
45 """An Inference Backend configuration value the deployment may not use."""
48def split_url(value: str, setting: str, *, quote_value: bool = True):
49 """Return the parsed URL, rejecting a value that is not a usable absolute URL.
51 `quote_value=False` keeps the rejected value out of the message, for a setting that can carry
52 a secret in its userinfo and whose failures are persisted.
53 """
54 if not isinstance(value, str):
55 raise InvalidInferenceConfiguration(f"'{setting}' must be a string, got {type(value).__name__}.")
56 got = f" Got '{value}'." if quote_value else ""
57 try:
58 # urlsplit raises a bare ValueError on malformed bracket syntax, which callers do not catch.
59 parts = urlsplit(value.strip())
60 except ValueError as exc:
61 raise InvalidInferenceConfiguration(f"'{setting}' is not a usable URL.{got}") from exc
62 if not parts.scheme:
63 raise InvalidInferenceConfiguration(f"'{setting}' must name a scheme, for example https://host:443.{got}")
64 if parts.scheme.lower() not in SUPPORTED_SCHEMES:
65 scheme = f" Got '{parts.scheme}'." if quote_value else ""
66 raise InvalidInferenceConfiguration(
67 f"'{setting}' must use the scheme {' or '.join(SUPPORTED_SCHEMES)}.{scheme}"
68 )
69 if parts.username or parts.password:
70 raise InvalidInferenceConfiguration(f"'{setting}' must not carry a credential in its userinfo component.")
71 if not parts.hostname:
72 raise InvalidInferenceConfiguration(f"'{setting}' must name a host.{got}")
73 if "*" in parts.netloc:
74 raise InvalidInferenceConfiguration(f"'{setting}' must not use a wildcard.{got}")
75 return parts
78def origin_of(value: str, setting: str) -> str:
79 """Return the scheme, host and port of one absolute URL, lower-cased."""
80 parts = split_url(value, setting)
81 unusable = f"'{setting}' must name a port between 1 and 65535. Got '{value}'."
82 try:
83 # urlsplit defers the cast, so a non-numeric or out-of-range port raises only here.
84 port = parts.port
85 except ValueError as exc:
86 raise InvalidInferenceConfiguration(unusable) from exc
87 if port is None:
88 raise InvalidInferenceConfiguration(
89 f"'{setting}' must name an explicit port, for example https://host:443. Got '{value}'."
90 )
91 if port < 1:
92 # urlsplit returns zero rather than raising, and no connection can use it.
93 raise InvalidInferenceConfiguration(unusable)
94 host = parts.hostname.lower()
95 # urlsplit strips the brackets, and only an IPv6 literal can leave a colon in a hostname.
96 if ":" in host:
97 host = f"[{host}]"
98 return urlunsplit((parts.scheme.lower(), f"{host}:{port}", "", "", ""))
101def validate_origin(value: str, setting: str) -> str:
102 """Return one exact allowlist origin, rejecting a path, query, fragment or wildcard."""
103 parts = split_url(value, setting)
104 if parts.path or parts.query or parts.fragment:
105 raise InvalidInferenceConfiguration(
106 f"'{setting}' entries carry no path, query or fragment component. Got '{value}'."
107 )
108 return origin_of(value, setting)
111def _allowlist_entry_for(origin: str, allowlist: Iterable[str]) -> str | None:
112 """Return the allowlist entry that approves *origin*, or None."""
113 for entry in allowlist:
114 if origin_of(entry, setting="allowlist") == origin:
115 return entry
116 return None
119def is_local_endpoint(origin: str) -> bool:
120 """Return whether an origin literally names a local address, which is the approval to reach one."""
121 host = urlsplit(origin).hostname or ""
122 if host.lower() in LOCAL_HOST_NAMES:
123 return True
124 try:
125 address = ipaddress.ip_address(host)
126 except ValueError:
127 return False
128 return address.is_loopback or address.is_private or address.is_link_local
131def _assert_origin_approved(url: str, allowlist: Sequence[str], authentication: str, setting: str) -> str:
132 """Reject a URL whose origin the deployment has not approved, or whose scheme it may not use."""
133 parts = split_url(url, setting)
134 origin = origin_of(url, setting)
135 if _allowlist_entry_for(origin, allowlist) is None:
136 raise InvalidInferenceConfiguration(
137 f"'{setting}' origin '{origin}' is not on the inference_backend_origin_allowlist."
138 )
139 if parts.scheme.lower() != "https" and authentication == "bearer" and not is_local_endpoint(origin):
140 raise InvalidInferenceConfiguration(
141 f"'{setting}' must use https when authentication is bearer, unless the allowlist approves it as a "
142 f"local endpoint. Got '{url}'."
143 )
144 return origin
147def validate_api_root(
148 api_root: str,
149 allowlist: Sequence[str],
150 authentication: str = "bearer",
151 setting: str = "api_root",
152) -> str:
153 """Return the validated API root, rejecting an origin the deployment has not approved."""
154 api_root = api_root.strip() if isinstance(api_root, str) else api_root
155 parts = split_url(api_root, setting)
156 if parts.path.endswith("/"):
157 raise InvalidInferenceConfiguration(
158 f"'{setting}' must have no trailing slash. The client appends /chat/completions. Got '{api_root}'."
159 )
160 # A bare `?` or `#` splits into an empty component, so the raw value is what shows it.
161 if parts.query or parts.fragment or "?" in api_root or "#" in api_root:
162 raise InvalidInferenceConfiguration(
163 f"'{setting}' carries no query or fragment component. The client appends /chat/completions to the "
164 f"path, which either one would swallow. Got '{api_root}'."
165 )
166 _assert_origin_approved(api_root, allowlist, authentication, setting)
167 return api_root
170def assert_resolved_address_allowed(
171 api_root: str,
172 allowlist: Sequence[str],
173 addresses: Iterable[str],
174 setting: str = "api_root",
175) -> None:
176 """Reject a destination whose resolved address NetBox must not reach."""
177 origin = origin_of(api_root, setting)
178 approved_local = is_local_endpoint(origin) and _allowlist_entry_for(origin, allowlist) is not None
179 resolved = tuple(addresses)
180 if not resolved:
181 raise InvalidInferenceConfiguration(f"'{setting}' host '{origin}' resolved to no address.")
182 for candidate in resolved:
183 address = ipaddress.ip_address(candidate)
184 if address in _CLOUD_METADATA_IPS:
185 raise InvalidInferenceConfiguration(
186 f"'{setting}' resolves to the cloud metadata address {candidate}, which NetBox must not call."
187 )
188 if approved_local:
189 continue
190 category = next(
191 (
192 label
193 for matches, label in (
194 (address.is_loopback, "loopback"),
195 (address.is_link_local, "link-local"),
196 (address.is_reserved, "reserved"),
197 (address.is_private, "private"),
198 )
199 if matches
200 ),
201 None,
202 )
203 if category:
204 raise InvalidInferenceConfiguration(
205 f"'{setting}' resolves to the {category} address {candidate}, which the allowlist does not approve "
206 f"as a local endpoint."
207 )
210def resolve_addresses(api_root: str, setting: str = "api_root") -> tuple[str, ...]:
211 """Return every address the API root's host answers with."""
212 parts = split_url(api_root, setting)
213 port = parts.port or (443 if parts.scheme.lower() == "https" else 80)
214 try:
215 answers = socket.getaddrinfo(parts.hostname, port, proto=socket.IPPROTO_TCP)
216 except OSError as exc:
217 raise InvalidInferenceConfiguration(f"'{setting}' host '{parts.hostname}' could not be resolved.") from exc
218 return tuple(dict.fromkeys(str(answer[4][0]) for answer in answers))
221__all__ = (
222 "CLOUD_METADATA_ADDRESSES",
223 "InvalidInferenceConfiguration",
224 "assert_resolved_address_allowed",
225 "is_local_endpoint",
226 "origin_of",
227 "resolve_addresses",
228 "split_url",
229 "validate_api_root",
230 "validate_origin",
231)