Security: brute-force limiter on the launch-PIN unlock (Tier 2)

A publicly-exposed instance gated only by the launch PIN was brute-forceable. Added
a lenient in-memory failed-attempt limiter (core/security/rate_limit.py): 10 wrong
PINs from one IP within 5 min → 429 with Retry-After, failures age out on their own
(self-heal, no persistent lockout), and a CORRECT entry clears that IP instantly.

Wired into /api/profiles/verify-launch-pin. By design it can only ever trigger on a
flood of WRONG PINs — correct entry, a couple of typos, or a no-PIN install are
never affected, so normal use sees no change. Keyed per-IP so an attacker can't
lock out a legit user.

Tests: limiter is lenient under threshold, trips on a flood, success clears it,
failures self-heal, per-IP isolation; endpoint returns 429 after 10 wrong PINs with
Retry-After. 6 tests pass.

core/security/rate_limit.py

@@ -0,0 +1,54 @@
+"""Lenient in-memory failed-attempt limiter for the launch-PIN unlock.
+
+Brute-force protection for a publicly-exposed instance. Deliberately lenient: only
+a FLOOD of failures from one client trips it, and a single success clears that
+client immediately — so a legitimate user typing their PIN (even with a few typos)
+never hits it. Failures age out on their own, so a tripped client self-heals
+without any persistent lockout state.
+
+Keyed by client IP. In-memory is fine here: the launch lock is a coarse gate, not
+per-account auth, and a process restart simply forgets attempts (fail-open, which
+is correct for a self-hosted convenience lock).
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Dict, List, Tuple
+
+
+class AttemptLimiter:
+    def __init__(self, max_attempts: int = 10, window_seconds: int = 300):
+        """``max_attempts`` failures within ``window_seconds`` → locked until the
+        oldest failure in the window ages out."""
+        self.max_attempts = max_attempts
+        self.window = window_seconds
+        self._failures: Dict[str, List[float]] = defaultdict(list)
+
+    def _prune(self, key: str, now: float) -> List[float]:
+        recent = [t for t in self._failures.get(key, []) if now - t < self.window]
+        if recent:
+            self._failures[key] = recent
+        else:
+            self._failures.pop(key, None)
+        return recent
+
+    def is_locked(self, key: str, now: float) -> Tuple[bool, int]:
+        """(locked, retry_after_seconds). retry_after is when the oldest in-window
+        failure expires, so the client unlocks naturally."""
+        recent = self._prune(key, now)
+        if len(recent) >= self.max_attempts:
+            retry_after = int(self.window - (now - min(recent))) + 1
+            return True, max(retry_after, 1)
+        return False, 0
+
+    def record_failure(self, key: str, now: float) -> None:
+        self._prune(key, now)
+        self._failures[key].append(now)
+
+    def record_success(self, key: str) -> None:
+        """A correct entry clears that client's failure history immediately."""
+        self._failures.pop(key, None)
+
+
+__all__ = ["AttemptLimiter"]

tests/test_credentials_endpoints.py

@@ -355,3 +355,27 @@ def test_real_app_not_in_reverse_proxy_mode_by_default():
     assert not isinstance(web_server.app.wsgi_app, ProxyFix)
     assert web_server.app.config.get('SESSION_COOKIE_SECURE') in (None, False)
     assert web_server.app.config.get('SESSION_COOKIE_SAMESITE') is None
+
+
+def test_verify_launch_pin_rate_limited_after_flood(client):
+    # A flood of WRONG PINs from one IP gets 429; cleaned up so neither the lock
+    # nor the temp PIN leaks to other tests (the limiter is a process singleton).
+    from werkzeug.security import generate_password_hash
+    db = web_server.get_database()
+    with db._get_connection() as conn:   # admin needs a PIN so wrong ones actually fail
+        conn.execute("UPDATE profiles SET pin_hash = ? WHERE id = 1",
+                     (generate_password_hash('1234', method='pbkdf2:sha256'),))
+        conn.commit()
+    web_server._launch_pin_limiter.record_success('127.0.0.1')  # clean slate
+    try:
+        for _ in range(10):
+            assert client.post('/api/profiles/verify-launch-pin',
+                               json={'pin': 'definitely-wrong'}).status_code == 401
+        r = client.post('/api/profiles/verify-launch-pin', json={'pin': 'definitely-wrong'})
+        assert r.status_code == 429
+        assert 'Retry-After' in r.headers
+    finally:
+        web_server._launch_pin_limiter.record_success('127.0.0.1')
+        with db._get_connection() as conn:
+            conn.execute("UPDATE profiles SET pin_hash = NULL WHERE id = 1")
+            conn.commit()

