0d1e949798
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.
54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
"""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"]
|