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) from derivepassphrase import exporter
19) 

20) if TYPE_CHECKING:
21)     from collections.abc import Iterator
22) 

23) STOREROOM_MASTER_KEYS_UUID = b'35b7c7ed-f71e-4adf-9051-02fb0f1e0e17'
24) VAULT_CIPHER_UUID = b'73e69e8a-cb05-4b50-9f42-59d76a511299'
25) IV_SIZE = 16
26) KEY_SIZE = MAC_SIZE = 32
27) ENCRYPTED_KEYPAIR_SIZE = 128
28) VERSION_SIZE = 1
29) 

30) logger = logging.getLogger(__name__)

31) 
32) 
33) class KeyPair(TypedDict):
34)     encryption_key: bytes
35)     signing_key: bytes
36) 
37) 
38) class MasterKeys(TypedDict):
39)     hashing_key: bytes
40)     encryption_key: bytes
41)     signing_key: bytes
42) 
43) 
44) def derive_master_keys_keys(password: str | bytes, iterations: int) -> KeyPair:

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

67)     if isinstance(password, str):
68)         password = password.encode('ASCII')
69)     master_keys_keys_blob = pbkdf2.PBKDF2HMAC(
70)         algorithm=hashes.SHA1(),  # noqa: S303

71)         length=2 * KEY_SIZE,

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

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

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

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

216)     ciphertext, claimed_mac = struct.unpack(
217)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
218)     )

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

220)     actual_mac.update(ciphertext)
221)     logger.debug(
222)         (

223)             'decrypt_bucket_item (session_keys): '

224)             'mac_key = bytes.fromhex(%s) (master), '
225)             'hashed_content = bytes.fromhex(%s), '
226)             'claimed_mac = bytes.fromhex(%s), '
227)             'actual_mac = bytes.fromhex(%s)'
228)         ),

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

230)         repr(ciphertext.hex(' ')),
231)         repr(claimed_mac.hex(' ')),
232)         repr(actual_mac.copy().finalize().hex(' ')),
233)     )
234)     actual_mac.verify(claimed_mac)
235) 
236)     iv, payload = struct.unpack(
237)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
238)     )
239)     decryptor = ciphers.Cipher(

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

241)     ).decryptor()
242)     padded_plaintext = bytearray()
243)     padded_plaintext.extend(decryptor.update(payload))
244)     padded_plaintext.extend(decryptor.finalize())
245)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
246)     plaintext = bytearray()
247)     plaintext.extend(unpadder.update(padded_plaintext))
248)     plaintext.extend(unpadder.finalize())
249) 
250)     session_encryption_key, session_signing_key, inner_payload = struct.unpack(
251)         f'{KEY_SIZE}s {KEY_SIZE}s {len(plaintext) - 2 * KEY_SIZE}s',
252)         plaintext,
253)     )
254)     session_keys: KeyPair = {
255)         'encryption_key': session_encryption_key,
256)         'signing_key': session_signing_key,
257)     }
258) 
259)     logger.debug(
260)         (

261)             'decrypt_bucket_item (session_keys): '

262)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
263)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
264)             '= bytes.fromhex(%s) '
265)             '= {"encryption_key": bytes.fromhex(%s), '
266)             '"signing_key": bytes.fromhex(%s)}'
267)         ),

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

269)         repr(iv.hex(' ')),
270)         repr(payload.hex(' ')),
271)         repr(plaintext.hex(' ')),
272)         repr(session_keys['encryption_key'].hex(' ')),
273)         repr(session_keys['signing_key'].hex(' ')),
274)     )
275) 
276)     if inner_payload:
277)         logger.debug(
278)             'ignoring misplaced inner payload bytes.fromhex(%s)',
279)             repr(inner_payload.hex(' ')),
280)         )
281) 
282)     return session_keys
283) 
284) 

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

314)     ciphertext, claimed_mac = struct.unpack(
315)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
316)     )

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

318)     actual_mac.update(ciphertext)
319)     logger.debug(
320)         (

321)             'decrypt_bucket_item (contents): '

322)             'mac_key = bytes.fromhex(%s), '
323)             'hashed_content = bytes.fromhex(%s), '
324)             'claimed_mac = bytes.fromhex(%s), '
325)             'actual_mac = bytes.fromhex(%s)'
326)         ),

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

328)         repr(ciphertext.hex(' ')),
329)         repr(claimed_mac.hex(' ')),
330)         repr(actual_mac.copy().finalize().hex(' ')),
331)     )
332)     actual_mac.verify(claimed_mac)
333) 
334)     iv, payload = struct.unpack(
335)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
336)     )
337)     decryptor = ciphers.Cipher(

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

339)     ).decryptor()
340)     padded_plaintext = bytearray()
341)     padded_plaintext.extend(decryptor.update(payload))
342)     padded_plaintext.extend(decryptor.finalize())
343)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
344)     plaintext = bytearray()
345)     plaintext.extend(unpadder.update(padded_plaintext))
346)     plaintext.extend(unpadder.finalize())
347) 
348)     logger.debug(
349)         (

350)             'decrypt_bucket_item (contents): '

