login: per-source-IP failure budget (auth.py) — username rotation no longer dodges the lockout; sliding window, XFF trusted only behind the proxy (ANIWEB_TRUST_XFF); lockout logic now unit-testable (tests/test_auth.py)

tests/test_auth.py

@@ -1,52 +1,61 @@
-"""test_auth.py — pure unit tests for server/auth.py (no server, no HTTP).
+"""test_auth.py — server/auth.py unit tests (no server, no network).
 
-The per-IP budget and the XFF trust rule live here; the per-username
-lockout's HTTP behaviour stays covered in test_api.py.
+Bucket logic in isolation: per-username lockout, per-source-IP budget,
+sliding-window expiry, and the spoof-gating of X-Forwarded-For. The
+module reads its env knobs at import; inside the pytest process they stay
+at defaults (5 / 20 / 900s) — the tests read auth.FAILS_USER/IP_FAILS/
+WINDOW rather than hardcoding, so they hold either way.
 """
-import time
-from types import SimpleNamespace
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from server import auth  # noqa: E402
 
 
-def _req(headers=None, host="203.0.113.7"):
-    return SimpleNamespace(headers=dict(headers or {}),
-                           client=SimpleNamespace(host=host))
+class _Peer:
+    host = "203.0.113.9"
 
 
-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")
+class _Req:
+    def __init__(self, headers=None):
+        self.headers = headers or {}
+        self.client = _Peer()
 
 
-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_username_lockout_and_sliding_expiry():
+    auth.reset()
+    t = 1_000_000.0
+    for _ in range(auth.FAILS_USER):
+        auth.record("locked-guy", "198.51.100.7", now=t)
+    assert auth.locked("locked-guy", "198.51.100.7", now=t)
+    # window is sliding: once every fail ages out, the name is free again
+    assert not auth.locked("locked-guy", "198.51.100.7", now=t + auth.WINDOW + 1)
+    # a locked username does not gate other usernames from another ip
+    assert not auth.locked("fresh-guy", "198.51.100.8", now=t)
+    # a correct password clears only that username's bucket
+    auth.clear("locked-guy")
+    assert not auth.locked("locked-guy", "198.51.100.9", now=t)
+    auth.reset()
 
 
-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_ip_budget_survives_username_rotation():
+    auth.reset()
+    t = 2_000_000.0
+    for i in range(auth.IP_FAILS):        # one fresh username per fail...
+        auth.record(f"sweep-{i}", "198.51.100.9", now=t)
+    assert auth.locked("brand-new-user", "198.51.100.9", now=t)     # ...still blocked
+    assert not auth.locked("brand-new-user", "198.51.100.10", now=t)  # other ip fine
+    assert not auth.locked("brand-new-user", "198.51.100.9", now=t + auth.WINDOW + 1)
+    auth.reset()
 
 
-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
+def test_client_ip_trust_gating():
+    # XFF is only consulted when the trust flag is on (it is off here — the
+    # flag is for the prod proxy path, not the test process)
+    r = _Req({"x-forwarded-for": "5.6.7.8, 10.0.0.1"})
+    assert auth.TRUST_XFF is False
+    assert auth.client_ip(r) == "203.0.113.9"   # spoofed header ignored
+    assert auth.client_ip(_Req()) == "203.0.113.9"