Compare commits
No commits in common. "a6a4ffdf7c72f98ab82d5a57e06de207c3276ae0" and "17e7c06b591cb5ffa8a0ed04a8fa759e900e8ec9" have entirely different histories.
a6a4ffdf7c
...
17e7c06b59
@ -1,9 +0,0 @@
|
|||||||
.git
|
|
||||||
data/
|
|
||||||
tests/
|
|
||||||
__pycache__/
|
|
||||||
*.pyc
|
|
||||||
.pytest_cache/
|
|
||||||
*.egg-info/
|
|
||||||
.env
|
|
||||||
.venv/
|
|
||||||
14
.env.example
14
.env.example
@ -1,14 +0,0 @@
|
|||||||
# 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=
|
|
||||||
@ -1,143 +0,0 @@
|
|||||||
name: CI
|
|
||||||
|
|
||||||
on:
|
|
||||||
push:
|
|
||||||
branches: [main, master]
|
|
||||||
pull_request:
|
|
||||||
branches: [main, master]
|
|
||||||
|
|
||||||
env:
|
|
||||||
GITEA_URL: https://git.example.com
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
lint:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
container:
|
|
||||||
image: gitea-job-image
|
|
||||||
steps:
|
|
||||||
- name: Clone repo
|
|
||||||
run: |
|
|
||||||
rm -rf $GITHUB_WORKSPACE/*
|
|
||||||
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
|
|
||||||
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
|
|
||||||
|
|
||||||
- name: Run ruff (Python lint)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f pyproject.toml ]]; then
|
|
||||||
pip3 install ruff
|
|
||||||
ruff check .
|
|
||||||
else
|
|
||||||
echo "No Python project detected, skipping ruff"
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Run npm lint (JS/TS)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f package.json ]]; then
|
|
||||||
npm ci
|
|
||||||
npm run lint --if-present || true
|
|
||||||
else
|
|
||||||
echo "No Node.js project detected, skipping npm lint"
|
|
||||||
fi
|
|
||||||
|
|
||||||
test:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
container:
|
|
||||||
image: gitea-job-image
|
|
||||||
steps:
|
|
||||||
- name: Clone repo
|
|
||||||
run: |
|
|
||||||
rm -rf $GITHUB_WORKSPACE/*
|
|
||||||
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
|
|
||||||
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
|
|
||||||
|
|
||||||
- name: Run pytest (Python)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f pyproject.toml ]]; then
|
|
||||||
python3 -m pip install --upgrade pip
|
|
||||||
pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true
|
|
||||||
pip3 install pytest
|
|
||||||
pytest tests/ -v --tb=short 2>/dev/null || true
|
|
||||||
else
|
|
||||||
echo "No Python project detected, skipping pytest"
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Run npm test (JS/TS)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f package.json ]]; then
|
|
||||||
npm ci
|
|
||||||
npm run test --if-present || true
|
|
||||||
else
|
|
||||||
echo "No Node.js project detected, skipping npm test"
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Run Go tests
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f go.mod ]]; then
|
|
||||||
go test ./...
|
|
||||||
else
|
|
||||||
echo "No Go project detected, skipping go test"
|
|
||||||
fi
|
|
||||||
|
|
||||||
docker-build:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
steps:
|
|
||||||
- name: Clone repo
|
|
||||||
run: |
|
|
||||||
rm -rf $GITHUB_WORKSPACE/*
|
|
||||||
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
|
|
||||||
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
|
|
||||||
|
|
||||||
- name: Build Docker image
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f Dockerfile ]]; then
|
|
||||||
docker build -t $GITHUB_REPOSITORY:test .
|
|
||||||
else
|
|
||||||
echo "No Dockerfile found, skipping docker build"
|
|
||||||
fi
|
|
||||||
|
|
||||||
security:
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
container:
|
|
||||||
image: gitea-job-image
|
|
||||||
steps:
|
|
||||||
- name: Clone repo
|
|
||||||
run: |
|
|
||||||
rm -rf $GITHUB_WORKSPACE/*
|
|
||||||
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
|
|
||||||
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
|
|
||||||
|
|
||||||
- name: Run bandit (Python SAST)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f pyproject.toml ]]; then
|
|
||||||
pip3 install bandit
|
|
||||||
bandit -r . --severity-level high --confidence-level high --exclude tests/,test_*
|
|
||||||
else
|
|
||||||
echo "No Python project detected, skipping bandit"
|
|
||||||
fi
|
|
||||||
|
|
||||||
- name: Run npm audit (JS/TS)
|
|
||||||
if: always()
|
|
||||||
run: |
|
|
||||||
if [[ -f package.json ]]; then
|
|
||||||
npm ci
|
|
||||||
npm audit --audit-level=high 2>/dev/null || echo "npm audit: vulnerabilities found (non-blocking)"
|
|
||||||
else
|
|
||||||
echo "No Node.js project detected, skipping npm audit"
|
|
||||||
fi
|
|
||||||
|
|
||||||
build-result:
|
|
||||||
needs: [lint, test, docker-build, security]
|
|
||||||
runs-on: ubuntu-latest
|
|
||||||
container:
|
|
||||||
image: gitea-job-image
|
|
||||||
if: always()
|
|
||||||
steps:
|
|
||||||
- name: Summary
|
|
||||||
run: echo "All CI checks completed"
|
|
||||||
7
.gitignore
vendored
7
.gitignore
vendored
@ -1,7 +0,0 @@
|
|||||||
.env
|
|
||||||
data/
|
|
||||||
__pycache__/
|
|
||||||
*.pyc
|
|
||||||
.pytest_cache/
|
|
||||||
*.egg-info/
|
|
||||||
.venv/
|
|
||||||
21
LICENSE
21
LICENSE
@ -1,21 +0,0 @@
|
|||||||
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.
|
|
||||||
107
README.md
107
README.md
@ -1,107 +0,0 @@
|
|||||||
# 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/<id>/access` | Reveal PIN. Body: `{}` after lock expiry, or `{"bypass_code": "..."}` to unlock early. 423 while locked, 429 when rate-limited |
|
|
||||||
| DELETE | `/api/pins/<id>` | 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/<filename>/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).
|
|
||||||
199
app.py
199
app.py
@ -1,17 +1,12 @@
|
|||||||
from flask import Flask, render_template, request, jsonify
|
from flask import Flask, render_template, request, jsonify
|
||||||
import hashlib
|
import hashlib
|
||||||
import hmac
|
|
||||||
import os
|
import os
|
||||||
import re
|
import random
|
||||||
import secrets
|
import secrets
|
||||||
import shutil
|
import shutil
|
||||||
import string
|
import string
|
||||||
import sys
|
|
||||||
import tempfile
|
|
||||||
import threading
|
import threading
|
||||||
import time
|
|
||||||
from datetime import datetime, timedelta, timezone
|
from datetime import datetime, timedelta, timezone
|
||||||
from functools import wraps
|
|
||||||
|
|
||||||
import pysqlcipher3.dbapi2 as sqlcipher
|
import pysqlcipher3.dbapi2 as sqlcipher
|
||||||
|
|
||||||
@ -19,63 +14,24 @@ app = Flask(__name__)
|
|||||||
app.secret_key = secrets.token_hex(32)
|
app.secret_key = secrets.token_hex(32)
|
||||||
|
|
||||||
DB_PATH = os.environ.get("PINVAULT_DB", "/data/pinvault.db")
|
DB_PATH = os.environ.get("PINVAULT_DB", "/data/pinvault.db")
|
||||||
LOCAL_BACKUP_DIR = os.environ.get("PINVAULT_LOCAL_BACKUP_DIR", "/data/backups")
|
LOCAL_BACKUP_DIR = "/data/backups"
|
||||||
NAS_BACKUP_DIR = os.environ.get("PINVAULT_NAS_BACKUP_DIR", "")
|
NAS_BACKUP_DIR = "/mnt/aidata/pinvault"
|
||||||
BACKUP_INTERVAL = int(os.environ.get("PINVAULT_BACKUP_INTERVAL", "3600"))
|
BACKUP_INTERVAL = 3600
|
||||||
MAX_BACKUPS = int(os.environ.get("PINVAULT_MAX_BACKUPS", "168"))
|
MAX_BACKUPS = 168
|
||||||
RECOVERY_CODE_COUNT = 4
|
RECOVERY_CODE_COUNT = 4
|
||||||
RECOVERY_CODE_LENGTH = 64
|
RECOVERY_CODE_LENGTH = 64
|
||||||
ALPHANUMERIC = string.ascii_letters + string.digits
|
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 _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 auth[7:] != API_KEY:
|
|
||||||
return jsonify({"error": "Unauthorized"}), 401
|
|
||||||
return f(*args, **kwargs)
|
|
||||||
return decorated
|
|
||||||
|
|
||||||
|
|
||||||
def dict_factory(cursor, row):
|
def dict_factory(cursor, row):
|
||||||
d = {}
|
d = {}
|
||||||
for idx, col in enumerate(cursor.description):
|
for idx, col in enumerate(cursor.description):
|
||||||
d[col[0].lower()] = row[idx]
|
d[col[0]] = row[idx]
|
||||||
return d
|
return d
|
||||||
|
|
||||||
|
|
||||||
def _derive_sqlcipher_key(bcrypt_hash: str) -> str:
|
def _derive_sqlcipher_key(bcrypt_hash: str) -> str:
|
||||||
"""Derive SQLCipher key from bcrypt hash using PBKDF2 (#5)."""
|
return hashlib.sha256(bcrypt_hash.encode()).hexdigest()[:64]
|
||||||
salt = b"pinvault-sqlcipher-v1" # Fixed salt - bcrypt hash provides entropy
|
|
||||||
derived = hashlib.pbkdf2_hmac("sha256", bcrypt_hash.encode(), salt, 100_000)
|
|
||||||
return derived.hex()[:64]
|
|
||||||
|
|
||||||
|
|
||||||
def get_conn():
|
def get_conn():
|
||||||
@ -87,13 +43,8 @@ def get_conn():
|
|||||||
return conn
|
return conn
|
||||||
|
|
||||||
|
|
||||||
LEGACY_SALT = "pinvault-recovery-v1"
|
def hash_val(value):
|
||||||
|
return hashlib.sha256(value.encode()).hexdigest()
|
||||||
|
|
||||||
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():
|
def generate_recovery_code():
|
||||||
@ -131,12 +82,6 @@ def init_db():
|
|||||||
conn.commit()
|
conn.commit()
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
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:
|
def _backup_db(dest_path: str) -> bool:
|
||||||
@ -149,22 +94,6 @@ def _backup_db(dest_path: str) -> bool:
|
|||||||
return False
|
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):
|
def _cleanup_backups(directory: str):
|
||||||
try:
|
try:
|
||||||
files = sorted(
|
files = sorted(
|
||||||
@ -177,35 +106,28 @@ def _cleanup_backups(directory: str):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def _backup_dirs():
|
|
||||||
return [d for d in (LOCAL_BACKUP_DIR, NAS_BACKUP_DIR) if d]
|
|
||||||
|
|
||||||
|
|
||||||
def _do_backup():
|
def _do_backup():
|
||||||
ts = now_utc().strftime("%Y%m%d-%H%M%S")
|
ts = now_utc().strftime("%Y%m%d-%H%M%S")
|
||||||
filename = f"pinvault-backup-{ts}.db"
|
filename = f"pinvault-backup-{ts}.db"
|
||||||
|
|
||||||
os.makedirs(LOCAL_BACKUP_DIR, exist_ok=True)
|
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)
|
local_path = os.path.join(LOCAL_BACKUP_DIR, filename)
|
||||||
|
nas_path = os.path.join(NAS_BACKUP_DIR, filename)
|
||||||
|
|
||||||
ok = _backup_db(local_path)
|
ok = _backup_db(local_path)
|
||||||
if ok:
|
if ok:
|
||||||
|
shutil.copy2(local_path, nas_path)
|
||||||
_cleanup_backups(LOCAL_BACKUP_DIR)
|
_cleanup_backups(LOCAL_BACKUP_DIR)
|
||||||
if NAS_BACKUP_DIR:
|
_cleanup_backups(NAS_BACKUP_DIR)
|
||||||
os.makedirs(NAS_BACKUP_DIR, exist_ok=True)
|
return ok
|
||||||
nas_path = os.path.join(NAS_BACKUP_DIR, filename)
|
|
||||||
shutil.copy2(local_path, nas_path)
|
|
||||||
_cleanup_backups(NAS_BACKUP_DIR)
|
|
||||||
return True
|
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def _backup_loop():
|
def _backup_loop():
|
||||||
while True:
|
while True:
|
||||||
try:
|
_do_backup()
|
||||||
_do_backup()
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Backup failed: {e}", file=sys.stderr, flush=True)
|
|
||||||
timer = threading.Timer(BACKUP_INTERVAL, _backup_loop)
|
timer = threading.Timer(BACKUP_INTERVAL, _backup_loop)
|
||||||
timer.daemon = True
|
timer.daemon = True
|
||||||
timer.start()
|
timer.start()
|
||||||
@ -214,7 +136,7 @@ def _backup_loop():
|
|||||||
|
|
||||||
def _get_backup_list():
|
def _get_backup_list():
|
||||||
result = []
|
result = []
|
||||||
for d in _backup_dirs():
|
for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]:
|
||||||
location = "local" if d == LOCAL_BACKUP_DIR else "nas"
|
location = "local" if d == LOCAL_BACKUP_DIR else "nas"
|
||||||
try:
|
try:
|
||||||
for f in sorted(os.listdir(d)):
|
for f in sorted(os.listdir(d)):
|
||||||
@ -239,13 +161,11 @@ def index():
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/status", methods=["GET"])
|
@app.route("/api/status", methods=["GET"])
|
||||||
@require_api_key
|
|
||||||
def api_status():
|
def api_status():
|
||||||
return jsonify({"locked": not getattr(app, "master_key", None)})
|
return jsonify({"locked": not getattr(app, "master_key", None)})
|
||||||
|
|
||||||
|
|
||||||
@app.route("/api/backups", methods=["GET"])
|
@app.route("/api/backups", methods=["GET"])
|
||||||
@require_api_key
|
|
||||||
def api_list_backups():
|
def api_list_backups():
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
@ -253,7 +173,6 @@ def api_list_backups():
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/backups", methods=["POST"])
|
@app.route("/api/backups", methods=["POST"])
|
||||||
@require_api_key
|
|
||||||
def api_trigger_backup():
|
def api_trigger_backup():
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
@ -262,15 +181,15 @@ def api_trigger_backup():
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/backups/<filename>/restore", methods=["POST"])
|
@app.route("/api/backups/<filename>/restore", methods=["POST"])
|
||||||
@require_api_key
|
|
||||||
def api_restore_backup(filename):
|
def api_restore_backup(filename):
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
|
import re
|
||||||
if not re.match(r'^pinvault-backup-\d{8}-\d{6}\.db$', filename):
|
if not re.match(r'^pinvault-backup-\d{8}-\d{6}\.db$', filename):
|
||||||
return jsonify({"error": "Invalid backup filename."}), 400
|
return jsonify({"error": "Invalid backup filename."}), 400
|
||||||
|
|
||||||
src = None
|
src = None
|
||||||
for d in _backup_dirs():
|
for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]:
|
||||||
candidate = os.path.join(d, filename)
|
candidate = os.path.join(d, filename)
|
||||||
if os.path.exists(candidate):
|
if os.path.exists(candidate):
|
||||||
src = candidate
|
src = candidate
|
||||||
@ -287,12 +206,11 @@ def api_restore_backup(filename):
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
return jsonify({"error": f"Backup verification failed: {e}"}), 500
|
return jsonify({"error": f"Backup verification failed: {e}"}), 500
|
||||||
|
|
||||||
_atomic_copy(src, DB_PATH)
|
shutil.copy2(src, DB_PATH)
|
||||||
return jsonify({"ok": True, "restored": filename})
|
return jsonify({"ok": True, "restored": filename})
|
||||||
|
|
||||||
|
|
||||||
@app.route("/api/pins", methods=["GET"])
|
@app.route("/api/pins", methods=["GET"])
|
||||||
@require_api_key
|
|
||||||
def api_list_pins():
|
def api_list_pins():
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
@ -325,19 +243,17 @@ def api_list_pins():
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/pins", methods=["POST"])
|
@app.route("/api/pins", methods=["POST"])
|
||||||
@require_api_key
|
|
||||||
def api_add_pin():
|
def api_add_pin():
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
data = request.get_json() or {}
|
data = request.get_json()
|
||||||
label = data.get("label", "").strip()
|
label = data.get("label", "").strip()
|
||||||
lock_days = data.get("lock_days", 30)
|
lock_days = data.get("lock_days", 30)
|
||||||
if isinstance(lock_days, bool) or not isinstance(lock_days, int) or lock_days < 0:
|
if lock_days < 0:
|
||||||
return jsonify({"error": "lock_days must be an integer >= 0"}), 400
|
return jsonify({"error": "lock_days must be >= 0"}), 400
|
||||||
|
|
||||||
pin = f"{secrets.randbelow(10000):04d}"
|
pin = f"{random.randint(0, 9999):04d}"
|
||||||
pin_salt = secrets.token_hex(16)
|
pin_hash = hash_val(pin)
|
||||||
pin_hash = hash_val(pin, pin_salt)
|
|
||||||
now = now_utc()
|
now = now_utc()
|
||||||
lock_until = (now + timedelta(days=lock_days)).isoformat()
|
lock_until = (now + timedelta(days=lock_days)).isoformat()
|
||||||
|
|
||||||
@ -345,15 +261,14 @@ def api_add_pin():
|
|||||||
|
|
||||||
with get_conn() as conn:
|
with get_conn() as conn:
|
||||||
cur = conn.execute(
|
cur = conn.execute(
|
||||||
"INSERT INTO pins (label, pin_hash, salt, created_at, lock_until) VALUES (?, ?, ?, ?, ?)",
|
"INSERT INTO pins (label, pin_hash, created_at, lock_until) VALUES (?, ?, ?, ?)",
|
||||||
(label, pin_hash, pin_salt, now.isoformat(), lock_until),
|
(label, pin_hash, now.isoformat(), lock_until),
|
||||||
)
|
)
|
||||||
pin_id = cur.lastrowid
|
pin_id = cur.lastrowid
|
||||||
for code in codes:
|
for code in codes:
|
||||||
code_salt = secrets.token_hex(16)
|
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"INSERT INTO recovery_codes (pin_id, code_hash, salt) VALUES (?, ?, ?)",
|
"INSERT INTO recovery_codes (pin_id, code_hash) VALUES (?, ?)",
|
||||||
(pin_id, hash_val(code, code_salt), code_salt),
|
(pin_id, hash_val(code)),
|
||||||
)
|
)
|
||||||
conn.commit()
|
conn.commit()
|
||||||
|
|
||||||
@ -367,13 +282,9 @@ def api_add_pin():
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/pins/<int:pin_id>/access", methods=["POST"])
|
@app.route("/api/pins/<int:pin_id>/access", methods=["POST"])
|
||||||
@require_api_key
|
|
||||||
def api_access_pin(pin_id):
|
def api_access_pin(pin_id):
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
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 {}
|
data = request.get_json() or {}
|
||||||
bypass_code = data.get("bypass_code", "").strip() if data.get("bypass_code") else None
|
bypass_code = data.get("bypass_code", "").strip() if data.get("bypass_code") else None
|
||||||
|
|
||||||
@ -386,14 +297,11 @@ def api_access_pin(pin_id):
|
|||||||
now = now_utc()
|
now = now_utc()
|
||||||
|
|
||||||
if bypass_code:
|
if bypass_code:
|
||||||
code_row = None
|
code_hash = hash_val(bypass_code)
|
||||||
for candidate in conn.execute(
|
code_row = conn.execute(
|
||||||
"SELECT * FROM recovery_codes WHERE pin_id = ? AND used = 0",
|
"SELECT * FROM recovery_codes WHERE pin_id = ? AND code_hash = ? AND used = 0",
|
||||||
(pin_id,),
|
(pin_id, code_hash),
|
||||||
).fetchall():
|
).fetchone()
|
||||||
if hmac.compare_digest(hash_val(bypass_code, candidate["salt"]), candidate["code_hash"]):
|
|
||||||
code_row = candidate
|
|
||||||
break
|
|
||||||
if not code_row:
|
if not code_row:
|
||||||
return jsonify({"error": "Invalid or already-used recovery code"}), 403
|
return jsonify({"error": "Invalid or already-used recovery code"}), 403
|
||||||
conn.execute(
|
conn.execute(
|
||||||
@ -405,15 +313,13 @@ def api_access_pin(pin_id):
|
|||||||
(now.isoformat(), pin_id),
|
(now.isoformat(), pin_id),
|
||||||
)
|
)
|
||||||
conn.commit()
|
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({"pin": pin})
|
||||||
|
|
||||||
if now >= lock_until:
|
if now >= lock_until:
|
||||||
conn.execute("UPDATE pins SET revealed = 1 WHERE id = ?", (pin_id,))
|
conn.execute("UPDATE pins SET revealed = 1 WHERE id = ?", (pin_id,))
|
||||||
conn.commit()
|
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({"pin": pin})
|
||||||
|
|
||||||
return jsonify({"error": "PIN is locked. Use a recovery code to bypass."}), 423
|
return jsonify({"error": "PIN is locked. Use a recovery code to bypass."}), 423
|
||||||
@ -422,7 +328,6 @@ def api_access_pin(pin_id):
|
|||||||
|
|
||||||
|
|
||||||
@app.route("/api/pins/<int:pin_id>", methods=["DELETE"])
|
@app.route("/api/pins/<int:pin_id>", methods=["DELETE"])
|
||||||
@require_api_key
|
|
||||||
def api_delete_pin(pin_id):
|
def api_delete_pin(pin_id):
|
||||||
if not getattr(app, "master_key", None):
|
if not getattr(app, "master_key", None):
|
||||||
return jsonify({"error": "Vault not configured."}), 401
|
return jsonify({"error": "Vault not configured."}), 401
|
||||||
@ -433,16 +338,15 @@ def api_delete_pin(pin_id):
|
|||||||
if not row["revealed"]:
|
if not row["revealed"]:
|
||||||
return jsonify({"error": "Cannot delete unrevealed PIN. Access it first."}), 409
|
return jsonify({"error": "Cannot delete unrevealed PIN. Access it first."}), 409
|
||||||
conn.execute("DELETE FROM recovery_codes WHERE pin_id = ?", (pin_id,))
|
conn.execute("DELETE FROM recovery_codes WHERE pin_id = ?", (pin_id,))
|
||||||
conn.execute("DELETE FROM pins WHERE id = ?", (pin_id,))
|
cur = conn.execute("DELETE FROM pins WHERE id = ?", (pin_id,))
|
||||||
conn.commit()
|
conn.commit()
|
||||||
return jsonify({"ok": True})
|
return jsonify({"ok": True})
|
||||||
|
|
||||||
|
|
||||||
def recover_pin(pin_hash, salt=None):
|
def recover_pin(pin_hash):
|
||||||
"""Recover a 4-digit PIN by exhaustive search (design: PINs are 10^4 space)."""
|
|
||||||
for i in range(10000):
|
for i in range(10000):
|
||||||
candidate = f"{i:04d}"
|
candidate = f"{i:04d}"
|
||||||
if hmac.compare_digest(hash_val(candidate, salt), pin_hash):
|
if hash_val(candidate) == pin_hash:
|
||||||
return candidate
|
return candidate
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@ -460,7 +364,7 @@ def _verify_db():
|
|||||||
|
|
||||||
def _latest_backup():
|
def _latest_backup():
|
||||||
candidates = []
|
candidates = []
|
||||||
for d in _backup_dirs():
|
for d in [LOCAL_BACKUP_DIR, NAS_BACKUP_DIR]:
|
||||||
try:
|
try:
|
||||||
for f in os.listdir(d):
|
for f in os.listdir(d):
|
||||||
if f.startswith("pinvault-backup-") and f.endswith(".db"):
|
if f.startswith("pinvault-backup-") and f.endswith(".db"):
|
||||||
@ -493,46 +397,39 @@ def _auto_restore():
|
|||||||
print(f"Backup {backup} also unreadable. Cannot auto-restore.", file=sys.stderr, flush=True)
|
print(f"Backup {backup} also unreadable. Cannot auto-restore.", file=sys.stderr, flush=True)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
_atomic_copy(backup, DB_PATH)
|
shutil.copy2(backup, DB_PATH)
|
||||||
print(f"Restored DB from backup: {backup}", flush=True)
|
print(f"Restored DB from backup: {backup}", flush=True)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
_bootstrap_ok = False
|
|
||||||
|
|
||||||
|
|
||||||
def bootstrap():
|
def bootstrap():
|
||||||
global _bootstrap_ok
|
import sys
|
||||||
bcrypt_hash = os.environ.get("PINVAULT_MASTER_HASH", "").strip()
|
bcrypt_hash = os.environ.get("PINVAULT_MASTER_HASH", "").strip()
|
||||||
os.environ.pop("PINVAULT_MASTER_HASH", None)
|
os.environ.pop("PINVAULT_MASTER_HASH", None)
|
||||||
|
|
||||||
if not bcrypt_hash:
|
if not bcrypt_hash:
|
||||||
print("ERROR: PINVAULT_MASTER_HASH environment variable is required.", file=sys.stderr)
|
print("ERROR: PINVAULT_MASTER_HASH environment variable is required.", file=sys.stderr)
|
||||||
print("Generate: python3 -c \"import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())\"", file=sys.stderr)
|
print("Generate one with: python3 -c \"import bcrypt; print(bcrypt.hashpw(b'YOUR_PASSWORD', bcrypt.gensalt()).decode())\"", file=sys.stderr)
|
||||||
return False
|
sys.exit(1)
|
||||||
|
|
||||||
if not bcrypt_hash.startswith("$2"):
|
if not bcrypt_hash.startswith("$2"):
|
||||||
print("ERROR: PINVAULT_MASTER_HASH must be a bcrypt hash (starts with $2b$ or $2a$)", file=sys.stderr)
|
print("ERROR: PINVAULT_MASTER_HASH must be a bcrypt hash (starts with $2b$ or $2a$)", file=sys.stderr)
|
||||||
return False
|
sys.exit(1)
|
||||||
|
|
||||||
app.master_key = _derive_sqlcipher_key(bcrypt_hash)
|
app.master_key = _derive_sqlcipher_key(bcrypt_hash)
|
||||||
|
|
||||||
if os.path.exists(DB_PATH):
|
if os.path.exists(DB_PATH):
|
||||||
if not _auto_restore():
|
if not _auto_restore():
|
||||||
print("FATAL: Cannot open database and no valid backup found.", file=sys.stderr, flush=True)
|
print("FATAL: Cannot open database and no valid backup found.", file=sys.stderr, flush=True)
|
||||||
return False
|
sys.exit(1)
|
||||||
|
|
||||||
init_db()
|
init_db()
|
||||||
_do_backup()
|
_do_backup()
|
||||||
t = threading.Thread(target=_backup_loop, daemon=True)
|
t = threading.Thread(target=_backup_loop, daemon=True)
|
||||||
t.start()
|
t.start()
|
||||||
_bootstrap_ok = True
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
_bootstrap_ok = bootstrap()
|
bootstrap()
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
if not _bootstrap_ok:
|
app.run(host="0.0.0.0", port=8765)
|
||||||
sys.exit(1)
|
|
||||||
app.run(host="0.0.0.0", port=8765)
|
|
||||||
@ -4,8 +4,11 @@ services:
|
|||||||
ports:
|
ports:
|
||||||
- "8765:8765"
|
- "8765:8765"
|
||||||
volumes:
|
volumes:
|
||||||
- ./data:/data
|
- pinvault-data:/data
|
||||||
- /mnt/aidata:/mnt/aidata
|
- /mnt/aidata:/mnt/aidata
|
||||||
environment:
|
environment:
|
||||||
- PINVAULT_MASTER_HASH=${PINVAULT_MASTER_HASH}
|
- PINVAULT_MASTER_HASH=${PINVAULT_MASTER_HASH}
|
||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
|
|
||||||
|
volumes:
|
||||||
|
pinvault-data:
|
||||||
@ -1,36 +0,0 @@
|
|||||||
[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"]
|
|
||||||
111
static/app.js
111
static/app.js
@ -133,84 +133,51 @@ document.getElementById("add-form").addEventListener("submit", async (e) => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
async function accessPin(id) {
|
function accessPin(id) {
|
||||||
const title = document.getElementById("modal-title");
|
const title = document.getElementById("modal-title");
|
||||||
const body = document.getElementById("modal-body");
|
const body = document.getElementById("modal-body");
|
||||||
title.textContent = `Access PIN #${id}`;
|
title.textContent = `Access PIN #${id}`;
|
||||||
|
body.innerHTML = `
|
||||||
|
<p class="muted" style="margin-bottom:1rem;font-size:0.85rem">
|
||||||
|
Enter a recovery code to bypass the lock:
|
||||||
|
</p>
|
||||||
|
<form id="access-form">
|
||||||
|
<div class="form-group">
|
||||||
|
<textarea id="bypass-input" placeholder="Type 64-character recovery code here..." rows="2" required onpaste="return false"></textarea>
|
||||||
|
</div>
|
||||||
|
<div id="access-error" class="error-msg" style="margin-bottom:0.5rem"></div>
|
||||||
|
<button type="submit" class="btn btn-primary" style="width:100%">Unlock PIN</button>
|
||||||
|
</form>`;
|
||||||
|
document.getElementById("modal-overlay").classList.remove("hidden");
|
||||||
|
|
||||||
const pins = await api("/api/pins");
|
const bypassInput = document.getElementById("bypass-input");
|
||||||
const pin = pins.find(p => p.id === id);
|
bypassInput.addEventListener("paste", (e) => { e.preventDefault(); });
|
||||||
|
bypassInput.addEventListener("drop", (e) => { e.preventDefault(); });
|
||||||
|
|
||||||
if (pin && !pin.locked) {
|
document.getElementById("access-form").addEventListener("submit", async (e) => {
|
||||||
body.innerHTML = `
|
e.preventDefault();
|
||||||
<p class="muted" style="margin-bottom:1rem;font-size:0.85rem">
|
const code = bypassInput.value.trim();
|
||||||
This PIN is unlocked and ready to view.
|
const errEl = document.getElementById("access-error");
|
||||||
</p>
|
errEl.textContent = "";
|
||||||
<button id="reveal-btn" class="btn btn-primary" style="width:100%">Reveal PIN</button>`;
|
|
||||||
document.getElementById("modal-overlay").classList.remove("hidden");
|
|
||||||
|
|
||||||
document.getElementById("reveal-btn").addEventListener("click", async () => {
|
try {
|
||||||
try {
|
const result = await api(`/api/pins/${id}/access`, {
|
||||||
const result = await api(`/api/pins/${id}/access`, {
|
method: "POST",
|
||||||
method: "POST",
|
body: JSON.stringify({ bypass_code: code }),
|
||||||
body: JSON.stringify({}),
|
});
|
||||||
});
|
body.innerHTML = `
|
||||||
body.innerHTML = `
|
<div class="text-center">
|
||||||
<div class="text-center">
|
<p class="muted" style="margin-bottom:0.3rem">Your PIN</p>
|
||||||
<p class="muted" style="margin-bottom:0.3rem">Your PIN</p>
|
<div class="pin-display">${result.pin}</div>
|
||||||
<div class="pin-display">${result.pin}</div>
|
<button class="btn btn-sm" onclick="copyTextDirect('${result.pin}')" style="margin-top:0.5rem">
|
||||||
<button class="btn btn-sm" onclick="copyTextDirect('${result.pin}')" style="margin-top:0.5rem">
|
Copy PIN 📋
|
||||||
Copy PIN 📋
|
</button>
|
||||||
</button>
|
</div>`;
|
||||||
</div>`;
|
loadPins();
|
||||||
loadPins();
|
} catch (err) {
|
||||||
} catch (err) {
|
errEl.textContent = err.message;
|
||||||
body.innerHTML = `<p class="error-msg">${esc(err.message)}</p>`;
|
}
|
||||||
}
|
});
|
||||||
});
|
|
||||||
} else {
|
|
||||||
body.innerHTML = `
|
|
||||||
<p class="muted" style="margin-bottom:1rem;font-size:0.85rem">
|
|
||||||
Enter a recovery code to bypass the lock:
|
|
||||||
</p>
|
|
||||||
<form id="access-form">
|
|
||||||
<div class="form-group">
|
|
||||||
<textarea id="bypass-input" placeholder="Type 64-character recovery code here..." rows="2" required onpaste="return false"></textarea>
|
|
||||||
</div>
|
|
||||||
<div id="access-error" class="error-msg" style="margin-bottom:0.5rem"></div>
|
|
||||||
<button type="submit" class="btn btn-primary" style="width:100%">Unlock PIN</button>
|
|
||||||
</form>`;
|
|
||||||
document.getElementById("modal-overlay").classList.remove("hidden");
|
|
||||||
|
|
||||||
const bypassInput = document.getElementById("bypass-input");
|
|
||||||
bypassInput.addEventListener("paste", (e) => { e.preventDefault(); });
|
|
||||||
bypassInput.addEventListener("drop", (e) => { e.preventDefault(); });
|
|
||||||
|
|
||||||
document.getElementById("access-form").addEventListener("submit", async (e) => {
|
|
||||||
e.preventDefault();
|
|
||||||
const code = bypassInput.value.trim();
|
|
||||||
const errEl = document.getElementById("access-error");
|
|
||||||
errEl.textContent = "";
|
|
||||||
|
|
||||||
try {
|
|
||||||
const result = await api(`/api/pins/${id}/access`, {
|
|
||||||
method: "POST",
|
|
||||||
body: JSON.stringify({ bypass_code: code }),
|
|
||||||
});
|
|
||||||
body.innerHTML = `
|
|
||||||
<div class="text-center">
|
|
||||||
<p class="muted" style="margin-bottom:0.3rem">Your PIN</p>
|
|
||||||
<div class="pin-display">${result.pin}</div>
|
|
||||||
<button class="btn btn-sm" onclick="copyTextDirect('${result.pin}')" style="margin-top:0.5rem">
|
|
||||||
Copy PIN 📋
|
|
||||||
</button>
|
|
||||||
</div>`;
|
|
||||||
loadPins();
|
|
||||||
} catch (err) {
|
|
||||||
errEl.textContent = err.message;
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async function deletePin(id, label) {
|
async function deletePin(id, label) {
|
||||||
|
|||||||
@ -1,89 +0,0 @@
|
|||||||
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()
|
|
||||||
@ -1,157 +0,0 @@
|
|||||||
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
|
|
||||||
@ -1,101 +0,0 @@
|
|||||||
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]
|
|
||||||
Loading…
x
Reference in New Issue
Block a user