Coverage for netbox_data_import/inference

1# SPDX-License-Identifier: Apache-2.0

2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>

3"""Shape of the three Inference Backend plugin settings, checked at startup (specification 8.2.1).

5Shape is all this module decides. Credential resolution and network liveness belong to the worker

6and fail at request time, so nothing here opens a socket.

7"""

9from collections.abc import Mapping, Sequence

10from typing import Any

12from .inference_trust import (

13 InvalidInferenceConfiguration,

14 split_url as _split_url,

15 validate_api_root,

16 validate_origin,

17)

19ORIGIN_ALLOWLIST_SETTING = "inference_backend_origin_allowlist"

20FILE_FALLBACK_SETTING = "inference_backend"

21VAULT_SETTING = "vault"

22PROPOSAL_CANDIDATE_LIMIT_SETTING = "inference_proposal_candidate_limit"

24# The fallback is one whole backend, so its key cannot be chosen per deployment.

25FILE_FALLBACK_KEY = "file-fallback"

27# The InferenceBackend row fields minus the backend key and enabled.

28FILE_FALLBACK_FIELDS = (

29 "display_name",

30 "adapter_type",

31 "api_root",

32 "model",

33 "authentication",

34 "response_mode",

35 "credential_reference",

36 "connect_timeout",

37 "read_timeout",

38)

40# The InferenceBackend column choices, here because settings load before the app registry.

41ADAPTER_TYPES = (("openai_compatible", "OpenAI compatible"),)

42AUTHENTICATION_METHODS = (("bearer", "Bearer token"),)

43RESPONSE_MODES = (

44 ("prompt_json", "JSON asked for in the prompt"),

45 ("json_object", "JSON object mode"),

46 ("json_schema", "JSON schema mode"),

47)

49# The InferenceBackend column widths the fallback has to respect.

50API_ROOT_MAX_LENGTH = 500

51DISPLAY_NAME_MAX_LENGTH = 200

52MODEL_MAX_LENGTH = 200

54# PositiveIntegerField stores up to this. One second is the smallest timeout that can make a call.

55TIMEOUT_MIN = 1

56POSITIVE_INTEGER_MAX = 2147483647

58#: One request at a time, because a local inference server answers its requests one after another.

59MAX_CONCURRENT_REQUESTS_DEFAULT = 1

60MAX_CONCURRENT_REQUESTS_MIN = 1

62# The InferenceBackend row fields the fallback may omit, with the default the row column has too.

63FILE_FALLBACK_DEFAULTS = {"max_concurrent_requests": MAX_CONCURRENT_REQUESTS_DEFAULT}

65#: A proposal offers the whole eligible set, so the bound is denser equipment, not the picker's page.

66PROPOSAL_CANDIDATE_LIMIT_DEFAULT = 64

67#: With supported 64-character termination names, even worst-case JSON escaping keeps this array below 1 MiB.

68PROPOSAL_CANDIDATE_LIMIT_MAX = 1024

70VAULT_AUTH_METHODS = ("proxy", "token")

71VAULT_FIELDS = ("address", "auth_method", "namespace", "ca_bundle", "connect_timeout", "read_timeout")

73# Named so a deployment that sets one is told why, rather than having it silently ignored.

74VAULT_FORBIDDEN_FIELDS = ("mount", "token", "role_id", "secret_id", "verify", "tls_skip_verify")

76CREDENTIAL_REFERENCE_FIELDS = ("backend", "mount", "path", "field")

77CREDENTIAL_REFERENCE_BACKEND = "vault_kv_v2"

80def _require_mapping(value: Any, label: str) -> Mapping[str, Any]:

81 """Return *value* as a mapping, or reject it."""

82 if not isinstance(value, Mapping):

83 raise InvalidInferenceConfiguration(f"'{label}' must be a mapping, got {type(value).__name__}.")

84 return value

87def validate_origin_allowlist(value: Any) -> tuple[str, ...]:

88 """Return the allowlist origins, rejecting anything that is not one exact origin."""

89 if isinstance(value, (str, bytes)) or not isinstance(value, Sequence):

90 raise InvalidInferenceConfiguration(

91 f"'{ORIGIN_ALLOWLIST_SETTING}' must be a list of origin strings, got {type(value).__name__}."

92 )

93 return tuple(validate_origin(entry, setting=ORIGIN_ALLOWLIST_SETTING) for entry in value)

96def validate_vault_settings(value: Any) -> Mapping[str, Any]:

