Coverage for src / derivepassphrase / _types.py: 100.000%
339 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-06 21:34 +0200
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-06 21:34 +0200
1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info>
2#
3# SPDX-License-Identifier: Zlib
5"""Types used by derivepassphrase."""
7from __future__ import annotations
9import enum
10import json
11import math
12import string
13import warnings
14from typing import TYPE_CHECKING, Generic, Protocol, TypeVar, cast
16from typing_extensions import (
17 Buffer,
18 NamedTuple,
19 NotRequired,
20 TypeAlias,
21 TypedDict,
22 deprecated,
23 get_overloads,
24 overload,
25 runtime_checkable,
26)
28if TYPE_CHECKING:
29 from collections.abc import Callable, Generator, Sequence
30 from typing import Literal
32 from typing_extensions import (
33 Any,
34 Required,
35 TypeIs,
36 )
38__all__ = (
39 "SSH_AGENT",
40 "SSH_AGENTC",
41 "SSHKeyCommentPair",
42 "VaultConfig",
43 "is_vault_config",
44)
47class _Omitted:
48 def __bool__(self) -> bool:
49 return False
51 def __repr__(self) -> str:
52 return "..."
55class VaultConstructorArgs(TypedDict, total=False):
56 r"""Common, optional arguments for the vault constructor.
58 (Also doubles as a base class for the vault configuration global
59 settings.)
61 Attributes:
62 length:
63 Desired passphrase length.
64 repeat:
65 The maximum number of immediate character repetitions
66 allowed in the passphrase. Disabled if set to 0.
67 lower:
68 Optional constraint on ASCII lowercase characters. If
69 positive, include this many lowercase characters
70 somewhere in the passphrase. If 0, avoid lowercase
71 characters altogether.
72 upper:
73 Same as `lower`, but for ASCII uppercase characters.
74 number:
75 Same as `lower`, but for ASCII digits.
76 space:
77 Same as `lower`, but for the space character.
78 dash:
79 Same as `lower`, but for the hyphen-minus and underscore
80 characters.
81 symbol:
82 Same as `lower`, but for all other hitherto unlisted
83 ASCII printable characters (except backquote).
85 """
87 length: NotRequired[int]
88 """"""
89 repeat: NotRequired[int]
90 """"""
91 lower: NotRequired[int]
92 """"""
93 upper: NotRequired[int]
94 """"""
95 number: NotRequired[int]
96 """"""
97 space: NotRequired[int]
98 """"""
99 dash: NotRequired[int]
100 """"""
101 symbol: NotRequired[int]
102 """"""
105class VaultConfigGlobalSettings(VaultConstructorArgs, total=False):
106 r"""Configuration for vault: global settings.
108 Attributes:
109 key:
110 The base64-encoded ssh public key to use, overriding the
111 master passphrase. Optional.
112 phrase:
113 The master passphrase. Optional.
114 unicode_normalization_form:
115 The preferred Unicode normalization form; we warn the user
116 if textual passphrases do not match their normalized forms.
117 Optional, and a `derivepassphrase` extension.
119 """
121 key: NotRequired[str]
122 """"""
123 phrase: NotRequired[str]
124 """"""
125 unicode_normalization_form: NotRequired[
126 Literal["NFC", "NFD", "NFKC", "NFKD"]
127 ]
128 """"""
131class VaultConfigServicesSettings(VaultConfigGlobalSettings, total=False):
132 r"""Configuration for vault: services settings.
134 Attributes:
135 notes:
136 Optional notes for this service, to display to the user when
137 generating the passphrase.
138 key:
139 As per the global settings.
140 phrase:
141 As per the global settings.
142 unicode_normalization_form:
143 As per the global settings.
144 length:
145 As per the global settings.
146 repeat:
147 As per the global settings.
148 lower:
149 As per the global settings.
150 upper:
151 As per the global settings.
152 number:
153 As per the global settings.
154 space:
155 As per the global settings.
156 dash:
157 As per the global settings.
158 symbol:
159 As per the global settings.
161 """
163 notes: NotRequired[str]
164 """"""
167_VaultConfig = TypedDict(
168 "_VaultConfig",
169 {"global": NotRequired[VaultConfigGlobalSettings]},
170 total=False,
171)
174class VaultConfig(TypedDict, _VaultConfig, total=False):
175 r"""Configuration for vault. For typing purposes.
177 Usually stored as JSON.
179 Attributes:
180 global (NotRequired[VaultConfigGlobalSettings]):
181 Global settings.
182 services (Required[dict[str, VaultConfigServicesSettings]]):
183 Service-specific settings.
185 """
187 services: Required[dict[str, VaultConfigServicesSettings]]
190def json_path(path: Sequence[str | int], /) -> str:
191 r"""Transform a series of keys and indices into a JSONPath selector.
193 The resulting JSONPath selector conforms to RFC 9535, is always
194 rooted at the JSON root node (i.e., starts with `$`), and only
195 contains name and index selectors (in shorthand dot notation, where
196 possible).
198 Args:
199 path:
200 A sequence of object keys or array indices to navigate to
201 the desired JSON value, starting from the root node.
203 Returns:
204 A valid JSONPath selector (a string) identifying the desired
205 JSON value.
207 Examples:
208 >>> json_path(["global", "phrase"])
209 '$.global.phrase'
210 >>> json_path(["services", "service name with spaces", "length"])
211 '$.services["service name with spaces"].length'
212 >>> json_path(["services", "special\u000acharacters", "notes"])
213 '$.services["special\\ncharacters"].notes'
214 >>> print(json_path(["services", "special\u000acharacters", "notes"]))
215 $.services["special\ncharacters"].notes
216 >>> json_path(["custom_array", 2, 0])
217 '$.custom_array[2][0]'
219 """
221 def needs_longhand(x: str | int) -> bool: 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
222 initial = ( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
223 frozenset(string.ascii_lowercase)
224 | frozenset(string.ascii_uppercase)
225 | frozenset("_")
226 )
227 chars = initial | frozenset(string.digits) 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
228 return not ( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
229 isinstance(x, str)
230 and x
231 and set(x).issubset(chars)
232 and x[:1] in initial
233 )
235 chunks = ["$"] 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
236 chunks.extend( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
237 f"[{json.dumps(x)}]" if needs_longhand(x) else f".{x}" for x in path
238 )
239 return "".join(chunks) 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
242class _VaultConfigValidator:
243 INVALID_CONFIG_ERROR = "vault config is invalid"
245 def __init__(self, maybe_config: Any) -> None: # noqa: ANN401
246 self.maybe_config = maybe_config 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
248 def traverse_path(self, path: tuple[str, ...]) -> Any: # noqa: ANN401
249 obj = self.maybe_config 1ampkqrlsjnohgdeicfb
250 for key in path: 1ampkqrlsjnohgdeicfb
251 obj = obj[key] 1ampkqrlsjnohgdeicfb
252 return obj 1ampkqrlsjnohgdeicfb
254 def walk_subconfigs(
255 self,
256 ) -> Generator[tuple[tuple[str] | tuple[str, str], str, Any], None, None]:
257 obj = cast("dict[str, dict[str, Any]]", self.maybe_config) 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
258 if isinstance(obj.get("global", False), dict): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
259 for k, v in list(obj["global"].items()): 1amkvBCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb
260 yield ("global",), k, v 1amkvBCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb
261 for sv_name, sv_obj in list(obj["services"].items()): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
262 for k, v in list(sv_obj.items()): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb
263 yield ("services", sv_name), k, v 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb
265 def validate( # noqa: C901,PLR0912
266 self,
267 *,
268 allow_unknown_settings: bool = False,
269 ) -> None:
270 err_obj_not_a_dict = "vault config is not a dict" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
271 err_non_str_service_name = ( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
272 "vault config contains non-string service name {sv_name!r}"
273 )
274 err_not_a_dict = "vault config entry {json_path_str} is not a dict" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
275 err_not_a_string = "vault config entry {json_path_str} is not a string" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
276 err_not_an_int = "vault config entry {json_path_str} is not an integer" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
277 err_unknown_setting = ( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
278 "vault config entry {json_path_str} uses unknown setting {key!r}"
279 )
280 err_bad_number0 = "vault config entry {json_path_str} is negative" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
281 err_bad_number1 = "vault config entry {json_path_str} is not positive" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
283 kwargs: dict[str, Any] = { 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
284 "allow_unknown_settings": allow_unknown_settings,
285 }
286 if not isinstance(self.maybe_config, dict): 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
287 raise TypeError(err_obj_not_a_dict.format(**kwargs)) 1,-(.c
288 if "global" in self.maybe_config: 1ampk*qTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
289 o_global = self.maybe_config["global"] 1amkv'BCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb
290 if not isinstance(o_global, dict): 1amkv'BCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb
291 kwargs["json_path_str"] = json_path(["global"]) 1'c
292 raise TypeError(err_not_a_dict.format(**kwargs)) 1'c
293 if not isinstance(self.maybe_config.get("services"), dict): 1ampk*qTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
294 kwargs["json_path_str"] = json_path(["services"]) 1'c
295 raise TypeError(err_not_a_dict.format(**kwargs)) 1'c
296 for sv_name, service in self.maybe_config["services"].items(): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
297 if not isinstance(sv_name, str): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb
298 kwargs["sv_name"] = sv_name 1cb
299 raise TypeError(err_non_str_service_name.format(**kwargs)) 1cb
300 if not isinstance(service, dict): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb
301 kwargs["json_path_str"] = json_path(["services", sv_name]) 1c
302 raise TypeError(err_not_a_dict.format(**kwargs)) 1c
303 for path, key, value in self.walk_subconfigs(): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb
304 kwargs["path"] = path 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
305 kwargs["key"] = key 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
306 kwargs["value"] = value 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
307 kwargs["json_path_str"] = json_path([*path, key]) 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
308 # TODO(the-13th-letter): Rewrite using structural pattern
309 # matching.
310 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
311 if key in {"key", "phrase"}: 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb
312 if not isinstance(value, str): 1pkvBCDEOlsjntFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb
313 raise TypeError(err_not_a_string.format(**kwargs)) 1cb
314 elif key == "unicode_normalization_form" and path == ("global",): 1ampkqTrUvVWXBCDEOljnoYZ0FGHIJ2KhgLMdeNPwQRxySzuAicfb
315 if not isinstance(value, str): 1hgdeicfb
316 raise TypeError(err_not_a_string.format(**kwargs)) 1cb
317 if not allow_unknown_settings: 1hgdeicfb
318 raise ValueError(err_unknown_setting.format(**kwargs)) 1cb
319 elif key == "notes" and path != ("global",): 1ampkqTrUvVWXBCDEOljnoYZ0FGHIJ2KhgLMdeNPwQRxySzuAicfb
320 if not isinstance(value, str): 1vOjnhgdePwQRxySzuAicfb
321 raise TypeError(err_not_a_string.format(**kwargs)) 1cb
322 elif key in { 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb
323 "length",
324 "repeat",
325 "lower",
326 "upper",
327 "number",
328 "space",
329 "dash",
330 "symbol",
331 }:
332 if not isinstance(value, int): 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb
333 raise TypeError(err_not_an_int.format(**kwargs)) 1cb
334 if key == "length" and value < 1: 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb
335 raise ValueError(err_bad_number1.format(**kwargs)) 1cb
336 if key != "length" and value < 0: 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb
337 raise ValueError(err_bad_number0.format(**kwargs)) 1cb
338 elif not allow_unknown_settings: 1hgdeicfb
339 raise ValueError(err_unknown_setting.format(**kwargs)) 1cb
341 def clean_up_falsy_values(self) -> Generator[CleanupStep, None, None]: # noqa: C901
342 obj = self.maybe_config 1ampkqrlsjnoth(gdeicfb
343 if ( 1ampkqrlsjnoth(gdeicfb
344 not isinstance(obj, dict)
345 or "services" not in obj
346 or not isinstance(obj["services"], dict)
347 ):
348 raise ValueError( 1(c
349 self.INVALID_CONFIG_ERROR
350 ) # pragma: no cover [failsafe]
351 if "global" in obj and not isinstance(obj["global"], dict): 1ampkqrlsjnothgdeicfb
352 raise ValueError( 1c
353 self.INVALID_CONFIG_ERROR
354 ) # pragma: no cover [failsafe]
355 if not all( 1ampkqrlsjnothgdeicfb
356 isinstance(service_obj, dict)
357 for service_obj in obj["services"].values()
358 ):
359 raise ValueError( 1c
360 self.INVALID_CONFIG_ERROR
361 ) # pragma: no cover [failsafe]
363 def falsy(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb
364 return not js_truthiness(value) 1ampkqrlsjnohgdeicfb
366 def falsy_but_not_zero(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb
367 return not js_truthiness(value) and not ( 1ampkljhgdeicfb
368 isinstance(value, int) and value == 0
369 )
371 def falsy_but_not_string(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb
372 return not js_truthiness(value) and value != "" # noqa: PLC1901 1pklsjnhgdeicfb
374 for path, key, value in self.walk_subconfigs(): 1ampkqrlsjnothgdeicfb
375 service_obj = self.traverse_path(path) 1ampkqrlsjnohgdeicfb
376 # TODO(the-13th-letter): Rewrite using structural pattern
377 # matching.
378 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
379 if key == "phrase" and falsy_but_not_string(value): 1ampkqrlsjnohgdeicfb
380 yield CleanupStep( 1defb
381 (*path, key), service_obj[key], "replace", ""
382 )
383 service_obj[key] = "" 1defb
384 elif key == "notes" and falsy(value): 1ampkqrlsjnohgdeicfb
385 yield CleanupStep( 1jndefb
386 (*path, key), service_obj[key], "remove", None
387 )
388 service_obj.pop(key) 1jndefb
389 elif key == "key" and falsy(value): 1ampkqrlsjnohgdeicfb
390 if path == ("global",): 1lsjndefb
391 yield CleanupStep( 1lsdefb
392 (*path, key), service_obj[key], "remove", None
393 )
394 service_obj.pop(key) 1lsdefb
395 else:
396 yield CleanupStep( 1jndefb
397 (*path, key), service_obj[key], "replace", ""
398 )
399 service_obj[key] = "" 1jndefb
400 elif key == "length" and falsy(value): 1ampkqrlsjnohgdeicfb
401 yield CleanupStep( 1defb
402 (*path, key), service_obj[key], "replace", 20
403 )
404 service_obj[key] = 20 1defb
405 elif key == "repeat" and falsy_but_not_zero(value): 1ampkqrlsjnohgdeicfb
406 yield CleanupStep((*path, key), service_obj[key], "replace", 0) 1defb
407 service_obj[key] = 0 1defb
408 elif key in { 1ampkqrlsjnohgdeicfb
409 "lower",
410 "upper",
411 "number",
412 "space",
413 "dash",
414 "symbol",
415 } and falsy_but_not_zero(value):
416 yield CleanupStep( 1defb
417 (*path, key), service_obj[key], "remove", None
418 )
419 service_obj.pop(key) 1defb
422@overload
423@deprecated(
424 "allow_derivepassphrase_extensions argument is deprecated since v0.4.0, "
425 "to be removed in v1.0: no extensions are defined"
426)
427def validate_vault_config(
428 obj: Any, # noqa: ANN401
429 /,
430 *,
431 allow_derivepassphrase_extensions: bool,
432 allow_unknown_settings: bool = False,
433) -> None: ...
436@overload
437def validate_vault_config(
438 obj: Any, # noqa: ANN401
439 /,
440 *,
441 allow_unknown_settings: bool = False,
442) -> None: ...
445def validate_vault_config(
446 obj: Any,
447 /,
448 *,
449 allow_unknown_settings: bool = False,
450 allow_derivepassphrase_extensions: bool = _Omitted(), # type: ignore[assignment]
451) -> None:
452 """Check that `obj` is a valid vault config.
454 Args:
455 obj:
456 The object to test.
457 allow_unknown_settings:
458 If false, abort on unknown settings.
459 allow_derivepassphrase_extensions:
460 (Deprecated.) Ignored since v0.4.0.
462 Raises:
463 TypeError:
464 An entry in the vault config, or the vault config itself,
465 has the wrong type.
466 ValueError:
467 An entry in the vault config is not allowed, or has a
468 disallowed value.
470 Warning: Deprecated argument
471 **v0.4.0**:
472 The `allow_derivepassphrase_extensions` keyword argument is
473 deprecated, and will be removed in v1.0. There are no
474 specified `derivepassphrase` extensions.
476 """
477 # TODO(the-13th-letter): Remove this block in v1.0.
478 # https://the13thletter.info/derivepassphrase/latest/upgrade-notes/#v1.0-allow-derivepassphrase-extensions
479 # TODO(the-13th-letter): Add tests that trigger the deprecation warning,
480 # then include this in coverage.
481 if not isinstance( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
482 allow_derivepassphrase_extensions, _Omitted
483 ): # pragma: no cover [unused]
484 warnings.warn(
485 get_overloads(validate_vault_config)[0].__deprecated__, # type: ignore[attr-defined]
486 DeprecationWarning,
487 stacklevel=2,
488 )
490 return _VaultConfigValidator(obj).validate( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb
491 allow_unknown_settings=allow_unknown_settings
492 )
495def is_vault_config(obj: Any) -> TypeIs[VaultConfig]: # noqa: ANN401
496 """Check if `obj` is a valid vault config, according to typing.
498 Args:
499 obj: The object to test.
501 Returns:
502 True if this is a vault config, false otherwise.
504 """ # noqa: DOC501
505 try: 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAif
506 validate_vault_config( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAif
507 obj,
508 allow_unknown_settings=True,
509 )
510 except (TypeError, ValueError) as exc: 1',-(.
511 if "vault config " not in str(exc): # pragma: no cover [failsafe] 1',-(.
512 raise
513 return False 1',-(.
514 return True 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAif
517def js_truthiness(value: Any, /) -> bool: # noqa: ANN401
518 """Return the truthiness of the value, according to JavaScript/ECMAScript.
520 Like Python, ECMAScript considers certain values to be false in
521 a boolean context, and every other value to be true. These
522 considerations do not agree: ECMAScript considers [`math.nan`][] to
523 be false too, and empty arrays and objects/dicts to be true,
524 contrary to Python. Because of these discrepancies, we cannot defer
525 to [`bool`][] for ECMAScript truthiness checking, and need
526 a separate, explicit predicate.
528 (Some falsy values in ECMAScript aren't defined in Python:
529 `undefined`, and `document.all`. We do not implement support for
530 those.)
532 !!! note
534 We cannot use a simple `value not in falsy_values` check,
535 because [`math.nan`][] behaves in annoying and obstructive ways.
536 In general, `float('NaN') == float('NaN')` is false, and
537 `float('NaN') != math.nan` and `math.nan != math.nan` are true.
538 CPython says `float('NaN') in [math.nan]` is false, PyPy3 says
539 it is true. Seemingly the only reliable and portable way to
540 check for [`math.nan`][] is to use [`math.isnan`][] directly.
542 Args:
543 value: The value to test.
545 """ # noqa: RUF002
546 try: 1a:mpkqrlsjnothgdeicfb;
547 if value in {None, False, 0, 0.0, ""}: # noqa: B033 1a:mpkqrlsjnothgdeicfb;
548 return False 1a:mkqrlsjnotgdefb
549 except TypeError: 1:cb
550 # All falsy values are hashable, so this can't be falsy.
551 return True 1:cb
552 return not (isinstance(value, float) and math.isnan(value)) 1a:mpkqrlsjnohgdeicfb;
555class CleanupStep(NamedTuple):
556 """A single executed step during vault config cleanup.
558 Attributes:
559 path:
560 A sequence of object keys or array indices to navigate to
561 the JSON value that was cleaned up.
562 old_value:
563 The old value.
564 action:
565 Either `'replace'` if `old_value` was replaced with
566 `new_value`, or `'remove'` if `old_value` was removed.
567 new_value:
568 The new value.
570 """
572 path: Sequence[str | int]
573 """"""
574 old_value: Any
575 """"""
576 action: Literal["replace", "remove"]
577 """"""
578 new_value: Any
579 """"""
582def clean_up_falsy_vault_config_values(
583 obj: Any, # noqa: ANN401
584) -> Sequence[CleanupStep] | None:
585 """Convert falsy values in a vault config to correct types, in-place.
587 Needed for compatibility with vault(1), which sometimes uses only
588 truthiness checks.
590 If vault(1) considered `obj` to be valid, then after clean up,
591 `obj` will be valid as per [`validate_vault_config`][].
593 Args:
594 obj:
595 A presumed valid vault configuration save for using falsy
596 values of the wrong type.
598 Returns:
599 A list of 4-tuples `(key_tup, old_value, action, new_value)`,
600 indicating the cleanup actions performed. `key_tup` is
601 a sequence of object keys and/or array indices indicating the
602 JSON path to the leaf value that was cleaned up, `old_value` is
603 the old value, `new_value` is the new value, and `action` is
604 either `replace` (`old_value` was replaced with `new_value`) or
605 `remove` (`old_value` was removed, and `new_value` is
606 meaningless).
608 If cleanup was never attempted because of an obviously invalid
609 vault configuration, then `None` is returned, directly.
611 """
612 try: 1ampkqrlsjnoth(gdeicfb
613 return list(_VaultConfigValidator(obj).clean_up_falsy_values()) 1ampkqrlsjnoth(gdeicfb
614 except ValueError: 1(c
615 return None 1(c
618# TODO(the-13th-letter): Use type variables local to each class.
619# https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.11
620T_Buffer = TypeVar("T_Buffer", bound=Buffer)
621"""
622A [`TypeVar`][] for classes implementing the [`Buffer`][] interface.
624Warning:
625 Non-public attribute, provided for didactical and educational
626 purposes only. Subject to change without notice, including
627 removal.
629"""
632class SSHKeyCommentPair(NamedTuple, Generic[T_Buffer]):
633 """SSH key plus comment pair. For typing purposes.
635 Attributes:
636 key: SSH key.
637 comment: SSH key comment.
639 """
641 key: T_Buffer
642 """"""
643 comment: T_Buffer
644 """"""
646 def toreadonly(self) -> SSHKeyCommentPair[bytes]:
647 """Return a copy with read-only entries."""
648 return SSHKeyCommentPair( 1^
649 key=bytes(self.key),
650 comment=bytes(self.comment),
651 )
654class SSH_AGENTC(int, enum.Enum): # noqa: N801
655 """SSH agent protocol numbers: client requests.
657 Attributes:
658 REQUEST_IDENTITIES (int):
659 List identities. Expecting
660 [`SSH_AGENT.IDENTITIES_ANSWER`][].
661 SIGN_REQUEST (int):
662 Sign data. Expecting [`SSH_AGENT.SIGN_RESPONSE`][].
663 ADD_IDENTITY (int):
664 Add an (SSH2) identity.
665 REMOVE_IDENTITY (int):
666 Remove an (SSH2) identity.
667 ADD_ID_CONSTRAINED (int):
668 Add an (SSH2) identity, including key constraints.
669 EXTENSION (int):
670 Issue a named request that isn't part of the core agent
671 protocol. Expecting [`SSH_AGENT.EXTENSION_RESPONSE`][] or
672 [`SSH_AGENT.EXTENSION_FAILURE`][] if the named request is
673 supported, [`SSH_AGENT.FAILURE`][] otherwise.
675 """
677 REQUEST_IDENTITIES = 11
678 """"""
679 SIGN_REQUEST = 13
680 """"""
681 ADD_IDENTITY = 17
682 """"""
683 REMOVE_IDENTITY = 18
684 """"""
685 ADD_ID_CONSTRAINED = 25
686 """"""
687 EXTENSION = 27
688 """"""
690 def __bytes__(self) -> bytes: # pragma: no cover [unused]
691 """Return the corresponding request code."""
692 return int.to_bytes(self, 1, "big", signed=False)
695class SSH_AGENT(int, enum.Enum): # noqa: N801
696 """SSH agent protocol numbers: server replies.
698 Attributes:
699 FAILURE (int):
700 Generic failure code.
701 SUCCESS (int):
702 Generic success code.
703 IDENTITIES_ANSWER (int):
704 Successful answer to [`SSH_AGENTC.REQUEST_IDENTITIES`][].
705 SIGN_RESPONSE (int):
706 Successful answer to [`SSH_AGENTC.SIGN_REQUEST`][].
707 EXTENSION_FAILURE (int):
708 Unsuccessful answer to [`SSH_AGENTC.EXTENSION`][].
709 EXTENSION_RESPONSE (int):
710 Successful answer to [`SSH_AGENTC.EXTENSION`][].
712 """
714 FAILURE = 5
715 """"""
716 SUCCESS = 6
717 """"""
718 IDENTITIES_ANSWER = 12
719 """"""
720 SIGN_RESPONSE = 14
721 """"""
722 EXTENSION_FAILURE = 28
723 """"""
724 EXTENSION_RESPONSE = 29
725 """"""
727 def __bytes__(self) -> bytes: # pragma: no cover [unused]
728 """Return the corresponding response code."""
729 return int.to_bytes(self, 1, "big", signed=False) 1a_`{|1}
732class StoreroomKeyPair(NamedTuple, Generic[T_Buffer]):
733 """A pair of AES256 keys, one for encryption and one for signing.
735 Attributes:
736 encryption_key:
737 AES256 key, used for encryption with AES256-CBC (with PKCS#7
738 padding).
739 signing_key:
740 AES256 key, used for signing with HMAC-SHA256.
742 """
744 encryption_key: T_Buffer
745 """"""
746 signing_key: T_Buffer
747 """"""
749 def toreadonly(self) -> StoreroomKeyPair[bytes]:
750 """Return a copy with read-only entries."""
751 return StoreroomKeyPair( 1u=?@[
752 encryption_key=bytes(self.encryption_key),
753 signing_key=bytes(self.signing_key),
754 )
757class StoreroomMasterKeys(NamedTuple, Generic[T_Buffer]):
758 """A triple of AES256 keys, for encryption, signing and hashing.
760 Attributes:
761 hashing_key:
762 AES256 key, used for hashing with HMAC-SHA256 to derive
763 a hash table slot for an item.
764 encryption_key:
765 AES256 key, used for encryption with AES256-CBC (with PKCS#7
766 padding).
767 signing_key:
768 AES256 key, used for signing with HMAC-SHA256.
770 """
772 hashing_key: T_Buffer
773 """"""
774 encryption_key: T_Buffer
775 """"""
776 signing_key: T_Buffer
777 """"""
779 def toreadonly(self) -> StoreroomMasterKeys[bytes]:
780 """Return a copy with read-only entries."""
781 return StoreroomMasterKeys( 2u ~ ab= ? @ [
782 hashing_key=bytes(self.hashing_key),
783 encryption_key=bytes(self.encryption_key),
784 signing_key=bytes(self.signing_key),
785 )
788class FeatureTestEnum(Protocol):
789 """An [`enum.Enum`][] subclass supporting feature tests.
791 Each value of the enum supports the `test` method, which tests
792 whether the feature is enabled or not. (The specific test function
793 may in general require arguments to actually execute the test,
794 though no function currently does.)
796 """
798 def test(self, *args: Any, **kwargs: Any) -> bool: ... # noqa: ANN401
801def _feature_test_function(f: Callable[..., bool]) -> Callable[..., bool]:
802 """Mark a function as a feature test function.
804 This decorator exists purely to hold some shared commentary.
806 Test functions may accept arbitrary arguments, dependent on which
807 feature they test for. They should make sure to raise no
808 exceptions. They may use delayed imports to avoid the import cost
809 until it is actually warranted.
811 Returns:
812 The callable.
814 """
815 return f
818class PEP508Extra(str, enum.Enum):
819 """PEP 508 extras supported by `derivepassphrase`.
821 Attributes:
822 EXPORT:
823 The necessary dependencies to allow the `export` subcommand
824 to handle as many foreign configuration formats as possible.
826 """
828 EXPORT = "export"
829 """"""
831 @_feature_test_function
832 @staticmethod
833 def _test_export() -> bool:
834 """Return true if [`PEP508Extra.EXPORT`][] is currently supported."""
835 import importlib.util # noqa: PLC0415 1%/
837 return importlib.util.find_spec("cryptography") is not None 1%/
839 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
840 """Return true if this extra is enabled."""
841 if self == PEP508Extra.EXPORT: 1%/
842 return self._test_export() 1%/
843 return False # pragma: no cover [unused]
845 __str__ = str.__str__
846 __format__ = str.__format__ # type: ignore[assignment]
849class Feature(str, enum.Enum):
850 """Optional features supported by `derivepassphrase`.
852 Attributes:
853 SSH_KEY:
854 The `vault` subcommand supports using a master SSH key,
855 instead of a master passphrase, if an SSH agent is running
856 and the master SSH key is loaded into it.
858 This feature requires Python support for the SSH agent's
859 chosen communication channel technology.
861 """
863 SSH_KEY = "master SSH key"
864 """"""
866 @_feature_test_function
867 @staticmethod
868 def _test_ssh_key() -> bool: # pragma: no cover [external]
869 """Return true if [`_types.Feature.SSH_KEY`][] is currently supported.
871 We test this by attempting to construct an SSH agent client,
872 reporting whether this can principally work, or not.
874 """
875 # Because BuiltinSSHAgentSocketProvider.test tests the same
876 # thing, just specific to a certain socket provider, we delegate
877 # to that test... not only out of laziness, but also out of
878 # consistency, avoiding two reimplementations of the same logic.
879 from derivepassphrase import ssh_agent # noqa: PLC0415 1%)
881 for provider_name in ssh_agent.SSHAgentClient.SOCKET_PROVIDERS: 1%)
882 try: 1%)
883 provider = BuiltinSSHAgentSocketProvider(provider_name) 1%)
884 except ValueError:
885 continue
886 else:
887 if provider.test(): 1%)
888 return True 1%)
889 return False
891 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
892 """Return true if this feature is enabled."""
893 if self == Feature.SSH_KEY: 1%)
894 return self._test_ssh_key() 1%)
895 return False # pragma: no cover [unused]
897 __str__ = str.__str__
898 __format__ = str.__format__ # type: ignore[assignment]
901class DerivationScheme(str, enum.Enum):
902 """Derivation schemes provided by `derivepassphrase`.
904 Attributes:
905 VAULT:
906 The derivation scheme used by James Coglan's `vault`.
908 """
910 VAULT = "vault"
911 """"""
913 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
914 """Return true if this derivation scheme is enabled."""
915 return self == DerivationScheme.VAULT 1%]
917 __str__ = str.__str__
918 __format__ = str.__format__ # type: ignore[assignment]
921class ForeignConfigurationFormat(str, enum.Enum):
922 """Configuration formats supported by `derivepassphrase export`.
924 Attributes:
925 VAULT_STOREROOM:
926 The vault "storeroom" format for the `export vault`
927 subcommand.
928 VAULT_V02:
929 The vault-native "v0.2" format for the `export vault`
930 subcommand.
931 VAULT_V03:
932 The vault-native "v0.3" format for the `export vault`
933 subcommand.
935 """
937 VAULT_STOREROOM = "vault storeroom"
938 """"""
940 @_feature_test_function
941 @staticmethod
942 def _test_vault_storeroom() -> bool:
943 """Return true if [`ForeignConfigurationFormat.VAULT_STOREROOM`][] is currently supported.""" # noqa: E501
944 from derivepassphrase.exporter import storeroom # noqa: PLC0415 1%/
946 return not storeroom.STUBBED 1%/
948 VAULT_V02 = "vault v0.2"
949 """"""
950 VAULT_V03 = "vault v0.3"
951 """"""
953 @_feature_test_function
954 @staticmethod
955 def _test_vault_v02_v03() -> bool:
956 """Return true if [`ForeignConfigurationFormat.VAULT_V02`][] and [`ForeignConfigurationFormat.VAULT_V03`][] is currently supported.""" # noqa: E501
957 from derivepassphrase.exporter import vault_native # noqa: PLC0415 1%/
959 return not vault_native.STUBBED 1%/
961 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
962 """Return true if this foreign configuration format is enabled."""
963 if self == ForeignConfigurationFormat.VAULT_STOREROOM: 1%/
964 return self._test_vault_storeroom() 1%/
965 if self in { 1%/
966 ForeignConfigurationFormat.VAULT_V02,
967 ForeignConfigurationFormat.VAULT_V03,
968 }:
969 return self._test_vault_v02_v03() 1%/
970 return False # pragma: no cover [unused]
972 __str__ = str.__str__
973 __format__ = str.__format__ # type: ignore[assignment]
976class ExportSubcommand(str, enum.Enum):
977 """Subcommands provided by `derivepassphrase export`.
979 Attributes:
980 VAULT:
981 The `export vault` subcommand.
983 """
985 VAULT = "vault"
986 """"""
988 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
989 """Return true if this `export` subcommand is enabled."""
990 return self == ExportSubcommand.VAULT 2% bb
992 __str__ = str.__str__
993 __format__ = str.__format__ # type: ignore[assignment]
996class Subcommand(str, enum.Enum):
997 """Subcommands provided by `derivepassphrase`.
999 Attributes:
1000 EXPORT:
1001 The `export` subcommand.
1002 VAULT:
1003 The `vault` subcommand.
1005 """
1007 EXPORT = "export"
1008 """"""
1009 VAULT = "vault"
1010 """"""
1012 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
1013 """Return true if this subcommand is enabled."""
1014 return self in {Subcommand.VAULT, Subcommand.EXPORT} 1%]
1016 __str__ = str.__str__
1017 __format__ = str.__format__ # type: ignore[assignment]
1020class BuiltinSSHAgentSocketProvider(str, enum.Enum):
1021 """SSH agent socket providers built into `derivepassphrase`.
1023 Attributes:
1024 SSH_AUTH_SOCK_ON_POSIX:
1025 A socket provider (on POSIX) that queries the
1026 `SSH_AUTH_SOCK` environment variable.
1027 SSH_AUTH_SOCK_ON_WINDOWS:
1028 A socket provider (on The Annoying OS, a.k.a. Microsoft
1029 Windows) that queries the `SSH_AUTH_SOCK` environment
1030 variable.
1031 PAGEANT_ON_WINDOWS:
1032 A socket provider (on The Annoying OS, a.k.a. Microsoft
1033 Windows) that connects to Pageant's standard socket. The
1034 socket address is computed by the socket provider.
1035 OPENSSH_ON_WINDOWS:
1036 A socket provider (on The Annoying OS, a.k.a. Microsoft
1037 Windows) that connects to OpenSSH on Windows's standard
1038 socket. The socket address is hardcoded by the socket
1039 provider.
1040 STUB_AGENT:
1041 A basic fake agent's socket provider that only reacts to
1042 known test keys. Used by the test suite.
1043 STUB_AGENT_WITH_ADDRESS:
1044 A more orchestratable fake agent's socket provider, compared
1045 to [`STUB_AGENT`][], that only reacts to known test keys.
1046 Used by the test suite.
1047 STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA:
1048 An elaborate fake agent's socket provider that only reacts
1049 to known test keys. Used by the test suite.
1050 SSH_AUTH_SOCK:
1051 A registry alias for [`SSH_AUTH_SOCK_ON_POSIX`][].
1052 UNIX_DOMAIN:
1053 A registry alias for [`SSH_AUTH_SOCK_ON_POSIX`][].
1054 POSIX:
1055 A registry alias for [`UNIX_DOMAIN`][].
1056 WINDOWS_NAMED_PIPE:
1057 A registry alias for [`SSH_AUTH_SOCK_ON_WINDOWS`][].
1058 WINDOWS:
1059 A registry alias for [`WINDOWS_NAMED_PIPE`][].
1060 NATIVE:
1061 A registry alias for [`WINDOWS`][] if on The Annoying OS, or
1062 for [`POSIX`][] otherwise.
1064 """
1066 SSH_AUTH_SOCK_ON_POSIX = "ssh_auth_sock_on_posix"
1067 """"""
1068 SSH_AUTH_SOCK_ON_WINDOWS = "ssh_auth_sock_on_windows"
1069 """"""
1070 PAGEANT_ON_WINDOWS = "pageant_on_windows"
1071 """"""
1072 OPENSSH_ON_WINDOWS = "openssh_on_windows"
1073 """"""
1075 STUB_AGENT = "stub_agent"
1076 """"""
1077 STUB_AGENT_WITH_ADDRESS = "stub_agent_with_address"
1078 """"""
1079 STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA = (
1080 "stub_agent_with_address_and_deterministic_dsa"
1081 )
1082 """"""
1084 SSH_AUTH_SOCK = "ssh_auth_sock"
1085 """"""
1086 UNIX_DOMAIN = "unix_domain"
1087 """"""
1088 POSIX = "posix"
1089 """"""
1090 WINDOWS_NAMED_PIPE = "windows_named_pipe"
1091 """"""
1092 WINDOWS = "windows"
1093 """"""
1094 NATIVE = "native"
1095 """"""
1097 def test(
1098 self,
1099 *_args: Any, # noqa: ANN401
1100 **_kwargs: Any, # noqa: ANN401
1101 ) -> bool: # pragma: no cover [external]
1102 """Return true if this SSH agent socket provider is available.
1104 This works by actually attempting to connect to an agent via the
1105 socket provider.
1107 """
1108 import sys # noqa: PLC0415 1%)
1110 import exceptiongroup # noqa: PLC0415 1%)
1112 from derivepassphrase import ssh_agent # noqa: PLC0415 1%)
1114 if sys.version_info < (3, 11): 1%)
1115 from exceptiongroup import BaseExceptionGroup # noqa: PLC0415 1%)
1117 ret = True 1%)
1119 def handle_notimplementederror( 1%)
1120 _exc: BaseExceptionGroup, 1%)
1121 ) -> None: # pragma: no cover [unused] 1%)
1122 nonlocal ret
1123 ret = False 1%)
1125 with exceptiongroup.catch({ # noqa: SIM117 1%)
1126 NotImplementedError: handle_notimplementederror, 1%)
1127 Exception: lambda _exc: None, 1%)
1128 }):
1129 with ssh_agent.SSHAgentClient.ensure_agent_subcontext(conn=[self]): 1%)
1130 pass 1%)
1131 return ret 1%)
1133 def is_known_fake_agent(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401
1134 """Return true if this SSH agent is a known fake agent."""
1135 return self in { 1%)
1136 BuiltinSSHAgentSocketProvider.STUB_AGENT,
1137 BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS,
1138 BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA,
1139 }
1141 __str__ = str.__str__
1142 __format__ = str.__format__ # type: ignore[assignment]
1145@runtime_checkable
1146class SSHAgentSocket(Protocol):
1147 """An abstract networking socket connected to an SSH agent.
1149 The abstract socket supports the [`sendall`][socket.socket.sendall]
1150 and a [`recv`][socket.socket.recv] operation, with the same
1151 signatures and semantics as for "real" sockets. The abstract socket
1152 also supports use as a context manager, for automatically closing
1153 the socket upon exiting the context.
1155 """
1157 def __enter__(self) -> Any: ... # noqa: ANN401
1159 # mypy/typeshed has a *very* lax annotation of
1160 # socket.socket.__exit__, which we need to be compatible with.
1161 # *sigh*
1162 def __exit__(
1163 self,
1164 *args: object,
1165 ) -> bool | None: ...
1167 def sendall(self, data: Buffer, flags: int = 0, /) -> None: ...
1169 def recv(self, bufsize: int, flags: int = 0, /) -> bytes: ...
1172SSHAgentSocketProvider: TypeAlias = "Callable[[], SSHAgentSocket]"
1173"""A callable that provides an SSH agent socket."""
1176class SSHAgentSocketProviderEntry(NamedTuple):
1177 """Registry information for the table of SSH agent socket providers.
1179 Third-party developers can register new socket providers for
1180 auto-discovery by setting up an entry point named
1181 [`derivepassphrase.ssh_agent_socket_providers`][derivepassphrase.ssh_agent.socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME],
1182 referencing an instance of this class. Upon startup of the `vault`
1183 subsystem, `derivepassphrase` will then add appropriate entries to
1184 the registry.
1186 Attributes:
1187 provider: The callable that provides the socket.
1188 key: The table key which this entry is registered as.
1189 aliases: Other keys that shall point to this entry's key.
1191 Note:
1192 The socket provider registry table uses the key as the key, and
1193 the provider as the value. It does not store this info object
1194 directly.
1196 """
1198 provider: SSHAgentSocketProvider
1199 """"""
1200 key: str
1201 """"""
1202 aliases: tuple[str, ...]