From 606b7c71c2e4a3b4a0140f7cef7692050e0bbb30 Mon Sep 17 00:00:00 2001 From: Jarian Cottingham Date: Sat, 4 Jul 2026 04:54:33 +0000 Subject: [PATCH] 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 | 79 ++++++++++++++++++++++++++++++++++++++++++++++++-------- tests.py | 58 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 11 deletions(-) create mode 100644 tests.py diff --git a/app.py b/app.py index 4c2058e..1d7dea4 100644 --- a/app.py +++ b/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" -NAS_BACKUP_DIR = "/mnt/aidata/pinvault" -BACKUP_INTERVAL = 3600 -MAX_BACKUPS = 168 +LOCAL_BACKUP_DIR = os.environ.get("PINVAULT_LOCAL_BACKUP_DIR", "/data/backups") +NAS_BACKUP_DIR = os.environ.get("PINVAULT_NAS_BACKUP_DIR", "") +BACKUP_INTERVAL = int(os.environ.get("PINVAULT_BACKUP_INTERVAL", "3600")) +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//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//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/", 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) - sys.exit(1) + print("Generate: python3 -c \"import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())\"", file=sys.stderr) + 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) \ No newline at end of file diff --git a/tests.py b/tests.py new file mode 100644 index 0000000..7ab3ad0 --- /dev/null +++ b/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()