tests/test_pin_rate_limit.py

@@ -0,0 +1,50 @@
+"""Launch-PIN brute-force limiter: only a flood of WRONG PINs trips it, a correct
+entry clears it instantly, and it self-heals as failures age out — so normal use
+is never affected."""
+
+from __future__ import annotations
+
+from core.security.rate_limit import AttemptLimiter
+
+
+def test_under_threshold_is_never_locked():
+    lim = AttemptLimiter(max_attempts=10, window_seconds=300)
+    for i in range(9):                       # 9 < 10 → never locked
+        lim.record_failure('1.2.3.4', now=1000 + i)
+    locked, _ = lim.is_locked('1.2.3.4', now=1010)
+    assert locked is False
+
+
+def test_flood_trips_the_lock_with_retry_after():
+    lim = AttemptLimiter(max_attempts=10, window_seconds=300)
+    for i in range(10):
+        lim.record_failure('1.2.3.4', now=1000 + i)
+    locked, retry_after = lim.is_locked('1.2.3.4', now=1010)
+    assert locked is True
+    assert retry_after > 0
+
+
+def test_success_clears_immediately():
+    lim = AttemptLimiter(max_attempts=10, window_seconds=300)
+    for i in range(10):
+        lim.record_failure('1.2.3.4', now=1000 + i)
+    assert lim.is_locked('1.2.3.4', now=1010)[0] is True
+    lim.record_success('1.2.3.4')            # correct PIN
+    assert lim.is_locked('1.2.3.4', now=1011)[0] is False
+
+
+def test_failures_age_out_self_heal():
+    lim = AttemptLimiter(max_attempts=10, window_seconds=300)
+    for i in range(10):
+        lim.record_failure('1.2.3.4', now=1000 + i)
+    assert lim.is_locked('1.2.3.4', now=1010)[0] is True
+    # well past the window → all failures expired → unlocked
+    assert lim.is_locked('1.2.3.4', now=2000)[0] is False
+
+
+def test_per_ip_isolation():
+    lim = AttemptLimiter(max_attempts=10, window_seconds=300)
+    for i in range(10):
+        lim.record_failure('attacker', now=1000 + i)
+    assert lim.is_locked('attacker', now=1010)[0] is True
+    assert lim.is_locked('legit-user', now=1010)[0] is False   # not punished for someone else

web_server.py

@@ -398,6 +398,11 @@ def inject_webui_assets():
         'vite_assets': build_webui_vite_assets,
     }
 
+# Brute-force limiter for the launch-PIN unlock (lenient; only a flood of wrong
+# PINs from one IP trips it — correct entry clears it instantly).
+from core.security.rate_limit import AttemptLimiter as _AttemptLimiter
+_launch_pin_limiter = _AttemptLimiter(max_attempts=10, window_seconds=300)
+
 # --- Launch PIN gate (before_request hook) ---
 @app.before_request
 def _enforce_launch_pin():
@@ -25255,6 +25260,15 @@ def get_current_profile():
 def verify_launch_pin():
     """Verify PIN for launch lock screen"""
     try:
+        # Brute-force guard: only a flood of WRONG PINs from one IP trips this; a
+        # correct entry clears it instantly, so normal use is never affected.
+        _ip = request.remote_addr or 'unknown'
+        _now = time.time()
+        _locked, _retry_after = _launch_pin_limiter.is_locked(_ip, _now)
+        if _locked:
+            return (jsonify({'success': False, 'error': 'Too many attempts — please wait and try again'}),
+                    429, {'Retry-After': str(_retry_after)})
+
         data = request.json or {}
         pin = data.get('pin', '')
         if not pin:
@@ -25263,8 +25277,10 @@ def verify_launch_pin():
         database = get_database()
         # Validate against admin profile (ID 1)
         if not database.verify_profile_pin(1, pin):
+            _launch_pin_limiter.record_failure(_ip, _now)
             return jsonify({'success': False, 'error': 'Invalid PIN'}), 401
 
+        _launch_pin_limiter.record_success(_ip)
         session['launch_pin_verified'] = True
         return jsonify({'success': True})
     except Exception as e: