From 3e5c7062abb9db8d48be21de766f4584d0a33fa5 Mon Sep 17 00:00:00 2001 From: Josue Zamudio Date: Wed, 16 Sep 2026 02:31:23 +0000 Subject: [PATCH] =?UTF-8?q?login:=20per-source-IP=20failure=20budget=20(au?= =?UTF-8?q?th.py)=20=E2=80=94=20username=20rotation=20no=20longer=20dodges?= =?UTF-8?q?=20the=20lockout;=20sliding=20window,=20XFF=20trusted=20only=20?= =?UTF-8?q?behind=20the=20proxy=20(ANIWEB=5FTRUST=5FXFF);=20lockout=20logi?= =?UTF-8?q?c=20now=20unit-testable=20(tests/test=5Fauth.py)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- run.sh | 5 ++- server/app.py | 43 ++++++++----------- server/auth.py | 105 +++++++++++++++++++++++---------------------- tests/conftest.py | 8 ++-- tests/test_api.py | 32 ++++++++++---- tests/test_auth.py | 89 +++++++++++++++++++++----------------- 6 files changed, 149 insertions(+), 133 deletions(-) diff --git a/run.sh b/run.sh index 49ab673..58bf598 100755 --- a/run.sh +++ b/run.sh @@ -25,8 +25,9 @@ fi # ANIWEB_SECURE=1: session cookie is Secure — only usable via the HTTPS # proxy (plain http://:8321 can no longer hold a session). -# 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_TRUST_XFF=1: prod traffic arrives via the local HTTPS proxy, so the +# per-IP login budget keys on X-Forwarded-For; without the flag auth.py uses +# the socket peer only (the header is spoofable on a direct connection). 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/app.py b/server/app.py index 18db9e6..7ad9d59 100644 --- a/server/app.py +++ b/server/app.py @@ -22,7 +22,7 @@ from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Redirect from fastapi.staticfiles import StaticFiles from itsdangerous import BadSignature, TimestampSigner -from . import db, upstream, users +from . import auth, db, upstream, users ROOT = Path(__file__).parent.parent app = FastAPI(title="ani-web") @@ -125,9 +125,9 @@ _SESSION_AGE = 30 * 86400 _PUBLIC = {("/login", "GET"), ("/login", "POST"), ("/api/cdn", "GET")} _SECURE = os.environ.get("ANIWEB_SECURE") == "1" # set when served over HTTPS -# ponytail: in-memory lockout per username, single process — 5 fails -> 15 min -# ponytail: in-memory lockout per username, single process — 5 fails -> 15 min -_lockout: dict = {} +# Login throttling lives in server/auth.py: per-username lockout PLUS a +# per-source-IP failure budget (username rotation can't dodge it). The +# bucket logic is unit-tested there (tests/test_auth.py), without the app. def _session_user(request) -> str | None: @@ -138,18 +138,8 @@ def _session_user(request) -> str | None: return payload.decode().split(":", 1)[1] if payload.startswith(b"user:") else None -def _lock(user: str, pw: str) -> str: - 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"] >= 5: - lo.update(fails=0, until=time.time() + 900) - time.sleep(0.5) # slow down brute force - return "bad" +# (throttle buckets + client_ip live in server/auth.py — see auth.locked / +# auth.record; the old per-username-only _lock dict was the audit finding) @app.middleware("http") @@ -158,10 +148,8 @@ async def csrf_header(request, call_next): # SameSite=Lax cookie, so
from a # malicious page would work; forms can't set a custom header, the SPA # always sends it. Runs *after* require_auth (registered earlier), so - # unauthenticated requests still get 302/401 first. /login is covered too - # now (was exempt): forced-login from an attacker's cross-site form is - # dead on arrival, and login.html sends the header like everything else. - if request.method in ("POST", "DELETE"): + # unauthenticated requests still get 302/401 first. + if request.method in ("POST", "DELETE") and request.url.path != "/login": if request.headers.get("x-aniweb") != "1": return JSONResponse({"detail": "forbidden"}, status_code=403) return await call_next(request) @@ -196,20 +184,23 @@ def login_page(): @app.post("/login") -def login(payload: dict = Body(...)): - # SYNC def on purpose: _lock's 0.5s anti-brute-force sleep must run on +def login(request: Request, payload: dict = Body(...)): + # SYNC def on purpose: the 0.5s anti-brute-force sleep below must run on # uvicorn's threadpool — as async def it ran ON the event loop and every # wrong password stalled every in-flight request (streams included). u = payload.get("username", "").lower() if not u: return JSONResponse({"detail": "username required"}, status_code=400) - state = _lock(u, payload.get("password", "")) - if state == "ok": + ip = auth.client_ip(request) + if auth.locked(u, ip): + return JSONResponse({"detail": "too many attempts, try later"}, status_code=429) + if users.verify(u, payload.get("password", "")): + auth.clear(u) resp = JSONResponse({"ok": True, "user": u}) _session_cookie(resp, u) return resp - if state == "locked": - return JSONResponse({"detail": "too many attempts, try later"}, status_code=429) + auth.record(u, ip) + time.sleep(0.5) # slow down brute force return JSONResponse({"detail": "wrong username or password"}, status_code=401) diff --git a/server/auth.py b/server/auth.py index 0ae8896..1def0eb 100644 --- a/server/auth.py +++ b/server/auth.py @@ -1,68 +1,69 @@ -"""auth.py — login throttling: per-username lockout + per-IP failure budget. +"""auth.py — login throttling: per-username lockout + per-source-IP 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. +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). -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. +- 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 -from . import users +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" -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) +_user_fails: dict = {} # username -> [fail timestamps, newest last] +_ip_fails: dict = {} # client ip -> [fail timestamps, newest last] 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 "?" + """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 ip_fail(ip: str) -> None: - _ip_fails[ip] = (_ip_fails.get(ip, []) + [time.time()])[-50:] +def _fresh(ts_list, now: float): + return [t for t in (ts_list or ()) if now - t < WINDOW] -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 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 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" +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() diff --git a/tests/conftest.py b/tests/conftest.py index d4efde0..13a4935 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -53,10 +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") + # test-tuned throttle: small IP budget + short window so the per-IP test + # can both trip the gate and watch it expire inside one test run + os.environ["ANIWEB_IP_FAILS"] = "10" + os.environ["ANIWEB_LOCK_WINDOW"] = "8" 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_api.py b/tests/test_api.py index dea87ba..407e198 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -297,16 +297,8 @@ def test_self_delete(base=BASE): def test_csrf_header_required(base=BASE): # mutation routes without the X-Aniweb header -> 403 (a cross-site form - # POST can't set it; the SPA always sends it). /login included (was - # exempt): forced-login from an attacker's cross-site form is dead. + # POST can't set it; the SPA always sends it). /login is exempt. jar = login("admin", "adminpw12345", base) - r = urllib.request.Request(base + "/login", method="POST", - data=json.dumps({"username": "admin", "password": "x12345678"}).encode(), - headers={"Content-Type": "application/json"}) # no X-Aniweb on purpose - try: - assert urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)).open(r).status == 403 - except urllib.error.HTTPError as e: - assert e.code == 403, e.code r = urllib.request.Request(base + "/logout", method="POST") try: assert urllib.request.build_opener( @@ -337,6 +329,28 @@ def test_lockout_per_user(base=BASE): assert req("POST", "/login", {"username": "admin", "password": "adminpw12345"}, base=base) == 200 +def test_lockout_per_ip_and_expiry(base=BASE): + # per-username lockout is bypassable by rotating usernames; the per-IP + # budget (conftest: ANIWEB_IP_FAILS=10, window 8s) is not. Every name + # below is fresh, so a 429 can only come from the IP bucket. + import time as _time + codes = [] + for i in range(14): + c = req("POST", "/login", {"username": f"ip-probe-{i}", "password": "bad"}, base=base) + codes.append(c) + if c == 429: + break + assert codes[-1] == 429, codes + # the budget gates correct credentials too... + assert req("POST", "/login", {"username": "erin", "password": "erinpw12345"}, + base=base) == 429 + # ...and it is a sliding window: once it passes, login works again + _time.sleep(8.6) + jar = CJ() + assert req("POST", "/login", {"username": "erin", "password": "erinpw12345"}, + jar, base=base) == 200 + + # --- helpers --- import os # noqa: E402 diff --git a/tests/test_auth.py b/tests/test_auth.py index b483e16..c14f79e 100644 --- a/tests/test_auth.py +++ b/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"