97 """Return the vault connection mapping, rejecting secret material and the KV v2 mount."""

98 mapping = _require_mapping(value, VAULT_SETTING)

99 for name in VAULT_FORBIDDEN_FIELDS:

100 if name in mapping:

101 raise InvalidInferenceConfiguration(

102 f"'{VAULT_SETTING}.{name}' is not accepted. Connection and machine identity data belong to the "

103 f"deployment, and the KV v2 mount belongs to the credential reference."

104 )

105 unknown = sorted(str(key) for key in set(mapping) - set(VAULT_FIELDS))

106 if unknown:

107 raise InvalidInferenceConfiguration(f"Unknown '{VAULT_SETTING}' key(s): {', '.join(unknown)}.")

108 if not mapping.get("address"):

109 raise InvalidInferenceConfiguration(f"'{VAULT_SETTING}.address' is required.")

110 _validate_vault_address(mapping["address"])

111 namespace = mapping.get("namespace")

112 if "namespace" in mapping and (not isinstance(namespace, str) or not namespace.strip()):

113 raise InvalidInferenceConfiguration(

114 f"'{VAULT_SETTING}.namespace' must be a non-empty string, got {type(namespace).__name__}."

115 )

116 bundle = mapping.get("ca_bundle")

117 if "ca_bundle" in mapping and (not isinstance(bundle, str) or not bundle.strip()):

118 # requests reads a bool here as "skip verification", which this setting must never mean.

119 raise InvalidInferenceConfiguration(

120 f"'{VAULT_SETTING}.ca_bundle' must be a path to a CA bundle, got {type(bundle).__name__}."

121 )

122 for field in ("connect_timeout", "read_timeout"):

123 # Optional here, unlike the fallback: absent means the backend's own default deadline.

124 if field in mapping:

125 _validate_timeout(mapping, field, setting=VAULT_SETTING)

126 method = mapping.get("auth_method", "proxy")

127 if method not in VAULT_AUTH_METHODS:

128 raise InvalidInferenceConfiguration(

129 f"'{VAULT_SETTING}.auth_method' must be one of {', '.join(VAULT_AUTH_METHODS)}, got '{method}'."

130 )

131 return mapping

132

133

134def _validate_vault_address(value: Any) -> None:

135 """Reject a Vault address that carries a secret or that the read path would misassemble."""

136 label = f"'{VAULT_SETTING}.address'"

137 if not isinstance(value, str):

138 raise InvalidInferenceConfiguration(f"{label} must be a string URL, got {type(value).__name__}.")

139 for character in "?#":

140 # A bare delimiter parses as an empty component, and the appended read path lands inside it.

141 if character in value:

142 raise InvalidInferenceConfiguration(

143 f"{label} cannot contain '{character}', which would put the appended read path in "

144 f"the query or fragment."

145 )

146 # Unquoted: the message is persisted, and an address can carry a token in its userinfo.

147 parts = _split_url(value, f"{VAULT_SETTING}.address", quote_value=False)

148 # Every read carries a Vault token or a resolved secret, including reads through a local Proxy.

149 if parts.scheme.lower() != "https":

150 raise InvalidInferenceConfiguration(

151 f"{label} must use https because the connection carries a Vault token or a resolved secret."

152 )

153

154

155_UNSAFE_PATH_SEGMENTS = frozenset({"", ".", ".."})

156

157

158def _require_text(value: Any, label: str) -> str:

159 """Return the value as text, rejecting one that only looks valid after coercion.

160

161 A number survives `str()` and reaches the Vault request as a path or a key name.

162 """

163 if not isinstance(value, str) or not value:

164 raise InvalidInferenceConfiguration(f"'{label}' must be a non-empty string.")

165 return value

166

167

168def _validate_vault_path(value: Any, label: str, *, segments: bool) -> None:

169 """Reject a Vault path value that could change the request it is interpolated into."""

170 text = _require_text(value, label)

171 for character in "?#%":

172 if character in text:

173 raise InvalidInferenceConfiguration(f"'{label}' cannot contain '{character}'.")

174 if not segments and "/" in text:

175 raise InvalidInferenceConfiguration(f"'{label}' names one path segment, so it cannot contain '/'.")

176 if any(part in _UNSAFE_PATH_SEGMENTS for part in text.split("/")):

177 raise InvalidInferenceConfiguration(f"'{label}' cannot hold an empty, '.' or '..' path segment.")

178

179

