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

16) if TYPE_CHECKING:
17)     from collections.abc import Iterator
18) 

19)     from cryptography.hazmat.primitives import ciphers, hashes, hmac, padding
20)     from cryptography.hazmat.primitives.ciphers import algorithms, modes
21)     from cryptography.hazmat.primitives.kdf import pbkdf2
22) else:
23)     try:
24)         from cryptography.hazmat.primitives import (
25)             ciphers,
26)             hashes,
27)             hmac,
28)             padding,
29)         )
30)         from cryptography.hazmat.primitives.ciphers import algorithms, modes
31)         from cryptography.hazmat.primitives.kdf import pbkdf2
32)     except ModuleNotFoundError as exc:
33) 
34)         class DummyModule:
35)             def __init__(self, exc: type[Exception]) -> None:
36)                 self.exc = exc
37) 
38)             def __getattr__(self, name: str) -> Any:
39)                 def func(*args: Any, **kwargs: Any) -> Any:  # noqa: ARG001
40)                     raise self.exc
41) 
42)                 return func
43) 
44)         ciphers = hashes = hmac = padding = DummyModule(exc)
45)         algorithms = modes = pbkdf2 = DummyModule(exc)
46)         STUBBED = True
47)     else:
48)         STUBBED = False
49) 

50) STOREROOM_MASTER_KEYS_UUID = b'35b7c7ed-f71e-4adf-9051-02fb0f1e0e17'
51) VAULT_CIPHER_UUID = b'73e69e8a-cb05-4b50-9f42-59d76a511299'
52) IV_SIZE = 16
53) KEY_SIZE = MAC_SIZE = 32
54) ENCRYPTED_KEYPAIR_SIZE = 128
55) VERSION_SIZE = 1
56) 

57) logger = logging.getLogger(__name__)

58) 
59) 
60) class KeyPair(TypedDict):
61)     encryption_key: bytes
62)     signing_key: bytes
63) 
64) 
65) class MasterKeys(TypedDict):
66)     hashing_key: bytes
67)     encryption_key: bytes
68)     signing_key: bytes
69) 
70) 
71) def derive_master_keys_keys(password: str | bytes, iterations: int) -> KeyPair:

72)     """Derive encryption and signing keys for the master keys data.
73) 
74)     The master password is run through a key derivation function to
75)     obtain a 64-byte string, which is then split to yield two 32-byte
76)     keys.  The key derivation function is PBKDF2, using HMAC-SHA1 and
77)     salted with the storeroom master keys UUID.
78) 
79)     Args:
80)         password:
81)             A master password for the storeroom instance.  Usually read
82)             from the `VAULT_KEY` environment variable, otherwise
83)             defaults to the username.
84)         iterations:
85)             A count of rounds for the underlying key derivation
86)             function.  Usually stored as a setting next to the encrypted
87)             master keys data.
88) 
89)     Returns:
90)         A 2-tuple of keys, the encryption key and the signing key, to
91)         decrypt and verify the master keys data with.
92) 
93)     """

94)     if isinstance(password, str):
95)         password = password.encode('ASCII')
96)     master_keys_keys_blob = pbkdf2.PBKDF2HMAC(
97)         algorithm=hashes.SHA1(),  # noqa: S303

98)         length=2 * KEY_SIZE,

99)         salt=STOREROOM_MASTER_KEYS_UUID,
100)         iterations=iterations,
101)     ).derive(password)
102)     encryption_key, signing_key = struct.unpack(
103)         f'{KEY_SIZE}s {KEY_SIZE}s', master_keys_keys_blob
104)     )
105)     logger.debug(
106)         (
107)             'derived master_keys_keys bytes.fromhex(%s) (encryption) '
108)             'and bytes.fromhex(%s) (signing) '
109)             'from password bytes.fromhex(%s), '
110)             'using call '
111)             'pbkdf2(algorithm=%s, length=%d, salt=%s, iterations=%d)'
112)         ),
113)         repr(encryption_key.hex(' ')),
114)         repr(signing_key.hex(' ')),
115)         repr(password.hex(' ')),
116)         repr('SHA256'),
117)         64,
118)         repr(STOREROOM_MASTER_KEYS_UUID),
119)         iterations,
120)     )
121)     return {
122)         'encryption_key': encryption_key,
123)         'signing_key': signing_key,
124)     }
125) 
126) 
127) def decrypt_master_keys_data(data: bytes, keys: KeyPair) -> MasterKeys:

