feat(auth): per-IP login brute-force throttle

Add an in-memory (single-worker) per-IP throttle that complements the per-user
account lockout: it stops password-spraying across many accounts from one IP
and prevents a single attacker draining the shared login rate-limit bucket.

- new app/services/auth_throttle_service.py — lock-guarded TTL map with
  is_blocked / register_failure / reset (+ opportunistic memory sweep)
- config: AUTH_IP_MAX_ATTEMPTS (10) / AUTH_IP_WINDOW_MINUTES (15) /
  AUTH_IP_BLOCK_MINUTES (15)
- wire into /login (block up front -> 429 + Retry-After BEFORE the password
  check; register on bad user/password; reset on correct password),
  /login-links/redeem, and the 2FA /verify endpoint
- per-user lockout and the flask-limiter 5/min decorator kept (defense in depth)
- tests: test_login_bruteforce.py (6)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -577,6 +577,15 @@ awaiting a stable release:
 
 ### Security
 
+- **Login brute-force is now throttled per client IP.** On top of the existing
+  per-user account lockout, ServerKit now blocks a *client IP* after repeated
+  failed logins (default 10 failures / 15 min → 15-minute block, returned as
+  `429` with `Retry-After`), checked before the password is verified. This stops
+  password-spraying across many usernames from one source and closes a gap where
+  one attacker could drain the shared login rate-limit for everyone. It also
+  guards the one-time login-link redeem and 2FA-code verification endpoints.
+  Tunable via `AUTH_IP_MAX_ATTEMPTS` / `AUTH_IP_WINDOW_MINUTES` /
+  `AUTH_IP_BLOCK_MINUTES`.
 - **Client IP is no longer spoofable behind the proxy.** Rate-limit buckets,
   login lockout and audit-log source IPs previously trusted the *leftmost*
   `X-Forwarded-For` token — a value the client fully controls, so an attacker

backend/app/api/auth.py

@@ -16,6 +16,8 @@ from app.middleware.rbac import admin_required, get_current_user as get_request_
 from app.services.settings_service import SettingsService
 from app.services.audit_service import AuditService
 from app.services import login_link_service
+from app.services import auth_throttle_service
+from app.utils.client_ip import get_client_ip
 
 logger = logging.getLogger(__name__)
 
@@ -220,6 +222,15 @@ def login():
     if not sso_service.is_password_login_allowed():
         return jsonify({'error': 'Password login is disabled. Please use SSO.'}), 403
 
+    # Per-IP brute-force throttle — checked up front, before any password work,
+    # so spraying wrong passwords across many usernames from one IP is blocked.
+    client_ip = get_client_ip()
+    blocked, retry_after = auth_throttle_service.is_blocked(client_ip)
+    if blocked:
+        return jsonify({
+            'error': 'Too many failed login attempts. Try again later.'
+        }), 429, {'Retry-After': str(retry_after)}
+
     data = request.get_json()
 
     if not data:
@@ -244,13 +255,19 @@ def login():
         }), 429
 
     if not user or not user.check_password(password):
-        # Record failed login attempt
+        # Record the failure against BOTH throttles: per-user (targeted guessing)
+        # and per-IP (spraying across accounts). Per-IP counts even bad usernames.
+        auth_throttle_service.register_failure(client_ip)
         if user:
             user.record_failed_login()
             AuditService.log_login(user.id, success=False, details={'reason': 'invalid_password'})
             db.session.commit()
         return jsonify({'error': 'Invalid username/email or password'}), 401
 
+    # Correct password — clear this IP's failure count (covers the 2FA-pending
+    # and no-2FA paths below).
+    auth_throttle_service.reset(client_ip)
+
     if not user.is_active:
         AuditService.log_login(user.id, success=False, details={'reason': 'account_deactivated'})
         db.session.commit()
@@ -368,14 +385,23 @@ def revoke_login_link(link_id):
 @limiter.limit("30 per minute")
 def redeem_login_link():
     """Redeem a one-time login link and issue normal JWT tokens."""
