70 lines
2.8 KiB
Python
70 lines
2.8 KiB
Python
"""auth.py — login throttling: per-username lockout + per-source-IP budget.
|
|
|
|
Lives outside app.py so it is unit-testable without booting the app
|
|
(tests/test_auth.py). Both buckets are in-memory timestamp lists — the app
|
|
is single-process by design, so no store is needed; every read prunes
|
|
timestamps older than the window (sliding, not fixed).
|
|
|
|
- per-username: FAILS_USER timestamps inside the window -> that account is
|
|
locked, right or wrong password, until the newest fail ages out.
|
|
- per-source-IP: IP_FAILS fails from one source across ANY usernames ->
|
|
every login from that IP bounces. This is what username-rotation sweeps
|
|
cannot step around.
|
|
|
|
Client IP is request.client.host. X-Forwarded-For is consulted ONLY when
|
|
ANIWEB_TRUST_XFF=1 (run.sh sets it, because production traffic arrives via
|
|
the local HTTPS proxy); a directly-connected client can spoof the header,
|
|
so it is never a silent trust — without the flag client.host rules even if
|
|
XFF is present.
|
|
"""
|
|
import os
|
|
import time
|
|
|
|
FAILS_USER = 5 # prod default
|
|
WINDOW = float(os.environ.get("ANIWEB_LOCK_WINDOW", "900")) # seconds
|
|
IP_FAILS = int(os.environ.get("ANIWEB_IP_FAILS", "20")) # prod default
|
|
TRUST_XFF = os.environ.get("ANIWEB_TRUST_XFF") == "1"
|
|
|
|
_user_fails: dict = {} # username -> [fail timestamps, newest last]
|
|
_ip_fails: dict = {} # client ip -> [fail timestamps, newest last]
|
|
|
|
|
|
def client_ip(request) -> str:
|
|
"""Source of a login attempt: socket peer, or the proxy-reported client
|
|
when ANIWEB_TRUST_XFF=1 (see module docstring — spoofable, hence gated)."""
|
|
if TRUST_XFF:
|
|
first = (request.headers.get("x-forwarded-for") or "").split(",", 1)[0].strip()
|
|
if first:
|
|
return first
|
|
return request.client.host if request.client is not None else "?"
|
|
|
|
|
|
def _fresh(ts_list, now: float):
|
|
return [t for t in (ts_list or ()) if now - t < WINDOW]
|
|
|
|
|
|
def locked(user: str, ip: str, now: float | None = None) -> bool:
|
|
"""True when either bucket is over budget — caller must not verify."""
|
|
now = time.time() if now is None else now
|
|
return (len(_fresh(_user_fails.get(user), now)) >= FAILS_USER
|
|
or len(_fresh(_ip_fails.get(ip), now)) >= IP_FAILS)
|
|
|
|
|
|
def record(user: str, ip: str, now: float | None = None) -> None:
|
|
"""Count one failed attempt against both buckets (pruning stale fails)."""
|
|
now = time.time() if now is None else now
|
|
_user_fails[user] = _fresh(_user_fails.get(user), now) + [now]
|
|
_ip_fails[ip] = _fresh(_ip_fails.get(ip), now) + [now]
|
|
|
|
|
|
def clear(user: str) -> None:
|
|
"""A correct password resets only that username's count; the IP bucket
|
|
keeps aging on its own (one success must not un-lock a sweep)."""
|
|
_user_fails.pop(user, None)
|
|
|
|
|
|
def reset() -> None:
|
|
"""Test hook: drop all buckets."""
|
|
_user_fails.clear()
|
|
_ip_fails.clear()
|