128)     """Decrypt the master keys data.
129) 
130)     The master keys data contains:
131) 
132)     - a 16-byte IV,
133)     - a 96-byte AES256-CBC-encrypted payload (using PKCS7 padding on the
134)       inside), and
135)     - a 32-byte MAC of the preceding 112 bytes.
136) 
137)     The decrypted payload itself consists of three 32-byte keys: the
138)     hashing, encryption and signing keys, in that order.
139) 
140)     The encrypted payload is encrypted with the encryption key, and the
141)     MAC is created based on the signing key.  As per standard
142)     cryptographic procedure, the MAC can be verified before attempting
143)     to decrypt the payload.
144) 
145)     Because the payload size is both fixed and a multiple of the
146)     cipher blocksize, in this case, the PKCS7 padding is a no-op.
147) 
148)     Args:
149)         data:
150)             The encrypted master keys data.
151)         keys:
152)             The encryption and signing keys for the master keys data.
153)             These should have previously been derived via the
154)             [`derivepassphrase.exporter.storeroom.derive_master_keys_keys`][]
155)             function.
156) 
157)     Returns:
158)         The master encryption, signing and hashing keys.
159) 
160)     """

161)     ciphertext, claimed_mac = struct.unpack(
162)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
163)     )
164)     actual_mac = hmac.HMAC(keys['signing_key'], hashes.SHA256())
165)     actual_mac.update(ciphertext)
166)     logger.debug(
167)         (
168)             'master_keys_data mac_key = bytes.fromhex(%s), '
169)             'hashed_content = bytes.fromhex(%s), '
170)             'claimed_mac = bytes.fromhex(%s), '
171)             'actual_mac = bytes.fromhex(%s)'
172)         ),
173)         repr(keys['signing_key'].hex(' ')),
174)         repr(ciphertext.hex(' ')),
175)         repr(claimed_mac.hex(' ')),
176)         repr(actual_mac.copy().finalize().hex(' ')),
177)     )
178)     actual_mac.verify(claimed_mac)
179) 
180)     iv, payload = struct.unpack(
181)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
182)     )
183)     decryptor = ciphers.Cipher(
184)         algorithms.AES256(keys['encryption_key']), modes.CBC(iv)
185)     ).decryptor()
186)     padded_plaintext = bytearray()
187)     padded_plaintext.extend(decryptor.update(payload))
188)     padded_plaintext.extend(decryptor.finalize())
189)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
190)     plaintext = bytearray()
191)     plaintext.extend(unpadder.update(padded_plaintext))
192)     plaintext.extend(unpadder.finalize())
193)     if len(plaintext) != 3 * KEY_SIZE:
194)         msg = (
195)             f'Expecting 3 encrypted keys at {3 * KEY_SIZE} bytes total, '
196)             f'but found {len(plaintext)} instead'
197)         )
198)         raise RuntimeError(msg)
199)     hashing_key, encryption_key, signing_key = struct.unpack(
200)         f'{KEY_SIZE}s {KEY_SIZE}s {KEY_SIZE}s', plaintext
201)     )
202)     return {
203)         'hashing_key': hashing_key,
204)         'encryption_key': encryption_key,
205)         'signing_key': signing_key,
206)     }
207) 
208) 

209) def decrypt_session_keys(data: bytes, master_keys: MasterKeys) -> KeyPair:
210)     """Decrypt the bucket item's session keys.
211) 
212)     The bucket item's session keys are single-use keys for encrypting
213)     and signing a single item in the storage bucket.  The encrypted
214)     session key data consists of:
215) 
216)     - a 16-byte IV,
217)     - a 64-byte AES256-CBC-encrypted payload (using PKCS7 padding on the
218)       inside), and
219)     - a 32-byte MAC of the preceding 80 bytes.
220) 
221)     The encrypted payload is encrypted with the master encryption key,
222)     and the MAC is created with the master signing key.  As per standard
223)     cryptographic procedure, the MAC can be verified before attempting
224)     to decrypt the payload.
225) 
226)     Because the payload size is both fixed and a multiple of the
227)     cipher blocksize, in this case, the PKCS7 padding is a no-op.
228) 
229)     Args:
230)         data:
231)             The encrypted bucket item session key data.
232)         master_keys:
233)             The master keys.  Presumably these have previously been
234)             obtained via the
235)             [`derivepassphrase.exporter.storeroom.decrypt_master_keys_data`][]
236)             function.
237) 
238)     Returns:
239)         The bucket item's encryption and signing keys.
240) 
241)     """
242) 

