derivepassphrase.git

1) #!/usr/bin/python3
2) 

3) from __future__ import annotations
4) 

5) import base64
6) import glob
7) import json
8) import logging
9) import os
10) import os.path
11) import struct

12) from typing import TYPE_CHECKING, Any, TypedDict

13) 
14) from cryptography.hazmat.primitives import ciphers, hashes, hmac, padding
15) from cryptography.hazmat.primitives.ciphers import algorithms, modes
16) from cryptography.hazmat.primitives.kdf import pbkdf2
17) 

18) if TYPE_CHECKING:
19)     from collections.abc import Iterator
20) 

21) STOREROOM_MASTER_KEYS_UUID = b'35b7c7ed-f71e-4adf-9051-02fb0f1e0e17'
22) VAULT_CIPHER_UUID = b'73e69e8a-cb05-4b50-9f42-59d76a511299'
23) IV_SIZE = 16
24) KEY_SIZE = MAC_SIZE = 32
25) ENCRYPTED_KEYPAIR_SIZE = 128
26) VERSION_SIZE = 1
27) MASTER_KEYS_KEY = (
28)     os.getenv('VAULT_KEY')
29)     or os.getenv('LOGNAME')
30)     or os.getenv('USER')
31)     or os.getenv('USERNAME')
32) )

33) VAULT_PATH = os.path.join(
34)     os.path.expanduser('~'), os.getenv('VAULT_PATH', '.vault')
35) )

36) 

37) logger = logging.getLogger(__name__)

38) 
39) 
40) class KeyPair(TypedDict):
41)     encryption_key: bytes
42)     signing_key: bytes
43) 
44) 
45) class MasterKeys(TypedDict):
46)     hashing_key: bytes
47)     encryption_key: bytes
48)     signing_key: bytes
49) 
50) 
51) def derive_master_keys_keys(password: str | bytes, iterations: int) -> KeyPair:

52)     """Derive encryption and signing keys for the master keys data.
53) 
54)     The master password is run through a key derivation function to
55)     obtain a 64-byte string, which is then split to yield two 32-byte
56)     keys.  The key derivation function is PBKDF2, using HMAC-SHA1 and
57)     salted with the storeroom master keys UUID.
58) 
59)     Args:
60)         password:
61)             A master password for the storeroom instance.  Usually read
62)             from the `VAULT_KEY` environment variable, otherwise
63)             defaults to the username.
64)         iterations:
65)             A count of rounds for the underlying key derivation
66)             function.  Usually stored as a setting next to the encrypted
67)             master keys data.
68) 
69)     Returns:
70)         A 2-tuple of keys, the encryption key and the signing key, to
71)         decrypt and verify the master keys data with.
72) 
73)     """

74)     if isinstance(password, str):
75)         password = password.encode('ASCII')
76)     master_keys_keys_blob = pbkdf2.PBKDF2HMAC(
77)         algorithm=hashes.SHA1(),  # noqa: S303

78)         length=2 * KEY_SIZE,

79)         salt=STOREROOM_MASTER_KEYS_UUID,
80)         iterations=iterations,
81)     ).derive(password)
82)     encryption_key, signing_key = struct.unpack(
83)         f'{KEY_SIZE}s {KEY_SIZE}s', master_keys_keys_blob
84)     )
85)     logger.debug(
86)         (
87)             'derived master_keys_keys bytes.fromhex(%s) (encryption) '
88)             'and bytes.fromhex(%s) (signing) '
89)             'from password bytes.fromhex(%s), '
90)             'using call '
91)             'pbkdf2(algorithm=%s, length=%d, salt=%s, iterations=%d)'
92)         ),
93)         repr(encryption_key.hex(' ')),
94)         repr(signing_key.hex(' ')),
95)         repr(password.hex(' ')),
96)         repr('SHA256'),
97)         64,
98)         repr(STOREROOM_MASTER_KEYS_UUID),
99)         iterations,
100)     )
101)     return {
102)         'encryption_key': encryption_key,
103)         'signing_key': signing_key,
104)     }
105) 
106) 
107) def decrypt_master_keys_data(data: bytes, keys: KeyPair) -> MasterKeys:

