From 0d1e949798b39e8af4b98afde51d46f2e70c66e1 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Wed, 10 Jun 2026 20:47:46 -0700 Subject: [PATCH] Security: brute-force limiter on the launch-PIN unlock (Tier 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 54 +++++++++++++++++++++++++++++ tests/test_credentials_endpoints.py | 24 +++++++++++++ tests/test_pin_rate_limit.py | 50 ++++++++++++++++++++++++++ web_server.py | 16 +++++++++ 4 files changed, 144 insertions(+) create mode 100644 core/security/rate_limit.py create mode 100644 tests/test_pin_rate_limit.py diff --git a/core/security/rate_limit.py b/core/security/rate_limit.py new file mode 100644 index 00000000..1fd7b56d --- /dev/null +++ b/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"] diff --git a/tests/test_credentials_endpoints.py b/tests/test_credentials_endpoints.py index 6f979779..1da0a4f6 100644 --- a/tests/test_credentials_endpoints.py +++ b/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() diff --git a/tests/test_pin_rate_limit.py b/tests/test_pin_rate_limit.py new file mode 100644 index 00000000..54ee6656 --- /dev/null +++ b/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 diff --git a/web_server.py b/web_server.py index a69858b4..4a4ea709 100644 --- a/web_server.py +++ b/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: