impl/python.py

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)