108)     """Decrypt the master keys data.
109) 
110)     The master keys data contains:
111) 
112)     - a 16-byte IV,
113)     - a 96-byte AES256-CBC-encrypted payload (using PKCS7 padding on the
114)       inside), and
115)     - a 32-byte MAC of the preceding 112 bytes.
116) 
117)     The decrypted payload itself consists of three 32-byte keys: the
118)     hashing, encryption and signing keys, in that order.
119) 
120)     The encrypted payload is encrypted with the encryption key, and the
121)     MAC is created based on the signing key.  As per standard
122)     cryptographic procedure, the MAC can be verified before attempting
123)     to decrypt the payload.
124) 
125)     Because the payload size is both fixed and a multiple of the
126)     cipher blocksize, in this case, the PKCS7 padding is a no-op.
127) 
128)     Args:
129)         data:
130)             The encrypted master keys data.
131)         keys:
132)             The encryption and signing keys for the master keys data.
133)             These should have previously been derived via the
134)             [`derivepassphrase.exporter.storeroom.derive_master_keys_keys`][]
135)             function.
136) 
137)     Returns:
138)         The master encryption, signing and hashing keys.
139) 
140)     """

141)     ciphertext, claimed_mac = struct.unpack(
142)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
143)     )
144)     actual_mac = hmac.HMAC(keys['signing_key'], hashes.SHA256())
145)     actual_mac.update(ciphertext)
146)     logger.debug(
147)         (
148)             'master_keys_data mac_key = bytes.fromhex(%s), '
149)             'hashed_content = bytes.fromhex(%s), '
150)             'claimed_mac = bytes.fromhex(%s), '
151)             'actual_mac = bytes.fromhex(%s)'
152)         ),
153)         repr(keys['signing_key'].hex(' ')),
154)         repr(ciphertext.hex(' ')),
155)         repr(claimed_mac.hex(' ')),
156)         repr(actual_mac.copy().finalize().hex(' ')),
157)     )
158)     actual_mac.verify(claimed_mac)
159) 
160)     iv, payload = struct.unpack(
161)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
162)     )
163)     decryptor = ciphers.Cipher(
164)         algorithms.AES256(keys['encryption_key']), modes.CBC(iv)
165)     ).decryptor()
166)     padded_plaintext = bytearray()
167)     padded_plaintext.extend(decryptor.update(payload))
168)     padded_plaintext.extend(decryptor.finalize())
169)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
170)     plaintext = bytearray()
171)     plaintext.extend(unpadder.update(padded_plaintext))
172)     plaintext.extend(unpadder.finalize())
173)     if len(plaintext) != 3 * KEY_SIZE:
174)         msg = (
175)             f'Expecting 3 encrypted keys at {3 * KEY_SIZE} bytes total, '
176)             f'but found {len(plaintext)} instead'
177)         )
178)         raise RuntimeError(msg)
179)     hashing_key, encryption_key, signing_key = struct.unpack(
180)         f'{KEY_SIZE}s {KEY_SIZE}s {KEY_SIZE}s', plaintext
181)     )
182)     return {
183)         'hashing_key': hashing_key,
184)         'encryption_key': encryption_key,
185)         'signing_key': signing_key,
186)     }
187) 
188) 

189) def decrypt_session_keys(data: bytes, master_keys: MasterKeys) -> KeyPair:
190)     """Decrypt the bucket item's session keys.
191) 
192)     The bucket item's session keys are single-use keys for encrypting
193)     and signing a single item in the storage bucket.  The encrypted
194)     session key data consists of:
195) 
196)     - a 16-byte IV,
197)     - a 64-byte AES256-CBC-encrypted payload (using PKCS7 padding on the
198)       inside), and
199)     - a 32-byte MAC of the preceding 80 bytes.
200) 
201)     The encrypted payload is encrypted with the master encryption key,
202)     and the MAC is created with the master signing key.  As per standard
203)     cryptographic procedure, the MAC can be verified before attempting
204)     to decrypt the payload.
205) 
206)     Because the payload size is both fixed and a multiple of the
207)     cipher blocksize, in this case, the PKCS7 padding is a no-op.
208) 
209)     Args:
210)         data:
211)             The encrypted bucket item session key data.
212)         master_keys:
213)             The master keys.  Presumably these have previously been
214)             obtained via the
215)             [`derivepassphrase.exporter.storeroom.decrypt_master_keys_data`][]
216)             function.
217) 
218)     Returns:
219)         The bucket item's encryption and signing keys.
220) 
221)     """
222) 

223)     ciphertext, claimed_mac = struct.unpack(
224)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
225)     )

226)     actual_mac = hmac.HMAC(master_keys['signing_key'], hashes.SHA256())

