Coverage for netbox_data_import/inference_credentials.py: 97%
126 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 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_trust import InvalidInferenceConfiguration
31# The deployment owns the token; the plugin never stores one.
32VAULT_TOKEN_ENVIRONMENT_VARIABLE = "VAULT_TOKEN" # noqa: S105 - This names an environment variable, not a token.
34TRANSPORT_LOGGER = "urllib3.connectionpool"
35_reading_vault = threading.local()
38class _QuietDuringVaultRead(logging.Filter):
39 """Drop transport records emitted while this thread is reading a credential.
41 urllib3 logs the connection and the request line at DEBUG, which names the Vault address and
42 the KV path of the secret being read. Neither is recoverable by redacting the message, because
43 the address is interpolated into several formats, so the read is silenced for its duration.
44 Every failure this module raises is still typed and still reaches the caller.
45 """
47 def filter(self, record: logging.LogRecord) -> bool:
48 """Return whether one transport record may be emitted."""
49 return not getattr(_reading_vault, "active", False)
52logging.getLogger(TRANSPORT_LOGGER).addFilter(_QuietDuringVaultRead())
55@contextmanager
56def _quiet_transport_logging():
57 """Silence the transport logger for this thread only, for the length of one read."""
58 _reading_vault.active = True
59 try:
60 yield
61 finally:
62 _reading_vault.active = False
65DEFAULT_CONNECT_TIMEOUT = 5
66DEFAULT_READ_TIMEOUT = 60
69class CredentialFailure(Exception):
70 """A credential resolution that failed, carrying its typed category and no secret."""
72 category = "credential_unavailable"
75class CredentialUnavailable(CredentialFailure):
76 """The credential store could not be reached, or held no answer for this reference."""
78 category = "credential_unavailable"
81class CredentialDenied(CredentialFailure):
82 """The credential store refused the read."""
84 category = "credential_denied"
87class InvalidCredentialReference(CredentialFailure):
88 """The reference is not a usable typed reference."""
90 category = "invalid_credential_reference"
93class InvalidSecretMaterial(CredentialFailure):
94 """The store answered, but the named field holds nothing usable as a key."""
96 category = "invalid_secret_material"
99class InvalidCredentialConfiguration(CredentialFailure):
100 """The deployment-owned credential settings cannot be used as given."""
102 category = "invalid_configuration"
105@dataclass(frozen=True)
106class CredentialReference:
107 """One typed Vault KV v2 reference: restricted configuration metadata, never secret material."""
109 backend: str
110 mount: str
111 path: str
112 field: str
114 @classmethod
115 def from_mapping(cls, mapping: Mapping[str, Any]) -> "CredentialReference":
116 """Return the validated reference, rejecting connection data and secret material."""
117 try:
118 validated = validate_credential_reference(mapping)
119 except InvalidInferenceConfiguration as exc:
120 raise InvalidCredentialReference(str(exc)) from exc
121 return cls(
122 backend=validated["backend"],
123 mount=validated["mount"],
124 path=validated["path"],
125 field=validated["field"],
126 )
129class CredentialBackend(Protocol):
130 """Resolve one typed reference into secret material for the lifetime of one request."""
132 name: str
134 def close(self) -> None:
135 """Release resources owned by this backend."""
136 ...
138 def __enter__(self) -> Self: ...
140 def __exit__(
141 self,
142 exc_type: type[BaseException] | None,
143 exc_value: BaseException | None,
144 traceback: TracebackType | None,
145 ) -> None: ...
147 def resolve(self, reference: CredentialReference) -> str:
148 """Return the secret the reference names."""
149 ...
152class VaultKvV2CredentialBackend:
153 """Read one named field from one KV v2 path, through Vault Proxy or a deployment token."""
155 name = CREDENTIAL_REFERENCE_BACKEND
157 def __init__(self, settings: Mapping[str, Any], session: requests.Session | None = None):
158 try:
159 self._settings = validate_vault_settings(settings)
160 except InvalidInferenceConfiguration as exc:
161 raise InvalidCredentialConfiguration(str(exc)) from exc
162 self._owns_session = session is None
163 self._session = requests.Session() if session is None else session
165 def close(self) -> None:
166 """Close the session only when this backend created it."""
167 if self._owns_session:
168 self._session.close()
170 def __enter__(self) -> Self:
171 return self
173 def __exit__(
174 self,
175 exc_type: type[BaseException] | None,
176 exc_value: BaseException | None,
177 traceback: TracebackType | None,
178 ) -> None:
179 self.close()
181 def _headers(self) -> dict[str, str]:
182 """Return the request headers, reading a token only when the deployment selected one."""
183 headers = {"Accept": "application/json"}
184 if namespace := self._settings.get("namespace"):
185 headers["X-Vault-Namespace"] = namespace
186 if self._settings.get("auth_method", "proxy") != "token":
187 return headers
188 token = os.environ.get(VAULT_TOKEN_ENVIRONMENT_VARIABLE, "")
189 if not token:
190 raise InvalidCredentialConfiguration(
191 f"vault.auth_method is 'token' but {VAULT_TOKEN_ENVIRONMENT_VARIABLE} is not set in the "
192 f"worker environment."
193 )
194 headers["X-Vault-Token"] = token
195 return headers
197 def _read(self, reference: CredentialReference) -> requests.Response:
198 """Perform the one KV v2 read this reference names."""
199 address = str(self._settings["address"]).rstrip("/")
200 # The reference is validated, and quoting keeps a stray character out of the request anyway.
201 mount = quote(reference.mount, safe="")
202 path = quote(reference.path, safe="/")
203 url = f"{address}/v1/{mount}/data/{path}"
204 timeout = (
205 self._settings.get("connect_timeout", DEFAULT_CONNECT_TIMEOUT),
206 self._settings.get("read_timeout", DEFAULT_READ_TIMEOUT),
207 )
208 try:
209 with _quiet_transport_logging():
210 return self._session.get(
211 url,
212 headers=self._headers(),
213 timeout=timeout,
214 verify=self._settings.get("ca_bundle", True),
215 allow_redirects=False,
216 )
217 except requests.RequestException as exc:
218 # This text reaches Job.data, so neither the address nor the URL is reported.
219 raise CredentialUnavailable(
220 f"The credential store could not be reached ({type(exc).__name__}). Check the configured vault address."
221 ) from None
223 def resolve(self, reference: CredentialReference) -> str:
224 """Return the secret the reference names, classifying every failure without quoting Vault."""
225 if reference.backend != self.name:
226 raise InvalidCredentialReference(f"This backend resolves '{self.name}' references only.")
227 response = self._read(reference)
228 # Every 3xx, not a list of them: a 300 or 305 body shaped like KV would read as the secret.
229 if 300 <= response.status_code < 400:
230 raise InvalidCredentialConfiguration(
231 f"The credential store redirected the read (HTTP {response.status_code}). "
232 f"Check the configured vault address."
233 )
234 if response.status_code in (401, 403):
235 raise CredentialDenied(f"The credential store refused the read (HTTP {response.status_code}).")
236 if response.status_code == 404:
237 raise CredentialUnavailable("The credential store holds no secret at the referenced path.")
238 if response.status_code >= 400:
239 raise CredentialUnavailable(f"The credential store answered HTTP {response.status_code}.")
240 try:
241 envelope = response.json()
242 data = envelope["data"]["data"]
243 except (ValueError, KeyError, TypeError):
244 raise CredentialUnavailable("The credential store answered with an unreadable KV v2 envelope.") from None
245 if not isinstance(data, Mapping):
246 raise CredentialUnavailable("The credential store answered with an unreadable KV v2 envelope.") from None
247 if reference.field not in data:
248 raise InvalidSecretMaterial("The referenced field is absent from the stored secret.")
249 value = data[reference.field]
250 if not isinstance(value, str):
251 raise InvalidSecretMaterial("The referenced field does not hold a string.")
252 if not value.strip():
253 raise InvalidSecretMaterial("The referenced field is empty.")
254 return value
257def credential_backend_for(reference: CredentialReference, vault_settings: Mapping[str, Any]) -> CredentialBackend:
258 """Return the credential backend that resolves this reference."""
259 if reference.backend == CREDENTIAL_REFERENCE_BACKEND:
260 return VaultKvV2CredentialBackend(vault_settings)
261 raise InvalidCredentialReference(f"No credential backend resolves '{reference.backend}' references.")
264__all__ = (
265 "VAULT_TOKEN_ENVIRONMENT_VARIABLE",
266 "CredentialBackend",
267 "CredentialDenied",
268 "CredentialFailure",
269 "CredentialReference",
270 "CredentialUnavailable",
271 "InvalidCredentialConfiguration",
272 "InvalidCredentialReference",
273 "InvalidSecretMaterial",
274 "VaultKvV2CredentialBackend",
275 "credential_backend_for",
276)