+    client_ip = get_client_ip()
+    blocked, retry_after = auth_throttle_service.is_blocked(client_ip)
+    if blocked:
+        return jsonify({
+            'error': 'Too many failed attempts. Try again later.'
+        }), 429, {'Retry-After': str(retry_after)}
+
     data = request.get_json() or {}
     token = data.get('token')
 
-    user, reason = login_link_service.redeem(token, request.remote_addr)
+    user, reason = login_link_service.redeem(token, client_ip)
     if not user:
-        logger.info(f"Login link redeem failed ({reason}) from {request.remote_addr}")
+        auth_throttle_service.register_failure(client_ip)
+        logger.info(f"Login link redeem failed ({reason}) from {client_ip}")
         return jsonify({'error': 'Invalid or expired link'}), 401
 
+    auth_throttle_service.reset(client_ip)
     user.reset_failed_login()
     user.last_login_at = datetime.utcnow()
     db.session.commit()

backend/app/api/two_factor.py

@@ -10,6 +10,8 @@ from flask_jwt_extended import jwt_required, get_jwt_identity
 from app import limiter
 from app.models import User
 from app.services.totp_service import TOTPService, TwoFactorSetup
+from app.services import auth_throttle_service
+from app.utils.client_ip import get_client_ip
 
 logger = logging.getLogger(__name__)
 
@@ -172,6 +174,15 @@ def verify_2fa_code():
     from flask_jwt_extended import create_access_token, create_refresh_token, decode_token
     from app import db
 
+    # Per-IP brute-force throttle — 2FA codes are a small keyspace, so guessing
+    # from one IP is throttled up front, before any code check.
+    client_ip = get_client_ip()
+    blocked, retry_after = auth_throttle_service.is_blocked(client_ip)
+    if blocked:
+        return jsonify({
+            'error': 'Too many failed attempts. Try again later.'
+        }), 429, {'Retry-After': str(retry_after)}
+
     data = request.get_json()
     if not data:
         return jsonify({'error': 'No data provided'}), 400
@@ -204,6 +215,7 @@ def verify_2fa_code():
     # Try TOTP code first
     if TOTPService.verify_code(user.totp_secret, code):
         # Success - issue real tokens
+        auth_throttle_service.reset(client_ip)
         user.reset_failed_login()
         db.session.commit()
 
@@ -222,6 +234,7 @@ def verify_2fa_code():
 
     if matched_hash:
         # Valid backup code - remove it (one-time use)
+        auth_throttle_service.reset(client_ip)
         user.use_backup_code(matched_hash)
         user.reset_failed_login()
         db.session.commit()
@@ -242,6 +255,7 @@ def verify_2fa_code():
         }), 200
 
     # Invalid code
+    auth_throttle_service.register_failure(client_ip)
     user.record_failed_login()
     db.session.commit()
 

backend/app/services/auth_throttle_service.py

