Coverage for src / lexigram / admin / auth / services / login_attempt_service.py: 20%
92 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Admin login attempt service — IP rate limiting and account lockout."""
3from __future__ import annotations
5from datetime import UTC, datetime
6import hashlib
8from lexigram.admin.auth.errors import AccountLockedError, RateLimitExceededError
9from lexigram.admin.auth.protocols import (
10 AdminAccountLockoutStoreProtocol,
11 AdminAuditLogServiceProtocol,
12 AdminLoginAttemptStoreProtocol,
13)
14from lexigram.admin.auth.types import (
15 AdminLockoutStatus,
16 AdminLoginAttempt,
17 AdminSecurityEventType,
18)
19from lexigram.contracts.infra.cache import CacheBackendProtocol
20from lexigram.di.decorators import inject
21from lexigram.logging import get_logger
23logger = get_logger(__name__)
26@inject
27class AdminLoginAttemptService:
28 """Manages IP rate limiting and progressive account lockout for admin auth.
30 IP rate limiting uses an optional CacheBackendProtocol. If cache is
31 unavailable (None), rate limiting is gracefully skipped (fail open).
32 Account lockout is DB-persisted and survives restarts.
33 """
35 def __init__(
36 self,
37 attempt_store: AdminLoginAttemptStoreProtocol,
38 lockout_store: AdminAccountLockoutStoreProtocol,
39 cache: CacheBackendProtocol | None = None,
40 ip_rate_limit_enabled: bool = True,
41 ip_limit_per_minute: int = 10,
42 ip_limit_per_15_minutes: int = 30,
43 ip_limit_per_hour: int = 60,
44 lockout_thresholds: list[tuple[int, int]] | None = None,
45 permanent_lockout_threshold: int = 50,
46 audit_service: AdminAuditLogServiceProtocol | None = None,
47 ) -> None:
48 """Initialize with stores and configuration.
50 Args:
51 attempt_store: Store for login attempt records.
52 lockout_store: Store for account lockout records.
53 cache: Optional cache backend for IP rate limiting.
54 ip_rate_limit_enabled: Whether IP rate limiting is active.
55 ip_limit_per_minute: Max failures per IP per minute.
56 ip_limit_per_15_minutes: Max failures per IP per 15 minutes.
57 ip_limit_per_hour: Max failures per IP per hour.
58 lockout_thresholds: List of (failure_count, lockout_minutes) pairs.
59 Defaults to [(5, 15), (10, 60), (15, 240), (20, 1440)].
60 permanent_lockout_threshold: Failures before permanent lockout.
61 """
62 self._attempt_store = attempt_store
63 self._lockout_store = lockout_store
64 self._cache = cache
65 self._ip_rate_limit_enabled = ip_rate_limit_enabled
66 self._ip_limit_per_minute = ip_limit_per_minute
67 self._ip_limit_per_15_minutes = ip_limit_per_15_minutes
68 self._ip_limit_per_hour = ip_limit_per_hour
69 self._lockout_thresholds = lockout_thresholds or [
70 (5, 15),
71 (10, 60),
72 (15, 240),
73 (20, 1440),
74 ]
75 self._permanent_lockout_threshold = permanent_lockout_threshold
76 self._audit_service = audit_service
78 async def check_ip_rate_limit(self, ip_address: str) -> None:
79 """Check if IP is rate-limited. Raises RateLimitExceededError if exceeded.
81 Uses cache for fast lookups. Gracefully skips if cache is unavailable.
83 Args:
84 ip_address: Client IP address.
86 Raises:
87 RateLimitExceededError: When the IP exceeds any rate limit tier.
88 """
89 if not self._ip_rate_limit_enabled or self._cache is None:
90 return
92 # Hash IP to avoid PII in Redis KEYS output
93 ip_hash = hashlib.sha256(ip_address.encode()).hexdigest()[:16]
95 try:
96 # Check three tiers using the DB-backed attempt store for accuracy
97 # (Cache is used for hard blocks only — set by prior blocked requests)
98 blocked_key = f"admin:blocked:ip:{ip_hash}"
99 is_blocked = await self._cache.get(blocked_key)
100 if is_blocked.is_ok() and is_blocked.unwrap():
101 logger.warning("ip_rate_limit.hard_blocked", ip_hash=ip_hash)
102 raise RateLimitExceededError(
103 "Too many failed login attempts from this IP. Please try again later.",
104 reason="rate_limit",
105 )
107 # Count recent failures from DB
108 failures_1min = await self._attempt_store.count_recent_failures_by_ip(
109 ip_address, 60
110 )
111 failures_15min = await self._attempt_store.count_recent_failures_by_ip(
112 ip_address, 900
113 )
114 failures_1hr = await self._attempt_store.count_recent_failures_by_ip(
115 ip_address, 3600
116 )
118 if failures_1min >= self._ip_limit_per_minute:
119 # Set a 5-minute hard block in cache
120 await self._cache.set(blocked_key, "1", ttl=300)
121 logger.warning(
122 "ip_rate_limit.exceeded_minute",
123 ip_hash=ip_hash,
124 count=failures_1min,
125 )
126 raise RateLimitExceededError(
127 "Too many login attempts. Please wait 5 minutes before trying again.",
128 retry_after=300,
129 reason="rate_limit",
130 )
132 if failures_15min >= self._ip_limit_per_15_minutes:
133 await self._cache.set(blocked_key, "1", ttl=900)
134 logger.warning(
135 "ip_rate_limit.exceeded_15min",
136 ip_hash=ip_hash,
137 count=failures_15min,
138 )
139 raise RateLimitExceededError(
140 "Too many login attempts. Please wait 15 minutes before trying again.",
141 retry_after=900,
142 reason="rate_limit",
143 )
145 if failures_1hr >= self._ip_limit_per_hour:
146 await self._cache.set(blocked_key, "1", ttl=3600)
147 logger.warning(
148 "ip_rate_limit.exceeded_hour",
149 ip_hash=ip_hash,
150 count=failures_1hr,
151 )
152 raise RateLimitExceededError(
153 "Too many login attempts. Please try again in 1 hour.",
154 retry_after=3600,
155 reason="rate_limit",
156 )
158 except RateLimitExceededError:
159 raise
160 except Exception: # noqa: BLE001
161 # Cache is unavailable — fail open (never block auth due to cache failure)
162 logger.warning("ip_rate_limit.cache_unavailable", ip_hash=ip_hash)
164 async def check_account_lockout(self, email: str) -> None:
165 """Check account lockout. Raises AccountLockedError if locked.
167 Args:
168 email: Email to check.
170 Raises:
171 AccountLockedError: When the account is locked.
172 """
173 lockout_info = await self._lockout_store.get_active_lockout(email)
174 if lockout_info is None:
175 return
177 if lockout_info.status == AdminLockoutStatus.PERMANENT:
178 raise AccountLockedError(
179 "This account has been permanently locked. Please contact an administrator.",
180 reason="lockout",
181 )
183 if lockout_info.status == AdminLockoutStatus.LOCKED:
184 if lockout_info.unlock_at:
185 retry_after = max(
186 0,
187 int((lockout_info.unlock_at - datetime.now(UTC)).total_seconds()),
188 )
189 raise AccountLockedError(
190 f"Account temporarily locked due to too many failed attempts. "
191 f"Please try again after {lockout_info.unlock_at.strftime('%H:%M UTC')}.",
192 unlock_at=lockout_info.unlock_at,
193 retry_after=retry_after,
194 reason="lockout",
195 )
196 raise AccountLockedError(
197 "Account is temporarily locked. Please try again later.",
198 reason="lockout",
199 )
201 async def record_attempt(
202 self,
203 email: str,
204 ip_address: str,
205 user_agent: str,
206 success: bool,
207 failure_reason: str | None = None,
208 ) -> None:
209 """Record a login attempt and update lockout state on failure.
211 On success, lockout counters are NOT cleared here — use clear_lockout().
212 On failure, checks thresholds and creates/updates lockout record.
214 Args:
215 email: Email that attempted login.
216 ip_address: Client IP.
217 user_agent: Client user agent.
218 success: Whether the attempt succeeded.
219 failure_reason: Short failure code when success=False.
220 """
221 from datetime import timedelta
222 import uuid
224 attempt = AdminLoginAttempt(
225 id=str(uuid.uuid4()),
226 email=email,
227 ip_address=ip_address,
228 user_agent=user_agent,
229 success=success,
230 failure_reason=failure_reason,
231 attempted_at=datetime.now(UTC),
232 )
233 await self._attempt_store.insert(attempt)
235 if success:
236 return
238 # Count total consecutive failures to determine lockout level
239 total_failures = await self._attempt_store.count_recent_failures(email, 86400)
241 if total_failures >= self._permanent_lockout_threshold:
242 await self._lockout_store.create_lockout(
243 email=email,
244 consecutive_failures=total_failures,
245 unlock_at=None,
246 is_permanent=True,
247 )
248 logger.warning(
249 "account_lockout.permanent",
250 email_hash=hashlib.sha256(email.encode()).hexdigest()[:8],
251 )
252 return
254 # Find applicable lockout threshold
255 lockout_minutes: int | None = None
256 for threshold_failures, minutes in sorted(
257 self._lockout_thresholds, reverse=True
258 ):
259 if total_failures >= threshold_failures:
260 lockout_minutes = minutes
261 break
263 if lockout_minutes is not None:
264 unlock_at = datetime.now(UTC) + timedelta(minutes=lockout_minutes)
265 await self._lockout_store.create_lockout(
266 email=email,
267 consecutive_failures=total_failures,
268 unlock_at=unlock_at,
269 is_permanent=False,
270 )
271 logger.warning(
272 "account_lockout.temporary",
273 email_hash=hashlib.sha256(email.encode()).hexdigest()[:8],
274 minutes=lockout_minutes,
275 failures=total_failures,
276 )
278 async def clear_lockout(self, email: str) -> None:
279 """Clear lockout and failure records on successful login.
281 Audits an ``ACCOUNT_UNLOCKED`` event when an active lockout existed.
283 Args:
284 email: Email to clear.
285 """
286 active = await self._lockout_store.get_active_lockout(email)
287 await self._lockout_store.clear_lockout(email)
288 await self._attempt_store.clear_failures(email)
289 if active and self._audit_service:
290 await self._audit_service.log_event(
291 event_type=AdminSecurityEventType.ACCOUNT_UNLOCKED,
292 ip_address="",
293 user_agent="",
294 success=True,
295 metadata={"email": email},
296 )
297 logger.debug(
298 "account_lockout.cleared",
299 email_hash=hashlib.sha256(email.encode()).hexdigest()[:8],
300 )
303__all__ = ["AdminLoginAttemptService"]