Add local access attempt limiting

Author: 01rabbit

README.md

@@ -295,6 +295,8 @@ PHANTASM_STATE_DIR=/path/to/state python3 main.py init
 | `PHANTASM_FIELD_MODE=1` | Reduce normal WebUI operational detail |
 | `PHANTASM_PROFILE` | Select local capability mode: `standard`, `field`, or `maintenance` |
 | `PHANTASM_MIN_PASSPHRASE_LENGTH` | Minimum Store passphrase length |
+| `PHANTASM_ACCESS_MAX_FAILURES` | Failed access attempts before temporary lockout |
+| `PHANTASM_ACCESS_LOCKOUT_SECONDS` | Temporary access lockout duration |
 | `PHANTASM_AUDIT=1` | Enable audit logging |
 
 ## Local-Only Trust Boundary

docs/REVIEW_VALIDATION_RECORD.md

@@ -24,7 +24,7 @@ python3 -m unittest discover -s tests
 Result:
 
 ```text
-2026-05-02, macOS Darwin arm64, python3 -m unittest discover -s tests, 141 tests passed.
+2026-05-02, macOS Darwin arm64, python3 -m unittest discover -s tests, 147 tests passed.
 ```
 
 Restricted-flow scenario matrix:

docs/SPECIFICATION.md

@@ -316,6 +316,8 @@ The physical object is an operational cue, not a high-entropy cryptographic fact
 | `PHANTASM_FIELD_MODE` | Reduce normal WebUI operational detail | `0` |
 | `PHANTASM_PROFILE` | Select local capability mode: `standard`, `field`, or `maintenance` | `standard` |
 | `PHANTASM_MIN_PASSPHRASE_LENGTH` | Minimum Store passphrase length | `10` |
+| `PHANTASM_ACCESS_MAX_FAILURES` | Failed access attempts before temporary lockout | `5` |
+| `PHANTASM_ACCESS_LOCKOUT_SECONDS` | Temporary access lockout duration | `60` |
 | `PHANTASM_AUDIT` | Enable audit logging | `0` |
 | `PHANTASM_AUDIT_FILENAMES` | Record filename hashes | unset |
 

docs/THREAT_MODEL.md

@@ -43,6 +43,7 @@ It is not a substitute for audited full-disk encryption, hardware-backed key sto
 - Reference keys are stored together in a single AES-GCM encrypted ORB state blob under the configured state directory, not as raw reference photos or semantic per-entry template filenames.
 - Image-key matching requires stable results across a short frame window rather than accepting a single-frame match.
 - Web mutation endpoints require `X-Phantasm-Token`, apply a simple per-client rate limit, and enforce upload size limits.
+- Access recovery flows count repeated local failures and apply a bounded temporary lockout. WebUI limiting is process-local; CLI limiting is stored in local state.
 - Web responses include no-store cache headers, frame denial, MIME-sniffing protection, no-referrer policy, constrained browser permissions, and a local-only content security policy. These reduce browser residue and common Web embedding risks but do not make the WebUI safe for untrusted networks.
 - Sensitive Web actions require a fresh restricted confirmation session in addition to the Web token and UI unlock state. Restricted action pages and entry maintenance details are withheld until that confirmation is active.
 - Optional UI face lock (`PHANTASM_UI_FACE_LOCK=1`) can gate normal WebUI routes with a short-lived local session. This is a UI access control only and is not used for vault encryption.
@@ -71,6 +72,7 @@ These surfaces should not reveal the internal disclosure model, internal trial o
 - Dual password slots duplicate encrypted payload material within the selected internal storage span. This improves operational control but reduces maximum payload size.
 - UI face lock can be affected by lighting, camera angle, false rejects, false accepts, and presentation attacks using photos or screens.
 - The in-memory Web rate limiter and restricted confirmation state reset on process restart and are not substitutes for a full access-control layer.
+- Access-attempt limiting slows repeated local failures but does not stop offline guessing against copied data, compromised hosts, or deliberate state rollback.
 - UI tokens can be read from a compromised browser or host session.
 - Passphrase policy cannot compensate for observed input, reused passwords, coercion, compromised hosts, or poor operational separation.
 - Metadata checks and metadata reduction are best-effort. They can miss embedded identifiers, thumbnails, histories, and application-specific fields.

src/phantasm/attempt_limiter.py

