Lockout -> server/auth.py + per-IP failure budget (XFF trusted only behind the proxy); unit tests

run.sh

@@ -25,6 +25,8 @@ fi
 
 # ANIWEB_SECURE=1: session cookie is Secure — only usable via the HTTPS
 # proxy (plain http://<this machine>:8321 can no longer hold a session).
-ANIWEB_SECURE=1 nohup .venv/bin/python -m uvicorn server.app:app --host 0.0.0.0 --port 8321 --timeout-graceful-shutdown 3 > server.log 2>&1 &
+# ANIWEB_TRUST_XFF=1: behind that same local proxy, so X-Forwarded-For may
+# be trusted for the per-IP login budget (server/auth.py).
+ANIWEB_SECURE=1 ANIWEB_TRUST_XFF=1 nohup .venv/bin/python -m uvicorn server.app:app --host 0.0.0.0 --port 8321 --timeout-graceful-shutdown 3 > server.log 2>&1 &
 echo $! > "$PIDF"
 echo "ani-web on http://localhost:8321 (pid $(cat "$PIDF")) — use the HTTPS proxy to log in"

server/auth.py

@@ -0,0 +1,68 @@
+"""auth.py — login throttling: per-username lockout + per-IP failure budget.
+
+Pure logic, no app imports, so it unit-tests without booting the app
+(tests/test_auth.py). The 0.5s wrong-password sleep is exactly why app.py's
+login handler must stay a sync def: on the event loop it would stall every
+stream in flight; on the threadpool it only holds that one request.
+
+Two in-memory layers (single process — same assumption as everything else):
+  * per-username: 5 fails -> 15 min lockout for that name (was: the only
+    layer, so spraying fails across many names bypassed it entirely)
+  * per-IP: IP_FAILS fails inside IP_WINDOW -> that IP is refused for the
+    rest of the window. Defaults are deliberately loose (15/300s): they
+    exist to stop username spraying, not to punish typos.
+"""
+import os
+import time
+
+from . import users
+
+LOCK_FAILS = 5
+LOCK_SECONDS = 900
+IP_FAILS = int(os.environ.get("ANIWEB_IP_FAILS", "15"))
+IP_WINDOW = float(os.environ.get("ANIWEB_IP_WINDOW", "300"))
+
+_lockout: dict = {}   # user -> {"fails": int, "until": float}
+_ip_fails: dict = {}  # ip -> [failure ts] (newest last, capped)
+
+
+def client_ip(request) -> str:
+    """Client IP. X-Forwarded-For is trusted ONLY when ANIWEB_TRUST_XFF=1
+    (run.sh sets it because production is always behind the local HTTPS
+    proxy); unconditional trust would let any client forge the header and
+    dodge the per-IP budget."""
+    if os.environ.get("ANIWEB_TRUST_XFF") == "1":
+        xff = request.headers.get("x-forwarded-for", "")
+        if xff:
+            return xff.split(",", 1)[0].strip()
+    return request.client.host if request.client else "?"
+
+
+def ip_fail(ip: str) -> None:
+    _ip_fails[ip] = (_ip_fails.get(ip, []) + [time.time()])[-50:]
+
+
+def ip_locked(ip: str) -> bool:
+    now = time.time()
+    ts = [t for t in _ip_fails.get(ip, ()) if now - t < IP_WINDOW]
+    if ts:
+        _ip_fails[ip] = ts
+    return len(ts) >= IP_FAILS
+
+
+def lock(user: str, pw: str, ip: str) -> str:
+    """-> "ok" | "bad" | "locked" (per-username lockout or per-IP budget)"""
+    if ip_locked(ip):
+        return "locked"
+    lo = _lockout.setdefault(user, {"fails": 0, "until": 0.0})
+    if time.time() < lo["until"]:
+        return "locked"
+    if users.verify(user, pw):
+        lo.update(fails=0, until=0.0)
+        return "ok"
+    lo["fails"] += 1
+    if lo["fails"] >= LOCK_FAILS:
+        lo.update(fails=0, until=time.time() + LOCK_SECONDS)
+    time.sleep(0.5)  # slow down brute force (sync handler -> threadpool)
+    ip_fail(ip)
+    return "bad"

tests/conftest.py

@@ -53,6 +53,10 @@ def server():
 
     data = Path(tempfile.mkdtemp(prefix="aniweb-test-"))
     os.environ["ANIWEB_DATA"] = str(data)
+    # the per-IP login budget must never trip during the suite (all tests
+    # share 127.0.0.1); per-username lockout stays at production defaults
+    os.environ.setdefault("ANIWEB_IP_FAILS", "1000")
+    os.environ.setdefault("ANIWEB_IP_WINDOW", "300")
     sys.path.insert(0, str(ROOT))
     from server import db, users  # noqa: E402
     db.init()  # schema before users.add / seeding

tests/test_auth.py

@@ -0,0 +1,52 @@
+"""test_auth.py — pure unit tests for server/auth.py (no server, no HTTP).
+
+The per-IP budget and the XFF trust rule live here; the per-username
+lockout's HTTP behaviour stays covered in test_api.py.
+"""
+import time
+from types import SimpleNamespace
+
+
+def _req(headers=None, host="203.0.113.7"):
+    return SimpleNamespace(headers=dict(headers or {}),
+                           client=SimpleNamespace(host=host))
+
+
+def test_ip_budget_counts_prunes_isolates(monkeypatch):
+    from server import auth
+    monkeypatch.setattr(auth, "IP_FAILS", 3)
+    monkeypatch.setattr(auth, "IP_WINDOW", 60.0)
+    auth._ip_fails.clear()
+    assert not auth.ip_locked("203.0.113.1")
+    for _ in range(3):
+        auth.ip_fail("203.0.113.1")
+    assert auth.ip_locked("203.0.113.1")
+    assert not auth.ip_locked("203.0.113.2")  # per-IP isolation
+    auth._ip_fails["203.0.113.1"] = [time.time() - 10_000]  # stale -> pruned
+    assert not auth.ip_locked("203.0.113.1")
+
+
+def test_client_ip_xff_trust_rule(monkeypatch):
+    from server import auth
+    r = _req({"x-forwarded-for": "9.9.9.9"})
+    monkeypatch.delenv("ANIWEB_TRUST_XFF", raising=False)
+    assert auth.client_ip(r) == "203.0.113.7"  # header NOT trusted by default
+    monkeypatch.setenv("ANIWEB_TRUST_XFF", "1")
+    assert auth.client_ip(r) == "9.9.9.9"       # honored only when explicitly on
+    assert auth.client_ip(_req()) == "203.0.113.7"  # no header -> socket peer
+
+
+def test_lock_locked_user_never_reaches_verify(monkeypatch):
+    from server import auth
+    monkeypatch.setattr(auth, "IP_FAILS", 1000)
+    auth._lockout["ghost"] = {"fails": 0, "until": time.time() + 900}
+    assert auth.lock("ghost", "x", "203.0.113.3") == "locked"
+
+
+def test_lock_wrong_password_is_bad_and_records_ip_failure(monkeypatch):
+    from server import auth
+    monkeypatch.setattr(auth, "IP_FAILS", 1000)  # keep the IP gate out of the way
+    auth._ip_fails.clear()
+    auth._lockout.clear()  # module state is shared across tests
+    assert auth.lock("ghost", "wrong123456", "203.0.113.4") == "bad"  # ghost: no user
+    assert len(auth._ip_fails.get("203.0.113.4", [])) == 1