227)     actual_mac.update(ciphertext)
228)     logger.debug(
229)         (

230)             'decrypt_bucket_item (session_keys): '

231)             'mac_key = bytes.fromhex(%s) (master), '
232)             'hashed_content = bytes.fromhex(%s), '
233)             'claimed_mac = bytes.fromhex(%s), '
234)             'actual_mac = bytes.fromhex(%s)'
235)         ),

236)         repr(master_keys['signing_key'].hex(' ')),

237)         repr(ciphertext.hex(' ')),
238)         repr(claimed_mac.hex(' ')),
239)         repr(actual_mac.copy().finalize().hex(' ')),
240)     )
241)     actual_mac.verify(claimed_mac)
242) 
243)     iv, payload = struct.unpack(
244)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
245)     )
246)     decryptor = ciphers.Cipher(

247)         algorithms.AES256(master_keys['encryption_key']), modes.CBC(iv)

248)     ).decryptor()
249)     padded_plaintext = bytearray()
250)     padded_plaintext.extend(decryptor.update(payload))
251)     padded_plaintext.extend(decryptor.finalize())
252)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
253)     plaintext = bytearray()
254)     plaintext.extend(unpadder.update(padded_plaintext))
255)     plaintext.extend(unpadder.finalize())
256) 
257)     session_encryption_key, session_signing_key, inner_payload = struct.unpack(
258)         f'{KEY_SIZE}s {KEY_SIZE}s {len(plaintext) - 2 * KEY_SIZE}s',
259)         plaintext,
260)     )
261)     session_keys: KeyPair = {
262)         'encryption_key': session_encryption_key,
263)         'signing_key': session_signing_key,
264)     }
265) 
266)     logger.debug(
267)         (

268)             'decrypt_bucket_item (session_keys): '

269)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
270)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
271)             '= bytes.fromhex(%s) '
272)             '= {"encryption_key": bytes.fromhex(%s), '
273)             '"signing_key": bytes.fromhex(%s)}'
274)         ),

275)         repr(master_keys['encryption_key'].hex(' ')),

276)         repr(iv.hex(' ')),
277)         repr(payload.hex(' ')),
278)         repr(plaintext.hex(' ')),
279)         repr(session_keys['encryption_key'].hex(' ')),
280)         repr(session_keys['signing_key'].hex(' ')),
281)     )
282) 
283)     if inner_payload:
284)         logger.debug(
285)             'ignoring misplaced inner payload bytes.fromhex(%s)',
286)             repr(inner_payload.hex(' ')),
287)         )
288) 
289)     return session_keys
290) 
291) 

292) def decrypt_contents(data: bytes, session_keys: KeyPair) -> bytes:
293)     """Decrypt the bucket item's contents.
294) 
295)     The data consists of:
296) 
297)     - a 16-byte IV,
298)     - a variable-sized AES256-CBC-encrypted payload (using PKCS7 padding
299)       on the inside), and
300)     - a 32-byte MAC of the preceding 80 bytes.
301) 
302)     The encrypted payload is encrypted with the bucket item's session
303)     encryption key, and the MAC is created with the bucket item's
304)     session signing key.  As per standard cryptographic procedure, the
305)     MAC can be verified before attempting to decrypt the payload.
306) 
307)     Args:
308)         data:
309)             The encrypted bucket item payload data.
310)         session_keys:
311)             The bucket item's session keys.  Presumably these have
312)             previously been obtained via the
313)             [`derivepassphrase.exporter.storeroom.decrypt_session_keys`][]
314)             function.
315) 
316)     Returns:
317)         The bucket item's payload.
318) 
319)     """
320) 

321)     ciphertext, claimed_mac = struct.unpack(
322)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
323)     )

324)     actual_mac = hmac.HMAC(session_keys['signing_key'], hashes.SHA256())

325)     actual_mac.update(ciphertext)
326)     logger.debug(
327)         (

328)             'decrypt_bucket_item (contents): '

329)             'mac_key = bytes.fromhex(%s), '
330)             'hashed_content = bytes.fromhex(%s), '
331)             'claimed_mac = bytes.fromhex(%s), '
332)             'actual_mac = bytes.fromhex(%s)'
333)         ),

334)         repr(session_keys['signing_key'].hex(' ')),

