Coverage for src / derivepassphrase / vault.py: 100.000%
154 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"""Python port of the vault(1) password generation scheme."""
7from __future__ import annotations
9import base64
10import collections
11import hashlib
12import hmac
13import math
14import types
15from typing import TYPE_CHECKING, Final
17from typing_extensions import TypeAlias, assert_type
19from derivepassphrase import _types, sequin, ssh_agent
21if TYPE_CHECKING:
22 from collections.abc import Callable, Sequence
24 from typing_extensions import Buffer
27class Vault:
28 """A work-alike of James Coglan's vault.
30 Store settings for generating (actually: deriving) passphrases for
31 named services, with various constraints, given only a master
32 passphrase. Also, actually generate the passphrase. The derivation
33 is deterministic and non-secret; only the master passphrase need be
34 kept secret. The implementation is compatible with [vault][].
36 [James Coglan explains the passphrase derivation algorithm in great
37 detail][ALGORITHM] in his blog post on said topic: A principally
38 infinite bit stream is obtained by running a key-derivation function
39 on the master passphrase and the service name, then this bit stream
40 is fed into a [sequin.Sequin][] to generate random numbers in the
41 correct range, and finally these random numbers select passphrase
42 characters until the desired length is reached.
44 [vault]: https://www.npmjs.com/package/vault
45 [ALGORITHM]: https://blog.jcoglan.com/2012/07/16/designing-vaults-generator-algorithm/
47 """
49 UUID: Final = b"e87eb0f4-34cb-46b9-93ad-766c5ab063e7"
50 """A tag used by vault in the bit stream generation."""
51 CHARSETS: Final = types.MappingProxyType(
52 collections.OrderedDict([
53 ("lower", b"abcdefghijklmnopqrstuvwxyz"),
54 ("upper", b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"),
55 (
56 "alpha",
57 (
58 # CHARSETS['lower']
59 b"abcdefghijklmnopqrstuvwxyz"
60 # CHARSETS['upper']
61 b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
62 ),
63 ),
64 ("number", b"0123456789"),
65 (
66 "alphanum",
67 (
68 # CHARSETS['lower']
69 b"abcdefghijklmnopqrstuvwxyz"
70 # CHARSETS['upper']
71 b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
72 # CHARSETS['number']
73 b"0123456789"
74 ),
75 ),
76 ("space", b" "),
77 ("dash", b"-_"),
78 ("symbol", b"!\"#$%&'()*+,./:;<=>?@[\\]^{|}~-_"),
79 (
80 "all",
81 (
82 # CHARSETS['lower']
83 b"abcdefghijklmnopqrstuvwxyz"
84 # CHARSETS['upper']
85 b"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
86 # CHARSETS['number']
87 b"0123456789"
88 # CHARSETS['space']
89 b" "
90 # CHARSETS['symbol']
91 b"!\"#$%&'()*+,./:;<=>?@[\\]^{|}~-_"
92 ),
93 ),
94 ])
95 )
96 """
97 Known character sets from which to draw passphrase characters.
98 Relies on a certain, fixed order for their definition and their
99 contents.
101 """
103 def __init__( # noqa: PLR0913
104 self,
105 *,
106 phrase: Buffer | str = b"",
107 length: int = 20,
108 repeat: int = 0,
109 lower: int | None = None,
110 upper: int | None = None,
111 number: int | None = None,
112 space: int | None = None,
113 dash: int | None = None,
114 symbol: int | None = None,
115 ) -> None:
116 """Initialize the Vault object.
118 Args:
119 phrase:
120 The master passphrase from which to derive the service
121 passphrases. If a string, then the UTF-8 encoding of
122 the string is used.
123 length:
124 Desired passphrase length.
125 repeat:
126 The maximum number of immediate character repetitions
127 allowed in the passphrase. Disabled if set to 0.
128 lower:
129 Optional constraint on ASCII lowercase characters. If
130 positive, include this many lowercase characters
131 somewhere in the passphrase. If 0, avoid lowercase
132 characters altogether.
133 upper:
134 Same as `lower`, but for ASCII uppercase characters.
135 number:
136 Same as `lower`, but for ASCII digits.
137 space:
138 Same as `lower`, but for the space character.
139 dash:
140 Same as `lower`, but for the hyphen-minus and underscore
141 characters.
142 symbol:
143 Same as `lower`, but for all other ASCII printable
144 characters except lowercase characters, uppercase
145 characters, digits, space and backquote.
147 Raises:
148 ValueError:
149 Conflicting passphrase constraints. Permit more
150 characters, or increase the desired passphrase length.
152 Warning:
153 Because of repetition constraints, it is not always possible
154 to detect conflicting passphrase constraints at construction
155 time.
157 """
158 self._phrase = self._get_binary_string(phrase) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
159 self._length = length 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
160 self._repeat = repeat 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
161 self._allowed = bytearray(self.CHARSETS["all"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
162 self._required: list[bytes] = [] 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
164 def subtract_or_require(count: int | None, characters: bytes) -> None: 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
165 if not isinstance(count, int): 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
166 return 1aghBniCDyzjkEcflmIKLHeFG
167 if count <= 0: 1aogphnqrsdtuvwxAyzjkcflmIJLb
168 self._allowed = self._subtract(characters, self._allowed) 1aghjkcflmIJb
169 else:
170 for _ in range(count): 1aopnqrsdtuvwxAyzLb
171 self._required.append(characters) 1aopnqrsdtuvwxAyzLb
173 subtract_or_require(lower, self.CHARSETS["lower"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
174 subtract_or_require(upper, self.CHARSETS["upper"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
175 subtract_or_require(number, self.CHARSETS["number"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
176 subtract_or_require(space, self.CHARSETS["space"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
177 subtract_or_require(dash, self.CHARSETS["dash"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
178 subtract_or_require(symbol, self.CHARSETS["symbol"]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
179 if len(self._required) > self._length: 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFG
180 msg = "requested passphrase length too short" 1L
181 raise ValueError(msg) 1L
182 if not self._allowed: 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJHbeFG
183 msg = "no allowed characters left" 1J
184 raise ValueError(msg) 1J
185 for _ in range(len(self._required), self._length): 1aogphBnqrsidtuvwxCDAyzjkEcflmIKHbeFG
186 self._required.append(bytes(self._allowed)) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKHbeFG
188 def _entropy(self) -> float:
189 """Estimate the passphrase entropy, given the current settings.
191 The entropy is the base 2 logarithm of the amount of
192 possibilities. We operate directly on the logarithms, and use
193 sorting and [`math.fsum`][] to keep high accuracy.
195 Note:
196 We actually overestimate the entropy here because of poor
197 handling of character repetitions. In the extreme, assuming
198 that only one character were allowed, then because there is
199 only one possible string of each given length, the entropy
200 of that string `s` is always be zero. However, we calculate
201 the entropy as `math.log2(math.factorial(len(s)))`, i.e. we
202 assume the characters at the respective string position are
203 distinguishable from each other.
205 Returns:
206 A valid (and somewhat close) upper bound to the entropy.
208 """
209 factors: list[int] = [] 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
210 if not self._required or any(not x for x in self._required): 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
211 return float("-inf") 1I
212 for i, charset in enumerate(self._required): 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
213 factors.extend([i + 1, len(charset)]) 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
214 factors.sort() 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
215 return math.fsum(math.log2(f) for f in factors) 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
217 def _estimate_sufficient_hash_length(
218 self,
219 safety_factor: float = 2.0,
220 ) -> int:
221 """Estimate the sufficient hash length, given the current settings.
223 Using the entropy (via [`_entropy`][]) and a safety factor, give
224 an initial estimate of the length to use for [`create_hash`][]
225 such that using a [`sequin.Sequin`][] with this hash will not
226 exhaust it during passphrase generation.
228 Args:
229 safety_factor: The safety factor. Must be at least 1.
231 Returns:
232 The estimated sufficient hash length.
234 Raises:
235 ValueError: The safety factor is less than 1, or not finite.
237 Warning:
238 This is a heuristic, not an exact computation; it may
239 underestimate the true necessary hash length. It is
240 intended as a starting point for searching for a sufficient
241 hash length, usually by doubling the hash length each time
242 it does not yet prove so.
244 """ # noqa: DOC501
245 try: 1aogphBnqrsidtuvwxCDAyzjkEcflmIKbeFG
246 safety_factor = float(safety_factor) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKbeFG
247 except TypeError as e: 1K
248 msg = f"invalid safety factor: not a float: {safety_factor!r}" 1K
249 raise TypeError(msg) from e 1K
250 if not math.isfinite(safety_factor) or safety_factor < 1.0: 1aogphBnqrsidtuvwxCDAyzjkEcflmIKbeFG
251 msg = f"invalid safety factor {safety_factor!r}" 1K
252 raise ValueError(msg) 1K
253 # Ensure the bound is strictly positive.
254 entropy_bound = max(1, self._entropy()) 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
255 return math.ceil(safety_factor * entropy_bound / 8) 1aogphBnqrsidtuvwxCDAyzjkEcflmIbeFG
257 @staticmethod
258 def _get_binary_string(s: Buffer | str, /) -> bytes:
259 """Convert the input string to a read-only, binary string.
261 If it is a text string, return the string's UTF-8
262 representation.
264 Args:
265 s: The string to (check and) convert.
267 Returns:
268 A read-only, binary copy of the string.
270 """
271 if isinstance(s, str): 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFXMUVGWN
272 return s.encode("UTF-8") 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFXMUVGWN
273 return bytes(s) 1aogphBnqrsidtuvwxCDAyzjkEcflmIKJLHbeFXMUVGN
275 @classmethod
276 def create_hash(
277 cls,
278 phrase: Buffer | str,
279 service: Buffer | str,
280 *,
281 length: int = 32,
282 ) -> bytes:
283 r"""Create a pseudorandom byte stream from phrase and service.
285 Create a pseudorandom byte stream from `phrase` and `service` by
286 feeding them into the key-derivation function PBKDF2
287 (8 iterations, using SHA-1).
289 Args:
290 phrase:
291 A master passphrase, or sometimes an SSH signature.
292 Used as the key for PBKDF2, the underlying cryptographic
293 primitive. If a string, then the UTF-8 encoding of the
294 string is used.
295 service:
296 A vault service name. Will be suffixed with the
297 [`UUID`][], and then used as the salt value for
298 PBKDF2. If a string, then the UTF-8 encoding of the
299 string is used.
300 length:
301 The length of the byte stream to generate.
303 Returns:
304 A pseudorandom byte string of length `length`.
306 Note:
307 Shorter values returned from this method (with the same key
308 and message) are prefixes of longer values returned from
309 this method. (This property is inherited from the
310 underlying PBKDF2 function.) It is thus safe (if slow) to
311 call this method with the same input with ever-increasing
312 target lengths.
314 Examples:
315 >>> # See also Vault.phrase_from_key examples.
316 >>> phrase = bytes.fromhex('''
317 ... 00 00 00 0b 73 73 68 2d 65 64 32 35 35 31 39
318 ... 00 00 00 40
319 ... f0 98 19 80 6c 1a 97 d5 26 03 6e cc e3 65 8f 86
320 ... 66 07 13 19 13 09 21 33 33 f9 e4 36 53 1d af fd
321 ... 0d 08 1f ec f8 73 9b 8c 5f 55 39 16 7c 53 54 2c
322 ... 1e 52 bb 30 ed 7f 89 e2 2f 69 51 55 d8 9e a6 02
323 ... ''')
324 >>> Vault.create_hash(phrase, "some_service", length=4)
325 b'M\xb1<S'
326 >>> Vault.create_hash(phrase, b"some_service", length=16)
327 b'M\xb1<S\x827E\xd1M\xaf\xf8~\xc8n\x10\xcc'
328 >>> Vault.create_hash(phrase, b"NOSUCHSERVICE", length=16)
329 b'\x1c\xc3\x9c\xd9\xb6\x1a\x99CS\x07\xc41\xf4\x85#s'
331 """
332 phrase = cls._get_binary_string(phrase) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFMUVGWN
333 assert isinstance(phrase, bytes) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFMUVGWN
334 salt = cls._get_binary_string(service) + cls.UUID 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFMUVGWN
335 return hashlib.pbkdf2_hmac( 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFMUVGWN
336 hash_name="sha1",
337 password=phrase,
338 salt=salt,
339 iterations=8,
340 dklen=length,
341 )
343 def generate(
344 self,
345 service_name: Buffer | str,
346 /,
347 *,
348 phrase: Buffer | str = b"",
349 ) -> bytes:
350 r"""Generate a service passphrase.
352 Args:
353 service_name:
354 The service name. If a string, then the UTF-8 encoding
355 of the string is used.
356 phrase:
357 If given, override the passphrase given during
358 construction. If a string, then the UTF-8 encoding of
359 the string is used.
361 Returns:
362 The service passphrase.
364 Raises:
365 ValueError:
366 Conflicting passphrase constraints. Permit more
367 characters, or increase the desired passphrase length.
369 Examples:
370 >>> phrase = b"She cells C shells bye the sea shoars"
371 >>> # Using default options in constructor.
372 >>> Vault(phrase=phrase).generate(b"google")
373 b': 4TVH#5:aZl8LueOT\\{'
374 >>> # Also possible:
375 >>> Vault().generate(b"google", phrase=phrase)
376 b': 4TVH#5:aZl8LueOT\\{'
378 Conflicting constraints are sometimes only found during
379 generation.
381 >>> # Note: no error here...
382 >>> v = Vault(
383 ... lower=0,
384 ... upper=0,
385 ... number=0,
386 ... space=2,
387 ... dash=0,
388 ... symbol=1,
389 ... repeat=2,
390 ... length=3,
391 ... )
392 >>> # ... but here.
393 >>> v.generate(
394 ... "0", phrase=b"\x00"
395 ... ) # doctest: +IGNORE_EXCEPTION_DETAIL
396 Traceback (most recent call last):
397 ...
398 ValueError: no allowed characters left
401 """
402 hash_length = self._estimate_sufficient_hash_length() 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
403 assert hash_length >= 1 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
404 # Ensure the phrase and the service name are bytes objects.
405 # This is needed later for safe concatenation.
406 service_name = self._get_binary_string(service_name) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
407 assert_type(service_name, bytes) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
408 if not phrase: 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
409 phrase = self._phrase 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
410 phrase = self._get_binary_string(phrase) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
411 assert_type(phrase, bytes) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
412 # Repeat the passphrase generation with ever-increasing hash
413 # lengths, until the passphrase can be formed without exhausting
414 # the sequin. See the guarantee in the create_hash method for
415 # why this works.
416 while True: 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
417 try: 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
418 required = self._required[:] 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
419 seq = sequin.Sequin( 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
420 self.create_hash(
421 phrase=phrase, service=service_name, length=hash_length
422 )
423 )
424 result = bytearray() 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
425 while len(result) < self._length: 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
426 pos = seq.generate(len(required)) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
427 charset = required.pop(pos) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
428 # Determine if an unlucky choice right now might
429 # violate the restriction on repeated characters.
430 # That is, check if the current partial passphrase
431 # ends with r - 1 copies of the same character
432 # (where r is the repeat limit that must not be
433 # reached), and if so, remove this same character
434 # from the current character's allowed set.
435 if self._repeat and result: 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
436 bad_suffix = bytes(result[-1:]) * (self._repeat - 1) 1aopnqrsdtuvwxcfbe
437 if result.endswith(bad_suffix): 1aopnqrsdtuvwxcfbe
438 charset = self._subtract( 1acfbe
439 bytes(result[-1:]), charset
440 )
441 pos = seq.generate(len(charset)) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
442 result.extend(charset[pos : pos + 1]) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
443 except ValueError as exc: 1aHbe
444 msg = "no allowed characters left" 1ab
445 raise ValueError(msg) from exc 1ab
446 except sequin.SequinExhaustedError: 1He
447 hash_length *= 2 1He
448 else:
449 return bytes(result) 1aogphBnqrsidtuvwxCDAyzjkEcflmHbeFG
451 @staticmethod
452 def is_suitable_ssh_key(
453 key: Buffer,
454 /,
455 *,
456 client: ssh_agent.SSHAgentClient | None = None,
457 ) -> bool:
458 """Check whether the key is suitable for passphrase derivation.
460 Some key types are guaranteed to be deterministic. Other keys
461 types are only deterministic if the SSH agent supports this
462 feature.
464 Args:
465 key:
466 SSH public key to check.
467 client:
468 An optional SSH agent client to check for additional
469 deterministic key types. If not given, assume no such
470 types.
472 Returns:
473 True if and only if the key is guaranteed suitable for use
474 in deriving a passphrase deterministically (perhaps
475 restricted to the indicated SSH agent).
477 """
478 key = bytes(key) 1aSQidOT
479 TestFunc: TypeAlias = "Callable[[bytes | bytearray], bool]" 1aSQidOT
480 deterministic_signature_types: dict[str, TestFunc] = { 1aSQidOT
481 "ssh-ed25519": lambda k: k.startswith(
482 b"\x00\x00\x00\x0bssh-ed25519"
483 ),
484 "ssh-ed448": lambda k: k.startswith(b"\x00\x00\x00\x09ssh-ed448"),
485 "ssh-rsa": lambda k: k.startswith(b"\x00\x00\x00\x07ssh-rsa"),
486 }
487 dsa_signature_types: dict[str, TestFunc] = { 1aSQidOT
488 "ssh-dss": lambda k: k.startswith(b"\x00\x00\x00\x07ssh-dss"),
489 "ecdsa-sha2-nistp256": lambda k: k.startswith(
490 b"\x00\x00\x00\x13ecdsa-sha2-nistp256"
491 ),
492 "ecdsa-sha2-nistp384": lambda k: k.startswith(
493 b"\x00\x00\x00\x13ecdsa-sha2-nistp384"
494 ),
495 "ecdsa-sha2-nistp521": lambda k: k.startswith(
496 b"\x00\x00\x00\x13ecdsa-sha2-nistp521"
497 ),
498 }
499 criteria: list[Callable[[], bool]] = [ 1aSQidOT
500 lambda: any(
501 v(key) for v in deterministic_signature_types.values()
502 ),
503 ]
504 if client is not None: 1aSQidOT
505 criteria.append( 1SQidOT
506 lambda: (
507 client.has_deterministic_dsa_signatures()
508 and any(v(key) for v in dsa_signature_types.values())
509 )
510 )
511 return any(crit() for crit in criteria) 1aSQidOT
513 @classmethod
514 def phrase_from_key(
515 cls,
516 key: Buffer,
517 /,
518 *,
519 conn: ssh_agent.SSHAgentClient
520 | _types.SSHAgentSocket
521 | Sequence[str]
522 | None = None,
523 ) -> bytes:
524 """Obtain the master passphrase from a configured SSH key.
526 vault allows the usage of certain SSH keys to derive a master
527 passphrase, by signing the vault [`UUID`][] with the SSH key.
528 The key type must ensure that signatures are deterministic
529 (perhaps only in conjunction with the given SSH agent).
531 Args:
532 key:
533 The (public) SSH key to use for signing.
534 conn:
535 An optional connection hint to the SSH agent. See
536 [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][].
538 Returns:
539 The signature of the vault [`UUID`][] under this key,
540 unframed but encoded in base64.
542 Raises:
543 derivepassphrase.ssh_agent.socketprovider.NoSuchProviderError:
544 As per
545 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient].
546 Only applicable if agent auto-discovery is used.
547 KeyError:
548 As per
549 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient].
550 Only applicable if agent auto-discovery is used.
551 NotImplementedError:
552 As per
553 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient],
554 including the mulitple raise as an exception group.
555 Only applicable if agent auto-discovery is used.
556 OSError:
557 If the connection hint was a socket, then there was an
558 error setting up the socket connection to the agent.
560 Otherwise, as per
561 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient].
562 Only applicable if agent auto-discovery is used.
563 ValueError:
564 The SSH key is principally unsuitable for this use case.
565 Usually this means that the signature is not
566 deterministic.
568 Examples:
569 >>> import base64
570 >>> # Actual Ed25519 test public key.
571 >>> public_key = bytes.fromhex('''
572 ... 00 00 00 0b 73 73 68 2d 65 64 32 35 35 31 39
573 ... 00 00 00 20
574 ... 81 78 81 68 26 d6 02 48 5f 0f ff 32 48 6f e4 c1
575 ... 30 89 dc 1c 6a 45 06 09 e9 09 0f fb c2 12 69 76
576 ... ''')
577 >>> expected_sig_raw = bytes.fromhex('''
578 ... 00 00 00 0b 73 73 68 2d 65 64 32 35 35 31 39
579 ... 00 00 00 40
580 ... f0 98 19 80 6c 1a 97 d5 26 03 6e cc e3 65 8f 86
581 ... 66 07 13 19 13 09 21 33 33 f9 e4 36 53 1d af fd
582 ... 0d 08 1f ec f8 73 9b 8c 5f 55 39 16 7c 53 54 2c
583 ... 1e 52 bb 30 ed 7f 89 e2 2f 69 51 55 d8 9e a6 02
584 ... ''')
585 >>> # Raw Ed25519 signatures are 64 bytes long.
586 >>> signature_blob = expected_sig_raw[-64:]
587 >>> phrase = base64.standard_b64encode(signature_blob)
588 >>> Vault.phrase_from_key(phrase) == expected # doctest:+SKIP
589 True
591 """
592 with ssh_agent.SSHAgentClient.ensure_agent_subcontext(conn) as client: 1QidO
593 if not cls.is_suitable_ssh_key(key, client=client): 1QidO
594 msg = ( 1O
595 "unsuitable SSH key: bad key, or "
596 "signature not deterministic under this agent"
597 )
598 raise ValueError(msg) 1O
599 raw_sig = client.sign(key, cls.UUID) 1QidO
600 _keytype, trailer = ssh_agent.SSHAgentClient.unstring_prefix(raw_sig) 1idO
601 signature_blob = ssh_agent.SSHAgentClient.unstring(trailer) 1idO
602 return bytes(base64.standard_b64encode(signature_blob)) 1idO
604 @classmethod
605 def phrases_are_interchangable(
606 cls,
607 phrase1: Buffer,
608 phrase2: Buffer,
609 /,
610 ) -> bool:
611 """Return true if the passphrases are interchangable to Vault.
613 Vault internally passes the passphrase as the key to HMAC-SHA1.
614 HMAC requires keys to have a certain fixed length, and therefore
615 transforms keys of other lengths suitably. Because of this, in
616 general, there exist multiple passphrases that behave
617 identically under Vault.
619 Note: HMAC key transformation
620 Keys strictly larger than the SHA1 block size (64 bytes) are
621 first hashed with SHA1, then the digest is used in place of
622 the original key. Then, any keys/digests smaller than the
623 block size are padded with NUL bytes on the right, up to the
624 block size.
626 As a result, keys smaller than the block size are padded,
627 keys larger than the block size are hashed and then padded,
628 and keys exactly as large as the block size are used as-is.
630 Args:
631 phrase1:
632 A passphrase to compare. Must be a binary string to
633 mitigate timing attacks.
634 phrase2:
635 A passphrase to compare. Must be a binary string to
636 mitigate timing attacks.
638 Warning: Likely non-resistant to timing attacks
639 This method makes some effort to be resistant to timing
640 attacks, but cannot guarantee that Python
641 micro-optimizations, version or platform differences affect
642 the effectiveness of these efforts.
644 Callers can definitely observe timing differences due to the
645 length of the passphrase passed in.
647 """
648 to_key = cls._phrase_to_hmac_key 1MN
649 return hmac.compare_digest(to_key(phrase1), to_key(phrase2)) 1MN
651 @classmethod
652 def _phrase_to_hmac_key(
653 cls,
654 phrase: Buffer | str,
655 /,
656 ) -> bytes:
657 r"""Return the HMAC key belonging to a passphrase.
659 This is the actual HMAC key this passphrase would be transformed
660 into when used within Vault.
662 See [`phrases_are_interchangable`][] for further explanations
663 and warnings about timing attack resistance.
665 Args:
666 phrase:
667 A passphrase to compare. Must be a binary string to
668 mitigate timing attacks.
670 """
671 phrase = cls._get_binary_string(phrase) 1MN
672 h = hashlib.sha1(phrase, usedforsecurity=False) 1MN
673 try: 1MN
674 key = bytearray(h.block_size) 1MN
675 for i, byte in enumerate(phrase): 1MN
676 key[i] = byte 1MN
677 return bytes(key) 1MN
678 except IndexError: 1M
679 return h.digest() + b"\x00" * (h.block_size - h.digest_size) 1M
681 @staticmethod
682 def _subtract(
683 charset: Buffer,
684 allowed: Buffer,
685 ) -> bytearray:
686 """Remove the characters in charset from allowed.
688 This preserves the relative order of characters in `allowed`.
690 Args:
691 charset:
692 Characters to remove. Must not contain duplicate
693 characters.
694 allowed:
695 Character set to remove the other characters from. Must
696 not contain duplicate characters.
698 Returns:
699 The pruned "allowed" character set.
701 Raises:
702 ValueError:
703 `allowed` or `charset` contained duplicate characters.
705 """
706 allowed = ( 1aghjkcflmPRIJbe
707 allowed if isinstance(allowed, bytearray) else bytearray(allowed)
708 )
709 assert_type(allowed, bytearray) 1aghjkcflmPRIJbe
710 charset = memoryview(charset).toreadonly().cast("c") 1aghjkcflmPRIJbe
711 assert_type(charset, "memoryview[bytes]") 1aghjkcflmPRIJbe
712 msg_dup_characters = "duplicate characters in set" 1aghjkcflmPRIJbe
713 if len(frozenset(allowed)) != len(allowed): 1aghjkcflmPRIJbe
714 raise ValueError(msg_dup_characters) 1R
715 if len(frozenset(charset)) != len(charset): 1aghjkcflmPRIJbe
716 raise ValueError(msg_dup_characters) 1R
717 for c in charset: 1aghjkcflmPIJbe
718 try: 1aghjkcflmPIJbe
719 pos = allowed.index(c) 1aghjkcflmPIJbe
720 except ValueError: 1acJb
721 pass 1acJb
722 else:
723 allowed[pos : pos + 1] = [] 1aghjkcflmPIJbe
724 return allowed 1aghjkcflmPIJbe