diff --git a/run.sh b/run.sh index d157f98..49ab673 100755 --- a/run.sh +++ b/run.sh @@ -25,6 +25,8 @@ fi # ANIWEB_SECURE=1: session cookie is Secure — only usable via the HTTPS # proxy (plain http://: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" diff --git a/server/auth.py b/server/auth.py new file mode 100644 index 0000000..0ae8896 --- /dev/null +++ b/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" diff --git a/tests/conftest.py b/tests/conftest.py index 1ad3cb2..d4efde0 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..b483e16 --- /dev/null +++ b/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