fix: add API key auth, PIN rate limiting, configurable NAS, bootstrap fix (#1,#3,#6,#7)

Require Bearer token on all API endpoints (PINVAULT_API_KEY env).
Rate limit PIN access to 5 attempts per 15min lockout per PIN.
Make NAS_BACKUP_DIR configurable via PINVAULT_NAS_BACKUP_DIR.
Replace bootstrap() sys.exit(1) with graceful False return.
Add tests for rate limiting and auth.

app.py

@@ -1,12 +1,16 @@
 from flask import Flask, render_template, request, jsonify
 import hashlib
+import hmac
 import os
 import random
 import secrets
 import shutil
 import string
+import sys
 import threading
+import time
 from datetime import datetime, timedelta, timezone
+from functools import wraps
 
 import pysqlcipher3.dbapi2 as sqlcipher
 
@@ -14,16 +18,49 @@ app = Flask(__name__)
 app.secret_key = secrets.token_hex(32)
 
 DB_PATH = os.environ.get("PINVAULT_DB", "/data/pinvault.db")
-LOCAL_BACKUP_DIR = "/data/backups"
+LOCAL_BACKUP_DIR = os.environ.get("PINVAULT_LOCAL_BACKUP_DIR", "/data/backups")
-NAS_BACKUP_DIR = "/mnt/aidata/pinvault"
+NAS_BACKUP_DIR = os.environ.get("PINVAULT_NAS_BACKUP_DIR", "")
-BACKUP_INTERVAL = 3600
+BACKUP_INTERVAL = int(os.environ.get("PINVAULT_BACKUP_INTERVAL", "3600"))
-MAX_BACKUPS = 168
+MAX_BACKUPS = int(os.environ.get("PINVAULT_MAX_BACKUPS", "168"))
 RECOVERY_CODE_COUNT = 4
 RECOVERY_CODE_LENGTH = 64
 ALPHANUMERIC = string.ascii_letters + string.digits
+API_KEY = os.environ.get("PINVAULT_API_KEY") or secrets.token_hex(32)
+
+_pin_attempt_locks = {}
+_PIN_MAX_ATTEMPTS = 5
+_PIN_LOCKOUT_SECONDS = 900
 
 
-def dict_factory(cursor, row):
+def _check_pin_rate_limit(pin_id):
+    now = time.monotonic()
+    if pin_id not in _pin_attempt_locks:
+        _pin_attempt_locks[pin_id] = {"attempts": 0, "lockout_until": 0}
+    entry = _pin_attempt_locks[pin_id]
+    if now < entry["lockout_until"]:
+        remaining = int(entry["lockout_until"] - now)
+        return False, remaining
+    entry["attempts"] += 1
+    if entry["attempts"] > _PIN_MAX_ATTEMPTS:
+        entry["lockout_until"] = now + _PIN_LOCKOUT_SECONDS
+        entry["attempts"] = 0
+        return False, _PIN_LOCKOUT_SECONDS
+    return True, 0
+
+
+def _reset_pin_rate_limit(pin_id):
+    if pin_id in _pin_attempt_locks:
+        _pin_attempt_locks[pin_id] = {"attempts": 0, "lockout_until": 0}
+
+
+def require_api_key(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        auth = request.headers.get("Authorization", "")
+        if not auth.startswith("Bearer ") or not hmac.compare_digest(auth[7:], API_KEY):
+            return jsonify({"error": "Invalid or missing API key"}), 401
+        return f(*args, **kwargs)
+    return decorated
     d = {}
     for idx, col in enumerate(cursor.description):
         d[col[0]] = row[idx]
@@ -161,11 +198,13 @@ def index():
 
 
 @app.route("/api/status", methods=["GET"])
+@require_api_key
 def api_status():
     return jsonify({"locked": not getattr(app, "master_key", None)})
 
 
 @app.route("/api/backups", methods=["GET"])
+@require_api_key
 def api_list_backups():
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -173,6 +212,7 @@ def api_list_backups():
 
 
 @app.route("/api/backups", methods=["POST"])
+@require_api_key
 def api_trigger_backup():
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -181,6 +221,7 @@ def api_trigger_backup():
 
 
 @app.route("/api/backups/<filename>/restore", methods=["POST"])
+@require_api_key
 def api_restore_backup(filename):
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -211,6 +252,7 @@ def api_restore_backup(filename):
 
 
 @app.route("/api/pins", methods=["GET"])
+@require_api_key
 def api_list_pins():
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -243,6 +285,7 @@ def api_list_pins():
 
 
 @app.route("/api/pins", methods=["POST"])
+@require_api_key
 def api_add_pin():
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -282,9 +325,13 @@ def api_add_pin():
 
 
 @app.route("/api/pins/<int:pin_id>/access", methods=["POST"])
+@require_api_key
 def api_access_pin(pin_id):
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
+    allowed, remaining = _check_pin_rate_limit(pin_id)
+    if not allowed:
+        return jsonify({"error": f"Too many attempts. Try again in {remaining}s."}), 429
     data = request.get_json() or {}
     bypass_code = data.get("bypass_code", "").strip() if data.get("bypass_code") else None
 
@@ -313,12 +360,14 @@ def api_access_pin(pin_id):
                 (now.isoformat(), pin_id),
             )
             conn.commit()
+            _reset_pin_rate_limit(pin_id)
             pin = recover_pin(row["pin_hash"])
             return jsonify({"pin": pin})
 
         if now >= lock_until:
             conn.execute("UPDATE pins SET revealed = 1 WHERE id = ?", (pin_id,))
             conn.commit()
+            _reset_pin_rate_limit(pin_id)
             pin = recover_pin(row["pin_hash"])
             return jsonify({"pin": pin})
 
@@ -328,6 +377,7 @@ def api_access_pin(pin_id):
 
 
 @app.route("/api/pins/<int:pin_id>", methods=["DELETE"])
+@require_api_key
 def api_delete_pin(pin_id):
     if not getattr(app, "master_key", None):
         return jsonify({"error": "Vault not configured."}), 401
@@ -402,34 +452,41 @@ def _auto_restore():
     return True
 
 
+_bootstrap_ok = False
+
+
 def bootstrap():
-    import sys
+    global _bootstrap_ok
     bcrypt_hash = os.environ.get("PINVAULT_MASTER_HASH", "").strip()
     os.environ.pop("PINVAULT_MASTER_HASH", None)
 
     if not bcrypt_hash:
         print("ERROR: PINVAULT_MASTER_HASH environment variable is required.", file=sys.stderr)
-        print("Generate one with: python3 -c \"import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())\"", file=sys.stderr)
+        print("Generate: python3 -c \"import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())\"", file=sys.stderr)
-        sys.exit(1)
+        return False
 
     if not bcrypt_hash.startswith("$2"):
         print("ERROR: PINVAULT_MASTER_HASH must be a bcrypt hash (starts with $2b$ or $2a$)", file=sys.stderr)
-        sys.exit(1)
+        return False
 
     app.master_key = _derive_sqlcipher_key(bcrypt_hash)
 
     if os.path.exists(DB_PATH):
         if not _auto_restore():
             print("FATAL: Cannot open database and no valid backup found.", file=sys.stderr, flush=True)
-            sys.exit(1)
+            return False
 
     init_db()
     _do_backup()
     t = threading.Thread(target=_backup_loop, daemon=True)
     t.start()
+    _bootstrap_ok = True
+    return True
 
 
-bootstrap()
+_bootstrap_ok = bootstrap()
 
 if __name__ == "__main__":
+    if not _bootstrap_ok:
+        sys.exit(1)
     app.run(host="0.0.0.0", port=8765)

tests.py

@@ -0,0 +1,58 @@
+import os
+import sys
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+
+class TestPinVaultSecurity(unittest.TestCase):
+
+    def test_api_key_required(self):
+        import app
+        self.assertTrue(hasattr(app, 'require_api_key'))
+        self.assertTrue(len(app.API_KEY) >= 32)
+
+    def test_rate_limit_exists(self):
+        import app
+        self.assertTrue(hasattr(app, '_check_pin_rate_limit'))
+        self.assertEqual(app._PIN_MAX_ATTEMPTS, 5)
+        self.assertEqual(app._PIN_LOCKOUT_SECONDS, 900)
+
+    def test_rate_limit_blocks_after_max(self):
+        import app
+        app._pin_attempt_locks.clear()
+        for i in range(app._PIN_MAX_ATTEMPTS):
+            allowed, _ = app._check_pin_rate_limit(999)
+            self.assertTrue(allowed, f"Attempt {i+1} should be allowed")
+        allowed, remaining = app._check_pin_rate_limit(999)
+        self.assertFalse(allowed, "Should be locked out after max attempts")
+        self.assertGreater(remaining, 0)
+        app._pin_attempt_locks.clear()
+
+    def test_rate_limit_resets_on_success(self):
+        import app
+        app._pin_attempt_locks.clear()
+        for i in range(3):
+            app._check_pin_rate_limit(999)
+        app._reset_pin_rate_limit(999)
+        allowed, _ = app._check_pin_rate_limit(999)
+        self.assertTrue(allowed, "Should allow after reset")
+        app._pin_attempt_locks.clear()
+
+    def test_nas_backup_configurable(self):
+        import app
+        self.assertEqual(app.NAS_BACKUP_DIR, os.environ.get("PINVAULT_NAS_BACKUP_DIR", ""))
+
+    def test_bootstrap_returns_bool(self):
+        import app
+        self.assertIsInstance(app._bootstrap_ok, bool)
+
+    def test_no_sys_exit_in_bootstrap(self):
+        import inspect, app
+        src = inspect.getsource(app.bootstrap)
+        self.assertNotIn("sys.exit", src, "bootstrap must not call sys.exit")
+
+
+if __name__ == "__main__":
+    unittest.main()