243)     ciphertext, claimed_mac = struct.unpack(
244)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
245)     )

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

247)     actual_mac.update(ciphertext)
248)     logger.debug(
249)         (

250)             'decrypt_bucket_item (session_keys): '

251)             'mac_key = bytes.fromhex(%s) (master), '
252)             'hashed_content = bytes.fromhex(%s), '
253)             'claimed_mac = bytes.fromhex(%s), '
254)             'actual_mac = bytes.fromhex(%s)'
255)         ),

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

257)         repr(ciphertext.hex(' ')),
258)         repr(claimed_mac.hex(' ')),
259)         repr(actual_mac.copy().finalize().hex(' ')),
260)     )
261)     actual_mac.verify(claimed_mac)
262) 
263)     iv, payload = struct.unpack(
264)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
265)     )
266)     decryptor = ciphers.Cipher(

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

268)     ).decryptor()
269)     padded_plaintext = bytearray()
270)     padded_plaintext.extend(decryptor.update(payload))
271)     padded_plaintext.extend(decryptor.finalize())
272)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
273)     plaintext = bytearray()
274)     plaintext.extend(unpadder.update(padded_plaintext))
275)     plaintext.extend(unpadder.finalize())
276) 
277)     session_encryption_key, session_signing_key, inner_payload = struct.unpack(
278)         f'{KEY_SIZE}s {KEY_SIZE}s {len(plaintext) - 2 * KEY_SIZE}s',
279)         plaintext,
280)     )
281)     session_keys: KeyPair = {
282)         'encryption_key': session_encryption_key,
283)         'signing_key': session_signing_key,
284)     }
285) 
286)     logger.debug(
287)         (

288)             'decrypt_bucket_item (session_keys): '

289)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
290)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
291)             '= bytes.fromhex(%s) '
292)             '= {"encryption_key": bytes.fromhex(%s), '
293)             '"signing_key": bytes.fromhex(%s)}'
294)         ),

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

296)         repr(iv.hex(' ')),
297)         repr(payload.hex(' ')),
298)         repr(plaintext.hex(' ')),
299)         repr(session_keys['encryption_key'].hex(' ')),
300)         repr(session_keys['signing_key'].hex(' ')),
301)     )
302) 
303)     if inner_payload:
304)         logger.debug(
305)             'ignoring misplaced inner payload bytes.fromhex(%s)',
306)             repr(inner_payload.hex(' ')),
307)         )
308) 
309)     return session_keys
310) 
311) 

312) def decrypt_contents(data: bytes, session_keys: KeyPair) -> bytes:
313)     """Decrypt the bucket item's contents.
314) 
315)     The data consists of:
316) 
317)     - a 16-byte IV,
318)     - a variable-sized AES256-CBC-encrypted payload (using PKCS7 padding
319)       on the inside), and
320)     - a 32-byte MAC of the preceding 80 bytes.
321) 
322)     The encrypted payload is encrypted with the bucket item's session
323)     encryption key, and the MAC is created with the bucket item's
324)     session signing key.  As per standard cryptographic procedure, the
325)     MAC can be verified before attempting to decrypt the payload.
326) 
327)     Args:
328)         data:
329)             The encrypted bucket item payload data.
330)         session_keys:
331)             The bucket item's session keys.  Presumably these have
332)             previously been obtained via the
333)             [`derivepassphrase.exporter.storeroom.decrypt_session_keys`][]
334)             function.
335) 
336)     Returns:
337)         The bucket item's payload.
338) 
339)     """
340) 

341)     ciphertext, claimed_mac = struct.unpack(
342)         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data
343)     )

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

345)     actual_mac.update(ciphertext)
346)     logger.debug(
347)         (

348)             'decrypt_bucket_item (contents): '

349)             'mac_key = bytes.fromhex(%s), '
350)             'hashed_content = bytes.fromhex(%s), '
351)             'claimed_mac = bytes.fromhex(%s), '
352)             'actual_mac = bytes.fromhex(%s)'
353)         ),

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

