Jarian Cottingham
4bd538b217
Critical: build_leaf_cert self-signed leaves with the leaf key instead of the YubiKey-held Intermediate CA key. Now extracts TBS, signs via pkcs11-tool (ECDSA-SHA384), reassembles, and verifies against the intermediate CA public key before returning. - setup-certauth.sh: ~1100 lines of stale inline api/ copies replaced with copy-from-repo (single source of truth); writes private /etc/certauth/certauth.env (0600); DB init loads env, no more swallowed errors; systemd unit gets EnvironmentFile= - config.py: YubiKey serials no longer hard-coded (env, fail closed); aarch64-only PKCS#11 path replaced with arch-neutral default; all paths env-overridable (CERTAUTH_*) - main.py: removed dead fastapi.security.CSRFProtection import (crashed startup); module-relative static/templates dirs; created_by resolved from the authenticated user instead of hard-coded 1; unclosed file handles fixed; domain_id 0 stored as NULL (FK bug) - models.py: certificates.domain_id FK pointed at users(id), now domains(id) - login: CSRF token now actually sent and validated - tests: 23 tests (auth, API flows, DER helpers, signing pipeline) - README, LICENSE, requirements.txt, pyproject.toml
92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
import sqlite3
|
|
import bcrypt
|
|
import logging
|
|
from config import DB_PATH, ADMIN_USERNAME, ADMIN_PASSWORD
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_db():
|
|
conn = sqlite3.connect(DB_PATH, timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA busy_timeout=30000")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
conn.execute("PRAGMA wal_autocheckpoint=1000")
|
|
return conn
|
|
|
|
def init_db():
|
|
conn = get_db()
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
CREATE TABLE IF NOT EXISTS domains (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT UNIQUE NOT NULL,
|
|
description TEXT,
|
|
status TEXT DEFAULT 'active',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS certificates (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
domain_id INTEGER REFERENCES domains(id),
|
|
subject TEXT NOT NULL,
|
|
san TEXT,
|
|
serial TEXT UNIQUE,
|
|
status TEXT DEFAULT 'pending',
|
|
cert_path TEXT,
|
|
issued_at TIMESTAMP,
|
|
expires_at TIMESTAMP,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS api_keys (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
key_hash TEXT UNIQUE NOT NULL,
|
|
prefix TEXT NOT NULL,
|
|
permissions TEXT DEFAULT 'read',
|
|
active BOOLEAN DEFAULT 1,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
action TEXT NOT NULL,
|
|
details TEXT,
|
|
user_id INTEGER REFERENCES users(id),
|
|
ip_address TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
CREATE TABLE IF NOT EXISTS crl (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
serial TEXT UNIQUE NOT NULL,
|
|
revoked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
reason TEXT
|
|
);
|
|
""")
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT id FROM users WHERE username = ?", (ADMIN_USERNAME,))
|
|
if not cur.fetchone():
|
|
pw_hash = bcrypt.hashpw(ADMIN_PASSWORD.encode(), bcrypt.gensalt())
|
|
if isinstance(pw_hash, bytes):
|
|
pw_hash = pw_hash.decode()
|
|
cur.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)",
|
|
(ADMIN_USERNAME, pw_hash))
|
|
conn.commit()
|
|
conn.close()
|
|
logger.info("Database initialized")
|
|
|
|
def hash_password(password):
|
|
h = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
|
|
return h.decode() if isinstance(h, bytes) else h
|
|
|
|
def verify_password(password, hash_):
|
|
if isinstance(hash_, str):
|
|
hash_ = hash_.encode()
|
|
return bcrypt.checkpw(password.encode(), hash_)
|