Coverage for src / derivepassphrase / exporter / storeroom.py: 100.000%

1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> 

2# 

3# SPDX-License-Identifier: Zlib 

4 

5"""Exporter for the vault "storeroom" configuration format. 

6 

7The "storeroom" format is the experimental format used in alpha and beta 

8versions of vault beyond v0.3.0. The configuration is stored as 

9a separate directory, which acts like a hash table (i.e. has named 

10slots) and provides an impure quasi-filesystem interface. Each hash 

11table entry is separately encrypted and authenticated. James Coglan 

12designed this format to avoid concurrent write issues when updating or 

13synchronizing the vault configuration with e.g. a cloud service. 

14 

15The public interface is the [`export_storeroom_data`][] function. 

16Multiple *non-public* functions are additionally documented here for 

17didactical and educational reasons, but they are not part of the module 

18API, are subject to change without notice (including removal), and 

19should *not* be used or relied on. 

20 

21""" 

22 

23# ruff: noqa: S303 

24 

25from __future__ import annotations (empty)

26 

27import base64 (empty)

28import importlib (empty)

29import json (empty)

30import logging (empty)

31import os (empty)

32import pathlib (empty)

33import struct (empty)

34from typing import TYPE_CHECKING, Any (empty)

35 

36from derivepassphrase import _types, exporter (empty)

37from derivepassphrase._internals import cli_messages as _msg (empty)

38 

39if TYPE_CHECKING: (empty)

40 from typing_extensions import Buffer 

41 

42if TYPE_CHECKING: (empty)

43 from collections.abc import Generator 

44 

45 from cryptography.hazmat.primitives import ciphers, hashes, hmac, padding 

46 from cryptography.hazmat.primitives.ciphers import algorithms, modes 

47 from cryptography.hazmat.primitives.kdf import pbkdf2 

48 

49 STUBBED = False 

50else: 

51 try: (empty)

52 importlib.import_module("cryptography") (empty)

53 except ModuleNotFoundError as exc: (empty)

54 

55 class _DummyModule: (empty)

56 def __init__(self, exc: type[Exception]) -> None: (empty)

57 self.exc = exc (empty)

58 

59 def __getattr__(self, name: str) -> Any: # noqa: ANN401 (empty)