355)         repr(ciphertext.hex(' ')),
356)         repr(claimed_mac.hex(' ')),
357)         repr(actual_mac.copy().finalize().hex(' ')),
358)     )
359)     actual_mac.verify(claimed_mac)
360) 
361)     iv, payload = struct.unpack(
362)         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext
363)     )
364)     decryptor = ciphers.Cipher(

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

366)     ).decryptor()
367)     padded_plaintext = bytearray()
368)     padded_plaintext.extend(decryptor.update(payload))
369)     padded_plaintext.extend(decryptor.finalize())
370)     unpadder = padding.PKCS7(IV_SIZE * 8).unpadder()
371)     plaintext = bytearray()
372)     plaintext.extend(unpadder.update(padded_plaintext))
373)     plaintext.extend(unpadder.finalize())
374) 
375)     logger.debug(
376)         (

377)             'decrypt_bucket_item (contents): '

378)             'decrypt_aes256_cbc_and_unpad(key=bytes.fromhex(%s), '
379)             'iv=bytes.fromhex(%s))(bytes.fromhex(%s)) '
380)             '= bytes.fromhex(%s)'
381)         ),

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

383)         repr(iv.hex(' ')),
384)         repr(payload.hex(' ')),
385)         repr(plaintext.hex(' ')),
386)     )
387) 
388)     return plaintext
389) 
390) 

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

392)     logger.debug(
393)         (

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

395)             'encryption_key = bytes.fromhex(%s), '
396)             'signing_key = bytes.fromhex(%s)'
397)         ),

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

399)         repr(master_keys['encryption_key'].hex(' ')),
400)         repr(master_keys['signing_key'].hex(' ')),
401)     )
402)     data_version, encrypted_session_keys, data_contents = struct.unpack(
403)         (
404)             f'B {ENCRYPTED_KEYPAIR_SIZE}s '

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

406)         ),

407)         bucket_item,

408)     )
409)     if data_version != 1:
410)         msg = f'Cannot handle version {data_version} encrypted data'
411)         raise RuntimeError(msg)
412)     session_keys = decrypt_session_keys(encrypted_session_keys, master_keys)
413)     return decrypt_contents(data_contents, session_keys)
414) 
415) 

416) def decrypt_bucket_file(

417)     filename: str,
418)     master_keys: MasterKeys,
419)     *,
420)     root_dir: str | bytes | os.PathLike = '.',

421) ) -> Iterator[bytes]:

422)     with open(
423)         os.path.join(os.fsdecode(root_dir), filename), 'rb'
424)     ) as bucket_file:

425)         header_line = bucket_file.readline()
426)         try:
427)             header = json.loads(header_line)
428)         except ValueError as exc:
429)             msg = f'Invalid bucket file: {filename}'
430)             raise RuntimeError(msg) from exc
431)         if header != {'version': 1}:
432)             msg = f'Invalid bucket file: {filename}'
433)             raise RuntimeError(msg) from None
434)         for line in bucket_file:

435)             yield decrypt_bucket_item(
436)                 base64.standard_b64decode(line), master_keys

437)             )
438) 
439) 

440) def store(config: dict[str, Any], path: str, json_contents: bytes) -> None:
441)     """Store the JSON contents at path in the config structure.
442) 
443)     Traverse the config structure according to path, and set the value
444)     of the leaf to the decoded JSON contents.
445) 
446)     A path `/foo/bar/xyz` translates to the JSON structure
447)     `{"foo": {"bar": {"xyz": ...}}}`.
448) 
449)     Args:
450)         config:
451)             The (top-level) configuration structure to update.
452)         path:
453)             The path within the configuration structure to traverse.
454)         json_contents:
455)             The contents to set the item to, after JSON-decoding.
456) 
457)     Raises:
458)         json.JSONDecodeError:
459)             There was an error parsing the JSON contents.
460) 
461)     """
462)     contents = json.loads(json_contents)
463)     path_parts = [part for part in path.split('/') if part]
464)     for part in path_parts[:-1]:
465)         config = config.setdefault(part, {})
466)     if path_parts:
467)         config[path_parts[-1]] = contents
468) 
469) 
470) def export_storeroom_data(

471)     storeroom_path: str | bytes | os.PathLike | None = None,
472)     master_keys_key: str | bytes | None = None,