180def validate_credential_reference(value: Any, label: str = "credential_reference") -> Mapping[str, Any]:

181 """Return the typed Vault KV v2 reference, rejecting connection data and secret material."""

182 mapping = _require_mapping(value, label)

183 unknown = sorted(str(key) for key in set(mapping) - set(CREDENTIAL_REFERENCE_FIELDS))

184 if unknown:

185 raise InvalidInferenceConfiguration(f"Unknown '{label}' key(s): {', '.join(unknown)}.")

186 missing = [name for name in CREDENTIAL_REFERENCE_FIELDS if not mapping.get(name)]

187 if missing:

188 raise InvalidInferenceConfiguration(f"'{label}' is missing required key(s): {', '.join(missing)}.")

189 if mapping["backend"] != CREDENTIAL_REFERENCE_BACKEND:

190 # The supplied value is not quoted back: this message is persisted and may hold secret material.

191 raise InvalidInferenceConfiguration(f"'{label}.backend' must be '{CREDENTIAL_REFERENCE_BACKEND}'.")

192 _validate_vault_path(mapping["mount"], f"{label}.mount", segments=False)

193 _validate_vault_path(mapping["path"], f"{label}.path", segments=True)

194 _require_text(mapping["field"], f"{label}.field")

195 return mapping

196

197

198def _validate_choice(mapping: Mapping[str, Any], field: str, choices) -> None:

199 """Reject a fallback value the matching InferenceBackend column would not accept."""

200 allowed = [value for value, _label in choices]

201 if mapping.get(field) not in allowed:

202 raise InvalidInferenceConfiguration(

203 f"'{FILE_FALLBACK_SETTING}.{field}' must be one of {', '.join(allowed)}, got '{mapping.get(field)}'."

204 )

205

206

207def _validate_text(mapping: Mapping[str, Any], field: str, max_length: int) -> None:

208 """Reject fallback text the matching column could not store, or that names nothing."""

209 value = mapping.get(field)

210 label = f"'{FILE_FALLBACK_SETTING}.{field}'"

211 if not isinstance(value, str) or not value.strip():

212 raise InvalidInferenceConfiguration(f"{label} must be a non-empty string, got {value!r}.")

213 if len(value) > max_length:

214 raise InvalidInferenceConfiguration(f"{label} is longer than the {max_length} characters the column holds.")

215

216

217def _validate_timeout(mapping: Mapping[str, Any], field: str, setting: str = FILE_FALLBACK_SETTING) -> None:

218 """Reject a timeout the matching PositiveIntegerField would not accept."""

219 value = mapping.get(field)

220 # bool is an int subclass, and True would otherwise read as a one second timeout.

221 if isinstance(value, bool) or not isinstance(value, int) or not TIMEOUT_MIN <= value <= POSITIVE_INTEGER_MAX:

222 raise InvalidInferenceConfiguration(

223 f"'{setting}.{field}' must be a whole number of seconds between "

224 f"{TIMEOUT_MIN} and {POSITIVE_INTEGER_MAX}, got {value!r}."

225 )

226

227

228def _validate_max_concurrent_requests(mapping: Mapping[str, Any]) -> None:

229 """Reject a request limit the matching InferenceBackend column would not accept."""

230 value = mapping["max_concurrent_requests"]

231 if (

232 isinstance(value, bool)

233 or not isinstance(value, int)

234 or not MAX_CONCURRENT_REQUESTS_MIN <= value <= POSITIVE_INTEGER_MAX

235 ):

236 raise InvalidInferenceConfiguration(

237 f"'{FILE_FALLBACK_SETTING}.max_concurrent_requests' must be a whole number between "

238 f"{MAX_CONCURRENT_REQUESTS_MIN} and {POSITIVE_INTEGER_MAX}, got {value!r}."

239 )

240

241

242def _validate_fallback_fields(mapping: Mapping[str, Any]) -> None:

243 """Apply the InferenceBackend column constraints the fallback bypasses by not being a row."""

244 _validate_choice(mapping, "adapter_type", ADAPTER_TYPES)

245 _validate_choice(mapping, "authentication", AUTHENTICATION_METHODS)

246 _validate_choice(mapping, "response_mode", RESPONSE_MODES)

247 _validate_text(mapping, "api_root", API_ROOT_MAX_LENGTH)

248 _validate_text(mapping, "display_name", DISPLAY_NAME_MAX_LENGTH)

249 _validate_text(mapping, "model", MODEL_MAX_LENGTH)

250 _validate_timeout(mapping, "connect_timeout")

