Coverage for netbox_data_import/inference_credentials.py: 95.39%
152 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 credential boundary and its Vault KV v2 implementation (specification 8.5, 8.6).
5The seam has four responsibilities: validate a typed reference, resolve it through the selected
6credential backend, return secret material for the lifetime of one outbound request, and classify
7failures. No message built here carries the secret or a Vault response body, so a caller may log
8any failure it catches.
9"""
11import logging
12import os
13import threading
14from urllib.parse import quote
16from collections.abc import Mapping
17from contextlib import contextmanager
18from dataclasses import dataclass
19from types import TracebackType
20from typing import Any, Protocol, Self
22import requests
24from .inference_settings import (
25 CREDENTIAL_REFERENCE_BACKEND,
26 validate_credential_reference,
27 validate_vault_settings,
28)
29from .inference_transport import (
30 ResponseBodyTooLarge,
31 ResponseProcessingFailure,
32 WallClockDeadline,
33 WallClockDeadlineExceeded,
34 is_preconnect_failure,
35 request_to_resolved_address,
36)
37from .inference_trust import InvalidInferenceConfiguration, VAULT_TOKEN_ENVIRONMENT_VARIABLE, resolve_addresses
39TRANSPORT_LOGGER = "urllib3.connectionpool"
40_reading_vault = threading.local()
43class _QuietDuringVaultRead(logging.Filter):
44 """Drop transport records emitted while this thread is reading a credential.
46 urllib3 logs the connection and the request line at DEBUG, which names the Vault address and
47 the KV path of the secret being read. Neither is recoverable by redacting the message, because
48 the address is interpolated into several formats, so the read is silenced for its duration.
49 Every failure this module raises is still typed and still reaches the caller.
50 """
52 def filter(self, record: logging.LogRecord) -> bool:
53 """Return whether one transport record may be emitted."""
54 return not getattr(_reading_vault, "active", False)
57logging.getLogger(TRANSPORT_LOGGER).addFilter(_QuietDuringVaultRead())
60@contextmanager
61def _quiet_transport_logging():
62 """Silence the transport logger for this thread only, for the length of one read."""
63 _reading_vault.active = True
64 try:
65 yield
66 finally:
67 _reading_vault.active = False
70DEFAULT_CONNECT_TIMEOUT = 5
71DEFAULT_READ_TIMEOUT = 60
72# A KV v2 envelope holding one secret is small; anything past this is refused unread.
73VAULT_RESPONSE_LIMIT = 65_536
76class CredentialFailure(Exception):
77 """A credential resolution that failed, carrying its typed category and no secret."""
79 category = "credential_unavailable"
82class CredentialUnavailable(CredentialFailure):
83 """The credential store could not be reached, or held no answer for this reference."""
85 category = "credential_unavailable"
88class CredentialDenied(CredentialFailure):
89 """The credential store refused the read."""
91 category = "credential_denied"
94class InvalidCredentialReference(CredentialFailure):
95 """The reference is not a usable typed reference."""
97 category = "invalid_credential_reference"
100class InvalidSecretMaterial(CredentialFailure):
101 """The store answered, but the named field holds nothing usable as a key."""
103 category = "invalid_secret_material"
106class InvalidCredentialConfiguration(CredentialFailure):
107 """The deployment-owned credential settings cannot be used as given."""
109 category = "invalid_configuration"
112@dataclass(frozen=True)
113class CredentialReference:
114 """One typed Vault KV v2 reference: restricted configuration metadata, never secret material."""
116 backend: str
117 mount: str
118 path: str
119 field: str
121 @classmethod
122 def from_mapping(cls, mapping: Mapping[str, Any]) -> "CredentialReference":
123 """Return the validated reference, rejecting connection data and secret material."""
124 try:
125 validated = validate_credential_reference(mapping)
126 except InvalidInferenceConfiguration as exc:
127 raise InvalidCredentialReference(str(exc)) from exc
128 return cls(
129 backend=validated["backend"],
130 mount=validated["mount"],
131 path=validated["path"],
132 field=validated["field"],
133 )
136class CredentialBackend(Protocol):
137 """Resolve one typed reference into secret material for the lifetime of one request."""
139 name: str
141 def close(self) -> None:
142 """Release resources owned by this backend."""
143 ...
145 def __enter__(self) -> Self: ...
147 def __exit__(
148 self,
149 exc_type: type[BaseException] | None,
150 exc_value: BaseException | None,
151 traceback: TracebackType | None,
152 ) -> None: ...
154 def resolve(self, reference: CredentialReference) -> str:
155 """Return the secret the reference names."""
156 ...
159def _status_failure(status: int) -> CredentialFailure | None:
160 """Return the typed credential failure one Vault status maps to, or None for a usable answer."""
161 # Every 3xx, not a list of them: a 300 or 305 body shaped like KV would read as the secret.
162 if 300 <= status < 400:
163 return InvalidCredentialConfiguration(
164 f"The credential store redirected the read (HTTP {status}). Check the configured vault address."
165 )
166 if status in (401, 403):
167 return CredentialDenied(f"The credential store refused the read (HTTP {status}).")
168 if status == 404:
169 return InvalidCredentialReference("The credential store holds no secret for this Inference Backend.")
170 if status >= 400:
171 return CredentialUnavailable(f"The credential store answered HTTP {status}.")
172 return None
175class VaultKvV2CredentialBackend:
176 """Read one named field from one KV v2 path, through Vault Proxy or a deployment token."""
178 name = CREDENTIAL_REFERENCE_BACKEND
180 def __init__(
181 self,
182 settings: Mapping[str, Any],
183 session: requests.Session | None = None,
184 deadline: WallClockDeadline | None = None,
185 ):
186 try:
187 self._settings = validate_vault_settings(settings)
188 except InvalidInferenceConfiguration as exc:
189 raise InvalidCredentialConfiguration(str(exc)) from exc
190 self._owns_session = session is None
191 self._session = requests.Session() if session is None else session
192 self._deadline = deadline
194 def close(self) -> None:
195 """Close the session only when this backend created it."""
196 if self._owns_session:
197 self._session.close()
199 def __enter__(self) -> Self:
200 return self
202 def __exit__(
203 self,
204 exc_type: type[BaseException] | None,
205 exc_value: BaseException | None,
206 traceback: TracebackType | None,
207 ) -> None:
208 self.close()
210 def _headers(self) -> dict[str, str]:
211 """Return the request headers, reading a token only when the deployment selected one."""
212 headers = {"Accept": "application/json", "X-Vault-Request": "true"}
213 if namespace := self._settings.get("namespace"):
214 headers["X-Vault-Namespace"] = namespace
215 if self._settings.get("auth_method", "proxy") != "token":
216 return headers
217 token = os.environ.get(VAULT_TOKEN_ENVIRONMENT_VARIABLE, "")
218 if not token:
219 raise InvalidCredentialConfiguration(
220 f"vault.auth_method is 'token' but {VAULT_TOKEN_ENVIRONMENT_VARIABLE} is not set in the "
221 f"NetBox process environment."
222 )
223 headers["X-Vault-Token"] = token
224 return headers
226 def _read(self, reference: CredentialReference) -> requests.Response:
227 """Perform the one KV v2 read this reference names."""
228 address = str(self._settings["address"]).rstrip("/")
229 # The reference is validated, and quoting keeps a stray character out of the request anyway.
230 mount = quote(reference.mount, safe="")
231 path = quote(reference.path, safe="/")
232 url = f"{address}/v1/{mount}/data/{path}"
233 timeout = (
234 self._settings.get("connect_timeout", DEFAULT_CONNECT_TIMEOUT),
235 self._settings.get("read_timeout", DEFAULT_READ_TIMEOUT),
236 )
237 try:
238 resolved_addresses = resolve_addresses(address, setting="vault.address", deadline=self._deadline)
239 except WallClockDeadlineExceeded:
240 raise
241 except InvalidInferenceConfiguration as exc:
242 raise CredentialUnavailable(
243 f"The credential store could not be reached ({type(exc).__name__}). Check the configured vault address."
244 ) from None
245 connection_failure = None
246 for resolved_address in resolved_addresses:
247 try:
248 with _quiet_transport_logging():
249 return request_to_resolved_address(
250 self._session,
251 "GET",
252 url,
253 resolved_address,
254 headers=self._headers(),
255 timeout=timeout,
256 deadline=self._deadline,
257 response_body_limit=VAULT_RESPONSE_LIMIT,
258 verify=self._settings.get("ca_bundle", True),
259 allow_redirects=False,
260 )
261 except WallClockDeadlineExceeded:
262 raise
263 except requests.RequestException as exc:
264 if isinstance(exc, ResponseProcessingFailure) and isinstance(exc.cause, WallClockDeadlineExceeded):
265 raise exc.cause from None
266 if isinstance(exc, ResponseProcessingFailure) and isinstance(exc.cause, ResponseBodyTooLarge):
267 # The body is refused unread, so the status is the only classification left.
268 refusal = _status_failure(exc.response.status_code)
269 if refusal is not None:
270 raise refusal from None
271 raise CredentialUnavailable(
272 "The credential store answered with a body this read cannot accept."
273 ) from None
274 if is_preconnect_failure(exc):
275 connection_failure = exc
276 continue
277 # This text reaches Job.data, so neither the address nor the URL is reported.
278 raise CredentialUnavailable(
279 f"The credential store could not be reached ({type(exc).__name__}). "
280 f"Check the configured vault address."
281 ) from None
282 except OSError:
283 raise InvalidCredentialConfiguration(
284 "The configured Vault TLS CA bundle could not be used. Check vault.ca_bundle."
285 ) from None
286 failure_name = type(connection_failure).__name__ if connection_failure is not None else "NoAddress"
287 raise CredentialUnavailable(
288 f"The credential store could not be reached ({failure_name}). Check the configured vault address."
289 ) from None
291 def resolve(self, reference: CredentialReference) -> str:
292 """Return the secret the reference names, classifying every failure without quoting Vault."""
293 if reference.backend != self.name:
294 raise InvalidCredentialReference(f"This backend resolves '{self.name}' references only.")
295 response = self._read(reference)
296 failure = _status_failure(response.status_code)
297 if failure is not None:
298 raise failure
299 try:
300 envelope = response.json()
301 data = envelope["data"]["data"]
302 except (ValueError, KeyError, TypeError):
303 raise CredentialUnavailable("The credential store answered with an unreadable KV v2 envelope.") from None
304 if not isinstance(data, Mapping):
305 raise CredentialUnavailable("The credential store answered with an unreadable KV v2 envelope.") from None
306 if reference.field not in data:
307 raise InvalidCredentialReference("The credential store holds no secret for this Inference Backend.")
308 value = data[reference.field]
309 if not isinstance(value, str):
310 raise InvalidSecretMaterial("The referenced field does not hold a string.")
311 if not value.strip():
312 raise InvalidSecretMaterial("The referenced field is empty.")
313 return value
316def credential_backend_for(
317 reference: CredentialReference,
318 vault_settings: Mapping[str, Any],
319 *,
320 deadline: WallClockDeadline | None = None,
321) -> CredentialBackend:
322 """Return the credential backend that resolves this reference."""
323 if reference.backend == CREDENTIAL_REFERENCE_BACKEND:
324 return VaultKvV2CredentialBackend(vault_settings, deadline=deadline)
325 raise InvalidCredentialReference(f"No credential backend resolves '{reference.backend}' references.")
328__all__ = (
329 "VAULT_TOKEN_ENVIRONMENT_VARIABLE",
330 "CredentialBackend",
331 "CredentialDenied",
332 "CredentialFailure",
333 "CredentialReference",
334 "CredentialUnavailable",
335 "InvalidCredentialConfiguration",
336 "InvalidCredentialReference",
337 "InvalidSecretMaterial",
338 "VaultKvV2CredentialBackend",
339 "credential_backend_for",
340)