473) ) -> dict[str, Any]:
474)     """Export the full configuration stored in the storeroom.
475) 
476)     Args:
477)         storeroom_path:

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

481)         master_keys_key:

482)             Encryption key/password for the master keys, usually the
483)             username, or passed via the `VAULT_KEY` environment
484)             variable.  If not given, then query
485)             [`derivepassphrase.exporter.get_vault_key`][] for the value.

486) 
487)     Returns:
488)         The full configuration, as stored in the storeroom.
489) 
490)         This may or may not be a valid configuration according to vault
491)         or derivepassphrase.
492) 
493)     Raises:
494)         RuntimeError:
495)             Something went wrong during data collection, e.g. we
496)             encountered unsupported or corrupted data in the storeroom.
497)         json.JSONDecodeError:
498)             An internal JSON data structure failed to parse from disk.
499)             The storeroom is probably corrupted.
500) 
501)     """
502) 

503)     if storeroom_path is None:
504)         storeroom_path = exporter.get_vault_path()

505)     if master_keys_key is None:

506)         master_keys_key = exporter.get_vault_key()

507)     with open(
508)         os.path.join(os.fsdecode(storeroom_path), '.keys'), encoding='utf-8'
509)     ) as master_keys_file:

510)         header = json.loads(master_keys_file.readline())
511)         if header != {'version': 1}:
512)             msg = 'bad or unsupported keys version header'
513)             raise RuntimeError(msg)
514)         raw_keys_data = base64.standard_b64decode(master_keys_file.readline())
515)         encrypted_keys_params, encrypted_keys = struct.unpack(
516)             f'B {len(raw_keys_data) - 1}s', raw_keys_data
517)         )
518)         if master_keys_file.read():
519)             msg = 'trailing data; cannot make sense of .keys file'
520)             raise RuntimeError(msg)
521)     encrypted_keys_version = encrypted_keys_params >> 4
522)     if encrypted_keys_version != 1:
523)         msg = f'cannot handle version {encrypted_keys_version} encrypted keys'
524)         raise RuntimeError(msg)
525)     encrypted_keys_iterations = 2 ** (10 + (encrypted_keys_params & 0x0F))
526)     master_keys_keys = derive_master_keys_keys(

527)         master_keys_key, encrypted_keys_iterations

528)     )
529)     master_keys = decrypt_master_keys_data(encrypted_keys, master_keys_keys)
530) 

531)     config_structure: dict[str, Any] = {}
532)     json_contents: dict[str, bytes] = {}

533)     for file in glob.glob(
534)         '[01][0-9a-f]', root_dir=os.fsdecode(storeroom_path)
535)     ):
536)         bucket_contents = list(
537)             decrypt_bucket_file(file, master_keys, root_dir=storeroom_path)
538)         )

539)         bucket_index = json.loads(bucket_contents.pop(0))
540)         for pos, item in enumerate(bucket_index):
541)             json_contents[item] = bucket_contents[pos]
542)             logger.debug(
543)                 'Found bucket item: %s -> %s', item, bucket_contents[pos]
544)             )
545)     dirs_to_check: dict[str, list[str]] = {}
546)     json_payload: Any
547)     for path, json_content in sorted(json_contents.items()):
548)         if path.endswith('/'):
549)             logger.debug(
550)                 'Postponing dir check: %s -> %s',
551)                 path,
552)                 json_content.decode('utf-8'),
553)             )
554)             json_payload = json.loads(json_content)
555)             if not isinstance(json_payload, list) or any(
556)                 not isinstance(x, str) for x in json_payload
557)             ):
558)                 msg = (
559)                     f'Directory index is not actually an index: '
560)                     f'{json_content!r}'
561)                 )
562)                 raise RuntimeError(msg)
563)             dirs_to_check[path] = json_payload
564)             logger.debug(
565)                 'Setting contents (empty directory): %s -> %s', path, '{}'
566)             )
567)             store(config_structure, path, b'{}')
568)         else:
569)             logger.debug(
570)                 'Setting contents: %s -> %s',
571)                 path,
572)                 json_content.decode('utf-8'),
573)             )
574)             store(config_structure, path, json_content)
575)     for _dir, namelist in dirs_to_check.items():
576)         namelist = [x.rstrip('/') for x in namelist]  # noqa: PLW2901
577)         try:
578)             obj = config_structure
579)             for part in _dir.split('/'):
580)                 if part:
581)                     obj = obj[part]
582)         except KeyError as exc:
583)             msg = f'Cannot traverse storage path: {_dir!r}'
584)             raise RuntimeError(msg) from exc
585)         if set(obj.keys()) != set(namelist):
586)             msg = f'Object key mismatch for path {_dir!r}'
587)             raise RuntimeError(msg)
588)     return config_structure

589) 
590) 
591) if __name__ == '__main__':