Coverage for src / lexigram / contracts / auth / protocols.py: 100%

32 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-08-19 05:41 +0800

1"""Auth password and provider protocol class definitions.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, Protocol, runtime_checkable 

6 

7from lexigram.contracts.core.provider import ProviderProtocol 

8 

9 

10@runtime_checkable 

11class LoginAttemptTrackerProtocol(Protocol): 

12 """Protocol for tracking failed login attempts and enforcing account lockout. 

13 

14 Implementations must handle both in-process and distributed state (e.g. 

15 backed by a cache) so that multiple application instances share a 

16 consistent view of the failure window. 

17 """ 

18 

19 async def is_locked(self, identifier: str) -> bool: 

20 """Return ``True`` if *identifier* has exceeded the failure threshold. 

21 

22 Args: 

23 identifier: Username, e-mail address, or IP used as the tracking key. 

24 

25 Returns: 

26 ``True`` when the account is currently locked out, ``False`` otherwise. 

27 """ 

28 ... 

29 

30 async def record_failure(self, identifier: str) -> None: 

31 """Record a failed authentication attempt for *identifier*. 

32 

33 Args: 

34 identifier: Username, e-mail address, or IP used as the tracking key. 

35 """ 

36 ... 

37 

38 async def clear(self, identifier: str) -> None: 

39 """Remove all recorded failures for *identifier* (call on successful login). 

40 

41 Args: 

42 identifier: Username, e-mail address, or IP used as the tracking key. 

43 """ 

44 ... 

45 

46 

47@runtime_checkable 

48class PasswordHasherProtocol(Protocol): 

49 """Protocol for password hashing services. 

50 

51 Responsible for hashing passwords and verifying them against hashes. 

52 """ 

53 

54 async def hash(self, password: str) -> str: 

55 """Hash a plain text password. 

56 

57 Args: 

58 password: Plain text password. 

59 

60 Returns: 

61 Hashed password string. 

62 """ 

63 ... 

64 

65 async def verify(self, password: str, hashed_password: str) -> bool: 

66 """Verify a password against a hash. 

67 

68 Args: 

69 password: Plain text password to check. 

70 hashed_password: Stored hash to compare against. 

71 

72 Returns: 

73 True if password matches hash, False otherwise. 

74 """ 

75 ... 

76 

77 def needs_rehash(self, hashed_password: str) -> bool: 

78 """Return True when the stored hash should be re-computed. 

79 

80 Implementations parse the cost parameters from the self-describing 

81 stored hash and compare them against their configured target. 

82 Unparseable or unknown formats return ``True`` (fail-closed) — safe 

83 because rehashing only ever runs after a successful ``verify()``. 

84 

85 Args: 

86 hashed_password: Stored hash string. 

87 

88 Returns: 

89 True when the hash is below the configured cost target. 

90 """ 

91 ... 

92 

93 async def rehash_if_needed( 

94 self, 

95 password: str, 

96 hashed_password: str | None, 

97 ) -> str | None: 

98 """Rehash *password* when *hashed_password* is below the cost target. 

99 

100 Django ``check_password(setter=...)`` pattern: run this after a 

101 successful ``verify()`` and persist the returned hash to upgrade 

102 stored credentials in place. Returns ``None`` when no upgrade is 

103 needed (or no stored hash exists). 

104 

105 Args: 

106 password: Plain text password (already verified against the hash). 

107 hashed_password: Stored hash string, or None. 

108 

109 Returns: 

110 A freshly computed hash when an upgrade is needed, else None. 

111 """ 

112 ... 

113 

114 

115@runtime_checkable 

116class PasswordPolicyProtocol(Protocol): 

117 """Protocol for password policy enforcement. 

118 

119 Validates that a plain-text password satisfies the application's 

120 complexity requirements (minimum length, character classes, etc.). 

121 """ 

122 

123 def validate(self, password: str) -> None: 

124 """Validate password against the policy. 

125 

126 Args: 

127 password: Plain text password to validate. 

128 

129 Raises: 

130 ValidationError: If the password violates the policy. 

131 """ 

132 ... 

133 

134 def is_valid(self, password: str) -> bool: 

135 """Return True if the password satisfies the policy without raising. 

136 

137 Args: 

138 password: Plain text password. 

139 

140 Returns: 

141 True if valid, False otherwise. 

142 """ 

143 ... 

144 

145 

146@runtime_checkable 

147class MFAManagerProtocol(Protocol): 

148 """Protocol for multi-factor authentication lifecycle management.""" 

149 

150 async def enroll(self, user_id: str, method: str) -> dict[str, Any]: ... 

151 async def verify(self, user_id: str, method: str, code: str) -> bool: ... 

152 async def revoke(self, user_id: str, method: str) -> None: ... 

153 async def list_methods(self, user_id: str) -> list[str]: ... 

154 async def get_mfa(self, user_id: str) -> Any | None: ... 

155 

156 

157@runtime_checkable 

158class AuthProviderProtocol(ProviderProtocol, Protocol): 

159 """Protocol for authentication and authorization providers. 

160 

161 Auth providers are responsible for user authentication, authorization, 

162 token management, and access control. 

163 """ 

164 

165 # Optional managers — None when the feature is disabled/not configured 

166 user_store: Any | None 

167 session_manager: Any | None 

168 delegation_manager: Any | None 

169 api_key_manager: Any | None 

170 mfa_manager: MFAManagerProtocol | None 

171 

172 async def get_user(self, user_id: str) -> Any | None: 

173 """Retrieve a user by their unique identifier. 

174 

175 Args: 

176 user_id: Unique user identifier. 

177 

178 Returns: 

179 User object or None if not found. 

180 """ 

181 ... 

182 

183 async def verify_token(self, token: str) -> Any: 

184 """Verify an encoded auth token and return verification details. 

185 

186 Args: 

187 token: Raw encoded auth token (JWT or similar). 

188 

189 Returns: 

190 Verification result (typically ``Result[VerifiedToken, TokenError]``). 

191 """ 

192 ... 

193 

194 def has_any_role(self, user: Any, roles: list[str]) -> bool: 

195 """Return True if *user* holds at least one of the given roles. 

196 

197 Args: 

198 user: Authenticated user object. 

199 roles: Role names to check. 

200 

201 Returns: 

202 True if the user has at least one of the supplied roles. 

203 """ 

204 ... 

205 

206 def has_any_permission(self, user: Any, permissions: list[str]) -> bool: 

207 """Return True if *user* has at least one of the given permissions. 

208 

209 Args: 

210 user: Authenticated user object. 

211 permissions: Permission names to check. 

212 

213 Returns: 

214 True if the user has at least one of the supplied permissions. 

215 """ 

216 ... 

217 

218 

219__all__ = [ 

220 "AuthProviderProtocol", 

221 "LoginAttemptTrackerProtocol", 

222 "MFAManagerProtocol", 

223 "PasswordHasherProtocol", 

224 "PasswordPolicyProtocol", 

225]