@@ -0,0 +1,131 @@
+"""Per-IP login brute-force throttle (in-memory, single-worker safe).
+
+Complements — does not replace — the per-*user* account lockout in
+``User.record_failed_login``. Per-user lockout stops targeted guessing at one
+account; this per-IP throttle stops password *spraying* across many accounts
+from one source, and closes the shared-bucket gap that let a single attacker
+drain everyone's login budget.
+
+State is process-local: after Phase 1 the app trusts the real client IP, and the
+panel runs a **single** gunicorn worker (mandated by the in-memory agent
+registry — see root ``CLAUDE.md``), so a plain in-process TTL map needs no shared
+store. It is guarded by a lock because the shipped worker is threaded.
+
+Thresholds are read from Flask config at call time so they can be tuned via env
+(``AUTH_IP_MAX_ATTEMPTS`` / ``AUTH_IP_WINDOW_MINUTES`` / ``AUTH_IP_BLOCK_MINUTES``)
+and overridden in tests.
+"""
+import math
+import threading
+import time
+
+from flask import current_app
+
+# ip -> list[failure_epoch_seconds] (only failures inside the window are kept)
+_attempts = {}
+# ip -> block_until_epoch_seconds
+_blocks = {}
+_lock = threading.Lock()
+_last_sweep = 0.0
+
+# Fallbacks used only outside an app context (config is the real source).
+_DEFAULT_MAX_ATTEMPTS = 10
+_DEFAULT_WINDOW_MINUTES = 15
+_DEFAULT_BLOCK_MINUTES = 15
+
+
+def _cfg(key, default):
+    try:
+        return current_app.config.get(key, default)
+    except RuntimeError:  # no application context
+        return default
+
+
+def _max_attempts():
+    return int(_cfg('AUTH_IP_MAX_ATTEMPTS', _DEFAULT_MAX_ATTEMPTS))
+
+
+def _window_seconds():
+    return int(_cfg('AUTH_IP_WINDOW_MINUTES', _DEFAULT_WINDOW_MINUTES)) * 60
+
+
+def _block_seconds():
+    return int(_cfg('AUTH_IP_BLOCK_MINUTES', _DEFAULT_BLOCK_MINUTES)) * 60
+
+
+def _sweep(now, window):
+    """Drop aged-out attempt lists and expired blocks so the maps can't grow
+    without bound from one-off IPs. Caller must hold the lock."""
+    global _last_sweep
+    if now - _last_sweep <= window:
+        return
+    for ip in list(_attempts.keys()):
+        fresh = [t for t in _attempts[ip] if now - t < window]
+        if fresh:
+            _attempts[ip] = fresh
+        else:
+            del _attempts[ip]
+    for ip in list(_blocks.keys()):
+        if _blocks[ip] <= now:
+            del _blocks[ip]
+    _last_sweep = now
+
+
+def is_blocked(ip):
+    """Return ``(blocked, retry_after_seconds)`` for ``ip``.
+
+    Does NOT record an attempt — call this up front, before checking the
+    password, and reject with 429 + ``Retry-After`` when blocked.
+    """
+    if not ip:
+        return False, 0
+    now = time.time()
+    with _lock:
+        until = _blocks.get(ip)
+        if until is None:
+            return False, 0
+        if until <= now:
+            _blocks.pop(ip, None)
+            return False, 0
+        return True, int(math.ceil(until - now))
+
+
+def register_failure(ip):
+    """Record one failed auth from ``ip``. Once failures within the rolling
+    window reach ``AUTH_IP_MAX_ATTEMPTS`` the IP is blocked for
+    ``AUTH_IP_BLOCK_MINUTES``. Returns ``(blocked_now, retry_after_seconds)``."""
+    if not ip:
+        return False, 0
+    now = time.time()
+    window = _window_seconds()
+    with _lock:
+        _sweep(now, window)
+        recent = [t for t in _attempts.get(ip, []) if now - t < window]
+        recent.append(now)
+        _attempts[ip] = recent
+        if len(recent) >= _max_attempts():
+            until = now + _block_seconds()
+            _blocks[ip] = until
+            # Window's failures have done their job; reset so the count starts
+            # fresh after the block lifts instead of instantly re-blocking.
+            _attempts.pop(ip, None)
+            return True, int(math.ceil(until - now))
+        return False, 0
+
+
+def reset(ip):
+    """Clear an IP's failure count and any block — call on a successful auth."""
+    if not ip:
+        return
+    with _lock:
+        _attempts.pop(ip, None)
+        _blocks.pop(ip, None)
+
+
+def reset_all():
+    """Wipe all throttle state. For tests and clean shutdowns."""
+    global _last_sweep
+    with _lock:
+        _attempts.clear()
+        _blocks.clear()
+        _last_sweep = 0.0

backend/config.py

@@ -141,6 +141,17 @@ class Config:
     # Raise it only if you add another proxy (e.g. Cloudflare in front of nginx).
     TRUSTED_PROXY_HOPS = _env_int('TRUSTED_PROXY_HOPS', 1)
 