335)         repr(ciphertext.hex(' ')),
336)         repr(claimed_mac.hex(' ')),
337)         repr(actual_mac.copy().finalize().hex(' ')),
338)     )
339)     actual_mac.verify(claimed_mac)
340) 
341)     iv, payload = struct.unpack(
342)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
343)     )
344)     decryptor = ciphers.Cipher(

345)         algorithms.AES256(session_keys['encryption_key']), modes.CBC(iv)

346)     ).decryptor()
347)     padded_plaintext = bytearray()
348)     padded_plaintext.extend(decryptor.update(payload))
349)     padded_plaintext.extend(decryptor.finalize())
350)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
351)     plaintext = bytearray()
352)     plaintext.extend(unpadder.update(padded_plaintext))
353)     plaintext.extend(unpadder.finalize())
354) 
355)     logger.debug(
356)         (

357)             'decrypt_bucket_item (contents): '

358)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
359)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
360)             '= bytes.fromhex(%s)'
361)         ),

362)         repr(session_keys['encryption_key'].hex(' ')),

363)         repr(iv.hex(' ')),
364)         repr(payload.hex(' ')),
365)         repr(plaintext.hex(' ')),
366)     )
367) 
368)     return plaintext
369) 
370) 

371) def decrypt_bucket_item(bucket_item: bytes, master_keys: MasterKeys) -> bytes:

372)     logger.debug(
373)         (

374)             'decrypt_bucket_item: data = bytes.fromhex(%s), '

375)             'encryption_key = bytes.fromhex(%s), '
376)             'signing_key = bytes.fromhex(%s)'
377)         ),

378)         repr(bucket_item.hex(' ')),

379)         repr(master_keys['encryption_key'].hex(' ')),
380)         repr(master_keys['signing_key'].hex(' ')),
381)     )
382)     data_version, encrypted_session_keys, data_contents = struct.unpack(
383)         (
384)             f'B {ENCRYPTED_KEYPAIR_SIZE}s '

385)             f'{len(bucket_item) - 1 - ENCRYPTED_KEYPAIR_SIZE}s'

386)         ),

387)         bucket_item,

388)     )
389)     if data_version != 1:
390)         msg = f'Cannot handle version {data_version} encrypted data'
391)         raise RuntimeError(msg)
392)     session_keys = decrypt_session_keys(encrypted_session_keys, master_keys)
393)     return decrypt_contents(data_contents, session_keys)
394) 
395) 

396) def decrypt_bucket_file(
397)     filename: str, master_keys: MasterKeys
398) ) -> Iterator[bytes]:
399)     with open(filename, 'rb') as bucket_file:

400)         header_line = bucket_file.readline()
401)         try:
402)             header = json.loads(header_line)
403)         except ValueError as exc:
404)             msg = f'Invalid bucket file: {filename}'
405)             raise RuntimeError(msg) from exc
406)         if header != {'version': 1}:
407)             msg = f'Invalid bucket file: {filename}'
408)             raise RuntimeError(msg) from None
409)         for line in bucket_file:

410)             yield decrypt_bucket_item(
411)                 base64.standard_b64decode(line), master_keys

412)             )
413) 
414) 

415) def store(config: dict[str, Any], path: str, json_contents: bytes) -> None:
416)     """Store the JSON contents at path in the config structure.
417) 
418)     Traverse the config structure according to path, and set the value
419)     of the leaf to the decoded JSON contents.
420) 
421)     A path `/foo/bar/xyz` translates to the JSON structure
422)     `{"foo": {"bar": {"xyz": ...}}}`.
423) 
424)     Args:
425)         config:
426)             The (top-level) configuration structure to update.
427)         path:
428)             The path within the configuration structure to traverse.
429)         json_contents:
430)             The contents to set the item to, after JSON-decoding.
431) 
432)     Raises:
433)         json.JSONDecodeError:
434)             There was an error parsing the JSON contents.
435) 
436)     """
437)     contents = json.loads(json_contents)
438)     path_parts = [part for part in path.split('/') if part]
439)     for part in path_parts[:-1]:
440)         config = config.setdefault(part, {})
441)     if path_parts:
442)         config[path_parts[-1]] = contents
443) 
444) 
445) def export_storeroom_data(
446)     storeroom_path: str | bytes | os.PathLike = VAULT_PATH,
447)     master_keys_key: str | bytes | None = MASTER_KEYS_KEY,
448) ) -> dict[str, Any]:
449)     """Export the full configuration stored in the storeroom.
450) 
451)     Args:
452)         storeroom_path:
453)             Path to the storeroom; usually `~/.vault`.
454)         master_keys_key:
455)             Encryption key/password for the master keys.  If not set via
456)             the `VAULT_KEY` environment variable, this usually is the
457)             user's username.
458) 
459)     Returns:
460)         The full configuration, as stored in the storeroom.
461) 
462)         This may or may not be a valid configuration according to vault
463)         or derivepassphrase.
464) 
465)     Raises:
466)         RuntimeError:
467)             Something went wrong during data collection, e.g. we
468)             encountered unsupported or corrupted data in the storeroom.
469)         json.JSONDecodeError:
470)             An internal JSON data structure failed to parse from disk.
471)             The storeroom is probably corrupted.
472) 
473)     """
474) 
475)     if master_keys_key is None:
476)         msg = 'Cannot determine master key; please set VAULT_KEY'
477)         raise RuntimeError(msg)
478)     with open(
479)         os.path.join(os.fsdecode(storeroom_path), '.keys'), encoding='utf-8'
480)     ) as master_keys_file:

