diff --git a/CLAUDE.md b/CLAUDE.md index 5023430..01dff92 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -98,7 +98,7 @@ Default: `~/.local/share/ssltui/` (overridden via `SSLTUI_DIR` env var or `--dir $SSLTUI_DIR/ ├── ca.key # CA private key (chmod 600) ├── ca.crt # CA certificate (self-signed) -├── index.json # issued cert metadata (CN, SANs, serial, expiry, paths) +├── ca.db # SQLite store: cert index, revocations, event log, counters ├── certs/ │ ├── / │ │ ├── cert.crt @@ -108,6 +108,24 @@ $SSLTUI_DIR/ └── ca.crl # certificate revocation list (PEM, regenerated on each revocation) ``` +### Metadata store (`ca.db`) + +All cert metadata, revocations, counters (serial / CRL number), and an +append-only **event log** live in a single SQLite database (`ssltui/store.py`), +replacing the earlier `index.json`. WAL mode plus a busy timeout let the TUI, +the cron `--renew` process, and the multi-threaded Flask API read and write +concurrently without explicit file locking. Cert/key material itself stays as +flat PEM files under `certs//`. + +Tables: `certs` (CN → metadata JSON), `revoked` (serial → JSON), `events` +(`id`, `ts`, `type`, `cn`, `method`, `detail`), and `meta` (serial, crl_number, +server_fqdn, and a `version` counter the dashboard/TUI watchers poll to refresh). + +Events record lifecycle actions (`issue`, `renew`, `revoke`) and every private +**key access** (`key_download`), each tagged with the originating `method` +(`api` / `tui` / `cli` / `cron`). The events table is the system of record for +the dashboard and TUI live activity logs. + ## Renewal Cron entry (installed by the app): diff --git a/ssltui/__main__.py b/ssltui/__main__.py index 8206044..73488fe 100644 --- a/ssltui/__main__.py +++ b/ssltui/__main__.py @@ -41,6 +41,9 @@ def _build_parser() -> argparse.ArgumentParser: issue.add_argument("--key-type", choices=["ec", "rsa"], default="ec") issue.add_argument("--days", type=int, default=180) + # audit + sub.add_parser("audit", help="Print the full stored event log") + # serve serve = sub.add_parser("serve", help="Start the REST API server (requires Flask)") serve.add_argument( @@ -104,7 +107,7 @@ def _build_parser() -> argparse.ArgumentParser: return p -_SUBCMDS = frozenset({"renew", "status", "issue", "serve", "get", "getroot"}) +_SUBCMDS = frozenset({"renew", "status", "issue", "audit", "serve", "get", "getroot"}) def main(argv: list[str] | None = None) -> None: @@ -129,6 +132,8 @@ def main(argv: list[str] | None = None) -> None: _cmd_status() elif args.cmd == "issue": _cmd_issue(args) + elif args.cmd == "audit": + _cmd_audit() elif args.cmd == "serve": _cmd_serve(args) elif args.cmd == "get": @@ -152,7 +157,7 @@ def _cmd_renew(args) -> None: if args.cert: try: - renew_cert(root, args.cert) + renew_cert(root, args.cert, method="cron") print(f"OK renewed {args.cert}") except CAError as exc: print(f"ERROR {args.cert}: {exc}", file=sys.stderr) @@ -215,6 +220,7 @@ def _cmd_issue(args) -> None: sans=args.san, key_type=args.key_type, validity_days=args.days, + method="cli", ) print(f"OK issued {meta['cn']}") print(f" cert: {meta['cert']}") @@ -225,6 +231,31 @@ def _cmd_issue(args) -> None: sys.exit(1) +def _cmd_audit() -> None: + from ssltui import config, store + + root = config.data_dir() + events = store.list_events(root, limit=None) + + if not events: + print("No events recorded.") + return + + fmt = "{:<34} {:<13} {:<8} {:<30} {}" + print(fmt.format("TIMESTAMP", "TYPE", "METHOD", "CN", "DETAIL")) + print("-" * 103) + for ev in events: + print( + fmt.format( + ev.get("ts") or "", + ev.get("type") or "", + ev.get("method") or "-", + ev.get("cn") or "-", + ev.get("detail") or "", + ) + ) + + def _cmd_serve(args) -> None: import sys diff --git a/ssltui/api.py b/ssltui/api.py index 009f685..65afa07 100644 --- a/ssltui/api.py +++ b/ssltui/api.py @@ -113,19 +113,34 @@ def since(self, after_seq: int) -> tuple[list[dict], int]: return result, self._seq +# Event type \u2192 (dashboard level, human label) for rendering stored events. +_EVENT_RENDER: dict[str, tuple[str, str]] = { + "issue": ("success", "issued"), + "renew": ("info", "renewed"), + "revoke": ("warning", "revoked"), + "key_download": ("warning", "key downloaded"), + "ca_init": ("error", "CA re-initialised"), +} + + +def _format_event(ev: dict) -> tuple[str, str]: + """Map a stored event row to a (level, message) pair for the dashboard log.""" + level, label = _EVENT_RENDER.get(ev["type"], ("dim", ev["type"])) + method = ev.get("method") + suffix = f" ({method})" if method else "" + cn = ev.get("cn") + msg = f"{label}{suffix}: {cn}" if cn else f"{label}{suffix}" + return level, msg + + def _write_historical_events(root: Path, event_log: EventLog) -> None: - """Seed the event log with recent cert history so the dashboard shows context.""" + """Seed the event log from the persisted events table so the dashboard has context.""" try: - events: list[tuple[str, str, str]] = [] - for cert in store.list_certs(root): - ts = cert["issued"][:10] - events.append((cert["issued"], "dim", f"[{ts}] issued: {cert['cn']}")) - for rev in store.list_revoked(root): - ts = rev["revoked_at"][:10] - events.append((rev["revoked_at"], "dim", f"[{ts}] revoked: {rev['cn']}")) - events.sort(key=lambda e: e[0]) - for _, level, msg in events[-20:]: - event_log.add(level, msg) + events = store.list_events(root, limit=20) + for ev in events: + level, msg = _format_event(ev) + day = (ev.get("ts") or "")[:10] + event_log.add(level, f"[{day}] {msg}" if day else msg) if events: event_log.add("dim", "\u2500" * 24 + " live") except Exception: @@ -133,7 +148,12 @@ def _write_historical_events(root: Path, event_log: EventLog) -> None: def _start_fs_watcher(root: Path, event_log: EventLog) -> None: - """Start a daemon thread that detects cert and CA changes and logs them.""" + """Daemon thread that surfaces new events (and CA/CRL changes) on the dashboard. + + Cert lifecycle and key-download events come from the persisted events table, + polled via the store version counter; CA re-init and CRL regeneration are + still detected by file mtime since they don't always write an event row. + """ def _mtime(p: Path) -> float: try: @@ -141,14 +161,16 @@ def _mtime(p: Path) -> float: except OSError: return 0.0 - def _snapshot() -> dict[str, dict]: + def _last_event_id() -> int: try: - return {c["cn"]: c for c in store.list_certs(root)} + evs = store.list_events(root, limit=1) + return evs[-1]["id"] if evs else 0 except Exception: - return {} + return 0 state: dict = { - "cns": _snapshot(), + "version": _safe_version(root), + "last_id": _last_event_id(), "ca_mtime": _mtime(config.ca_cert_path(root)), "crl_mtime": _mtime(config.crl_path(root)), } @@ -157,16 +179,15 @@ def _poll() -> None: while True: time.sleep(5) try: - current = _snapshot() - old = state["cns"] - for cn in set(current) - set(old): - event_log.add("success", f"cert issued: {cn}") - for cn in set(old) - set(current): - event_log.add("warning", f"cert revoked: {cn}") - for cn in set(current) & set(old): - if current[cn].get("serial") != old[cn].get("serial"): - event_log.add("info", f"cert renewed: {cn}") - state["cns"] = current + version = _safe_version(root) + if version != state["version"]: + state["version"] = version + for ev in store.list_events(root, limit=100): + if ev["id"] <= state["last_id"]: + continue + level, msg = _format_event(ev) + event_log.add(level, msg) + state["last_id"] = ev["id"] ca_mtime = _mtime(config.ca_cert_path(root)) if ca_mtime and ca_mtime != state["ca_mtime"]: @@ -183,6 +204,13 @@ def _poll() -> None: threading.Thread(target=_poll, daemon=True, name="ssltui-fs-watcher").start() +def _safe_version(root: Path) -> int: + try: + return store.get_version(root) + except Exception: + return 0 + + # --------------------------------------------------------------------------- # Dashboard HTML templates # --------------------------------------------------------------------------- @@ -400,7 +428,6 @@ def _poll() -> None:
- Logout
@@ -408,8 +435,8 @@ def _poll() -> None: ● API on {{ server_url | e }}  ·  CA root: {{ ca_root | e }} -  ·  ↓ Root CA cert -  ·  ↓ CRL +  ·  ↓ Root CA cert +  ·  ↓ CRL
@@ -530,7 +557,11 @@ def _poll() -> None: const ESC = s => s.replace(/&/g,'&').replace(//g,'>').replace(/"/g,'"'); const _tok = {{ server_token | tojson }}; -const CERT_EVENTS = /^(cert issued|cert renewed|cert revoked|API: issued|API: renewed)/; +// Event messages that change the cert list (see _format_event server-side): +// "issued (api): cn", "renewed (cron): cn", "revoked (tui): cn", and the +// historical "[2026-06-14] issued (api): cn" variant. "key downloaded" and +// "CA re-initialised" deliberately don't match — they don't alter the table. +const CERT_EVENTS = /(?:^|\\] )(?:issued|renewed|revoked)\\b/; function dcls(d) { if (d < 0) return 'expired'; @@ -598,6 +629,14 @@ def _poll() -> None: } } +// Coalesce reloads: connecting to the SSE stream replays recent history, so a +// burst of cert events shouldn't fire one fetch each. +let _certReloadTimer = null; +function scheduleCertReload() { + if (_certReloadTimer) return; + _certReloadTimer = setTimeout(() => { _certReloadTimer = null; loadCerts(); }, 150); +} + const elog = document.getElementById('elog'); function appendEv(ev) { @@ -620,7 +659,7 @@ def _poll() -> None: evSrc.onmessage = e => { const ev = JSON.parse(e.data); appendEv(ev); - if (CERT_EVENTS.test(ev.msg)) loadCerts(); + if (CERT_EVENTS.test(ev.msg)) scheduleCertReload(); }; evSrc.onerror = () => { evSrc.close(); @@ -1000,6 +1039,39 @@ def _poll() -> None: document.getElementById('admodal-bg').addEventListener('click', e => { if (e.target === document.getElementById('admodal-bg')) closeDesigner(); }); +// Root CA cert / CRL: fetch into a blob and save it, rather than navigating +// the browser straight to the HTTPS URL. A direct download over a connection +// whose certificate the browser doesn't trust yet (the local CA being exactly +// what's downloaded here) is blocked by Chrome as a "network error"; a same-page +// blob save isn't subject to the origin connection's trust state. +async function caDownload(url, filename) { + try { + const r = await fetch(url); + if (r.status === 401) { showAuthError(); return; } + if (!r.ok) { + let msg = 'Download failed (' + r.status + ')'; + try { const j = await r.json(); if (j && j.error) msg = j.error; } catch (e) {} + alert(msg); + return; + } + const blob = await r.blob(); + const a = document.createElement('a'); + a.href = URL.createObjectURL(blob); + a.download = filename; + document.body.appendChild(a); + a.click(); + setTimeout(() => { URL.revokeObjectURL(a.href); a.remove(); }, 0); + } catch (e) { + alert('Download error: ' + e.message); + } +} +document.querySelectorAll('a.ca-dl').forEach(a => { + a.addEventListener('click', e => { + e.preventDefault(); + caDownload(a.dataset.dl, a.dataset.name); + }); +}); + document.addEventListener('keydown', e => { if (e.key === 'Escape') { closeKeyModal(); closePemModal(); closeIssueModal(); closeDesigner(); } }); loadCerts(); @@ -1171,6 +1243,14 @@ def issue(): cn = (body.get("cn") or "").strip() if not cn: abort(400, description="cn is required") + if store.get_cert(root, cn) is not None: + abort( + 409, + description=( + f"a certificate for {cn!r} already exists; " + "revoke it before issuing a new one" + ), + ) try: meta = issue_cert( root, @@ -1178,10 +1258,10 @@ def issue(): sans=body.get("sans") or [], key_type=body.get("key_type", "ec"), validity_days=int(body.get("validity_days", config.LEAF_VALIDITY_DAYS)), + method="api", ) except (CAError, ValueError) as exc: abort(400, description=str(exc)) - _event_log.add("success", f"API: issued {meta['cn']}") # type: ignore[index] return jsonify(meta), 201 # type: ignore[return-value] @app.get("/api/v1/certs/") @@ -1192,10 +1272,9 @@ def cert_meta(cn: str): def renew(cn: str): _entry_or_404(cn) try: - meta = renew_cert(root, cn) + meta = renew_cert(root, cn, method="api") except CAError as exc: abort(400, description=str(exc)) - _event_log.add("info", f"API: renewed {cn}") return jsonify(meta) # type: ignore[return-value] @app.get("/api/v1/certs//cert.pem") @@ -1206,7 +1285,7 @@ def download_cert(cn: str): @app.get("/api/v1/certs//key.pem") def download_key(cn: str): entry = _entry_or_404(cn) - _event_log.add("warning", f"key downloaded: {cn}") + store.add_event(root, "key_download", cn=cn, method="api") return _pem_response(Path(entry["key"]), f"{_safe(cn)}.key") @app.get("/api/v1/certs//chain.pem") diff --git a/ssltui/ca.py b/ssltui/ca.py index 19d3459..4e82fd0 100644 --- a/ssltui/ca.py +++ b/ssltui/ca.py @@ -165,10 +165,14 @@ def issue_cert( sans: list[str] | None = None, key_type: str = "ec", validity_days: int = config.LEAF_VALIDITY_DAYS, + method: str = "tui", + event_type: str = "issue", ) -> dict: """Issue a signed leaf certificate. - Returns the metadata dict that was written to the store. + Returns the metadata dict that was written to the store. ``method`` records + which interface triggered the issue (api/tui/cli/cron); ``event_type`` lets + renew_cert log a ``renew`` instead of a duplicate ``issue``. """ if not config.ca_key_path(root).exists(): raise CAError("CA not initialised. Run init_ca() first.") @@ -293,9 +297,10 @@ def issue_cert( "chain": str(chain_path), } - from ssltui.store import add_cert + from ssltui.store import add_cert, add_event add_cert(root, metadata) + add_event(root, event_type, cn=cn, method=method) return metadata @@ -305,7 +310,7 @@ def issue_cert( # --------------------------------------------------------------------------- -def renew_cert(root: Path, cn: str) -> dict: +def renew_cert(root: Path, cn: str, method: str = "tui") -> dict: """Re-issue a cert for *cn*, preserving its key type and SANs.""" from ssltui.store import get_cert @@ -334,6 +339,8 @@ def renew_cert(root: Path, cn: str) -> dict: ], key_type=entry.get("key_type", "ec"), validity_days=entry.get("validity_days", config.LEAF_VALIDITY_DAYS), + method=method, + event_type="renew", ) @@ -342,9 +349,9 @@ def renew_cert(root: Path, cn: str) -> dict: # --------------------------------------------------------------------------- -def revoke_cert(root: Path, cn: str) -> None: +def revoke_cert(root: Path, cn: str, method: str = "tui") -> None: """Revoke a cert, update ca.crl, delete its files, and remove it from the store.""" - from ssltui.store import add_revoked, get_cert, remove_cert + from ssltui.store import add_event, add_revoked, get_cert, remove_cert entry = get_cert(root, cn) if entry is None: @@ -366,6 +373,7 @@ def revoke_cert(root: Path, cn: str) -> None: shutil.rmtree(cd) remove_cert(root, cn) + add_event(root, "revoke", cn=cn, method=method) def generate_crl(root: Path) -> Path: diff --git a/ssltui/config.py b/ssltui/config.py index 6e865f6..192e4e8 100644 --- a/ssltui/config.py +++ b/ssltui/config.py @@ -35,8 +35,8 @@ def ca_cert_path(root: Path) -> Path: return root / "ca.crt" -def index_path(root: Path) -> Path: - return root / "index.json" +def db_path(root: Path) -> Path: + return root / "ca.db" def cert_dir(root: Path, cn: str) -> Path: diff --git a/ssltui/renewal.py b/ssltui/renewal.py index 03807e2..4861094 100644 --- a/ssltui/renewal.py +++ b/ssltui/renewal.py @@ -42,7 +42,7 @@ def renew_all( for cert in certs_expiring_within(root, threshold_days): cn = cert["cn"] try: - renew_cert(root, cn) + renew_cert(root, cn, method="cron") results.append((cn, True, "renewed")) except CAError as exc: results.append((cn, False, str(exc))) diff --git a/ssltui/store.py b/ssltui/store.py index cd9046a..1d580f9 100644 --- a/ssltui/store.py +++ b/ssltui/store.py @@ -1,114 +1,248 @@ -"""Cert/key storage and index (JSON).""" +"""Cert/key metadata and event store (SQLite). + +A single ``ca.db`` in the CA root holds the cert index, revocation list, an +append-only event log, and small counters (serial, CRL number, version). All +access goes through the module functions below; callers never touch the DB +directly. WAL mode plus a busy timeout let the TUI, the cron ``--renew`` +process, and the multi-threaded Flask API read and write concurrently without +explicit file locking. +""" from __future__ import annotations -import fcntl import json -import threading +import sqlite3 +from collections.abc import Iterator from contextlib import contextmanager +from datetime import UTC, datetime from pathlib import Path from ssltui import config -_thread_lock = threading.Lock() - @contextmanager -def _locked(root: Path): - """Exclusive lock for read-modify-write transactions on index.json. +def _connect(root: Path) -> Iterator[sqlite3.Connection]: + """Open a connection to the CA database, creating the schema if needed. - Combines a threading.Lock (in-process) with fcntl.flock (cross-process) - so concurrent Flask threads and cron processes serialise correctly. + A fresh connection per call keeps the module functions stateless and + sidesteps SQLite's cross-thread restrictions under Flask. The transaction + is committed on clean exit (rolled back on error) and the connection is + always closed. """ - lock_path = config.index_path(root).with_suffix(".lock") - with _thread_lock: - with open(lock_path, "a") as fh: - fcntl.flock(fh, fcntl.LOCK_EX) - yield - - -def _load(root: Path) -> dict: - p = config.index_path(root) - if not p.exists(): - return {"certs": {}} - return json.loads(p.read_text()) - - -def _save(root: Path, data: dict) -> None: - p = config.index_path(root) - tmp = p.with_suffix(".tmp") - tmp.write_text(json.dumps(data, indent=2)) - tmp.chmod(0o600) - tmp.rename(p) # atomic on POSIX — readers see old or new, never partial + conn = sqlite3.connect(config.db_path(root)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS certs ( + cn TEXT PRIMARY KEY, + data TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS revoked ( + serial TEXT PRIMARY KEY, + data TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + type TEXT NOT NULL, + cn TEXT, + method TEXT, + detail TEXT + ); + CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT + ); + """ + ) + try: + yield conn + conn.commit() + except BaseException: + conn.rollback() + raise + finally: + conn.close() + + +def _bump_version(conn: sqlite3.Connection) -> None: + """Increment the change counter that watchers poll for refresh.""" + conn.execute( + "INSERT INTO meta (key, value) VALUES ('version', '1') " + "ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1" + ) + + +def _incr(conn: sqlite3.Connection, key: str) -> int: + """Atomically increment an integer counter in ``meta`` and return it.""" + row = conn.execute( + "INSERT INTO meta (key, value) VALUES (?, '1') " + "ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1 " + "RETURNING value", + (key,), + ).fetchone() + return int(row[0]) + + +# --------------------------------------------------------------------------- +# Certs +# --------------------------------------------------------------------------- + + +def _exists(root: Path) -> bool: + """True once the database has been created (i.e. after CA init / first write). + + Reads tolerate a missing DB so the TUI can render an empty list before the + CA exists; writes always run after init_ca has created the root directory. + """ + return config.db_path(root).exists() def list_certs(root: Path) -> list[dict]: - data = _load(root) - return list(data.get("certs", {}).values()) + if not _exists(root): + return [] + with _connect(root) as conn: + rows = conn.execute("SELECT data FROM certs ORDER BY cn").fetchall() + return [json.loads(r["data"]) for r in rows] def get_cert(root: Path, cn: str) -> dict | None: - data = _load(root) - return data.get("certs", {}).get(cn) + if not _exists(root): + return None + with _connect(root) as conn: + row = conn.execute("SELECT data FROM certs WHERE cn = ?", (cn,)).fetchone() + return json.loads(row["data"]) if row else None def add_cert(root: Path, metadata: dict) -> None: - with _locked(root): - data = _load(root) - data.setdefault("certs", {})[metadata["cn"]] = metadata - _save(root, data) + with _connect(root) as conn: + conn.execute( + "INSERT INTO certs (cn, data) VALUES (?, ?) " + "ON CONFLICT(cn) DO UPDATE SET data = excluded.data", + (metadata["cn"], json.dumps(metadata)), + ) + _bump_version(conn) def remove_cert(root: Path, cn: str) -> None: - with _locked(root): - data = _load(root) - data.get("certs", {}).pop(cn, None) - _save(root, data) + with _connect(root) as conn: + conn.execute("DELETE FROM certs WHERE cn = ?", (cn,)) + _bump_version(conn) + + +# --------------------------------------------------------------------------- +# Revocation +# --------------------------------------------------------------------------- def list_revoked(root: Path) -> list[dict]: - data = _load(root) - return list(data.get("revoked", [])) + if not _exists(root): + return [] + with _connect(root) as conn: + rows = conn.execute("SELECT data FROM revoked ORDER BY serial").fetchall() + return [json.loads(r["data"]) for r in rows] def add_revoked(root: Path, entry: dict) -> None: - with _locked(root): - data = _load(root) - revoked = data.setdefault("revoked", []) - if not any(r["serial"] == entry["serial"] for r in revoked): - revoked.append(entry) - _save(root, data) + with _connect(root) as conn: + # PRIMARY KEY on serial makes this idempotent — a repeated revoke is a no-op. + conn.execute( + "INSERT OR IGNORE INTO revoked (serial, data) VALUES (?, ?)", + (str(entry["serial"]), json.dumps(entry)), + ) + _bump_version(conn) + + +# --------------------------------------------------------------------------- +# Events (audit log + dashboard live feed) +# --------------------------------------------------------------------------- + + +def add_event( + root: Path, + type: str, + *, + cn: str | None = None, + method: str | None = None, + detail: str | None = None, +) -> None: + """Append an event row. ``type`` is e.g. issue/revoke/renew/key_download.""" + ts = datetime.now(UTC).isoformat() + with _connect(root) as conn: + conn.execute( + "INSERT INTO events (ts, type, cn, method, detail) VALUES (?, ?, ?, ?, ?)", + (ts, type, cn, method, detail), + ) + _bump_version(conn) + + +def list_events(root: Path, limit: int | None = 50) -> list[dict]: + """Return the most recent events, oldest first. ``limit=None`` returns all.""" + if not _exists(root): + return [] + with _connect(root) as conn: + if limit is None: + rows = conn.execute( + "SELECT id, ts, type, cn, method, detail FROM events ORDER BY id DESC" + ).fetchall() + else: + rows = conn.execute( + "SELECT id, ts, type, cn, method, detail FROM events " + "ORDER BY id DESC LIMIT ?", + (limit,), + ).fetchall() + return [dict(r) for r in reversed(rows)] + + +# --------------------------------------------------------------------------- +# Server FQDN / counters / version +# --------------------------------------------------------------------------- def get_server_fqdn(root: Path) -> str | None: """Return the FQDN the dashboard/API server should present a cert for.""" - data = _load(root) - return data.get("server_fqdn") + if not _exists(root): + return None + with _connect(root) as conn: + row = conn.execute( + "SELECT value FROM meta WHERE key = 'server_fqdn'" + ).fetchone() + return row["value"] if row else None def set_server_fqdn(root: Path, fqdn: str | None) -> None: - with _locked(root): - data = _load(root) + with _connect(root) as conn: if fqdn: - data["server_fqdn"] = fqdn + conn.execute( + "INSERT INTO meta (key, value) VALUES ('server_fqdn', ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (fqdn,), + ) else: - data.pop("server_fqdn", None) - _save(root, data) + conn.execute("DELETE FROM meta WHERE key = 'server_fqdn'") + _bump_version(conn) def next_crl_number(root: Path) -> int: - with _locked(root): - data = _load(root) - n = data.get("crl_number", 0) + 1 - data["crl_number"] = n - _save(root, data) + with _connect(root) as conn: + n = _incr(conn, "crl_number") + _bump_version(conn) return n def next_serial(root: Path) -> int: - with _locked(root): - data = _load(root) - n = data.get("serial", 0) + 1 - data["serial"] = n - _save(root, data) + with _connect(root) as conn: + n = _incr(conn, "serial") + _bump_version(conn) return n + + +def get_version(root: Path) -> int: + """Monotonic counter bumped on every write — watchers poll this to refresh.""" + if not _exists(root): + return 0 + with _connect(root) as conn: + row = conn.execute("SELECT value FROM meta WHERE key = 'version'").fetchone() + return int(row["value"]) if row else 0 diff --git a/ssltui/tui.py b/ssltui/tui.py index 494c34b..263db82 100644 --- a/ssltui/tui.py +++ b/ssltui/tui.py @@ -74,6 +74,26 @@ def _expiry_style(days: int) -> str: return "green" +# Event type → (Rich style, human label) for the dashboard server log. +_EVENT_STYLE: dict[str, tuple[str, str]] = { + "issue": ("green", "cert issued"), + "renew": ("cyan", "cert renewed"), + "revoke": ("yellow", "cert revoked"), + "key_download": ("bold yellow", "key downloaded"), + "ca_init": ("bold red", "CA re-initialised"), +} + + +def _format_tui_event(ev: dict) -> tuple[str, str]: + """Map a stored event row to a (style, message) pair for the request log.""" + style, label = _EVENT_STYLE.get(ev["type"], ("dim", ev["type"])) + method = ev.get("method") + suffix = f" ({method})" if method else "" + cn = ev.get("cn") + msg = f"{label}{suffix}: {cn}" if cn else f"{label}{suffix}" + return style, msg + + # --------------------------------------------------------------------------- # Confirm modal (used for destructive actions) # --------------------------------------------------------------------------- @@ -127,6 +147,44 @@ def action_cancel(self) -> None: self.dismiss(False) +class ErrorScreen(ModalScreen): + """Show a blocking error message with a single dismiss action.""" + + BINDINGS = [ + Binding("escape", "close", "Close"), + Binding("enter", "close", "Close"), + ] + + DEFAULT_CSS = """ + ErrorScreen { align: center middle; } + ErrorScreen > Vertical { + width: 60; height: auto; + background: $surface; border: thick $error; padding: 1 2; + } + ErrorScreen .title { text-align: center; text-style: bold; color: $error; margin-bottom: 1; } + ErrorScreen .message { text-align: center; margin-bottom: 1; } + ErrorScreen .hint { text-align: center; color: $text-muted; } + """ + + def __init__(self, title: str, message: str) -> None: + super().__init__() + self._title = title + self._message = message + + def compose(self) -> ComposeResult: + with Vertical(): + yield Label(self._title, classes="title") + yield Label(self._message, classes="message") + yield Label("[dim]Enter / Esc[/dim] close", markup=True, classes="hint") + yield Button("OK", variant="primary", id="ok") + + def on_button_pressed(self, event: Button.Pressed) -> None: + self.dismiss(None) + + def action_close(self) -> None: + self.dismiss(None) + + # --------------------------------------------------------------------------- # Init CA modal # --------------------------------------------------------------------------- @@ -295,6 +353,18 @@ def _do_issue(self) -> None: self.query_one("#cn", Input).focus() return + if store.get_cert(_root(), cn) is not None: + self.app.push_screen( + ErrorScreen( + title="Certificate already exists", + message=( + f"A certificate for '{cn}' already exists.\n" + "Revoke it before issuing a new one." + ), + ) + ) + return + try: validity = int(validity_raw) except ValueError: @@ -729,9 +799,20 @@ def check_action(self, action: str, parameters: tuple) -> bool | None: return True if self._key_path else False return True + def _record_key_access(self, detail: str) -> None: + """Audit a private-key exposure in the TUI (reveal/copy/save).""" + cn = self._title.removeprefix("Certificate: ") + try: + store.add_event(_root(), "key_download", cn=cn, method="tui", detail=detail) + except Exception: + pass + def action_copy(self) -> None: - pem = self._key_pem if (self._showing_key and self._key_pem) else self._pem + showing_key = self._showing_key and self._key_pem is not None + pem = self._key_pem if showing_key else self._pem self.app.copy_to_clipboard(pem) + if showing_key: + self._record_key_access("copy") self.notify("Copied to clipboard.") def action_toggle_key(self) -> None: @@ -741,6 +822,8 @@ def action_toggle_key(self) -> None: if self._key_pem is None: self._key_pem = self._key_path.read_text() self._showing_key = not self._showing_key + if self._showing_key: + self._record_key_access("reveal") pem_widget = self.query_one("#pem-content", Static) if self._showing_key: combined = self._pem.rstrip("\n") + "\n" + self._key_pem @@ -759,6 +842,7 @@ def action_save(self) -> None: combined = self._pem.rstrip("\n") + "\n" + self._key_pem content = combined default = str(Path.home() / f"{safe_name}.pem") + self._record_key_access("save") else: content = self._pem default = str(Path.home() / self._filename) @@ -861,8 +945,13 @@ def _update_ca_status(self) -> None: def _build_table(self) -> None: table = self.query_one("#cert-table", DataTable) - table.clear(columns=True) - table.add_columns("CN", "SANs", "Key", "Expires", "Days left") + # Add the columns once; on later refreshes clear rows only. Re-adding + # columns (clear(columns=True)) makes Textual recompute auto-widths from + # the header labels alone when called outside a fresh mount, which + # collapses the column widths. + if not table.columns: + table.add_columns("CN", "SANs", "Key", "Expires", "Days left") + table.clear() for cert in store.list_certs(_root()): days = days_until_expiry(cert["expiry"]) @@ -1091,7 +1180,8 @@ def __init__(self, server: APIServer, token: str) -> None: self._token = token self._log_handler: _WerkzeugCapture | None = None self._fs_state: dict[str, float] = {} - self._cached_cns: dict[str, dict] = {} + self._last_version: int = 0 + self._last_event_id: int = 0 def compose(self) -> ComposeResult: yield Header(show_clock=True) @@ -1122,7 +1212,8 @@ def on_mount(self) -> None: self._server.start() self._fs_state = self._fs_snapshot() - self._cached_cns = {c["cn"]: c for c in store.list_certs(self._server.root)} + self._last_version = self._safe_version() + self._last_event_id = self._latest_event_id() self._write_history() self.set_interval(0.1, self._drain_log) self.set_interval(1.0, self._poll_fs) @@ -1144,33 +1235,40 @@ def _drain_log(self) -> None: # text and corrupt the line, so decode them into a Rich Text. log.write(Text.from_ansi(line)) + def _safe_version(self) -> int: + try: + return store.get_version(self._server.root) + except Exception: + return self._last_version + + def _latest_event_id(self) -> int: + try: + evs = store.list_events(self._server.root, limit=1) + return evs[-1]["id"] if evs else 0 + except Exception: + return 0 + def _write_history(self) -> None: - root = self._server.root - events: list[tuple[str, str, str]] = [] - for cert in store.list_certs(root): - events.append((cert["issued"], "cert issued", cert["cn"])) - for rev in store.list_revoked(root): - events.append((rev["revoked_at"], "cert revoked", rev["cn"])) - - events.sort(key=lambda e: e[0]) - recent = events[-5:] - if not recent: + events = store.list_events(self._server.root, limit=5) + if not events: return log = self.query_one("#request-log", RichLog) - for ts_iso, label, cn in recent: + for ev in events: + _, msg = _format_tui_event(ev) + ts_iso = ev.get("ts") or "" try: dt = datetime.fromisoformat(ts_iso.replace("Z", "+00:00")) ts = dt.strftime("%Y-%m-%d %H:%M") except ValueError: ts = ts_iso[:16] - log.write(Text(f"{ts} HISTORICAL {label}: {cn}", style="dim italic")) + log.write(Text(f"{ts} HISTORICAL {msg}", style="dim italic")) log.write(Text("─" * 60 + " live", style="dim")) def _fs_snapshot(self) -> dict[str, float]: root = self._server.root state: dict[str, float] = {} - for name in ("index.json", "ca.crt", "ca.crl", "api_token"): + for name in ("ca.crt", "ca.crl", "api_token"): p = root / name if p.exists(): state[name] = p.stat().st_mtime @@ -1184,26 +1282,25 @@ def _poll_fs(self) -> None: log = self.query_one("#request-log", RichLog) ts = datetime.now().strftime("%H:%M:%S") - root = self._server.root - if now.get("index.json") != self._fs_state.get("index.json"): + version = self._safe_version() + version_changed = version != self._last_version + if version_changed: + self._last_version = version try: - current = {c["cn"]: c for c in store.list_certs(root)} + new_events = store.list_events(self._server.root, limit=100) except Exception: - current = {} - old = self._cached_cns - for cn in set(current) - set(old): - log.write(Text(f"{ts} cert issued: {cn}", style="green")) - for cn in set(old) - set(current): - log.write(Text(f"{ts} cert revoked: {cn}", style="yellow")) - for cn in set(current) & set(old): - if current[cn].get("serial") != old[cn].get("serial"): - log.write(Text(f"{ts} cert renewed: {cn}", style="cyan")) - self._cached_cns = current - - if now.get("ca.crl") != self._fs_state.get("ca.crl") and now.get( - "index.json" - ) == self._fs_state.get("index.json"): + new_events = [] + for ev in new_events: + if ev["id"] <= self._last_event_id: + continue + style, msg = _format_tui_event(ev) + log.write(Text(f"{ts} {msg}", style=style)) + self._last_event_id = ev["id"] + + # CRL regeneration accompanies revokes (already logged above), so only + # surface a standalone "CRL regenerated" when no event was recorded. + if not version_changed and now.get("ca.crl") != self._fs_state.get("ca.crl"): log.write(Text(f"{ts} CRL regenerated", style="dim")) if now.get("ca.crt") != self._fs_state.get("ca.crt"): diff --git a/tests/test_2_store.py b/tests/test_2_store.py new file mode 100644 index 0000000..b5d3fc9 --- /dev/null +++ b/tests/test_2_store.py @@ -0,0 +1,209 @@ +"""Unit tests for the SQLite-backed metadata/event store. + +These exercise ``ssltui.store`` directly against a temporary CA root — no tmux +and (mostly) no openssl required. A small lifecycle section uses the real CA +helpers and is skipped where openssl is unavailable. +""" + +from __future__ import annotations + +import importlib.util +import shutil +import threading +from pathlib import Path + +import pytest + +from ssltui import store + +openssl_required = pytest.mark.skipif( + shutil.which("openssl") is None, reason="requires openssl on PATH" +) +flask_required = pytest.mark.skipif( + importlib.util.find_spec("flask") is None, + reason="requires the optional 'api' extra (flask)", +) + + +def _meta(cn: str, serial: int = 1) -> dict: + return { + "cn": cn, + "sans": [f"DNS:{cn}"], + "key_type": "ec", + "serial": serial, + "issued": "2026-01-01T00:00:00+00:00", + "expiry": "Jun 13 12:00:00 2026 GMT", + "validity_days": 180, + "cert": f"/tmp/{cn}/cert.crt", + "key": f"/tmp/{cn}/cert.key", + "chain": f"/tmp/{cn}/chain.crt", + } + + +# --------------------------------------------------------------------------- +# Certs +# --------------------------------------------------------------------------- + + +def test_add_get_list_cert(tmp_path: Path) -> None: + assert store.list_certs(tmp_path) == [] + assert store.get_cert(tmp_path, "a.local") is None + + store.add_cert(tmp_path, _meta("a.local")) + store.add_cert(tmp_path, _meta("b.local", serial=2)) + + cns = {c["cn"] for c in store.list_certs(tmp_path)} + assert cns == {"a.local", "b.local"} + got = store.get_cert(tmp_path, "a.local") + assert got is not None and got["serial"] == 1 + + +def test_add_cert_upsert(tmp_path: Path) -> None: + store.add_cert(tmp_path, _meta("a.local", serial=1)) + store.add_cert(tmp_path, _meta("a.local", serial=9)) # same CN -> replace + assert len(store.list_certs(tmp_path)) == 1 + assert store.get_cert(tmp_path, "a.local")["serial"] == 9 + + +def test_remove_cert(tmp_path: Path) -> None: + store.add_cert(tmp_path, _meta("a.local")) + store.remove_cert(tmp_path, "a.local") + assert store.get_cert(tmp_path, "a.local") is None + store.remove_cert(tmp_path, "missing.local") # no-op, must not raise + + +# --------------------------------------------------------------------------- +# Revocation +# --------------------------------------------------------------------------- + + +def test_revoked_dedupe(tmp_path: Path) -> None: + entry = {"cn": "a.local", "serial": 5, "expiry": "x", "revoked_at": "t"} + store.add_revoked(tmp_path, entry) + store.add_revoked(tmp_path, entry) # same serial -> idempotent + revoked = store.list_revoked(tmp_path) + assert len(revoked) == 1 + assert revoked[0]["serial"] == 5 + + +# --------------------------------------------------------------------------- +# Counters / version / server fqdn +# --------------------------------------------------------------------------- + + +def test_counters_monotonic(tmp_path: Path) -> None: + assert [store.next_serial(tmp_path) for _ in range(3)] == [1, 2, 3] + assert [store.next_crl_number(tmp_path) for _ in range(2)] == [1, 2] + + +def test_version_increments_on_writes(tmp_path: Path) -> None: + assert store.get_version(tmp_path) == 0 + store.add_cert(tmp_path, _meta("a.local")) + v1 = store.get_version(tmp_path) + assert v1 >= 1 + store.add_event(tmp_path, "issue", cn="a.local", method="tui") + assert store.get_version(tmp_path) > v1 + + +def test_server_fqdn_roundtrip(tmp_path: Path) -> None: + assert store.get_server_fqdn(tmp_path) is None + store.set_server_fqdn(tmp_path, "host.local") + assert store.get_server_fqdn(tmp_path) == "host.local" + store.set_server_fqdn(tmp_path, None) + assert store.get_server_fqdn(tmp_path) is None + + +# --------------------------------------------------------------------------- +# Events +# --------------------------------------------------------------------------- + + +def test_events_order_and_fields(tmp_path: Path) -> None: + store.add_event(tmp_path, "issue", cn="a.local", method="cli") + store.add_event(tmp_path, "key_download", cn="a.local", method="api") + store.add_event(tmp_path, "revoke", cn="a.local", method="tui", detail="x") + + events = store.list_events(tmp_path) + assert [e["type"] for e in events] == ["issue", "key_download", "revoke"] + assert events[1]["method"] == "api" + assert events[2]["detail"] == "x" + # ids are strictly increasing in chronological order + assert events[0]["id"] < events[1]["id"] < events[2]["id"] + + +def test_events_limit_returns_recent(tmp_path: Path) -> None: + for i in range(10): + store.add_event(tmp_path, "issue", cn=f"c{i}.local", method="cli") + recent = store.list_events(tmp_path, limit=3) + assert [e["cn"] for e in recent] == ["c7.local", "c8.local", "c9.local"] + + +# --------------------------------------------------------------------------- +# Concurrency +# --------------------------------------------------------------------------- + + +def test_concurrent_next_serial_unique(tmp_path: Path) -> None: + # WAL + busy_timeout must serialise writers so no two callers get the same + # serial even under heavy contention from multiple threads. + results: list[int] = [] + lock = threading.Lock() + + def worker() -> None: + local = [store.next_serial(tmp_path) for _ in range(20)] + with lock: + results.extend(local) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(results) == 160 + assert len(set(results)) == 160 # all unique + assert sorted(results) == list(range(1, 161)) # contiguous, no gaps + + +# --------------------------------------------------------------------------- +# Lifecycle events via the real CA helpers +# --------------------------------------------------------------------------- + + +@openssl_required +def test_lifecycle_records_events(tmp_path: Path) -> None: + from ssltui.ca import init_ca, issue_cert, revoke_cert + + init_ca(tmp_path) + issue_cert(tmp_path, cn="app.local", method="api") + revoke_cert(tmp_path, "app.local", method="tui") + + events = store.list_events(tmp_path) + issued = [e for e in events if e["type"] == "issue" and e["cn"] == "app.local"] + revoked = [e for e in events if e["type"] == "revoke" and e["cn"] == "app.local"] + assert issued and issued[0]["method"] == "api" + assert revoked and revoked[0]["method"] == "tui" + + +@openssl_required +@flask_required +def test_api_key_download_records_event(tmp_path: Path) -> None: + from ssltui.api import create_app + from ssltui.ca import init_ca, issue_cert + from ssltui.config import api_token_path + + init_ca(tmp_path) + issue_cert(tmp_path, cn="app.local", method="api") + token = api_token_path(tmp_path).read_text() + + app = create_app(tmp_path, token) + client = app.test_client() + resp = client.get( + "/api/v1/certs/app.local/key.pem", + headers={"Authorization": f"Bearer {token}"}, + ) + assert resp.status_code == 200 + + downloads = [e for e in store.list_events(tmp_path) if e["type"] == "key_download"] + assert downloads and downloads[-1]["cn"] == "app.local" + assert downloads[-1]["method"] == "api"