Coverage for src/lexigram/admin/auth/store/mfa_sql.py: 0%

1"""SQL-backed implementation of AdminMfaStoreProtocol. 

2 

3Owns all DDL and DML for the ``admin_mfa_totp`` table. The service 

4layer depends only on ``AdminMfaStoreProtocol`` from 

5``lexigram.admin.auth.protocols`` — never on this class directly. 

6""" 

7 

8from __future__ import annotations 

9 

10from lexigram.admin.sql_dialect import is_postgres, now_expr 

11from lexigram.contracts.data import DatabaseProviderProtocol 

12from lexigram.di.decorators import inject 

13from lexigram.logging import get_logger 

14from lexigram.security.encryption import EncryptionService 

15from lexigram.security.exceptions import DecryptionError 

16 

17logger = get_logger(__name__) 

18 

19_TABLE = "admin_mfa_totp" 

20 

21 

22@inject 

23class AdminMfaSqlStore: 

24 """SQL-backed store for per-user TOTP secrets. 

25 

26 Implements ``AdminMfaStoreProtocol`` via structural subtyping. 

27 Manages the ``admin_mfa_totp`` table including DDL bootstrap and 

28 idempotent enable/disable semantics. A row's existence means 2FA 

29 is enabled for that user; disabling deletes the row. 

30 """ 

31 

32 def __init__( 

33 self, 

34 db: DatabaseProviderProtocol, 

35 encryption_service: EncryptionService | None = None, 

36 ) -> None: 

37 """Initialise with a resolved database provider. 

38 

39 Args: 

40 db: Framework database provider exposing ``execute`` and 

41 ``execute_query``. 

42 encryption_service: Optional AES-256-GCM service used to 

43 encrypt/decrypt ``secret`` at rest. ``None`` keeps the 

44 legacy plaintext behavior (tests and no-crypto builds). 

45 """ 

46 self._db = db 

47 self._encryption_service = encryption_service 

48 self._initialized = False 

49 

50 # ------------------------------------------------------------------ 

51 # Schema bootstrap (DDL) 

52 # ------------------------------------------------------------------ 

53 

54 async def ensure_schema(self) -> None: 

55 """Create the MFA table if it does not exist (idempotent). 

56 

57 The ``secret`` column is ``VARCHAR(512)`` — AES-256-GCM hex 

58 ciphertext (nonce 12 + tag 16 + data bytes) exceeds the legacy 

59 ``VARCHAR(64)`` width; on Postgres an idempotent ``ALTER`` widens 

60 pre-existing tables. 

61 """ 

62 if self._initialized: 

63 return 

64 if is_postgres(self._db): 

65 create_sql = f""" 

66 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

67 user_id TEXT PRIMARY KEY, 

68 secret VARCHAR(512) NOT NULL, 

69 enabled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

70 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 

71 ) 

72 """ 

73 await self._db.execute(create_sql, []) 

74 await self._db.execute( 

75 f"ALTER TABLE {_TABLE} ALTER COLUMN secret TYPE VARCHAR(512)", 

76 [], 

77 ) 

78 else: 

79 create_sql = f""" 

80 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

81 user_id TEXT PRIMARY KEY, 

82 secret VARCHAR(512) NOT NULL, 

83 enabled_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, 

84 updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP 

85 ) 

86 """ 

87 await self._db.execute(create_sql, []) 

88 self._initialized = True 

89 

90 # ------------------------------------------------------------------ 

91 # AdminMfaStoreProtocol 

92 # ------------------------------------------------------------------ 

93 

94 async def is_enabled(self, user_id: str) -> bool: 

95 """Return True when the user has a stored TOTP secret.""" 

96 return await self.get_secret(user_id) is not None 

97 

98 async def get_secret(self, user_id: str) -> str | None: 

99 """Return the stored TOTP secret for a user (None when disabled). 

100 

101 Decrypts the stored value when an ``encryption_service`` is 

102 configured. Rows written before encryption (raw base32) fall 

103 back to their raw value — a one-time read path; they are 

104 re-encrypted on the next ``save_secret``. 

105 """ 

106 result = await self._db.execute_query( 

107 f"SELECT secret FROM {_TABLE} WHERE user_id = ?", # noqa: S608 — table name is module constant "admin_mfa_totp", never user input 

108 [user_id], 

109 ) 

110 row = None 

111 if hasattr(result, "rows") and result.rows: 

112 row = result.rows[0] 

113 elif isinstance(result, list) and result: 

114 row = result[0] 

115 elif isinstance(result, dict): 

116 row = result 

117 if not row: 

118 return None 

119 value = str(row.get("secret", "")) 

120 if self._encryption_service is None: 

121 return value 

122 try: 

123 return self._encryption_service.decrypt(value) 

124 except DecryptionError: 

125 return value 

126 

127 async def save_secret(self, user_id: str, secret: str) -> None: 

128 """Persist (or refresh) the TOTP secret for a user. 

129 

130 The value is encrypted at rest (ciphertext in ``secret``) when 

131 an ``encryption_service`` is configured, otherwise stored raw. 

132 """ 

133 stored = ( 

134 self._encryption_service.encrypt(secret) 

135 if self._encryption_service is not None 

136 else secret 

137 ) 

138 await self._db.execute( 

139 f""" 

140 INSERT INTO {_TABLE} (user_id, secret) 

141 VALUES (?, ?) 

142 ON CONFLICT (user_id) DO UPDATE SET 

143 secret = excluded.secret, 

144 updated_at = {now_expr(self._db)} 

145 """, # noqa: S608 — table name is module constant, now_expr yields fixed NOW()/CURRENT_TIMESTAMP 

146 [user_id, stored], 

147 ) 

148 

149 async def disable(self, user_id: str) -> None: 

150 """Remove the TOTP secret for a user (2FA off).""" 

151 await self._db.execute( 

152 f"DELETE FROM {_TABLE} WHERE user_id = ?", # noqa: S608 — table name is module constant "admin_mfa_totp", never user input 

153 [user_id], 

154 ) 

155 

156 

157__all__ = ["AdminMfaSqlStore"]