from typing import Any, List, Optional, Sequence from .auth_login_throttle_types import ThrottleDecision, ThrottlePolicy def _is_whole(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) def login_throttle(failures: Sequence[int], now: int, policy: ThrottlePolicy) -> ThrottleDecision: """Allowed or locked, by replaying the failures in time order: max_attempts failures within window_seconds of the latest lock the account for lockout_seconds from that failure, and the count starts again.""" if not _is_whole(policy.max_attempts) or policy.max_attempts < 1: raise ValueError("maxAttempts must be a whole number of at least 1") if not _is_whole(policy.window_seconds) or policy.window_seconds < 1: raise ValueError("windowSeconds must be a whole number of at least 1") if not _is_whole(policy.lockout_seconds) or policy.lockout_seconds < 1: raise ValueError("lockoutSeconds must be a whole number of at least 1") if not _is_whole(now): raise TypeError("now must be a whole number of Unix seconds") if isinstance(failures, (str, dict)) or not all(_is_whole(t) for t in failures): raise TypeError("failures must be whole Unix seconds") locked_until: Optional[int] = None streak: List[int] = [] for t in sorted(t for t in failures if t <= now): if locked_until is not None and t < locked_until: continue streak = [s for s in streak if s > t - policy.window_seconds] streak.append(t) if len(streak) >= policy.max_attempts: locked_until = t + policy.lockout_seconds streak = [] if locked_until is not None and now < locked_until: return ThrottleDecision(allowed=False, retry_after_seconds=locked_until - now, remaining_attempts=0, locked_until=locked_until) live = sum(1 for s in streak if s > now - policy.window_seconds) return ThrottleDecision(allowed=True, retry_after_seconds=0, remaining_attempts=policy.max_attempts - live, locked_until=None)