Coverage for src/local_deep_research/security/account_lockout.py: 99%

1"""Per-user account lockout after repeated failed login attempts. 

2 

3Complements the per-IP rate limiting in ``security/rate_limiter.py`` by 

4tracking failures at the *username* level. A user who exceeds the 

5configured threshold is temporarily locked out regardless of the source IP. 

6 

7Note: lockout state is in-memory and per-process. In multi-worker 

8deployments (e.g. gunicorn), each worker maintains separate state. 

9""" 

10 

11import threading 

12from collections import OrderedDict 

13from datetime import datetime, timedelta, timezone 

14 

15from loguru import logger 

16 

17from .security_settings import get_security_default 

18 

19 

20class AccountLockoutManager: 

21 """Track failed login attempts and lock accounts after a threshold. 

22 

23 A failure counter lasts one lockout window from its last failure, the 

24 same as a lockout. Within its window an entry is kept, with one 

25 exception that bounds memory: counters holding a single failure, the 

26 only kind a spray of throwaway usernames creates for one request each, 

27 are capped at ``_MAX_SINGLE_FAILURES``, and a new one past that evicts 

28 the oldest. A counter that has reached two failures, and an active 

29 lockout, stay until their window ends or the user logs in, so flooding 

30 the table cannot reset them. 

31 """ 

32 

33 _MAX_SINGLE_FAILURES = 10_000 

34 

35 def __init__( 

36 self, 

37 threshold: int | None = None, 

38 lockout_minutes: int | None = None, 

39 ) -> None: 

40 if threshold is None: 

41 threshold = get_security_default( 

42 "security.account_lockout_threshold", 10 

43 ) 

44 if lockout_minutes is None: 

45 lockout_minutes = get_security_default( 

46 "security.account_lockout_duration_minutes", 15 

47 ) 

48 

49 self.threshold: int = threshold 

50 self.lockout_minutes: int = lockout_minutes 

51 

52 # {username: {"count": int, "locked_until": datetime | None, 

53 # "last_failure": datetime}}, ordered by last failure 

54 self._state: OrderedDict[str, dict] = OrderedDict() 

55 # The usernames in _state whose counter holds a single failure and 

56 # no lockout, in the order of that failure. 

57 self._singles: OrderedDict[str, None] = OrderedDict() 

58 self._lock = threading.Lock() 

59 

60 # ------------------------------------------------------------------ 

61 # Public API 

62 # ------------------------------------------------------------------ 

63 

64 def is_locked(self, username: str) -> bool: 

65 """Return ``True`` if *username* is currently locked out.""" 

66 with self._lock: 

67 entry = self._state.get(username) 

68 if entry is None: 

69 return False 

70 locked_until = entry.get("locked_until") 

71 if locked_until is None: 

72 return False 

73 if datetime.now(timezone.utc) >= locked_until: 

74 # Lockout expired — reset automatically 

75 self._state.pop(username, None) 

76 logger.info("Account lockout expired") 

77 return False 

78 return True 

79 

80 def _drop_expired(self, now: datetime) -> None: 

81 """Drop the entries whose last failure is a full window old. 

82 

83 The table is ordered by last failure, so they are all at its front. 

84 A lockout ends one window after the failure that set it, so this 

85 covers expired lockouts as well as counters whose window has passed. 

86 

87 Must be called while ``self._lock`` is held. 

88 """ 

89 cutoff = now - timedelta(minutes=self.lockout_minutes) 

90 while self._state: 

91 username, entry = next(iter(self._state.items())) 

92 if entry["last_failure"] > cutoff: 

93 break 

94 del self._state[username] 

95 self._singles.pop(username, None) 

96 

97 def record_failure(self, username: str) -> None: 

98 """Record a failed login attempt for *username*.""" 

99 with self._lock: 

100 # Read under the lock, so the table's order by last failure is 

101 # the order in which failures take the lock — assuming the 

102 # wall clock doesn't step backwards (#6943). A backward step 

103 # of d (NTP correction, manual clock change) can leave an older 

104 # stamp after a newer one. _drop_expired stops at the first 

105 # unexpired entry, so an expired counter behind it can linger 

106 # up to d past its window; repeated steps add up, and nothing 

107 # caps the linger at the lockout window. While it lingers, its 

108 # count carries into that user's next failure. The direction 

109 # is conservative: a backward step drops nothing early, and 

110 # is_locked compares locked_until with the clock directly, so 

111 # no lockout ends early (one set before the step lasts up to d 

112 # longer). If that ever needs hardening, clamp stamps 

113 # (max(latest_seen, now)) or order on a monotonic clock instead. 

114 now = datetime.now(timezone.utc) 

115 self._drop_expired(now) 

116 entry = self._state.get(username) 

117 if entry is None: 

118 entry = {"count": 0, "locked_until": None} 

119 self._state[username] = entry 

120 else: 

121 self._state.move_to_end(username) 

122 entry["last_failure"] = now 

123 entry["count"] += 1 

124 if entry["count"] >= self.threshold: 

125 entry["locked_until"] = now + timedelta( 

126 minutes=self.lockout_minutes 

127 ) 

128 logger.warning( 

129 "Account locked after {} failed attempts", 

130 self.threshold, 

131 ) 

132 

133 if entry["count"] == 1 and entry["locked_until"] is None: 

134 self._singles[username] = None 

135 if len(self._singles) > self._MAX_SINGLE_FAILURES: 

136 oldest, _ = self._singles.popitem(last=False) 

137 del self._state[oldest] 

138 else: 

139 self._singles.pop(username, None) 

140 

141 def record_success(self, username: str) -> None: 

142 """Clear the failure counter for *username* after a successful login.""" 

143 with self._lock: 

144 removed = self._state.pop(username, None) 

145 self._singles.pop(username, None) 

146 if removed is not None: 

147 logger.info("Account lockout cleared after successful login") 

148 

149 

150# ------------------------------------------------------------------ 

151# Module-level singleton 

152# ------------------------------------------------------------------ 

153 

154_manager: AccountLockoutManager | None = None 

155_singleton_lock = threading.Lock() 

156 

157 

158def get_account_lockout_manager() -> AccountLockoutManager: 

159 """Return the module-level singleton ``AccountLockoutManager``.""" 

160 global _manager 

161 if _manager is None: 

162 with _singleton_lock: 

163 if _manager is None: 163 ↛ 165line 163 didn't jump to line 165

164 _manager = AccountLockoutManager() 

165 return _manager