rate-limit: extract generic helper, reuse from login

Pulled the per-IP sliding-window check out of auth_routes so the
upcoming /profile/password endpoint can share it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
mwiegand 2026-05-11 21:45:51 +02:00
parent a5982941df
commit 26a6a9d7b0
No known key found for this signature in database
3 changed files with 62 additions and 11 deletions

View file

@ -1,16 +1,15 @@
import time
from flask import Blueprint, Response, redirect, render_template, request from flask import Blueprint, Response, redirect, render_template, request
from sqlalchemy import select from sqlalchemy import select
from l4d2web.auth import hash_password, is_safe_next, login_user, logout_user, verify_password from l4d2web.auth import hash_password, is_safe_next, login_user, logout_user, verify_password
from l4d2web.db import session_scope from l4d2web.db import session_scope
from l4d2web.models import User from l4d2web.models import User
from l4d2web.services.rate_limit import check_rate_limit
bp = Blueprint("auth", __name__) bp = Blueprint("auth", __name__)
_TIMING_DUMMY_DIGEST = hash_password("__timing_dummy__") _TIMING_DUMMY_DIGEST = hash_password("__timing_dummy__")
LOGIN_RATE_LIMIT_WINDOW_SECONDS = 60 LOGIN_RATE_LIMIT_WINDOW_SECONDS = 60.0
LOGIN_RATE_LIMIT_MAX_ATTEMPTS = 20 LOGIN_RATE_LIMIT_MAX_ATTEMPTS = 20
LOGIN_ATTEMPTS_BY_IP: dict[str, list[float]] = {} LOGIN_ATTEMPTS_BY_IP: dict[str, list[float]] = {}
@ -20,14 +19,12 @@ def reset_login_rate_limits() -> None:
def is_login_rate_limited(remote_addr: str) -> bool: def is_login_rate_limited(remote_addr: str) -> bool:
now = time.time() return check_rate_limit(
attempts = LOGIN_ATTEMPTS_BY_IP.setdefault(remote_addr, []) LOGIN_ATTEMPTS_BY_IP,
cutoff = now - LOGIN_RATE_LIMIT_WINDOW_SECONDS remote_addr,
attempts[:] = [ts for ts in attempts if ts >= cutoff] window=LOGIN_RATE_LIMIT_WINDOW_SECONDS,
if len(attempts) >= LOGIN_RATE_LIMIT_MAX_ATTEMPTS: max_attempts=LOGIN_RATE_LIMIT_MAX_ATTEMPTS,
return True )
attempts.append(now)
return False
@bp.get("/login") @bp.get("/login")

View file

@ -0,0 +1,23 @@
import time
def check_rate_limit(
bucket: dict[str, list[float]],
remote_addr: str,
*,
window: float,
max_attempts: int,
) -> bool:
"""Return True if this request should be rate-limited (rejected).
The bucket is owned by the caller so each endpoint can keep an
independent count and tests can reset state cleanly.
"""
now = time.time()
attempts = bucket.setdefault(remote_addr, [])
cutoff = now - window
attempts[:] = [ts for ts in attempts if ts >= cutoff]
if len(attempts) >= max_attempts:
return True
attempts.append(now)
return False

View file

@ -0,0 +1,31 @@
import time
from l4d2web.services.rate_limit import check_rate_limit
def test_under_threshold_allows():
bucket: dict[str, list[float]] = {}
for _ in range(3):
assert check_rate_limit(bucket, "1.2.3.4", window=60.0, max_attempts=5) is False
def test_at_threshold_blocks():
bucket: dict[str, list[float]] = {}
for _ in range(5):
assert check_rate_limit(bucket, "1.2.3.4", window=60.0, max_attempts=5) is False
assert check_rate_limit(bucket, "1.2.3.4", window=60.0, max_attempts=5) is True
def test_other_ips_independent():
bucket: dict[str, list[float]] = {}
for _ in range(5):
check_rate_limit(bucket, "1.2.3.4", window=60.0, max_attempts=5)
assert check_rate_limit(bucket, "5.6.7.8", window=60.0, max_attempts=5) is False
def test_old_attempts_expire():
bucket: dict[str, list[float]] = {}
for _ in range(5):
check_rate_limit(bucket, "1.2.3.4", window=0.05, max_attempts=5)
time.sleep(0.1)
assert check_rate_limit(bucket, "1.2.3.4", window=0.05, max_attempts=5) is False