60 def func(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401,ARG001 (empty)

61 raise self.exc 

62 

63 return func (empty)

64 

65 ciphers = hashes = hmac = padding = _DummyModule(exc) (empty)

66 algorithms = modes = pbkdf2 = _DummyModule(exc) (empty)

67 STUBBED = True (empty)

68 else: 

69 from cryptography.hazmat.primitives import ( (empty)

70 ciphers, 

71 hashes, 

72 hmac, 

73 padding, 

74 ) 

75 from cryptography.hazmat.primitives.ciphers import algorithms, modes (empty)

76 from cryptography.hazmat.primitives.kdf import pbkdf2 (empty)

77 

78 STUBBED = False (empty)

79 

80STOREROOM_MASTER_KEYS_UUID = b"35b7c7ed-f71e-4adf-9051-02fb0f1e0e17" (empty)

81VAULT_CIPHER_UUID = b"73e69e8a-cb05-4b50-9f42-59d76a511299" (empty)

82IV_SIZE = 16 (empty)

83KEY_SIZE = MAC_SIZE = 32 (empty)

84ENCRYPTED_KEYPAIR_SIZE = 128 (empty)

85VERSION_SIZE = 1 (empty)

86 

87__all__ = ("export_storeroom_data",) (empty)

88 

89logger = logging.getLogger(__name__) (empty)

90 

91 

92@exporter.register_export_vault_config_data_handler("storeroom") (empty)

93def export_storeroom_data( # noqa: C901,D417,PLR0912,PLR0914,PLR0915 (empty)

94 path: str | bytes | os.PathLike | None = None, 

95 key: str | Buffer | None = None, 

96 *, 

97 format: str = "storeroom", # noqa: A002 

98) -> dict[str, Any]: 

99 """Export the full configuration stored in the storeroom. 

100 

101 See [`exporter.ExportVaultConfigDataFunction`][] for an explanation 

102 of the call signature, and the exceptions to expect. 

103 

104 Other Args: 

105 format: 

106 The only supported format is `storeroom`. 

107 

108 """ # noqa: DOC201,DOC501 

109 # Trigger import errors if necessary. 

110 importlib.import_module("cryptography") 7 ctx1ijcabfl

111 if path is None: 6 ctx1ijcabf

112 path = exporter.get_vault_path() 4 ctx1cabf

113 else: 

114 path = pathlib.Path(os.fsdecode(path)) 3 ctx1ija

115 if key is None: 6 ctx1ijcabf

116 key = exporter.get_vault_key() 5 ctx1ijabf

117 if format != "storeroom": # pragma: no cover [failsafe] 6 ctx1ijcabf

118 msg = exporter.INVALID_VAULT_NATIVE_CONFIGURATION_FORMAT.format( 

119 fmt=format 

120 ) 

121 raise ValueError(msg) 

122 try: 6 ctx1ijcabf

123 master_keys_file = pathlib.Path(path, ".keys").open( # noqa: SIM115 6 ctx1ijcabf

124 encoding="utf-8", 

125 ) 

126 except FileNotFoundError as exc: 2 ctx1ij

127 raise exporter.NotAVaultConfigError(path, format="storeroom") from exc 2 ctx1ij

128 with master_keys_file: 4 ctx1cabf

129 header = json.loads(master_keys_file.readline()) 4 ctx1cabf

130 if header != {"version": 1}: 4 ctx1cabf

131 msg = "bad or unsupported keys version header" 1 ctx1f

132 raise RuntimeError(msg) 1 ctx1f

133 raw_keys_data = base64.standard_b64decode(master_keys_file.readline()) 4 ctx1cabf

134 encrypted_keys_params, encrypted_keys = struct.unpack( 4 ctx1cabf

135 f"B {len(raw_keys_data) - 1}s", raw_keys_data 

136 ) 

137 if master_keys_file.read(): 4 ctx1cabf

138 msg = "trailing data; cannot make sense of .keys file" 1 ctx1f

139 raise RuntimeError(msg) 1 ctx1f

140 encrypted_keys_version = encrypted_keys_params >> 4 4 ctx1cabf

141 if encrypted_keys_version != 1: 4 ctx1cabf

142 msg = f"cannot handle version {encrypted_keys_version} encrypted keys" 1 ctx1f

143 raise RuntimeError(msg) 1 ctx1f

144 logger.info( 3 ctx1cab

145 _msg.TranslatedString(_msg.InfoMsgTemplate.PARSING_MASTER_KEYS_DATA) 

146 ) 

147 encrypted_keys_iterations = 2 ** (10 + (encrypted_keys_params & 0x0F)) 3 ctx1cab

148 master_keys_keys = _derive_master_keys_keys(key, encrypted_keys_iterations) 3 ctx1cab

149 master_keys = _decrypt_master_keys_data(encrypted_keys, master_keys_keys) 3 ctx1cab

150 

151 config_structure: dict[str, Any] = {} 3 ctx1cab

152 json_contents: dict[str, bytes] = {} 3 ctx1cab

153 valid_hashdirs = list(path.glob("[01][0-9a-f]")) 3 ctx1cab

154 for file in valid_hashdirs: 3 ctx1cab

155 logger.info( 3 ctx1cab

156 _msg.TranslatedString( 

157 _msg.InfoMsgTemplate.DECRYPTING_BUCKET, 

158 bucket_number=file, 

159 ) 

160 ) 

161 bucket_contents = [ 3 ctx1cab

162 bytes(item) for item in _decrypt_bucket_file(file, master_keys) 

163 ] 

164 bucket_index = json.loads(bucket_contents.pop(0)) 3 ctx1cab

165 for pos, item in enumerate(bucket_index): 3 ctx1cab

166 json_contents[item] = bucket_contents[pos] 3 ctx1cab

167 logger.debug( 3 ctx1cab

168 _msg.TranslatedString( 

169 _msg.DebugMsgTemplate.BUCKET_ITEM_FOUND, 

170 path=item, 

171 value=bucket_contents[pos], 

172 ) 

173 ) 

174 dirs_to_check: dict[str, list[str]] = {} 3 ctx1cab

175 json_payload: Any 

176 logger.info( 3 ctx1cab

177 _msg.TranslatedString(_msg.InfoMsgTemplate.ASSEMBLING_CONFIG_STRUCTURE) 

178 ) 

179 for item_path, json_content in sorted(json_contents.items()): 3 ctx1cab

180 if item_path.endswith("/"): 3 ctx1cab

181 logger.debug( 3 ctx1cab

182 _msg.TranslatedString( 

183 _msg.DebugMsgTemplate.POSTPONING_DIRECTORY_CONTENTS_CHECK, 

184 path=item_path, 

185 contents=json_content.decode("utf-8"), 

186 ) 

187 ) 

188 json_payload = json.loads(json_content) 3 ctx1cab

189 if not isinstance(json_payload, list) or any( 3 ctx1cab

190 not isinstance(x, str) for x in json_payload 

191 ): 

192 msg = ( 1 ctx1b

193 f"Directory index is not actually an index: " 

194 f"{json_content!r}" 

195 ) 

196 raise RuntimeError(msg) 1 ctx1b

197 dirs_to_check[item_path] = json_payload 3 ctx1cab

198 logger.debug( 3 ctx1cab

199 _msg.TranslatedString( 

200 _msg.DebugMsgTemplate.SETTING_CONFIG_STRUCTURE_CONTENTS_EMPTY_DIRECTORY, 

201 path=item_path, 

202 ), 

203 ) 

204 _store(config_structure, item_path, b"{}") 3 ctx1cab

205 else: 

206 logger.debug( 2 ctx1ca

207 _msg.TranslatedString( 

208 _msg.DebugMsgTemplate.SETTING_CONFIG_STRUCTURE_CONTENTS, 

209 path=item_path, 

210 value=json_content.decode("utf-8"), 

211 ), 

212 ) 

213 _store(config_structure, item_path, json_content) 2 ctx1ca

214 logger.info( 3 ctx1cab

215 _msg.TranslatedString( 

216 _msg.InfoMsgTemplate.CHECKING_CONFIG_STRUCTURE_CONSISTENCY, 

217 ) 

218 ) 

219 # Sorted order is important; see `maybe_obj` below. 

220 for dir_, namelist_ in sorted(dirs_to_check.items()): 3 ctx1cab

221 namelist = [x.rstrip("/") for x in namelist_] 3 ctx1cab

222 obj: dict[Any, Any] = config_structure 3 ctx1cab

223 for part in dir_.split("/"): 3 ctx1cab

224 if part: 3 ctx1cab

225 # Because we iterate paths in sorted order, parent 

226 # directories are encountered before child directories. 

227 # So parent directories always exist (lest we would have 

228 # aborted earlier). 

229 # 

230 # Of course, the type checker doesn't necessarily know 

231 # this, so we need to use assertions anyway. 

232 maybe_obj = obj.get(part) 3 ctx1cab

233 assert isinstance(maybe_obj, dict), ( 3 ctx1cab

234 f"Cannot traverse storage path {dir_!r}" 

235 ) 

236 obj = maybe_obj 3 ctx1cab

237 if set(obj.keys()) != set(namelist): 3 ctx1cab

238 msg = f"Object key mismatch for path {dir_!r}" 1 ctx1b

239 raise RuntimeError(msg) 1 ctx1b

240 logger.debug( 3 ctx1cab

241 _msg.TranslatedString( 

242 _msg.DebugMsgTemplate.DIRECTORY_CONTENTS_CHECK_OK, 

243 path=dir_, 

244 contents=json.dumps(namelist_), 

245 ) 

246 ) 

247 return config_structure 2 ctx1ca

248 

249 

250def _h(bs: Buffer) -> str: (empty)

251 return "<{}>".format(memoryview(bs).hex(" ")) 6 ctx1ckeabg

252 

253 

254def _derive_master_keys_keys( (empty)

255 password: str | Buffer, 

256 iterations: int, 

257) -> _types.StoreroomKeyPair: 

258 """Derive encryption and signing keys for the master keys data. 

259 

260 The master password is run through a key derivation function to 

261 obtain a 64-byte string, which is then split to yield two 32-byte 

262 keys. The key derivation function is PBKDF2, using HMAC-SHA1 and 

263 salted with the storeroom master keys UUID. 

264 

265 Args: 

266 password: 

267 A master password for the storeroom instance. Usually read 

268 from the `VAULT_KEY` environment variable, otherwise 

269 defaults to the username. 

270 iterations: 

271 A count of rounds for the underlying key derivation 

272 function. Usually stored as a setting next to the encrypted 

273 master keys data. 

274 

275 Returns: 

276 A 2-tuple of keys, the encryption key and the signing key, to 

277 decrypt and verify the master keys data with. 

278 

279 Warning: 

280 Non-public function, provided for didactical and educational 

281 purposes only. Subject to change without notice, including 

282 removal. 

283 

284 """ 

285 if isinstance(password, str): 3 ctx1cab

286 password = password.encode("ASCII") 1 ctx1a

287 master_keys_keys_blob = pbkdf2.PBKDF2HMAC( 3 ctx1cab

288 algorithm=hashes.SHA1(), 

289 length=2 * KEY_SIZE, 

290 salt=STOREROOM_MASTER_KEYS_UUID, 

291 iterations=iterations, 

292 ).derive(bytes(password)) 

293 encryption_key, signing_key = struct.unpack( 3 ctx1cab

294 f"{KEY_SIZE}s {KEY_SIZE}s", master_keys_keys_blob 

295 ) 

296 logger.debug( 3 ctx1cab

297 _msg.TranslatedString( 

298 _msg.DebugMsgTemplate.DERIVED_MASTER_KEYS_KEYS, 

299 enc_key=_h(encryption_key), 

300 sign_key=_h(signing_key), 

301 pw_bytes=_h(password), 

302 algorithm="SHA256", 

303 length=64, 

304 salt=STOREROOM_MASTER_KEYS_UUID, 

305 iterations=iterations, 

306 ), 

307 ) 

308 return _types.StoreroomKeyPair( 3 ctx1cab

309 encryption_key=encryption_key, 

310 signing_key=signing_key, 

311 ).toreadonly() 

312 

313 

314def _decrypt_master_keys_data( (empty)

315 data: Buffer, 

316 keys: _types.StoreroomKeyPair, 

317) -> _types.StoreroomMasterKeys: 

318 r"""Decrypt the master keys data. 

319 

320 The master keys data contains: 

321 

322 - a 16-byte IV, 

323 - a 96-byte AES256-CBC-encrypted payload, plus 16 further bytes of 

324 PKCS7 padding, and 

325 - a 32-byte MAC of the preceding 128 bytes. 

326 

327 The decrypted payload itself consists of three 32-byte keys: the 

328 hashing, encryption and signing keys, in that order. 

329 

330 The encrypted payload is encrypted with the encryption key, and the 

331 MAC is created based on the signing key. As per standard 

332 cryptographic procedure, the MAC can be verified before attempting 

333 to decrypt the payload. 

334 

335 Because the payload size is both fixed and a multiple of the cipher 

336 blocksize, in this case, the PKCS7 padding always is `b'\x10' * 16`. 

337 

338 Args: 

339 data: 

340 The encrypted master keys data. 

341 keys: 

342 The encryption and signing keys for the master keys data. 

343 These should have previously been derived via the 

344 [`_derive_master_keys_keys`][] function. 

345 

346 Returns: 

347 The master encryption, signing and hashing keys. 

348 

349 Raises: 

350 cryptography.exceptions.InvalidSignature: 

351 The data does not contain a valid signature under the given 

352 key. 

353 ValueError: 

354 The format is invalid, in a non-cryptographic way. (For 

355 example, it contains an unsupported version marker, or 

356 unexpected extra contents, or invalid padding.) 

357 

358 Warning: 

359 Non-public function, provided for didactical and educational 

360 purposes only. Subject to change without notice, including 

361 removal. 

362 

363 """ 

364 data = memoryview(data).toreadonly().cast("c") 5 ctx1ceabg

365 keys = keys.toreadonly() 5 ctx1ceabg

366 ciphertext, claimed_mac = struct.unpack( 5 ctx1ceabg

367 f"{len(data) - MAC_SIZE}s {MAC_SIZE}s", data 

368 ) 

369 actual_mac = hmac.HMAC(keys.signing_key, hashes.SHA256()) 5 ctx1ceabg

370 actual_mac.update(ciphertext) 5 ctx1ceabg

371 logger.debug( 5 ctx1ceabg

372 _msg.TranslatedString( 

373 _msg.DebugMsgTemplate.MASTER_KEYS_DATA_MAC_INFO, 

374 sign_key=_h(keys.signing_key), 

375 ciphertext=_h(ciphertext), 

376 claimed_mac=_h(claimed_mac), 

377 actual_mac=_h(actual_mac.copy().finalize()), 

378 ), 

379 ) 

380 actual_mac.verify(claimed_mac) 5 ctx1ceabg

381 

382 try: 4 ctx1ceab

383 iv, payload = struct.unpack( 4 ctx1ceab

384 f"{IV_SIZE}s {len(ciphertext) - IV_SIZE}s", ciphertext 

385 ) 

386 decryptor = ciphers.Cipher( 4 ctx1ceab

387 algorithms.AES256(keys.encryption_key), modes.CBC(iv) 

388 ).decryptor() 

389 padded_plaintext = bytearray() 4 ctx1ceab

390 padded_plaintext.extend(decryptor.update(payload)) 4 ctx1ceab

391 padded_plaintext.extend(decryptor.finalize()) 4 ctx1ceab

392 unpadder = padding.PKCS7(IV_SIZE * 8).unpadder() 4 ctx1ceab

393 plaintext = bytearray() 4 ctx1ceab

394 plaintext.extend(unpadder.update(bytes(padded_plaintext))) 4 ctx1ceab

395 plaintext.extend(unpadder.finalize()) 4 ctx1ceab

396 hashing_key, encryption_key, signing_key = struct.unpack( 4 ctx1ceab

397 f"{KEY_SIZE}s {KEY_SIZE}s {KEY_SIZE}s", plaintext 

398 ) 

399 except (ValueError, struct.error) as exc: 1 ctx1e

400 msg = "Invalid encrypted master keys payload" 1 ctx1e

401 raise ValueError(msg) from exc 1 ctx1e

402 return _types.StoreroomMasterKeys( 3 ctx1cab

403 hashing_key=hashing_key, 

404 encryption_key=encryption_key, 

405 signing_key=signing_key, 

406 ).toreadonly() 

407 

408 

409def _decrypt_session_keys( (empty)

410 data: Buffer, 

411 master_keys: _types.StoreroomMasterKeys, 

412) -> _types.StoreroomKeyPair: 

413 r"""Decrypt the bucket item's session keys. 

414 

415 The bucket item's session keys are single-use keys for encrypting 

416 and signing a single item in the storage bucket. The encrypted 

417 session key data consists of: 

418 

419 - a 16-byte IV, 

420 - a 64-byte AES256-CBC-encrypted payload, plus 16 further bytes of 

421 PKCS7 padding, and 

422 - a 32-byte MAC of the preceding 96 bytes. 

423 

424 The encrypted payload is encrypted with the master encryption key, 

425 and the MAC is created with the master signing key. As per standard 

426 cryptographic procedure, the MAC can be verified before attempting 

427 to decrypt the payload. 

428 

429 Because the payload size is both fixed and a multiple of the cipher 

430 blocksize, in this case, the PKCS7 padding always is `b'\x10' * 16`. 

431 

432 Args: 

433 data: 

434 The encrypted bucket item session key data. 

435 master_keys: 

436 The master keys. Presumably these have previously been 

437 obtained via the [`_decrypt_master_keys_data`][] function. 

438 

439 Returns: 

440 The bucket item's encryption and signing keys. 

441 

442 Raises: 

443 cryptography.exceptions.InvalidSignature: 

444 The data does not contain a valid signature under the given 

445 key. 

446 ValueError: 

447 The format is invalid, in a non-cryptographic way. (For 

448 example, it contains an unsupported version marker, or 

449 unexpected extra contents, or invalid padding.) 

450 

451 Warning: 

452 Non-public function, provided for didactical and educational 

453 purposes only. Subject to change without notice, including 

454 removal. 

455 

456 """ 

457 data = memoryview(data).toreadonly().cast("c") 5 ctx1ceabg

458 master_keys = master_keys.toreadonly() 5 ctx1ceabg

459 ciphertext, claimed_mac = struct.unpack( 5 ctx1ceabg

460 f"{len(data) - MAC_SIZE}s {MAC_SIZE}s", data 

461 ) 

462 actual_mac = hmac.HMAC(master_keys.signing_key, hashes.SHA256()) 5 ctx1ceabg

463 actual_mac.update(ciphertext) 5 ctx1ceabg

464 logger.debug( 5 ctx1ceabg

465 _msg.TranslatedString( 

466 _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_SESSION_KEYS_MAC_INFO, 

467 sign_key=_h(master_keys.signing_key), 

468 ciphertext=_h(ciphertext), 

469 claimed_mac=_h(claimed_mac), 

470 actual_mac=_h(actual_mac.copy().finalize()), 

471 ), 

472 ) 

473 actual_mac.verify(claimed_mac) 5 ctx1ceabg

474 

475 try: 4 ctx1ceab

476 iv, payload = struct.unpack( 4 ctx1ceab

477 f"{IV_SIZE}s {len(ciphertext) - IV_SIZE}s", ciphertext 

478 ) 

479 decryptor = ciphers.Cipher( 4 ctx1ceab

480 algorithms.AES256(master_keys.encryption_key), modes.CBC(iv) 

481 ).decryptor() 

482 padded_plaintext = bytearray() 4 ctx1ceab

483 padded_plaintext.extend(decryptor.update(payload)) 4 ctx1ceab

484 padded_plaintext.extend(decryptor.finalize()) 4 ctx1ceab

485 unpadder = padding.PKCS7(IV_SIZE * 8).unpadder() 4 ctx1ceab

486 plaintext = bytearray() 4 ctx1ceab

487 plaintext.extend(unpadder.update(bytes(padded_plaintext))) 4 ctx1ceab

488 plaintext.extend(unpadder.finalize()) 4 ctx1ceab

489 session_encryption_key, session_signing_key = struct.unpack( 4 ctx1ceab

490 f"{KEY_SIZE}s {KEY_SIZE}s", plaintext 

491 ) 

492 except (ValueError, struct.error) as exc: 1 ctx1e

493 msg = "Invalid encrypted session keys payload" 1 ctx1e

494 raise ValueError(msg) from exc 1 ctx1e

495 

496 session_keys = _types.StoreroomKeyPair( 3 ctx1cab

497 encryption_key=session_encryption_key, 

498 signing_key=session_signing_key, 

499 ).toreadonly() 

500 

501 logger.debug( 3 ctx1cab

502 _msg.TranslatedString( 

503 _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_SESSION_KEYS_INFO, 

504 enc_key=_h(master_keys.encryption_key), 

505 iv=_h(iv), 

506 ciphertext=_h(payload), 

507 plaintext=_h(plaintext), 

508 code=_msg.TranslatedString( 

509 "StoreroomKeyPair(encryption_key=bytes.fromhex({enc_key!r}), " 

510 "signing_key=bytes.fromhex({sign_key!r}))", 

511 enc_key=session_keys.encryption_key.hex(" "), 

512 sign_key=session_keys.signing_key.hex(" "), 

513 ), 

514 ), 

515 ) 

516 

517 return session_keys 3 ctx1cab

518 

519 

520def _decrypt_contents( (empty)

521 data: Buffer, 

522 session_keys: _types.StoreroomKeyPair, 

523) -> Buffer: 

524 """Decrypt the bucket item's contents. 

525 

526 The data consists of: 

527 

528 - a 16-byte IV, 

529 - a variable-sized AES256-CBC-encrypted payload (using PKCS7 padding 

530 on the inside), and 

531 - a 32-byte MAC of the preceding bytes. 

532 

533 The encrypted payload is encrypted with the bucket item's session 

534 encryption key, and the MAC is created with the bucket item's 

535 session signing key. As per standard cryptographic procedure, the 

536 MAC can be verified before attempting to decrypt the payload. 

537 

538 Args: 

539 data: 

540 The encrypted bucket item payload data. 

541 session_keys: 

542 The bucket item's session keys. Presumably these have 

543 previously been obtained via the [`_decrypt_session_keys`][] 

544 function. 

545 

546 Returns: 

547 The bucket item's payload. 

548 

549 Raises: 

550 cryptography.exceptions.InvalidSignature: 

551 The data does not contain a valid signature under the given 

552 key. 

553 ValueError: 

554 The format is invalid, in a non-cryptographic way. (For 

555 example, it contains an unsupported version marker, or 

556 unexpected extra contents, or invalid padding.) 

557 

558 Warning: 

559 Non-public function, provided for didactical and educational 

560 purposes only. Subject to change without notice, including 

561 removal. 

562 

563 """ 

564 data = memoryview(data).toreadonly().cast("c") 3 ctx1cab

565 session_keys = session_keys.toreadonly() 3 ctx1cab

566 ciphertext, claimed_mac = struct.unpack( 3 ctx1cab

567 f"{len(data) - MAC_SIZE}s {MAC_SIZE}s", data 

568 ) 

569 actual_mac = hmac.HMAC(session_keys.signing_key, hashes.SHA256()) 3 ctx1cab

570 actual_mac.update(ciphertext) 3 ctx1cab

571 logger.debug( 3 ctx1cab

572 _msg.TranslatedString( 

573 _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_MAC_INFO, 

574 sign_key=_h(session_keys.signing_key), 

575 ciphertext=_h(ciphertext), 

576 claimed_mac=_h(claimed_mac), 

577 actual_mac=_h(actual_mac.copy().finalize()), 

578 ), 

579 ) 

580 actual_mac.verify(claimed_mac) 3 ctx1cab

581 

582 iv, payload = struct.unpack( 3 ctx1cab

583 f"{IV_SIZE}s {len(ciphertext) - IV_SIZE}s", ciphertext 

584 ) 

585 decryptor = ciphers.Cipher( 3 ctx1cab

586 algorithms.AES256(session_keys.encryption_key), modes.CBC(iv) 

587 ).decryptor() 

588 padded_plaintext = bytearray() 3 ctx1cab

589 padded_plaintext.extend(decryptor.update(payload)) 3 ctx1cab

590 padded_plaintext.extend(decryptor.finalize()) 3 ctx1cab

591 unpadder = padding.PKCS7(IV_SIZE * 8).unpadder() 3 ctx1cab

592 plaintext = bytearray() 3 ctx1cab

593 plaintext.extend(unpadder.update(bytes(padded_plaintext))) 3 ctx1cab

594 plaintext.extend(unpadder.finalize()) 3 ctx1cab

595 

596 logger.debug( 3 ctx1cab

597 _msg.TranslatedString( 

598 _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_INFO, 

599 enc_key=_h(session_keys.encryption_key), 

600 iv=_h(iv), 

601 ciphertext=_h(payload), 

602 plaintext=_h(plaintext), 

603 ), 

604 ) 

605 

606 return plaintext 3 ctx1cab

607 

608 

609def _decrypt_bucket_item( (empty)

610 bucket_item: Buffer, 

611 master_keys: _types.StoreroomMasterKeys, 

612) -> Buffer: 

613 """Decrypt a bucket item. 

614 

615 Args: 

616 bucket_item: 

617 The encrypted bucket item. 

618 master_keys: 

619 The master keys. Presumably these have previously been 

620 obtained via the [`_decrypt_master_keys_data`][] function. 

621 

622 Returns: 

623 The decrypted bucket item. 

624 

625 Raises: 

626 cryptography.exceptions.InvalidSignature: 

627 The data does not contain a valid signature under the given 

628 key. 

629 ValueError: 

630 The format is invalid, in a non-cryptographic way. (For 

631 example, it contains an unsupported version marker, or 

632 unexpected extra contents, or invalid padding.) 

633 

634 Warning: 

635 Non-public function, provided for didactical and educational 

636 purposes only. Subject to change without notice, including 

637 removal. 

638 

639 """ 

640 bucket_item = memoryview(bucket_item).toreadonly().cast("c") 4 ctx1ckab

641 master_keys = master_keys.toreadonly() 4 ctx1ckab

642 logger.debug( 4 ctx1ckab

643 _msg.TranslatedString( 

644 _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_KEY_INFO, 

645 plaintext=_h(bucket_item), 

646 enc_key=_h(master_keys.encryption_key), 

647 sign_key=_h(master_keys.signing_key), 

648 ), 

649 ) 

650 data_version, encrypted_session_keys, data_contents = struct.unpack( 4 ctx1ckab

651 ( 

652 f"B {ENCRYPTED_KEYPAIR_SIZE}s " 

653 f"{len(bucket_item) - 1 - ENCRYPTED_KEYPAIR_SIZE}s" 

654 ), 

655 bucket_item, 

656 ) 

657 if data_version != 1: 4 ctx1ckab

658 msg = f"Cannot handle version {data_version} encrypted data" 1 ctx1k

659 raise ValueError(msg) 1 ctx1k

660 session_keys = _decrypt_session_keys(encrypted_session_keys, master_keys) 3 ctx1cab

661 return _decrypt_contents(data_contents, session_keys) 3 ctx1cab

662 

663 

664def _decrypt_bucket_file( (empty)

665 filename: str | bytes | os.PathLike, 

666 master_keys: _types.StoreroomMasterKeys, 

667 *, 

668 root_dir: str | bytes | os.PathLike = ".", 

669) -> Generator[Buffer, None, None]: 

670 """Decrypt a complete bucket. 

671 

672 Args: 

673 filename: 

674 The bucket file's filename. 

675 master_keys: 

676 The master keys. Presumably these have previously been 

677 obtained via the [`_decrypt_master_keys_data`][] function. 

678 root_dir: 

679 The root directory of the data store. The filename is 

680 interpreted relatively to this directory. 

681 

682 Yields: 

683 A decrypted bucket item. 

684 

685 Raises: 

686 cryptography.exceptions.InvalidSignature: 

687 The data does not contain a valid signature under the given 

688 key. 

689 ValueError: 

690 The format is invalid, in a non-cryptographic way. (For 

691 example, it contains an unsupported version marker, or 

692 unexpected extra contents, or invalid padding.) 

693 

694 Warning: 

695 Non-public function, provided for didactical and educational 

696 purposes only. Subject to change without notice, including 

697 removal. 

698 

699 """ 

700 master_keys = master_keys.toreadonly() 4 ctx1chab

701 root_dir = pathlib.Path(os.fsdecode(root_dir)) 4 ctx1chab

702 filename = pathlib.Path(os.fsdecode(filename)) 4 ctx1chab

703 with (root_dir / filename).open("rb") as bucket_file: 4 ctx1chab

704 header_line = bucket_file.readline() 4 ctx1chab

705 try: 4 ctx1chab

706 header = json.loads(header_line) 4 ctx1chab

707 except ValueError as exc: 1 ctx1h

708 msg = f"Invalid bucket file: {filename}" 1 ctx1h

709 raise ValueError(msg) from exc 1 ctx1h

710 if header != {"version": 1}: 4 ctx1chab

711 msg = f"Invalid bucket file: {filename}" 1 ctx1h

712 raise ValueError(msg) from None 1 ctx1h

713 for line in bucket_file: 3 ctx1cab

714 yield _decrypt_bucket_item( 4 ctx1chab

715 base64.standard_b64decode(line), master_keys 

716 ) 

717 

718 

719def _store(config: dict[str, Any], path: str, json_contents: bytes) -> None: (empty)

720 """Store the JSON contents at path in the config structure. 

721 

722 Traverse the config structure according to path, and set the value 

723 of the leaf to the decoded JSON contents. 

724 

725 A path `/foo/bar/xyz` translates to the JSON structure 

726 `{"foo": {"bar": {"xyz": ...}}}`. 

727 

728 Args: 

729 config: 

730 The (top-level) configuration structure to update. 

731 path: 

732 The path within the configuration structure to traverse. 

733 json_contents: 

734 The contents to set the item to, after JSON-decoding. 

735 

736 Raises: 

737 json.JSONDecodeError: 

738 There was an error parsing the JSON contents. 

739 

740 """ 

741 contents = json.loads(json_contents) 3 ctx1cab

742 path_parts = [part for part in path.split("/") if part] 3 ctx1cab

743 for part in path_parts[:-1]: 3 ctx1cab

744 config = config.setdefault(part, {}) 3 ctx1cab

745 if path_parts: 3 ctx1cab

746 config[path_parts[-1]] = contents 3 ctx1cab

747 

748 

749if __name__ == "__main__": (empty)

750 logging.basicConfig(level=("DEBUG" if os.getenv("DEBUG") else "WARNING")) 

751 config_structure = export_storeroom_data(format="storeroom") 

752 print(json.dumps(config_structure, indent=2, sort_keys=True)) # noqa: T201