351)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
352)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
353)             '= bytes.fromhex(%s)'
354)         ),

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

356)         repr(iv.hex(' ')),
357)         repr(payload.hex(' ')),
358)         repr(plaintext.hex(' ')),
359)     )
360) 
361)     return plaintext
362) 
363) 

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

365)     logger.debug(
366)         (

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

368)             'encryption_key = bytes.fromhex(%s), '
369)             'signing_key = bytes.fromhex(%s)'
370)         ),

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

372)         repr(master_keys['encryption_key'].hex(' ')),
373)         repr(master_keys['signing_key'].hex(' ')),
374)     )
375)     data_version, encrypted_session_keys, data_contents = struct.unpack(
376)         (
377)             f'B {ENCRYPTED_KEYPAIR_SIZE}s '

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

379)         ),

380)         bucket_item,

381)     )
382)     if data_version != 1:
383)         msg = f'Cannot handle version {data_version} encrypted data'
384)         raise RuntimeError(msg)
385)     session_keys = decrypt_session_keys(encrypted_session_keys, master_keys)
386)     return decrypt_contents(data_contents, session_keys)
387) 
388) 

389) def decrypt_bucket_file(

390)     filename: str,
391)     master_keys: MasterKeys,
392)     *,
393)     root_dir: str | bytes | os.PathLike = '.',

394) ) -> Iterator[bytes]:

395)     with open(
396)         os.path.join(os.fsdecode(root_dir), filename), 'rb'
397)     ) as bucket_file:

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

408)             yield decrypt_bucket_item(
409)                 base64.standard_b64decode(line), master_keys

410)             )
411) 
412) 

413) def store(config: dict[str, Any], path: str, json_contents: bytes) -> None:
414)     """Store the JSON contents at path in the config structure.
415) 
416)     Traverse the config structure according to path, and set the value
417)     of the leaf to the decoded JSON contents.
418) 
419)     A path `/foo/bar/xyz` translates to the JSON structure
420)     `{"foo": {"bar": {"xyz": ...}}}`.
421) 
422)     Args:
423)         config:
424)             The (top-level) configuration structure to update.
425)         path:
426)             The path within the configuration structure to traverse.
427)         json_contents:
428)             The contents to set the item to, after JSON-decoding.
429) 
430)     Raises:
431)         json.JSONDecodeError:
432)             There was an error parsing the JSON contents.
433) 
434)     """
435)     contents = json.loads(json_contents)
436)     path_parts = [part for part in path.split('/') if part]
437)     for part in path_parts[:-1]:
438)         config = config.setdefault(part, {})
439)     if path_parts:
440)         config[path_parts[-1]] = contents
441) 
442) 
443) def export_storeroom_data(

444)     storeroom_path: str | bytes | os.PathLike | None = None,
445)     master_keys_key: str | bytes | None = None,

446) ) -> dict[str, Any]:
447)     """Export the full configuration stored in the storeroom.
448) 
449)     Args:
450)         storeroom_path:

451)             Path to the storeroom; usually `~/.vault`.  If not given,
452)             then query [`derivepassphrase.exporter.get_vault_path`][]
453)             for the value.

454)         master_keys_key:

455)             Encryption key/password for the master keys, usually the
456)             username, or passed via the `VAULT_KEY` environment
457)             variable.  If not given, then query
458)             [`derivepassphrase.exporter.get_vault_key`][] for the value.

459) 
460)     Returns:
461)         The full configuration, as stored in the storeroom.
462) 
463)         This may or may not be a valid configuration according to vault
464)         or derivepassphrase.
465) 
466)     Raises:
467)         RuntimeError:
468)             Something went wrong during data collection, e.g. we
469)             encountered unsupported or corrupted data in the storeroom.
470)         json.JSONDecodeError:
471)             An internal JSON data structure failed to parse from disk.
472)             The storeroom is probably corrupted.
473) 
474)     """
475) 

476)     if storeroom_path is None:
477)         storeroom_path = exporter.get_vault_path()

478)     if master_keys_key is None:

479)         master_keys_key = exporter.get_vault_key()

480)     with open(
481)         os.path.join(os.fsdecode(storeroom_path), '.keys'), encoding='utf-8'
482)     ) as master_keys_file:

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

500)         master_keys_key, encrypted_keys_iterations

501)     )
502)     master_keys = decrypt_master_keys_data(encrypted_keys, master_keys_keys)
503) 

504)     config_structure: dict[str, Any] = {}
505)     json_contents: dict[str, bytes] = {}

506)     for file in glob.glob(
507)         '[01][0-9a-f]', root_dir=os.fsdecode(storeroom_path)
508)     ):
509)         bucket_contents = list(
510)             decrypt_bucket_file(file, master_keys, root_dir=storeroom_path)
511)         )

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

562) 
563) 
564) if __name__ == '__main__':