+    # ── Per-IP login brute-force throttle ───────────────────────────────
+    # In-memory (single-worker), complements the per-user account lockout: it
+    # stops password-spraying across many accounts from one IP and prevents a
+    # single attacker draining the shared login rate-limit bucket. After
+    # AUTH_IP_MAX_ATTEMPTS failed auths from one client IP within
+    # AUTH_IP_WINDOW_MINUTES, that IP is blocked (429 + Retry-After) for
+    # AUTH_IP_BLOCK_MINUTES. See app/services/auth_throttle_service.py.
+    AUTH_IP_MAX_ATTEMPTS = _env_int('AUTH_IP_MAX_ATTEMPTS', 10)
+    AUTH_IP_WINDOW_MINUTES = _env_int('AUTH_IP_WINDOW_MINUTES', 15)
+    AUTH_IP_BLOCK_MINUTES = _env_int('AUTH_IP_BLOCK_MINUTES', 15)
+
     # ── Build packs ─────────────────────────────────────────────────────
     # Path to the optional nixpacks binary (used by build_service's opaque
     # nixpacks path). The transparent build-pack layer (buildpack_service)

backend/tests/test_login_bruteforce.py

@@ -0,0 +1,175 @@
+"""Proving tests for the per-IP login brute-force throttle (plan 48, Phase 2).
+
+The per-IP throttle (app/services/auth_throttle_service.py) complements the
+existing per-user account lockout: it blocks an IP after AUTH_IP_MAX_ATTEMPTS
+failed auths, *before* the password is checked, so spraying wrong passwords
+across many usernames from one source is stopped and a single attacker can't
+silently drain the shared login rate-limit bucket.
+
+Notes:
+- These build a trust-ON app so the real client IP is driven via the rightmost
+  X-Forwarded-For hop (Phase 1), letting one test exercise several distinct
+  clients and prove a forged leftmost value can't dodge the block.
+- Thresholds are lowered via config so the throttle trips well within
+  flask-limiter's own 5/min on /login. Every test keeps each IP at <=5 requests
+  for the same reason, and uses TEST-NET-2 (198.51.100.x) addresses distinct
+  from other suites to avoid cross-test limiter contamination.
+"""
+import pytest
+from datetime import datetime, timedelta
+
+from app import create_app, db
+from app.services import auth_throttle_service
+
+
+@pytest.fixture(autouse=True)
+def _reset_throttle():
+    auth_throttle_service.reset_all()
+    yield
+    auth_throttle_service.reset_all()
+
+
+@pytest.fixture
+def make_app(monkeypatch):
+    """Factory: build a trust-ON testing app with a tunable per-IP threshold.
+
+    Monkeypatching the class attribute before create_app drives the real config
+    (from_object reads it at create time). Cleans up DB + app context after.
+    """
+    created = []
+
+    def _make(max_attempts=3, hops=1):
+        from config import TestingConfig
+        monkeypatch.setattr(TestingConfig, 'TRUST_PROXY_HEADERS', True)
+        monkeypatch.setattr(TestingConfig, 'TRUSTED_PROXY_HOPS', hops)
+        monkeypatch.setattr(TestingConfig, 'AUTH_IP_MAX_ATTEMPTS', max_attempts)
+        app = create_app('testing')
+        ctx = app.app_context()
+        ctx.push()
+        db.create_all()
+        created.append(ctx)
+        return app
+
+    yield _make
+
+    for ctx in created:
+        db.session.remove()
+        db.drop_all()
+        ctx.pop()
+
+
+def _mk_user(username='victim', password='correct-horse-battery'):
+    from app.models import User
+    u = User(email=f'{username}@t.local', username=username,
+             role=User.ROLE_ADMIN, is_active=True)
+    u.set_password(password)
+    db.session.add(u)
+    db.session.commit()
+    return u
+
+
+def _login(client, password, peer, leftmost='9.9.9.9'):
+    return client.post(
+        '/api/v1/auth/login',
+        json={'email': 'victim@t.local', 'password': password},
+        headers={'X-Forwarded-For': f'{leftmost}, {peer}'},
+    )
+
+
+THROTTLE_MSG = 'Too many failed login attempts. Try again later.'
+
+
+def test_ip_blocked_after_max_failures_before_password_check(make_app):
+    app = make_app(max_attempts=3)
+    _mk_user(password='rightpass')
+    client = app.test_client()
+    peer = '198.51.100.10'
+
+    # Three wrong-password failures from one IP arm the block.
+    for _ in range(3):
+        assert _login(client, 'wrong', peer).status_code == 401
+
+    # The next attempt is rejected with 429 + Retry-After even with the CORRECT
+    # password — proving the IP block precedes the password check.
+    r = _login(client, 'rightpass', peer)
+    assert r.status_code == 429
+    assert r.get_json()['error'] == THROTTLE_MSG
+    assert int(r.headers.get('Retry-After', '0')) > 0
+
+    # A different real client IP is unaffected (still reaches the password check).
+    r2 = _login(client, 'wrong', '198.51.100.99')
+    assert r2.status_code == 401
+
+
+def test_forged_leftmost_xff_cannot_dodge_the_block(make_app):
+    """Regression tie-in to Phase 1: rotating the client-controlled leftmost
+    X-Forwarded-For value can neither spread the failures across buckets nor
+    escape the block, because the trusted IP is the rightmost proxy hop."""
+    app = make_app(max_attempts=3)
+    _mk_user(password='rightpass')
+    client = app.test_client()
+    peer = '198.51.100.20'
+
+    # Each failure forges a DIFFERENT leftmost value, but the real peer is one.
+    for i in range(3):
+        assert _login(client, 'wrong', peer, leftmost=f'{i}.{i}.{i}.{i}').status_code == 401
+
+    # A brand-new forged prefix + the correct password is still blocked.
+    r = _login(client, 'rightpass', peer, leftmost='250.250.250.250')
+    assert r.status_code == 429
+    assert r.get_json()['error'] == THROTTLE_MSG
+
+
+def test_per_user_lockout_still_enforced_at_endpoint(make_app):
+    """The pre-existing per-user lockout must still fire (defense in depth).
+    Per-IP is set high here so it can't mask the per-user path."""
+    app = make_app(max_attempts=100)
+    u = _mk_user(password='rightpass')
+    u.locked_until = datetime.utcnow() + timedelta(minutes=5)
+    db.session.commit()
+
+    client = app.test_client()
+    r = _login(client, 'rightpass', '198.51.100.30')
+    assert r.status_code == 429
+    assert 'locked' in r.get_json()['error'].lower()
+
+
+def test_record_failed_login_locks_user_after_max(make_app):
+    """Model-level: the per-user lockout mechanism itself is untouched."""
+    make_app(max_attempts=100)
+    from app.models import User
+    u = _mk_user()
+    for _ in range(User.MAX_FAILED_ATTEMPTS):
+        u.record_failed_login()
+    assert u.is_locked is True
+
+
+def test_successful_login_clears_ip_failure_count(make_app):
+    app = make_app(max_attempts=2)
+    _mk_user(password='rightpass')
+    client = app.test_client()
+    peer = '198.51.100.40'
+
+    assert _login(client, 'wrong', peer).status_code == 401       # count = 1
+    assert _login(client, 'rightpass', peer).status_code == 200   # success -> reset
+    # Fresh window after the reset: it takes a NEW max failures to re-arm.
+    assert _login(client, 'wrong', peer).status_code == 401       # count = 1
+    r4 = _login(client, 'wrong', peer)                            # count = 2 -> arms
+    assert r4.status_code == 401                                   # not blocked yet
+    r5 = _login(client, 'wrong', peer)                            # now blocked
+    assert r5.status_code == 429
+    assert r5.get_json()['error'] == THROTTLE_MSG
+
+
+def test_different_ip_not_blocked_by_another_ips_failures(make_app):
+    app = make_app(max_attempts=2)
+    _mk_user(password='rightpass')
+    client = app.test_client()
+
+    # Arm the block for peer A.
+    assert _login(client, 'wrong', '198.51.100.50').status_code == 401
+    assert _login(client, 'wrong', '198.51.100.50').status_code == 401
+    assert _login(client, 'wrong', '198.51.100.50').status_code == 429
+
+    # Peer B, sharing the same forged leftmost, is a clean bucket.
+    assert _login(client, 'wrong', '198.51.100.51').status_code == 401