Coverage for src/local_deep_research/web/auth/password_utils.py: 100%

1""" 

2Shared utilities for resolving the current user's DB password. 

3 

4``get_user_password`` centralises the 3-source fallback chain (session 

5password store → Flask g → temp auth store) so every route that needs the 

6current user's DB password uses the same logic. Without this, each route 

7reimplemented the chain independently, risking subtle divergence (e.g. one 

8route forgetting to check temp_auth, or using a different method alias). 

9 

10``resolve_user_password`` builds on it with the encryption-aware guard the 

11research entry points share: it returns ``(password, session_expired)`` so 

12a caller can reject a run when an encrypted DB has no available password. 

13""" 

14 

15from typing import Optional, Tuple 

16 

17from flask import g, session 

18from loguru import logger 

19 

20 

21def get_user_password(username: str) -> Optional[str]: 

22 """Retrieve user password from available session sources. 

23 

24 Checks, in order: 

25 1. SessionPasswordStore (persistent per-session passwords) 

26 2. Flask ``g.user_password`` (set by middleware when temp_auth was used) 

27 3. TempAuthStore (one-time tokens stored during login redirect) 

28 

29 Returns ``None`` when no password can be found — callers must decide 

30 whether that is acceptable (e.g. non-encrypted databases) or an error 

31 (encrypted databases → 401). 

32 """ 

33 from ...database.session_passwords import session_password_store 

34 

35 session_id = session.get("session_id") 

36 if session_id: 

37 password = session_password_store.get_session_password( 

38 username, session_id 

39 ) 

40 if password: 

41 return password 

42 

43 password = getattr(g, "user_password", None) 

44 if password: 

45 return password 

46 

47 from ...database.temp_auth import temp_auth_store 

48 

49 auth_token = session.get("temp_auth_token") 

50 if auth_token: 

51 auth_data = temp_auth_store.peek_auth(auth_token) 

52 if auth_data and auth_data[0] == username: 

53 return auth_data[1] 

54 

55 return None 

56 

57 

58def resolve_user_password(username: str) -> Tuple[Optional[str], bool]: 

59 """Resolve the user's DB password for starting a research run. 

60 

61 Returns ``(password, session_expired)``: 

62 

63 - ``session_expired`` is ``True`` only when the database is encrypted 

64 and no password is available. The caller MUST reject the request 

65 (e.g. a 401 telling the user to log back in) because the research's 

66 background DB and metric writes would otherwise be silently dropped 

67 (issue #4457) — research would appear to run while every metric write 

68 fails. Trigger: session-password-store TTL expiry or a server/ 

69 container restart while the session cookie is still valid. 

70 - For unencrypted databases ``session_expired`` is always ``False``; 

71 the returned ``password`` may legitimately be ``None``. 

72 

73 Centralises the guard that the direct (``/start_research``), follow-up, 

74 and chat research entry points all need, so the encryption-aware 

75 decision and its logging live in one place instead of being copied 

76 (and risking divergence) across routes. Each route still formats its 

77 own error response, because their frontends expect different shapes. 

78 """ 

79 from ...database.encrypted_db import db_manager 

80 

81 password = get_user_password(username) # gitleaks:allow 

82 if not password and db_manager.has_encryption: 

83 logger.error( 

84 f"No password available for user {username} with encrypted " 

85 "database - cannot start research (session password expired or " 

86 "lost after server restart)" 

87 ) 

88 return None, True 

89 if not password: 

90 logger.warning( 

91 f"No password available for metrics access for user {username} " 

92 "(unencrypted database)" 

93 ) 

94 return password, False