481)         header = json.loads(master_keys_file.readline())
482)         if header != {'version': 1}:
483)             msg = 'bad or unsupported keys version header'
484)             raise RuntimeError(msg)
485)         raw_keys_data = base64.standard_b64decode(master_keys_file.readline())
486)         encrypted_keys_params, encrypted_keys = struct.unpack(
487)             f'B {len(raw_keys_data) - 1}s', raw_keys_data
488)         )
489)         if master_keys_file.read():
490)             msg = 'trailing data; cannot make sense of .keys file'
491)             raise RuntimeError(msg)
492)     encrypted_keys_version = encrypted_keys_params >> 4
493)     if encrypted_keys_version != 1:
494)         msg = f'cannot handle version {encrypted_keys_version} encrypted keys'
495)         raise RuntimeError(msg)
496)     encrypted_keys_iterations = 2 ** (10 + (encrypted_keys_params & 0x0F))
497)     master_keys_keys = derive_master_keys_keys(

498)         master_keys_key, encrypted_keys_iterations

499)     )
500)     master_keys = decrypt_master_keys_data(encrypted_keys, master_keys_keys)
501) 

502)     config_structure: dict[str, Any] = {}
503)     json_contents: dict[str, bytes] = {}

504)     for file in glob.glob('[01][0-9a-f]'):

505)         bucket_contents = list(decrypt_bucket_file(file, master_keys))
506)         bucket_index = json.loads(bucket_contents.pop(0))
507)         for pos, item in enumerate(bucket_index):
508)             json_contents[item] = bucket_contents[pos]
509)             logger.debug(
510)                 'Found bucket item: %s -> %s', item, bucket_contents[pos]
511)             )
512)     dirs_to_check: dict[str, list[str]] = {}
513)     json_payload: Any
514)     for path, json_content in sorted(json_contents.items()):
515)         if path.endswith('/'):
516)             logger.debug(
517)                 'Postponing dir check: %s -> %s',
518)                 path,
519)                 json_content.decode('utf-8'),
520)             )
521)             json_payload = json.loads(json_content)
522)             if not isinstance(json_payload, list) or any(
523)                 not isinstance(x, str) for x in json_payload
524)             ):
525)                 msg = (
526)                     f'Directory index is not actually an index: '
527)                     f'{json_content!r}'
528)                 )
529)                 raise RuntimeError(msg)
530)             dirs_to_check[path] = json_payload
531)             logger.debug(
532)                 'Setting contents (empty directory): %s -> %s', path, '{}'
533)             )
534)             store(config_structure, path, b'{}')
535)         else:
536)             logger.debug(
537)                 'Setting contents: %s -> %s',
538)                 path,
539)                 json_content.decode('utf-8'),
540)             )
541)             store(config_structure, path, json_content)
542)     for _dir, namelist in dirs_to_check.items():
543)         namelist = [x.rstrip('/') for x in namelist]  # noqa: PLW2901
544)         try:
545)             obj = config_structure
546)             for part in _dir.split('/'):
547)                 if part:
548)                     obj = obj[part]
549)         except KeyError as exc:
550)             msg = f'Cannot traverse storage path: {_dir!r}'
551)             raise RuntimeError(msg) from exc
552)         if set(obj.keys()) != set(namelist):
553)             msg = f'Object key mismatch for path {_dir!r}'
554)             raise RuntimeError(msg)
555)     return config_structure

556) 
557) 
558) if __name__ == '__main__':