Coverage for netbox_data_import/inference_settings.py: 99%

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). 

4 

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""" 

8 

9from collections.abc import Mapping, Sequence 

10from typing import Any 

11 

12from .inference_trust import ( 

13 InvalidInferenceConfiguration, 

14 is_local_endpoint, 

15 split_url as _split_url, 

16 validate_api_root, 

17 validate_origin, 

18) 

19 

20ORIGIN_ALLOWLIST_SETTING = "inference_backend_origin_allowlist" 

21FILE_FALLBACK_SETTING = "inference_backend" 

22VAULT_SETTING = "vault" 

23 

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

25FILE_FALLBACK_KEY = "file-fallback" 

26 

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) 

39 

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) 

48 

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

50API_ROOT_MAX_LENGTH = 500 

51DISPLAY_NAME_MAX_LENGTH = 200 

52MODEL_MAX_LENGTH = 200 

53 

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

55TIMEOUT_MIN = 1 

56TIMEOUT_MAX = 2147483647 

57 

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

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

60 

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

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

63 

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

65CREDENTIAL_REFERENCE_BACKEND = "vault_kv_v2" 

66 

67 

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

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

70 if not isinstance(value, Mapping): 

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

72 return value 

73 

74 

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

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

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

78 raise InvalidInferenceConfiguration( 

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

80 ) 

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

82 

83 

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

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

86 mapping = _require_mapping(value, VAULT_SETTING) 

87 for name in VAULT_FORBIDDEN_FIELDS: 

88 if name in mapping: 

89 raise InvalidInferenceConfiguration( 

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

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

92 ) 

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

94 if unknown: 

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

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

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

98 _validate_vault_address(mapping["address"]) 

99 namespace = mapping.get("namespace") 

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

101 raise InvalidInferenceConfiguration( 

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

103 ) 

104 bundle = mapping.get("ca_bundle") 

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

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

107 raise InvalidInferenceConfiguration( 

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

109 ) 

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

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

112 if field in mapping: 

113 _validate_timeout(mapping, field, setting=VAULT_SETTING) 

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

115 if method not in VAULT_AUTH_METHODS: 

116 raise InvalidInferenceConfiguration( 

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

118 ) 

119 return mapping 

120 

121 

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

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

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

125 if not isinstance(value, str): 

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

127 for character in "?#": 

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

129 if character in value: 

130 raise InvalidInferenceConfiguration( 

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

132 f"the query or fragment." 

133 ) 

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

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

136 # The read sends a token to this address, so it follows the api_root rule for a bearer token. 

137 if parts.scheme.lower() != "https" and not is_local_endpoint(value): 

138 raise InvalidInferenceConfiguration( 

139 f"{label} must use https, unless it names a local endpoint, because the read sends a token to it." 

140 ) 

141 

142 

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

144 

145 

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

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

148 

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

150 """ 

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

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

153 return value 

154 

155 

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

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

158 text = _require_text(value, label) 

159 for character in "?#%": 

160 if character in text: 

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

162 if not segments and "/" in text: 

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

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

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

166 

167 

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

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

170 mapping = _require_mapping(value, label) 

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

172 if unknown: 

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

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

175 if missing: 

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

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

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

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

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

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

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

183 return mapping 

184 

185 

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

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

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

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

190 raise InvalidInferenceConfiguration( 

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

192 ) 

193 

194 

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

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

197 value = mapping.get(field) 

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

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

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

201 if len(value) > max_length: 

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

203 

204 

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

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

207 value = mapping.get(field) 

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

209 if isinstance(value, bool) or not isinstance(value, int) or not TIMEOUT_MIN <= value <= TIMEOUT_MAX: 

210 raise InvalidInferenceConfiguration( 

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

212 f"{TIMEOUT_MIN} and {TIMEOUT_MAX}, got {value!r}." 

213 ) 

214 

215 

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

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

218 _validate_choice(mapping, "adapter_type", ADAPTER_TYPES) 

219 _validate_choice(mapping, "authentication", AUTHENTICATION_METHODS) 

220 _validate_choice(mapping, "response_mode", RESPONSE_MODES) 

221 _validate_text(mapping, "api_root", API_ROOT_MAX_LENGTH) 

222 _validate_text(mapping, "display_name", DISPLAY_NAME_MAX_LENGTH) 

223 _validate_text(mapping, "model", MODEL_MAX_LENGTH) 

224 _validate_timeout(mapping, "connect_timeout") 

225 _validate_timeout(mapping, "read_timeout") 

226 

227 

228def validate_file_fallback(value: Any, allowlist: Sequence[str]) -> Mapping[str, Any]: 

229 """Return the whole-backend fallback, rejecting a field set that is not exactly the row's.""" 

230 mapping = _require_mapping(value, FILE_FALLBACK_SETTING) 

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

232 if unknown: 

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

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

235 if missing: 

236 raise InvalidInferenceConfiguration( 

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

238 ) 

239 _validate_fallback_fields(mapping) 

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

241 validate_api_root( 

242 mapping["api_root"], 

243 allowlist=allowlist, 

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

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

246 ) 

247 return mapping 

248 

249 

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

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

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

253 if VAULT_SETTING in user_config: 

254 validate_vault_settings(user_config[VAULT_SETTING]) 

255 if FILE_FALLBACK_SETTING in user_config: 

256 validate_file_fallback(user_config[FILE_FALLBACK_SETTING], allowlist) 

257 

258 

259__all__ = ( 

260 "CREDENTIAL_REFERENCE_BACKEND", 

261 "CREDENTIAL_REFERENCE_FIELDS", 

262 "FILE_FALLBACK_FIELDS", 

263 "FILE_FALLBACK_KEY", 

264 "FILE_FALLBACK_SETTING", 

265 "ORIGIN_ALLOWLIST_SETTING", 

266 "VAULT_AUTH_METHODS", 

267 "VAULT_SETTING", 

268 "InvalidInferenceConfiguration", 

269 "validate_credential_reference", 

270 "validate_file_fallback", 

271 "validate_origin_allowlist", 

272 "validate_plugin_settings", 

273 "validate_vault_settings", 

274)