251 _validate_timeout(mapping, "read_timeout")

252 _validate_max_concurrent_requests(mapping)

253

254

255def validate_file_fallback(value: Any, allowlist: Sequence[str]) -> dict[str, Any]:

256 """Return the whole-backend fallback with its defaults, rejecting a field set the row does not have."""

257 mapping = {**FILE_FALLBACK_DEFAULTS, **_require_mapping(value, FILE_FALLBACK_SETTING)}

258 unknown = sorted(str(key) for key in set(mapping) - set(FILE_FALLBACK_FIELDS) - set(FILE_FALLBACK_DEFAULTS))

259 if unknown:

260 raise InvalidInferenceConfiguration(f"Unknown '{FILE_FALLBACK_SETTING}' key(s): {', '.join(unknown)}.")

261 missing = [name for name in FILE_FALLBACK_FIELDS if name not in mapping]

262 if missing:

263 raise InvalidInferenceConfiguration(

264 f"'{FILE_FALLBACK_SETTING}' is missing required key(s): {', '.join(missing)}."

265 )

266 _validate_fallback_fields(mapping)

267 validate_credential_reference(mapping["credential_reference"], f"{FILE_FALLBACK_SETTING}.credential_reference")

268 validate_api_root(

269 mapping["api_root"],

270 allowlist=allowlist,

271 authentication=mapping.get("authentication", "bearer"),

272 setting=f"{FILE_FALLBACK_SETTING}.api_root",

273 )

274 return mapping

275

276

277def validate_proposal_candidate_limit(value: Any) -> int:

278 """Reject a candidate bound the retrieval cannot honour (section 7.3, operator decision).

279

280 `True` is numerically 1 and would silently admit a one-candidate set, so bool is excluded before

281 the range check. A value past the range never reaches a database slice.

282 """

283 label = f"'{PROPOSAL_CANDIDATE_LIMIT_SETTING}'"

284 if isinstance(value, bool) or not isinstance(value, int):

285 raise InvalidInferenceConfiguration(f"{label} must be a whole number, got {value!r}.")

286 if not 0 < value <= PROPOSAL_CANDIDATE_LIMIT_MAX:

287 raise InvalidInferenceConfiguration(

288 f"{label} must be between 1 and {PROPOSAL_CANDIDATE_LIMIT_MAX}, got {value!r}."

289 )

290 return value

291

292

293def validate_plugin_settings(user_config: Mapping[str, Any]) -> None:

294 """Reject a malformed Inference Backend configuration before the application serves a request."""

295 allowlist = validate_origin_allowlist(user_config.get(ORIGIN_ALLOWLIST_SETTING, ()))

296 if VAULT_SETTING in user_config:

297 validate_vault_settings(user_config[VAULT_SETTING])

298 if FILE_FALLBACK_SETTING in user_config:

299 validate_file_fallback(user_config[FILE_FALLBACK_SETTING], allowlist)

300 # Validated even when inference is otherwise unconfigured, and defaulted only when omitted.

301 if PROPOSAL_CANDIDATE_LIMIT_SETTING in user_config:

302 validate_proposal_candidate_limit(user_config[PROPOSAL_CANDIDATE_LIMIT_SETTING])

303

304

305__all__ = (

306 "CREDENTIAL_REFERENCE_BACKEND",

307 "CREDENTIAL_REFERENCE_FIELDS",

308 "FILE_FALLBACK_DEFAULTS",

309 "FILE_FALLBACK_FIELDS",

310 "FILE_FALLBACK_KEY",

311 "FILE_FALLBACK_SETTING",

312 "MAX_CONCURRENT_REQUESTS_DEFAULT",

313 "MAX_CONCURRENT_REQUESTS_MIN",

314 "ORIGIN_ALLOWLIST_SETTING",

315 "PROPOSAL_CANDIDATE_LIMIT_DEFAULT",

316 "PROPOSAL_CANDIDATE_LIMIT_MAX",

317 "PROPOSAL_CANDIDATE_LIMIT_SETTING",

318 "VAULT_AUTH_METHODS",

319 "VAULT_SETTING",

320 "InvalidInferenceConfiguration",

321 "validate_credential_reference",

322 "validate_file_fallback",

323 "validate_origin_allowlist",

324 "validate_plugin_settings",

325 "validate_proposal_candidate_limit",

326 "validate_vault_settings",

327)

Coverage for netbox_data_import/inference_settings.py: 99.35%

154 statements