@@ -0,0 +1,86 @@
+"""Local access-attempt limiter for recovery flows."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import time
+
+from .config import access_lockout_seconds, access_max_failures
+from .state_store import LocalStateStore
+
+ATTEMPT_STATE_NAME = "access_attempts.json"
+
+
+@dataclass(frozen=True)
+class AttemptDecision:
+    allowed: bool
+    wait_seconds: int = 0
+
+
+class AttemptLimiter:
+    def __init__(
+        self,
+        max_failures: int | None = None,
+        lockout_seconds: int | None = None,
+        clock=None,
+    ):
+        self.max_failures = max_failures or access_max_failures()
+        self.lockout_seconds = lockout_seconds or access_lockout_seconds()
+        self.clock = clock or time.time
+        self._state = {}
+
+    def check(self, scope: str):
+        state = self._state.get(scope, {"failures": 0, "locked_until": 0})
+        now = int(self.clock())
+        locked_until = int(state.get("locked_until", 0))
+        if locked_until > now:
+            return AttemptDecision(False, locked_until - now)
+        return AttemptDecision(True, 0)
+
+    def record_failure(self, scope: str):
+        state = self._state.get(scope, {"failures": 0, "locked_until": 0})
+        failures = int(state.get("failures", 0)) + 1
+        locked_until = 0
+        if failures >= self.max_failures:
+            locked_until = int(self.clock()) + self.lockout_seconds
+        self._state[scope] = {
+            "failures": failures,
+            "locked_until": locked_until,
+        }
+
+    def record_success(self, scope: str):
+        self._state.pop(scope, None)
+
+
+class FileAttemptLimiter(AttemptLimiter):
+    def __init__(self, store: LocalStateStore | None = None, **kwargs):
+        super().__init__(**kwargs)
+        self.store = store or LocalStateStore()
+        self._state = self._load()
+
+    def record_failure(self, scope: str):
+        super().record_failure(scope)
+        self._save()
+
+    def record_success(self, scope: str):
+        super().record_success(scope)
+        self._save()
+
+    def _load(self):
+        try:
+            data = self.store.read_json(ATTEMPT_STATE_NAME)
+        except Exception:
+            return {}
+        if not isinstance(data, dict):
+            return {}
+        return {
+            str(scope): {
+                "failures": int(value.get("failures", 0)),
+                "locked_until": int(value.get("locked_until", 0)),
+            }
+            for scope, value in data.items()
+            if isinstance(value, dict)
+        }
+
+    def _save(self):
+        self.store.write_json_atomic(ATTEMPT_STATE_NAME, self._state)

src/phantasm/cli.py

@@ -5,6 +5,7 @@
 import time
 
 from .ai_gate import get_gesture_sequence, gate
+from .attempt_limiter import FileAttemptLimiter
 from .audit import audit_event
 from .bridge_ui import ui
 from .config import duress_mode_enabled, purge_confirmation_required
@@ -13,6 +14,7 @@
 from .gv_core import GhostVault
 from .operations import doctor, export_redacted_log, verify_audit_log, verify_state
 from .passphrase_policy import check_store_passphrases
+from . import strings as text
 
 CAMERA_WARMUP_TIMEOUT = 10
 REFERENCE_MATCH_TIMEOUT = 10
@@ -323,17 +325,25 @@ def main():
             print(" DEVICE STATUS: READY")
             print("=" * 55)
 
+            attempt_limiter = FileAttemptLimiter()
+            attempt_scope = "cli-retrieve"
+            if not attempt_limiter.check(attempt_scope).allowed:
+                print(f"\n[!] {text.ACCESS_TEMPORARILY_UNAVAILABLE}")
+                return
+
             pw = getpass.getpass("\n[AUTH] Access password: ")
 
             print("\n[LOCAL] Requesting bound object verification...")
             user_gesture_seq = _collect_auth_sequence()
 
             if gate.last_match_mode == gate.MATCH_AMBIGUOUS:
                 ui.show_alert("ACCESS ERROR\nAMBIGUOUS OBJECT")
+                attempt_limiter.record_failure(attempt_scope)
                 print("[!] Access rejected because the object match is ambiguous.")
                 return
             if not user_gesture_seq or user_gesture_seq[0] == gate.MATCH_NONE:
                 ui.show_alert("ACCESS ERROR\nOBJECT NOT FOUND")
+                attempt_limiter.record_failure(attempt_scope)
                 print("[!] No bound object matched.")
                 return
 
@@ -355,6 +365,7 @@ def main():
                 accessed_mode = gate.MODES[1]
 
             if result is not None:
+                attempt_limiter.record_success(attempt_scope)
                 ui.show_alert("ACCESS GRANTED")
 
                 show_loading("Preparing recovered payload", 2)
@@ -410,6 +421,7 @@ def main():
             else:
                 ui.show_alert("ACCESS DENIED\nINVALID CREDENTIALS")
                 audit_event("retrieve_failed")
+                attempt_limiter.record_failure(attempt_scope)
                 print("\n[!] Access failed.")
 
         elif args.action == "brick":

src/phantasm/config.py

@@ -50,3 +50,19 @@ def passphrase_min_length():
         return max(1, int(value))
     except ValueError:
         return 10
+
+
+def access_max_failures():
+    value = os.environ.get("PHANTASM_ACCESS_MAX_FAILURES", "5")
+    try:
+        return max(1, int(value))
+    except ValueError:
+        return 5
+
+
+def access_lockout_seconds():
+    value = os.environ.get("PHANTASM_ACCESS_LOCKOUT_SECONDS", "60")
+    try:
+        return max(1, int(value))
+    except ValueError:
+        return 60

src/phantasm/strings.py

@@ -8,6 +8,7 @@
 INVALID_WEB_TOKEN = "invalid web token"
 RATE_LIMIT_EXCEEDED = "rate limit exceeded"
 UPLOAD_TOO_LARGE = "upload too large"
+ACCESS_TEMPORARILY_UNAVAILABLE = "Access temporarily unavailable."
 
 CONFIRMATION_REJECTED_DISPLAY = "Confirmation rejected."
 RESTRICTED_CONFIRMATION_ACCEPTED = "Restricted confirmation accepted."

src/phantasm/web_server.py

@@ -24,6 +24,7 @@
 from fastapi.templating import Jinja2Templates
 
 from .ai_gate import gate, get_gesture_sequence
+from .attempt_limiter import AttemptLimiter
 from .audit import audit_event
 from .capabilities import Capability, active_policy, capability_enabled
 from .config import (
@@ -62,6 +63,7 @@
 )
 _rate_limit = {}
 _restricted_sessions = {}
+_access_attempts = AttemptLimiter()
 
 ENTRY_TO_MODE = {
     "entry_1": gate.MODES[0],
@@ -815,10 +817,15 @@ async def metadata_scrub(request: Request, file: UploadFile = File(...)):
 )
 async def retrieve(request: Request, password: str = Form(...)):
     enforce_rate_limit(request)
+    attempt_scope = f"web:{_client_id(request)}"
+    if not _access_attempts.check(attempt_scope).allowed:
+        return {"error": text.ACCESS_TEMPORARILY_UNAVAILABLE}
     auth_sequence = get_gesture_sequence(length=1)
     if gate.last_match_mode == gate.MATCH_AMBIGUOUS:
+        _access_attempts.record_failure(attempt_scope)
         return {"error": text.AMBIGUOUS_OBJECT_MATCH}
     if auth_sequence[0] == gate.MATCH_NONE:
+        _access_attempts.record_failure(attempt_scope)
         return {"error": text.NO_VALID_ENTRY_FOUND}
 
     for mode in gate.MODES:
@@ -840,6 +847,7 @@ async def retrieve(request: Request, password: str = Form(...)):
         if result is not None:
             # Release camera to save power and heat after successful retrieval
             gate.close()
+        _access_attempts.record_success(attempt_scope)
         if not purge_applied:
             purge_applied = _maybe_auto_purge(mode, source="web")
         return create_file_response(
@@ -849,6 +857,7 @@ async def retrieve(request: Request, password: str = Form(...)):
         )
 
     audit_event("retrieve_failed", source="web")
+    _access_attempts.record_failure(attempt_scope)
     return {"error": text.NO_VALID_ENTRY_FOUND}
 
 

tests/test_attempt_limiter.py

@@ -0,0 +1,63 @@
+import os
+import tempfile
+import unittest
+
+ROOT = os.path.dirname(os.path.dirname(__file__))
+import sys
+
+sys.path.insert(0, os.path.join(ROOT, "src"))
+
+from phantasm.attempt_limiter import AttemptLimiter, FileAttemptLimiter
+from phantasm.state_store import LocalStateStore
+
+
+class AttemptLimiterTests(unittest.TestCase):
+    def test_repeated_failures_trigger_lockout(self):
+        now = [1000]
+        limiter = AttemptLimiter(
+            max_failures=2,
+            lockout_seconds=30,
+            clock=lambda: now[0],
+        )
+
+        self.assertTrue(limiter.check("local").allowed)
+        limiter.record_failure("local")
+        self.assertTrue(limiter.check("local").allowed)
+        limiter.record_failure("local")
+
+        decision = limiter.check("local")
+        self.assertFalse(decision.allowed)
+        self.assertEqual(decision.wait_seconds, 30)
+
+    def test_success_resets_attempt_state(self):
+        limiter = AttemptLimiter(max_failures=1, lockout_seconds=30, clock=lambda: 1000)
+
+        limiter.record_failure("local")
+        self.assertFalse(limiter.check("local").allowed)
+        limiter.record_success("local")
+
+        self.assertTrue(limiter.check("local").allowed)
+
+    def test_file_limiter_persists_state(self):
+        tmpdir = tempfile.mkdtemp()
+        store = LocalStateStore(tmpdir)
+        limiter = FileAttemptLimiter(
+            store=store,
+            max_failures=1,
+            lockout_seconds=30,
+            clock=lambda: 1000,
+        )
+        limiter.record_failure("cli")
+
+        restored = FileAttemptLimiter(
+            store=store,
+            max_failures=1,
+            lockout_seconds=30,
+            clock=lambda: 1000,
+        )
+
+        self.assertFalse(restored.check("cli").allowed)
+
+
+if __name__ == "__main__":
+    unittest.main()