diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..5a87c52 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +# bcrypt hash of the master password (required). +# Generate with: +# python3 -c "import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())" +PINVAULT_MASTER_HASH= + +# Optional overrides (defaults shown): +#PINVAULT_DB=/data/pinvault.db +#PINVAULT_LOCAL_BACKUP_DIR=/data/backups +#PINVAULT_NAS_BACKUP_DIR= +#PINVAULT_BACKUP_INTERVAL=3600 +#PINVAULT_MAX_BACKUPS=168 +# If unset, a random API key is generated per process (printed nowhere; +# set this explicitly for stable automation access): +#PINVAULT_API_KEY= diff --git a/.gitignore b/.gitignore index 24f9f8d..e51d658 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,7 @@ .env data/ +__pycache__/ +*.pyc +.pytest_cache/ +*.egg-info/ +.venv/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..eba8c1d --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Jarian Cottingham + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index e69de29..c251f1d 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,107 @@ +# PinVault + +Encrypted PIN lock-vault. Store 4-digit PINs (door codes, safe combinations, +router admin PINs, ...) behind a timed lock. Each PIN is sealed for a +configurable lock period; after the lock expires — or immediately, using one +of the one-time recovery codes — the PIN is revealed via the API or web UI. + +The database is encrypted with [SQLCipher](https://www.zetetic.net/sqlcipher/). +The SQLCipher key is derived from a bcrypt master hash via PBKDF2-SHA256 +(100k iterations); the plaintext master password is never stored. + +## Design notes + +- **PINs are a 10^4 space on purpose.** "Recovery" of a PIN means exhaustive + re-derivation (≤10,000 HMAC checks). The security model is *time-locked + access + one-time bypass codes + brute-force rate limiting*, not PIN + secrecy. Do not store high-entropy secrets here — use a password manager. +- **Per-row salts.** Every PIN and every recovery code is HMAC-SHA256-hashed + with its own random 16-byte salt, stored next to the hash. Precomputed + tables do not transfer across rows or installations. +- **One-time codes.** Each PIN ships with 4 recovery codes (64 chars, + generated with the `secrets` module). A code is single-use: it unlocks the + PIN, is marked used, and the bypass is counted. +- **Rate limiting.** 5 failed access attempts per PIN → 15-minute lockout + (in-process). +- **Backups.** The DB is copied to a local backup dir hourly (configurable) + and, if configured, mirrored to a NAS path. A rolling window keeps the + most recent N backups. If the primary DB is unreadable at startup, the most + recent valid backup is auto-restored. Restores are atomic (temp file + + rename), so a crash mid-restore cannot tear the database. + +## API + +All endpoints require `Authorization: Bearer $PINVAULT_API_KEY` (except the +web UI at `/`). + +| Method | Path | Description | +| ------ | ---- | ----------- | +| GET | `/api/status` | Vault status (`locked` = master key not loaded) | +| GET | `/api/pins` | List PINs (never includes PINs or codes) | +| POST | `/api/pins` | Create PIN. Body: `{"label": "...", "lock_days": 30}`. Returns the PIN and 4 recovery codes **once** | +| POST | `/api/pins//access` | Reveal PIN. Body: `{}` after lock expiry, or `{"bypass_code": "..."}` to unlock early. 423 while locked, 429 when rate-limited | +| DELETE | `/api/pins/` | Delete a revealed PIN (409 if never revealed) | +| GET | `/api/backups` | List backup files (local + NAS) | +| POST | `/api/backups` | Trigger an immediate backup | +| POST | `/api/backups//restore` | Restore a backup (strict filename validation, key-verified before swap) | + +## Configuration + +Environment variables (see `.env.example`): + +| Variable | Default | Required | +| -------- | ------- | -------- | +| `PINVAULT_MASTER_HASH` | — | **yes** (bcrypt hash of the master password) | +| `PINVAULT_API_KEY` | random per process | recommended | +| `PINVAULT_DB` | `/data/pinvault.db` | no | +| `PINVAULT_LOCAL_BACKUP_DIR` | `/data/backups` | no | +| `PINVAULT_NAS_BACKUP_DIR` | *(empty = local only)* | no | +| `PINVAULT_BACKUP_INTERVAL` | `3600` (seconds) | no | +| `PINVAULT_MAX_BACKUPS` | `168` | no | + +## Run + +### Docker + +```sh +cp .env.example .env # fill in PINVAULT_MASTER_HASH +docker compose up -d --build +``` + +The compose file mounts `./data` for the DB/backups and `/mnt/aidata` as an +optional NAS target — point `PINVAULT_NAS_BACKUP_DIR` at it to enable +off-box backups. + +### Bare metal (Python ≥3.10) + +Requires `libsqlcipher-dev` (or equivalent) so `pysqlcipher3` can build: + +```sh +python3 -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt +export PINVAULT_MASTER_HASH='$(python3 -c "import bcrypt; print(bcrypt.hashpw(b\"secret\", bcrypt.gensalt()).decode())")' +gunicorn -b 0.0.0.0:8765 --timeout 120 app:app +``` + +## Tests + +The suite runs without the SQLCipher native library: when `pysqlcipher3` is +not installed, `tests/conftest.py` shims it with plain `sqlite3` (swallowing +the `PRAGMA key` statement) so all endpoint logic is still exercised. + +```sh +pip install -e ".[dev]" +pytest tests/ -v +ruff check . +``` + +## Security notes + +- The master password itself is never stored, only its bcrypt hash; the + SQLCipher key is PBKDF2-derived from that hash. +- Recovery codes and PIN hashes use per-row random salts; comparisons are + constant-time (`hmac.compare_digest`). +- Backup filenames are strictly validated before use, and restore paths are + confined to the configured backup directories. +- PIN generation uses the `secrets` module. +- PIN access is rate-limited per PIN (5 attempts / 15 min). diff --git a/app.py b/app.py index 1e85f6d..d67eef4 100644 --- a/app.py +++ b/app.py @@ -2,11 +2,12 @@ from flask import Flask, render_template, request, jsonify import hashlib import hmac import os -import random +import re import secrets import shutil import string import sys +import tempfile import threading import time from datetime import datetime, timedelta, timezone @@ -57,13 +58,16 @@ 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 + if not auth.startswith("Bearer ") or auth[7:] != API_KEY: + return jsonify({"error": "Unauthorized"}), 401 return f(*args, **kwargs) return decorated + + +def dict_factory(cursor, row): d = {} for idx, col in enumerate(cursor.description): - d[col[0]] = row[idx] + d[col[0].lower()] = row[idx] return d @@ -83,9 +87,13 @@ def get_conn(): return conn -def hash_val(value, salt: str = "pinvault-recovery-v1"): - """Hash recovery code with HMAC to prevent precomputed attacks (#4).""" - return hmac.new(salt.encode(), value.encode(), hashlib.sha256).hexdigest() +LEGACY_SALT = "pinvault-recovery-v1" + + +def hash_val(value, salt=None): + """Hash a PIN or recovery code with HMAC-SHA256 using a per-row salt (#4).""" + effective_salt = salt if salt else LEGACY_SALT + return hmac.new(effective_salt.encode(), value.encode(), hashlib.sha256).hexdigest() def generate_recovery_code(): @@ -123,6 +131,12 @@ def init_db(): conn.commit() except Exception: pass + for table in ("pins", "recovery_codes"): + try: + conn.execute(f"ALTER TABLE {table} ADD COLUMN salt TEXT") + conn.commit() + except Exception: + pass def _backup_db(dest_path: str) -> bool: @@ -135,6 +149,22 @@ def _backup_db(dest_path: str) -> bool: return False +def _atomic_copy(src, dst): + """Copy src over dst atomically (temp file + rename) to avoid a torn DB.""" + dst_dir = os.path.dirname(dst) or "." + fd, tmp = tempfile.mkstemp(dir=dst_dir, prefix=".pinvault-restore-") + try: + with os.fdopen(fd, "wb") as out, open(src, "rb") as f: + shutil.copyfileobj(f, out) + os.replace(tmp, dst) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + def _cleanup_backups(directory: str): try: files = sorted( @@ -147,28 +177,35 @@ def _cleanup_backups(directory: str): pass +def _backup_dirs(): + return [d for d in (LOCAL_BACKUP_DIR, NAS_BACKUP_DIR) if d] + + def _do_backup(): ts = now_utc().strftime("%Y%m%d-%H%M%S") filename = f"pinvault-backup-{ts}.db" os.makedirs(LOCAL_BACKUP_DIR, exist_ok=True) - os.makedirs(NAS_BACKUP_DIR, exist_ok=True) - local_path = os.path.join(LOCAL_BACKUP_DIR, filename) - nas_path = os.path.join(NAS_BACKUP_DIR, filename) ok = _backup_db(local_path) if ok: - shutil.copy2(local_path, nas_path) _cleanup_backups(LOCAL_BACKUP_DIR) - _cleanup_backups(NAS_BACKUP_DIR) - return ok + if NAS_BACKUP_DIR: + os.makedirs(NAS_BACKUP_DIR, exist_ok=True) + nas_path = os.path.join(NAS_BACKUP_DIR, filename) + shutil.copy2(local_path, nas_path) + _cleanup_backups(NAS_BACKUP_DIR) + return True return False def _backup_loop(): while True: - _do_backup() + try: + _do_backup() + except Exception as e: + print(f"Backup failed: {e}", file=sys.stderr, flush=True) timer = threading.Timer(BACKUP_INTERVAL, _backup_loop) timer.daemon = True timer.start() @@ -177,7 +214,7 @@ def _backup_loop(): def _get_backup_list(): result = [] - for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]: + for d in _backup_dirs(): location = "local" if d == LOCAL_BACKUP_DIR else "nas" try: for f in sorted(os.listdir(d)): @@ -229,12 +266,11 @@ def api_trigger_backup(): def api_restore_backup(filename): if not getattr(app, "master_key", None): return jsonify({"error": "Vault not configured."}), 401 - import re if not re.match(r'^pinvault-backup-\d{8}-\d{6}\.db$', filename): return jsonify({"error": "Invalid backup filename."}), 400 src = None - for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]: + for d in _backup_dirs(): candidate = os.path.join(d, filename) if os.path.exists(candidate): src = candidate @@ -251,7 +287,7 @@ def api_restore_backup(filename): except Exception as e: return jsonify({"error": f"Backup verification failed: {e}"}), 500 - shutil.copy2(src, DB_PATH) + _atomic_copy(src, DB_PATH) return jsonify({"ok": True, "restored": filename}) @@ -293,14 +329,15 @@ def api_list_pins(): def api_add_pin(): if not getattr(app, "master_key", None): return jsonify({"error": "Vault not configured."}), 401 - data = request.get_json() + data = request.get_json() or {} label = data.get("label", "").strip() lock_days = data.get("lock_days", 30) - if lock_days < 0: - return jsonify({"error": "lock_days must be >= 0"}), 400 + if isinstance(lock_days, bool) or not isinstance(lock_days, int) or lock_days < 0: + return jsonify({"error": "lock_days must be an integer >= 0"}), 400 - pin = f"{random.randint(0, 9999):04d}" - pin_hash = hash_val(pin) + pin = f"{secrets.randbelow(10000):04d}" + pin_salt = secrets.token_hex(16) + pin_hash = hash_val(pin, pin_salt) now = now_utc() lock_until = (now + timedelta(days=lock_days)).isoformat() @@ -308,14 +345,15 @@ def api_add_pin(): with get_conn() as conn: cur = conn.execute( - "INSERT INTO pins (label, pin_hash, created_at, lock_until) VALUES (?, ?, ?, ?)", - (label, pin_hash, now.isoformat(), lock_until), + "INSERT INTO pins (label, pin_hash, salt, created_at, lock_until) VALUES (?, ?, ?, ?, ?)", + (label, pin_hash, pin_salt, now.isoformat(), lock_until), ) pin_id = cur.lastrowid for code in codes: + code_salt = secrets.token_hex(16) conn.execute( - "INSERT INTO recovery_codes (pin_id, code_hash) VALUES (?, ?)", - (pin_id, hash_val(code)), + "INSERT INTO recovery_codes (pin_id, code_hash, salt) VALUES (?, ?, ?)", + (pin_id, hash_val(code, code_salt), code_salt), ) conn.commit() @@ -348,11 +386,14 @@ def api_access_pin(pin_id): now = now_utc() if bypass_code: - code_hash = hash_val(bypass_code) - code_row = conn.execute( - "SELECT * FROM recovery_codes WHERE pin_id = ? AND code_hash = ? AND used = 0", - (pin_id, code_hash), - ).fetchone() + code_row = None + for candidate in conn.execute( + "SELECT * FROM recovery_codes WHERE pin_id = ? AND used = 0", + (pin_id,), + ).fetchall(): + if hmac.compare_digest(hash_val(bypass_code, candidate["salt"]), candidate["code_hash"]): + code_row = candidate + break if not code_row: return jsonify({"error": "Invalid or already-used recovery code"}), 403 conn.execute( @@ -365,14 +406,14 @@ def api_access_pin(pin_id): ) conn.commit() _reset_pin_rate_limit(pin_id) - pin = recover_pin(row["pin_hash"]) + pin = recover_pin(row["pin_hash"], row["salt"]) 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"]) + pin = recover_pin(row["pin_hash"], row["salt"]) return jsonify({"pin": pin}) return jsonify({"error": "PIN is locked. Use a recovery code to bypass."}), 423 @@ -392,15 +433,16 @@ def api_delete_pin(pin_id): if not row["revealed"]: return jsonify({"error": "Cannot delete unrevealed PIN. Access it first."}), 409 conn.execute("DELETE FROM recovery_codes WHERE pin_id = ?", (pin_id,)) - cur = conn.execute("DELETE FROM pins WHERE id = ?", (pin_id,)) + conn.execute("DELETE FROM pins WHERE id = ?", (pin_id,)) conn.commit() return jsonify({"ok": True}) -def recover_pin(pin_hash): +def recover_pin(pin_hash, salt=None): + """Recover a 4-digit PIN by exhaustive search (design: PINs are 10^4 space).""" for i in range(10000): candidate = f"{i:04d}" - if hash_val(candidate) == pin_hash: + if hmac.compare_digest(hash_val(candidate, salt), pin_hash): return candidate return None @@ -418,7 +460,7 @@ def _verify_db(): def _latest_backup(): candidates = [] - for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]: + for d in _backup_dirs(): try: for f in os.listdir(d): if f.startswith("pinvault-backup-") and f.endswith(".db"): @@ -451,7 +493,7 @@ def _auto_restore(): print(f"Backup {backup} also unreadable. Cannot auto-restore.", file=sys.stderr, flush=True) return False - shutil.copy2(backup, DB_PATH) + _atomic_copy(backup, DB_PATH) print(f"Restored DB from backup: {backup}", flush=True) return True @@ -493,4 +535,4 @@ _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 + app.run(host="0.0.0.0", port=8765) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..d644a93 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,36 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "pinvault" +version = "1.1.0" +description = "Encrypted PIN lock-vault: SQLCipher storage, timed locks, one-time recovery codes, local/NAS backup" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +dependencies = [ + "flask>=3.1,<4", + "gunicorn>=23", + "pysqlcipher3>=1.2", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8", + "ruff>=0.4", +] + +[tool.setuptools] +py-modules = ["app"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff] +line-length = 120 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "W"] +ignore = ["E501"] diff --git a/tests.py b/tests.py deleted file mode 100644 index 7ab3ad0..0000000 --- a/tests.py +++ /dev/null @@ -1,58 +0,0 @@ -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() diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..868e7be --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,89 @@ +import os +import sqlite3 +import sys +import tempfile +import types + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +_TEST_ROOT = tempfile.mkdtemp(prefix="pinvault-test-") + + +def _install_sqlcipher_shim(): + """Fall back to plain sqlite3 when pysqlcipher3 is not installed. + + The SQLCipher-specific `PRAGMA key` statement is swallowed so the app's + full endpoint logic can be exercised without the native library. + """ + try: + import pysqlcipher3.dbapi2 # noqa: F401 + return + except ImportError: + pass + + class _Connection: + def __init__(self, raw): + self._raw = raw + + @property + def row_factory(self): + return self._raw.row_factory + + @row_factory.setter + def row_factory(self, value): + self._raw.row_factory = value + + def execute(self, sql, *args): + if sql.strip().upper().startswith("PRAGMA KEY"): + return self._raw.execute("SELECT 1", *args) + return self._raw.execute(sql, *args) + + def commit(self): + self._raw.commit() + + def close(self): + self._raw.close() + + def __enter__(self): + return self + + def __exit__(self, *exc): + self._raw.close() + + def connect(path, *args, **kwargs): + raw = sqlite3.connect(path, *args, **kwargs) + return _Connection(raw) + + dbapi = types.ModuleType("pysqlcipher3.dbapi2") + dbapi.connect = connect + pkg = types.ModuleType("pysqlcipher3") + pkg.dbapi2 = dbapi + sys.modules["pysqlcipher3"] = pkg + sys.modules["pysqlcipher3.dbapi2"] = dbapi + + +_install_sqlcipher_shim() + +os.environ.setdefault("PINVAULT_MASTER_HASH", "$2b$12$" + "a" * 53) +os.environ.setdefault("PINVAULT_DB", os.path.join(_TEST_ROOT, "pinvault.db")) +os.environ.setdefault("PINVAULT_LOCAL_BACKUP_DIR", os.path.join(_TEST_ROOT, "backups")) +os.environ.setdefault("PINVAULT_NAS_BACKUP_DIR", "") +os.environ.setdefault("PINVAULT_API_KEY", "test-api-key-0123456789abcdef0123456789abcd") + +import app as pinvault # noqa: E402 + +import pytest # noqa: E402 + +AUTH = {"Authorization": f"Bearer {os.environ['PINVAULT_API_KEY']}"} + + +def create_pin(client, label="test", lock_days=30): + resp = client.post("/api/pins", json={"label": label, "lock_days": lock_days}, headers=AUTH) + assert resp.status_code == 201, resp.get_json() + return resp.get_json() + + +@pytest.fixture() +def client(): + pinvault._pin_attempt_locks.clear() + return pinvault.app.test_client() diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..163d207 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,157 @@ +import os +import re + +from conftest import AUTH, create_pin + +BACKUP_RE = re.compile(r"^pinvault-backup-\d{8}-\d{6}\.db$") + + +def test_bootstrap_ok(client): + import app + assert app._bootstrap_ok is True + + +def test_backup_ran_with_empty_nas_dir(client): + """Regression: bootstrap must not crash when PINVAULT_NAS_BACKUP_DIR is empty.""" + import app + assert app.NAS_BACKUP_DIR == "" + files = os.listdir(app.LOCAL_BACKUP_DIR) + assert any(BACKUP_RE.match(f) for f in files) + + +def test_missing_api_key_rejected(client): + assert client.get("/api/pins").status_code == 401 + assert client.post("/api/pins", json={}).status_code == 401 + + +def test_status(client): + resp = client.get("/api/status", headers=AUTH) + assert resp.status_code == 200 + assert resp.get_json()["locked"] is False + + +def test_list_pins(client): + resp = client.get("/api/pins", headers=AUTH) + assert resp.status_code == 200 + assert isinstance(resp.get_json(), list) + + +def test_create_pin(client): + data = create_pin(client, label="home", lock_days=14) + assert data["label"] == "home" + assert len(data["pin"]) == 4 and data["pin"].isdigit() + assert len(data["recovery_codes"]) == 4 + assert all(len(c) == 64 for c in data["recovery_codes"]) + assert data["id"] > 0 + + +def test_create_pin_lock_days_validation(client): + for bad in ("-1", "abc", 1.5, True, None): + resp = client.post("/api/pins", json={"lock_days": bad}, headers=AUTH) + assert resp.status_code == 400, f"lock_days={bad!r} should be rejected" + + +def test_access_locked_pin_denied(client): + data = create_pin(client, lock_days=30) + resp = client.post(f"/api/pins/{data['id']}/access", json={}, headers=AUTH) + assert resp.status_code == 423 + + +def test_access_unlocked_pin(client): + data = create_pin(client, lock_days=0) + resp = client.post(f"/api/pins/{data['id']}/access", json={}, headers=AUTH) + assert resp.status_code == 200 + assert resp.get_json()["pin"] == data["pin"] + + +def test_access_nonexistent_pin(client): + resp = client.post("/api/pins/999999/access", json={}, headers=AUTH) + assert resp.status_code == 404 + + +def test_recovery_code_unlocks_pin(client): + data = create_pin(client, lock_days=30) + code = data["recovery_codes"][0] + resp = client.post( + f"/api/pins/{data['id']}/access", json={"bypass_code": code}, headers=AUTH + ) + assert resp.status_code == 200 + assert resp.get_json()["pin"] == data["pin"] + + +def test_recovery_code_single_use(client): + data = create_pin(client, lock_days=30) + code = data["recovery_codes"][0] + url = f"/api/pins/{data['id']}/access" + first = client.post(url, json={"bypass_code": code}, headers=AUTH) + assert first.status_code == 200 + second = client.post(url, json={"bypass_code": code}, headers=AUTH) + assert second.status_code == 403 + + +def test_invalid_recovery_code(client): + data = create_pin(client, lock_days=30) + resp = client.post( + f"/api/pins/{data['id']}/access", json={"bypass_code": "x" * 64}, headers=AUTH + ) + assert resp.status_code == 403 + + +def test_delete_unrevealed_pin_refused(client): + data = create_pin(client, lock_days=30) + resp = client.delete(f"/api/pins/{data['id']}", headers=AUTH) + assert resp.status_code == 409 + + +def test_delete_revealed_pin(client): + data = create_pin(client, lock_days=0) + assert client.post(f"/api/pins/{data['id']}/access", json={}, headers=AUTH).status_code == 200 + resp = client.delete(f"/api/pins/{data['id']}", headers=AUTH) + assert resp.status_code == 200 + ids = [p["id"] for p in client.get("/api/pins", headers=AUTH).get_json()] + assert data["id"] not in ids + + +def test_backups_listed(client): + resp = client.get("/api/backups", headers=AUTH) + assert resp.status_code == 200 + items = resp.get_json() + assert items, "expected at least the bootstrap backup" + for item in items: + assert BACKUP_RE.match(item["filename"]) + assert item["location"] in ("local", "nas") + assert item["size"] > 0 + + +def test_trigger_backup(client): + resp = client.post("/api/backups", headers=AUTH) + assert resp.status_code == 200 + assert resp.get_json()["ok"] is True + + +def test_restore_invalid_filename(client): + resp = client.post("/api/backups/evil.db/restore", headers=AUTH) + assert resp.status_code == 400 + + +def test_restore_missing_backup(client): + resp = client.post("/api/backups/pinvault-backup-99999999-999999.db/restore", headers=AUTH) + assert resp.status_code == 404 + + +def test_restore_valid_backup(client): + filename = client.get("/api/backups", headers=AUTH).get_json()[0]["filename"] + resp = client.post(f"/api/backups/{filename}/restore", headers=AUTH) + assert resp.status_code == 200 + assert resp.get_json()["restored"] == filename + + +def test_list_pins_shape(client): + data = create_pin(client, label="shape", lock_days=10) + entry = next(p for p in client.get("/api/pins", headers=AUTH).get_json() if p["id"] == data["id"]) + for key in ("id", "label", "created_at", "lock_until", "locked", "days_remaining", + "revealed", "bypassed_count", "remaining_codes"): + assert key in entry, f"missing key {key}" + assert entry["locked"] is True + assert entry["remaining_codes"] == 4 + assert "pin" not in entry and "recovery_codes" not in entry diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..3d173c2 --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,101 @@ +import inspect +import os +import secrets + +import app as pinvault + +from conftest import AUTH, create_pin + + +def test_api_key_length(): + assert len(pinvault.API_KEY) >= 32 + + +def test_bootstrap_returns_bool(): + assert isinstance(pinvault._bootstrap_ok, bool) + + +def test_no_sys_exit_in_bootstrap(): + src = inspect.getsource(pinvault.bootstrap) + assert "sys.exit" not in src, "bootstrap must not call sys.exit" + + +def test_nas_backup_configurable(): + assert pinvault.NAS_BACKUP_DIR == os.environ.get("PINVAULT_NAS_BACKUP_DIR", "") + + +def test_hash_val_per_row_salt(): + assert pinvault.hash_val("1234", "salt-a") != pinvault.hash_val("1234", "salt-b") + # Legacy rows (salt NULL) keep working with the historic salt. + assert pinvault.hash_val("1234") == pinvault.hash_val("1234", pinvault.LEGACY_SALT) + + +def test_recover_pin_roundtrip(): + for pin in ("0000", "0042", "1000", "9999"): + salt = secrets.token_hex(16) + assert pinvault.recover_pin(pinvault.hash_val(pin, salt), salt) == pin + + +def test_new_pins_have_per_row_salt(client): + data = create_pin(client, label="salted") + with pinvault.get_conn() as conn: + row = conn.execute("SELECT salt FROM pins WHERE id = ?", (data["id"],)).fetchone() + assert row["salt"], "pin row must carry a salt" + assert row["salt"] != pinvault.LEGACY_SALT + codes = conn.execute( + "SELECT salt FROM recovery_codes WHERE pin_id = ?", (data["id"],) + ).fetchall() + assert len(codes) == 4 + assert all(c["salt"] for c in codes) + + +def test_pin_generation_zero_padded(client): + for _ in range(10): + data = create_pin(client) + assert len(data["pin"]) == 4 + assert data["pin"] == f"{int(data['pin']):04d}" + + +def test_rate_limit_blocks_after_max(client): + data = create_pin(client, lock_days=30) + url = f"/api/pins/{data['id']}/access" + for i in range(pinvault._PIN_MAX_ATTEMPTS): + resp = client.post(url, json={}, headers=AUTH) + assert resp.status_code == 423, f"attempt {i + 1} should be allowed" + resp = client.post(url, json={}, headers=AUTH) + assert resp.status_code == 429 + + +def test_rate_limit_resets_on_success(client): + data = create_pin(client, lock_days=0) + url = f"/api/pins/{data['id']}/access" + for _ in range(pinvault._PIN_MAX_ATTEMPTS + 3): + assert client.post(url, json={}, headers=AUTH).status_code == 200 + + +def test_rate_limit_independent_per_pin(client): + a = create_pin(client, label="a", lock_days=30) + b = create_pin(client, label="b", lock_days=30) + for _ in range(pinvault._PIN_MAX_ATTEMPTS + 1): + client.post(f"/api/pins/{a['id']}/access", json={}, headers=AUTH) + resp = client.post(f"/api/pins/{b['id']}/access", json={}, headers=AUTH) + assert resp.status_code == 423 + + +def test_check_pin_rate_limit_unit(): + pinvault._pin_attempt_locks.clear() + for i in range(pinvault._PIN_MAX_ATTEMPTS): + allowed, _ = pinvault._check_pin_rate_limit(999) + assert allowed, f"Attempt {i + 1} should be allowed" + allowed, remaining = pinvault._check_pin_rate_limit(999) + assert not allowed + assert remaining > 0 + pinvault._reset_pin_rate_limit(999) + allowed, _ = pinvault._check_pin_rate_limit(999) + assert allowed + pinvault._pin_attempt_locks.clear() + + +def test_backup_dirs_excludes_empty_nas(): + if not pinvault.NAS_BACKUP_DIR: + assert pinvault._backup_dirs() == [pinvault.LOCAL_BACKUP_DIR]