Coverage for src/local_deep_research/database/rekey_epoch.py: 100%

1"""Durable authorization identity projection for encrypted user databases. 

2 

3Nothing in the application imports this module yet: it is the read side 

4for session and socket identity fencing, which the FastAPI layer does not 

5perform yet (a planned follow-up). Its readers do not take the rekey 

6sidecar lock, which is why ``read_rekey_epoch_state`` reads pending-first. 

7""" 

8 

9from pathlib import Path 

10 

11from .rekey_lock import RekeyRecoveryRequiredError, read_rekey_epoch_state 

12 

13 

14def read_authorization_epoch(db_path: Path) -> str: 

15 """Project the epoch that may authorize identities without repairing state.""" 

16 state = read_rekey_epoch_state(db_path) 

17 committed_epoch = state.committed_epoch 

18 if committed_epoch is None or ( 

19 state.pending_epoch is not None 

20 and state.pending_epoch != committed_epoch 

21 ): 

22 raise RekeyRecoveryRequiredError(db_path) 

23 return committed_epoch 

24 

25 

26def read_user_authorization_epoch(username: str) -> str: 

27 """Resolve a user's database path and project its authorization epoch. 

28 

29 The path MUST be resolved through ``db_manager`` (lazy-imported to 

30 keep this module's import graph shallow): the manager may carry a 

31 pinned ``data_dir`` override while ``get_data_directory()`` resolves 

32 dynamically from the environment, and epochs are written beside the 

33 databases under the MANAGER's resolution. Reading via the env-only 

34 path projected a different directory whenever the two diverged 

35 (any process that pinned the override), which would fence perfectly 

36 healthy sessions with ``RekeyRecoveryRequiredError``. 

37 """ 

38 from .encrypted_db import db_manager 

39 

40 db_path = db_manager._get_user_db_path(username) 

41 return read_authorization_epoch(db_path)