Merge pull request 'fix: YubiKey CA signing, remove inline API copies, env-driven config' (#42) from improve/v1 into master

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

.env.example

@@ -1,19 +1,50 @@
-# CertAuth Environment Variables
+# CertAuth API environment variables.
-# Copy this to .env and fill in your real values
+# Installed to /etc/certauth/certauth.env (mode 600, owned by the API user)
+# and loaded via EnvironmentFile= in certauth-api.service.
+#
+# The API fails closed on startup if any of the REQUIRED variables below
+# are missing.
 
-# Admin web login password
+# ---- REQUIRED -------------------------------------------------------------
-ADMIN_PASS=your-secure-admin-password
 
-# YubiKey 1 (Root CA) credentials
+# YubiKey hardware assignments (serials are printed by `ykman list`;
-YK_ROOT_PIN=your-yk1-pin
+# setup-certauth.sh auto-detects them at provision time)
-YK_ROOT_PUK=your-yk1-puk
+YK_ROOT_SERIAL=your-root-yubikey-serial
+YK_INT_SERIAL=your-intermediate-yubikey-serial
 
-# YubiKey 2 (Intermediate CA) credentials
+# YubiKey PINs (change from the defaults with:
-YK_INT_PIN=your-yk2-pin
+#   ykman piv access change-pin -P <old> --new-pin <new>)
-YK_INT_PUK=your-yk2-puk
+YK_ROOT_PIN=your-yk-root-pin
+YK_INT_PIN=your-yk-intermediate-pin
 
 # JWT signing secret (generate with: python3 -c "import secrets; print(secrets.token_hex(32))")
 JWT_SECRET=your-64-char-hex-secret
 
-# PFX download password
+# Admin web/API login password
+ADMIN_PASSWORD=your-secure-admin-password
+
+# ---- OPTIONAL (defaults shown) ---------------------------------------------
+
+# Admin username (default: certauth)
+ADMIN_USERNAME=certauth
+
+# Filesystem layout (defaults match setup-certauth.sh production layout)
+CERTAUTH_CA_BASE=/etc/ssl/ca
+CERTAUTH_ROOT_CA=/etc/ssl/ca/root/root-ca.crt
+CERTAUTH_INT_CA=/etc/ssl/ca/intermediate/intermediate-ca.crt
+CERTAUTH_CA_CHAIN=/etc/ssl/ca/ca-chain.crt
+CERTAUTH_ISSUED_DIR=/etc/ssl/ca/issued
+CERTAUTH_TMP_DIR=/var/lib/certauth/tmp
+CERTAUTH_DB_PATH=/var/lib/certauth/certauth.db
+
+# PKCS#11 (defaults work on aarch64 and x86_64 Ubuntu)
+PKCS11_MODULE=/usr/lib/opensc-pkcs11.so
+# pkcs11-tool --token-label; leave empty to match by key label only
+PKCS11_TOKEN_LABEL=
+
+# Leaf certificate subject (matches setup-certauth.sh defaults)
+CA_ORG=Home
+CA_COUNTRY=US
+
+# PFX download default password (overridable per download)
 PFX_PASS=certauth

.gitea/workflows/ci.yml

@@ -0,0 +1,143 @@
+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"

.gitignore

@@ -1,21 +1,11 @@
-# Environment variables containing real secrets
-.env
-.env.local
-.env.*.local
-
-# Python
 __pycache__/
-*.py[cod]
+*.pyc
-*.egg-info/
+.env
-
-# OS
-.DS_Store
-Thumbs.db
-
-# Local overrides
-*.local
-
-# Secrets
 .password
-ssl/
+*.pem
-ssl-home/
+*.key
+*.p12
+*.pfx
+*.db
+*.sqlite
+*.log

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 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.

README.md

@@ -0,0 +1,107 @@
+# CertAuth Key Vault
+
+Self-contained Certificate Authority with YubiKey-backed signing. Two
+YubiKey 5 Nano devices act as HSMs — one for the Root CA, one for the
+Intermediate CA. Every certificate sign requires physical touch + PIN, so
+compromising the server does not compromise the CA keys.
+
+```
+Root CA (YubiKey 1)  →  Intermediate CA (YubiKey 2)  →  Leaf certs
+   25-year validity        15-year validity              1-year validity
+```
+
+## Components
+
+| Path | Purpose |
+| --- | --- |
+| `api/` | FastAPI service: REST API + Jinja2 web UI |
+| `setup-certauth.sh` | Idempotent provisioner: fresh Ubuntu 24.04+ → full CA (aarch64/x86_64) |
+| `certauth-api.service` | systemd unit for the API |
+| `Caddyfile` | Caddy TLS-terminating reverse proxy in front of the API |
+| `nginx-example.com` | Reference nginx config for the wider homelab vhost layout |
+| `landing/` | Landing pages for the CA domain |
+| `SKILL.md` | Operator/agent runbook: API usage, endpoints, workflows |
+| `.env.example` | All environment variables documented |
+
+## Architecture
+
+- **Signing**: `pkcs11-tool` against the OpenSC PKCS#11 module; ECDSA-SHA384.
+  Private keys never leave the YubiKeys; only extracted public keys are
+  cached on disk.
+- **Auth**: JWT bearer tokens for the REST API, session cookie for the web
+  UI, per-request CSRF tokens on all state-changing web endpoints.
+- **Storage**: SQLite (WAL) for domains, certificate requests, audit log,
+  and CRL entries.
+- **CRL**: revoked serials tracked in the DB; `scripts`-style daily CRL
+  refresh is documented in `SKILL.md`.
+
+## Quick start (provision a node)
+
+```bash
+sudo bash setup-certauth.sh
+```
+
+Prerequisites: fresh Ubuntu 24.04+, two YubiKey 5 Nanos plugged in, root.
+The script configures everything (CA hierarchy, API, Caddy, systemd) and
+prints the generated PINs. Configuration is overridable via environment
+(`CA_ORG`, `YK1_SERIAL`, `NETWORK_CIDR`, ...).
+
+## Running the API standalone
+
+```bash
+pip install -r requirements.txt
+export YK_ROOT_SERIAL=... YK_INT_SERIAL=...
+export YK_ROOT_PIN=... YK_INT_PIN=...
+export JWT_SECRET=$(python3 -c "import secrets; print(secrets.token_hex(32))")
+export ADMIN_PASSWORD=...
+python3 -m uvicorn main:app --host 127.0.0.1 --port 8000   # from api/
+```
+
+All filesystem paths default to the production layout (`/etc/ssl/ca`,
+`/var/lib/certauth`) and are overridable via `CERTAUTH_*` environment
+variables — see `.env.example`.
+
+## API (summary)
+
+| Method | Path | Purpose |
+| --- | --- | --- |
+| POST | `/api/token` | Login → JWT |
+| GET | `/api/me` | Validate token |
+| GET/POST | `/api/domains` | List / register domain |
+| GET | `/api/certs` | List certificate requests |
+| POST | `/api/certs/request` | Create pending request (cn, sans) |
+| POST | `/api/certs/{id}/sign` | Sign via YubiKey (touch + PIN) |
+| GET | `/api/certs/{id}/pem` | Download leaf + chain |
+| GET | `/api/certs/{id}/pfx` | Download PKCS#12 (password-protected) |
+| GET | `/api/health` | Liveness |
+| GET | `/api/ca-chain` | Download CA chain |
+
+Web UI: `/` dashboard, `/login`, `/domains`, `/certs`, `/history`,
+`/setup` (provisioning helper pages).
+
+Full usage examples: see `SKILL.md`.
+
+## Tests
+
+```bash
+pip install pytest
+pytest tests/ -v
+```
+
+Unit tests cover auth (JWT round-trip, rejection of invalid tokens),
+login/token endpoints, domain + certificate-request flows, and password
+hashing. Signing paths that require a physical YubiKey are exercised at
+the integration level on the provisioned node.
+
+## Security notes
+
+- YubiKey serials, PINs, JWT secret, and admin password are environment
+  driven and **fail closed** at import time — the service will not start
+  with missing configuration.
+- Leaf keys are written `0600`, certs `0640`, under the issued directory.
+- CSRF tokens use constant-time comparison (`secrets.compare_digest`).
+- Error strings are HTML-escaped before rendering into pages.
+
+## License
+
+MIT — see [LICENSE](LICENSE).

api/auth.py

@@ -1,15 +1,14 @@
+from datetime import datetime, timedelta, timezone
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 from jose import jwt, JWTError
-from datetime import datetime, timedelta
 from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
-from models import get_db, verify_password
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/token")
 
 def create_access_token(data: dict, expires_delta: timedelta = None):
     to_encode = data.copy()
-    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
+    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
     to_encode.update({"exp": expire})
     return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
 

api/config.py

@@ -1,18 +1,61 @@
+"""CertAuth configuration.
+
+All hardware identifiers and filesystem paths are environment-driven so the
+service can run on aarch64/x86_64 and in test environments. Secrets and
+YubiKey assignments fail closed at import time.
+"""
+
 import os
 
-YK_ROOT_SERIAL = "35450561"
+
-YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "CHANGE_ME_YK1_PIN")
+def _required(name: str) -> str:
-YK_INT_SERIAL = "33930436"
+    value = os.environ.get(name)
-YK_INT_PIN = os.environ.get("YK_INT_PIN", "CHANGE_ME_YK2_PIN")
+    if not value:
-ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
+        raise RuntimeError(f"{name} environment variable is required")
-INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
+    return value
-CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
+
-ISSUED_DIR = "/etc/ssl/ca/issued"
+
-DB_PATH = "/var/lib/certauth/certauth.db"
+# YubiKey hardware: serials must be supplied by the operator (auto-detected at
-SECRET_KEY = os.environ.get("JWT_SECRET", "CHANGE_ME_JWT_SECRET")
+# setup time by setup-certauth.sh). Never hard-code device serials in source.
+YK_ROOT_SERIAL = _required("YK_ROOT_SERIAL")
+YK_INT_SERIAL = _required("YK_INT_SERIAL")
+YK_ROOT_PIN = _required("YK_ROOT_PIN")
+YK_INT_PIN = _required("YK_INT_PIN")
+
+# Filesystem layout (defaults match the production layout created by
+# setup-certauth.sh; override via environment for testing or custom installs).
+CA_BASE = os.environ.get("CERTAUTH_CA_BASE", "/etc/ssl/ca")
+ROOT_CA_PATH = os.environ.get("CERTAUTH_ROOT_CA", f"{CA_BASE}/root/root-ca.crt")
+INT_CA_PATH = os.environ.get("CERTAUTH_INT_CA", f"{CA_BASE}/intermediate/intermediate-ca.crt")
+CA_CHAIN_PATH = os.environ.get("CERTAUTH_CA_CHAIN", f"{CA_BASE}/ca-chain.crt")
+ISSUED_DIR = os.environ.get("CERTAUTH_ISSUED_DIR", f"{CA_BASE}/issued")
+TMP_DIR = os.environ.get("CERTAUTH_TMP_DIR", "/var/lib/certauth/tmp")
+DB_PATH = os.environ.get("CERTAUTH_DB_PATH", "/var/lib/certauth/certauth.db")
+
+# Extracted YubiKey public keys (written by setup-certauth.sh).
+YK_PUB_ROOT = os.environ.get("YK_PUB_ROOT", "/tmp/yk1-root-pub.pem")
+YK_PUB_INT = os.environ.get("YK_PUB_INT", "/tmp/yk2-int-pub.pem")
+
+# PKCS#11 module for pkcs11-tool. Default works on both aarch64 and x86_64
+# Ubuntu; override for nonstandard installs.
+PKCS11_MODULE = os.environ.get("PKCS11_MODULE", "/usr/lib/opensc-pkcs11.so")
+# Optional pkcs11-tool --token-label. Empty = match the setup-script
+# invocation (key label only), which works with both YubiKeys attached.
+PKCS11_TOKEN_LABEL = os.environ.get("PKCS11_TOKEN_LABEL", "")
+
+# Leaf certificate subject defaults (match setup-certauth.sh defaults)
+CA_ORG = os.environ.get("CA_ORG", "Home")
+CA_COUNTRY = os.environ.get("CA_COUNTRY", "US")
+
+# JWT / web session
+JWT_SECRET = _required("JWT_SECRET")
+SECRET_KEY = JWT_SECRET
 ALGORITHM = "HS256"
 ACCESS_TOKEN_EXPIRE_MINUTES = 60
-ADMIN_USERNAME = "certauth"
+
-PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
+# Admin bootstrap account
-YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
+ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "certauth")
-YK_PUB_INT = "/tmp/yk2-int-pub.pem"
+ADMIN_PASSWORD = _required("ADMIN_PASSWORD")
+
+# PFX download default password (overridable per download)
+PFX_DEFAULT_PASSWORD = os.environ.get("PFX_PASS", "certauth")

api/main.py

@@ -1,39 +1,85 @@
-import os, sqlite3, datetime, secrets, hashlib, subprocess, json
+import os
+import secrets
+import logging
+import html
+from datetime import datetime, timezone
 from fastapi import FastAPI, Request, Depends, HTTPException, Form
-from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse
+from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, PlainTextResponse
 from fastapi.staticfiles import StaticFiles
 from pydantic import BaseModel
 from jose import jwt
 from jinja2 import Environment, FileSystemLoader, select_autoescape
-from config import *
+from config import (
-from models import get_db, init_db, hash_password, verify_password
+    ALGORITHM,
+    CA_CHAIN_PATH,
+    INT_CA_PATH,
+    ISSUED_DIR,
+    PFX_DEFAULT_PASSWORD,
+    ROOT_CA_PATH,
+    SECRET_KEY,
+    TMP_DIR,
+)
+from models import get_db, init_db, verify_password
 from auth import create_access_token, get_current_user
 from signing import build_leaf_cert
 from cryptography.hazmat.primitives import serialization
 
-app = FastAPI(title="CertAuth Key Vault")
+logger = logging.getLogger(__name__)
-app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static")
 
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+STATIC_DIR = os.environ.get("CERTAUTH_STATIC_DIR", os.path.join(BASE_DIR, "static"))
+TEMPLATES_DIR = os.environ.get("CERTAUTH_TEMPLATES_DIR", os.path.join(BASE_DIR, "templates"))
+
+app = FastAPI(title="CertAuth Key Vault")
+app.mount("/static", StaticFiles(directory=STATIC_DIR, check_dir=False), name="static")
+
+_csrf_secrets = {}
+
+def get_csrf_token(session_id: str) -> str:
+    if session_id not in _csrf_secrets:
+        _csrf_secrets[session_id] = secrets.token_hex(32)
+    return _csrf_secrets[session_id]
+
+def verify_csrf_token(session_id: str, token: str) -> bool:
+    stored = _csrf_secrets.get(session_id)
+    if not stored:
+        return False
+    return secrets.compare_digest(stored, token)
+
+def sanitize_error(msg: str) -> str:
+    return html.escape(str(msg))
 
 def get_user_from_cookie(request: Request):
     token = request.cookies.get("token")
-    if not token: return None
+    if not token:
-    try: return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        return None
-    except: return None
+    try:
+        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+    except Exception:
+        return None
 
 jinja_env = Environment(
-    loader=FileSystemLoader("/opt/certauth/api/templates"),
+    loader=FileSystemLoader(TEMPLATES_DIR),
-    autoescape=select_autoescape(["html"])
+    autoescape=select_autoescape(["html", "xml"]),
 )
 
 @app.on_event("startup")
 def startup():
     init_db()
     try:
-        with open(ROOT_CA_PATH) as f: root = f.read()
+        with open(ROOT_CA_PATH) as f:
-        with open(INT_CA_PATH) as f: inter = f.read()
+            root = f.read()
-        with open(CA_CHAIN_PATH, "w") as f: f.write(inter + "\n" + root)
+        with open(INT_CA_PATH) as f:
-    except: pass
+            inter = f.read()
+        with open(CA_CHAIN_PATH, "w") as f:
+            f.write(inter + "\n" + root)
+    except Exception as e:
+        logger.warning("CA chain setup failed: %s", sanitize_error(str(e)))
+
+def get_user_id(conn, username: str) -> int:
+    row = conn.execute("SELECT id FROM users WHERE username = ?", (username,)).fetchone()
+    return row["id"] if row else 1
+
 
 def render(name, ctx):
     return HTMLResponse(jinja_env.get_template(name).render(**ctx))
@@ -64,12 +110,17 @@ async def list_domains(user: str = Depends(get_current_user)):
     return [dict(r) for r in rows]
 
 @app.post("/api/domains")
-async def create_domain(name: str = Form(...), description: str = Form(""),
+async def create_domain(
-                        user: str = Depends(get_current_user)):
+    name: str = Form(...),
+    description: str = Form(""),
+    user: str = Depends(get_current_user),
+):
     conn = get_db()
     cur = conn.cursor()
-    cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
+    cur.execute(
-                 (name, description, 1))
+        "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
+        (name, description, get_user_id(conn, user)),
+    )
     conn.commit()
     conn.close()
     return {"status": "ok"}
@@ -77,18 +128,28 @@ async def create_domain(name: str = Form(...), description: str = Form(""),
 @app.get("/api/certs")
 async def list_certs(user: str = Depends(get_current_user)):
     conn = get_db()
-    rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall()
+    rows = conn.execute(
+        "SELECT c.*, d.name as domain_name FROM certificates c "
+        "LEFT JOIN domains d ON c.domain_id = d.id "
+        "ORDER BY c.created_at DESC"
+    ).fetchall()
     conn.close()
     return [dict(r) for r in rows]
 
 @app.post("/api/certs/request")
-async def request_cert(cn: str = Form(...), sans: str = Form(""),
+async def request_cert(
-                       days: int = Form(365), domain_id: int = Form(0),
+    cn: str = Form(...),
-                       user: str = Depends(get_current_user)):
+    sans: str = Form(""),
+    days: int = Form(365),
+    domain_id: int = Form(0),
+    user: str = Depends(get_current_user),
+):
     conn = get_db()
     cur = conn.cursor()
-    cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)",
+    cur.execute(
-                 (domain_id, cn, sans, "pending", 1))
+        "INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)",
+        (domain_id or None, cn, sans, "pending", get_user_id(conn, user)),
+    )
     conn.commit()
     cid = cur.lastrowid
     conn.close()
@@ -97,34 +158,50 @@ async def request_cert(cn: str = Form(...), sans: str = Form(""),
 @app.post("/api/certs/{cert_id}/sign")
 async def sign_cert(cert_id: int, user: str = Depends(get_current_user)):
     conn = get_db()
-    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
+    row = conn.execute(
+        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
+    ).fetchone()
     if not row or row["status"] != "pending":
         conn.close()
         raise HTTPException(400, "Not found or already signed")
     conn.close()
+    try:
         result, err = build_leaf_cert(row["subject"], row["san"], 365)
-    if err: raise HTTPException(500, f"Signing failed: {err}")
+    except Exception as e:
-    cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
+        logger.error("Signing failed: %s", sanitize_error(str(e)))
-    kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
+        raise HTTPException(500, "Signing failed")
-    open(cf, "w").write(result["cert_pem"])
+    if err:
-    open(kf, "w").write(result["key_pem"])
+        raise HTTPException(500, "Signing failed")
-    os.chmod(cf, 0o640); os.chmod(kf, 0o600)
+    cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt"
+    kf = f"{ISSUED_DIR}/cert-{result['serial']}.key"
+    with open(cf, "w") as f:
+        f.write(result["cert_pem"])
+    with open(kf, "w") as f:
+        f.write(result["key_pem"])
+    os.chmod(cf, 0o640)
+    os.chmod(kf, 0o600)
     conn = get_db()
-    conn.execute("UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
+    conn.execute(
-                 ("issued", result["serial"], cf, datetime.datetime.now().isoformat(), result["expires_at"], cert_id))
+        "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
-    conn.commit(); conn.close()
+        ("issued", result["serial"], cf, datetime.now(timezone.utc).isoformat(), result["expires_at"], cert_id),
+    )
+    conn.commit()
+    conn.close()
     return {"status": "ok", "serial": result["serial"]}
 
 @app.get("/api/certs/{cert_id}/pem")
 async def download_pem(cert_id: int, request: Request = None):
-    """Download cert + chain as bundled PEM."""
     user = get_user_from_cookie(request)
-    if not user: raise HTTPException(401, "Login required")
+    if not user:
+        raise HTTPException(401, "Login required")
     conn = get_db()
-    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
+    row = conn.execute(
+        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
+    ).fetchone()
     conn.close()
-    if not row or row["status"] != "issued": raise HTTPException(404)
+    if not row or row["status"] != "issued":
-    pem_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pem"
+        raise HTTPException(404)
+    pem_path = f"{TMP_DIR}/cert-{row['serial']}.pem"
     with open(row["cert_path"]) as f:
         cert_pem = f.read()
     with open(CA_CHAIN_PATH) as f:
@@ -134,16 +211,24 @@ async def download_pem(cert_id: int, request: Request = None):
     return FileResponse(pem_path, media_type="application/x-pem-file", filename=f"cert-{row['serial']}.pem")
 
 @app.get("/api/certs/{cert_id}/pfx")
-async def download_pfx(cert_id: int, password: str = "certauth", request: Request = None):
+async def download_pfx(
-    """Download cert + key + chain as PKCS12/PFX."""
+    cert_id: int,
+    password: str = PFX_DEFAULT_PASSWORD,
+    request: Request = None,
+):
     user = get_user_from_cookie(request)
-    if not user: raise HTTPException(401, "Login required")
+    if not user:
+        raise HTTPException(401, "Login required")
     conn = get_db()
-    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
+    row = conn.execute(
+        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
+    ).fetchone()
     conn.close()
-    if not row or row["status"] != "issued": raise HTTPException(404)
+    if not row or row["status"] != "issued":
+        raise HTTPException(404)
     kf = row["cert_path"].replace(".crt", ".key")
-    if not os.path.exists(kf): raise HTTPException(404)
+    if not os.path.exists(kf):
+        raise HTTPException(404)
     from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption
     from cryptography import x509
     with open(row["cert_path"], "rb") as f:
@@ -155,33 +240,30 @@ async def download_pfx(cert_id: int, password: str = "certauth", request: Reques
         for cert_pem in f.read().split(b"-----END CERTIFICATE-----"):
             cert_pem = cert_pem.strip()
             if cert_pem:
-                chain_certs.append(x509.load_pem_x509_certificate(cert_pem + b"\n-----END CERTIFICATE-----"))
+                chain_certs.append(
+                    x509.load_pem_x509_certificate(
+                        cert_pem + b"\n-----END CERTIFICATE-----"
+                    )
+                )
     pfx_data = pkcs12.serialize_key_and_certificates(
         name=row["subject"].encode(),
         key=key,
         cert=leaf,
         cas=chain_certs or None,
-        encryption_algorithm=BestAvailableEncryption(password.encode())
+        encryption_algorithm=BestAvailableEncryption(password.encode()),
     )
-    pfx_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pfx"
+    pfx_path = f"{TMP_DIR}/cert-{row['serial']}.pfx"
     with open(pfx_path, "wb") as f:
         f.write(pfx_data)
     return FileResponse(pfx_path, media_type="application/x-pkcs12", filename=f"cert-{row['serial']}.pfx")
 
 @app.get("/api/health")
-async def health(): return {"status": "ok"}
+async def health():
+    return {"status": "ok"}
 
 @app.get("/api/ca-chain")
-async def ca_chain(): return FileResponse(CA_CHAIN_PATH, filename="ca-chain.crt")
+async def ca_chain():
-
+    return FileResponse(CA_CHAIN_PATH, filename="ca-chain.crt")
-def get_user_from_cookie(request: Request):
-    token = request.cookies.get("token")
-    if not token:
-        return None
-    try:
-        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
-    except:
-        return None
 
 @app.get("/", response_class=HTMLResponse)
 async def dashboard(request: Request):
@@ -189,63 +271,115 @@ async def dashboard(request: Request):
     if not user:
         return RedirectResponse("/login", status_code=302)
     conn = get_db()
-    certs = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC LIMIT 20").fetchall()
+    certs = conn.execute(
+        "SELECT c.*, d.name as domain_name FROM certificates c "
+        "LEFT JOIN domains d ON c.domain_id = d.id "
+        "ORDER BY c.created_at DESC LIMIT 20"
+    ).fetchall()
     domains = conn.execute("SELECT * FROM domains").fetchall()
-    p = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("pending",)).fetchone()["c"]
+    p = conn.execute(
-    i = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("issued",)).fetchone()["c"]
+        "SELECT COUNT(*) as c FROM certificates WHERE status = ?",
+        ("pending",),
+    ).fetchone()["c"]
+    i = conn.execute(
+        "SELECT COUNT(*) as c FROM certificates WHERE status = ?",
+        ("issued",),
+    ).fetchone()["c"]
     conn.close()
-    return render("dashboard.html", {"request": request, "user": user,
+    return render(
-        "certs": [dict(r) for r in certs], "domains": [dict(r) for r in domains],
+        "dashboard.html",
-        "pending": p, "issued": i})
+        {
+            "request": request,
+            "user": user,
+            "certs": [dict(r) for r in certs],
+            "domains": [dict(r) for r in domains],
+            "pending": p,
+            "issued": i,
+            "csrf_token": get_csrf_token(user.get("sub", "anon")),
+        },
+    )
 
 @app.get("/login", response_class=HTMLResponse)
 async def login_page(request: Request):
-    return render("login.html", {"request": request, "error": None})
+    return render("login.html", {"request": request, "error": None, "csrf_token": get_csrf_token("anon")})
 
 @app.post("/login")
-async def login_post(username: str = Form(...), password: str = Form(...)):
+async def login_post(
+    username: str = Form(...),
+    password: str = Form(...),
+    csrf_token: str = Form(""),
+):
+    if not verify_csrf_token("anon", csrf_token):
+        return render(
+            "login.html",
+            {"request": None, "error": "Invalid request", "csrf_token": get_csrf_token("anon")},
+        )
     conn = get_db()
-    row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
+    row = conn.execute(
+        "SELECT * FROM users WHERE username = ?", (username,)
+    ).fetchone()
     conn.close()
     if not row or not verify_password(password, row["password_hash"]):
-        return render("login.html", {"request": None, "error": "Invalid credentials"})
+        return render(
+            "login.html",
+            {"request": None, "error": "Invalid credentials", "csrf_token": get_csrf_token("anon")},
+        )
     token = create_access_token({"sub": username})
     resp = RedirectResponse("/", status_code=302)
-    resp.set_cookie("token", token, httponly=True, samesite="lax", path="/")
+    resp.set_cookie("token", token, httponly=True, samesite="lax", secure=True, path="/")
     return resp
 
-
-
 @app.post("/api/certs/{cert_id}/sign/web")
 async def sign_cert_web(cert_id: int, request: Request = None):
     user = get_user_from_cookie(request)
     if not user:
         return RedirectResponse("/login", status_code=302)
+    csrf = request.form.get("csrf_token", "")
+    if not verify_csrf_token(user.get("sub", "anon"), csrf):
+        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
     conn = get_db()
-    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
+    row = conn.execute(
+        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
+    ).fetchone()
     if not row or row["status"] != "pending":
         conn.close()
-        raise HTTPException(400, "Not found or already issued")
+        return HTMLResponse("<span class='text-red-400'>Not found or already issued</span>", status_code=400)
     conn.close()
     try:
         result, err = build_leaf_cert(row["subject"], row["san"], 365)
         if err:
-            return HTMLResponse(f'<span class="text-red-400">Issue failed: {err}</span>')
+            return HTMLResponse("<span class='text-red-400'>Issue failed</span>")
-        cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
+        cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt"
-        kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
+        kf = f"{ISSUED_DIR}/cert-{result['serial']}.key"
-        open(cf, "w").write(result["cert_pem"])
+        with open(cf, "w") as f:
-        open(kf, "w").write(result["key_pem"])
+            f.write(result["cert_pem"])
+        with open(kf, "w") as f:
+            f.write(result["key_pem"])
         os.chmod(cf, 0o640)
         os.chmod(kf, 0o600)
         conn2 = get_db()
-        conn2.execute("UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
+        conn2.execute(
-                      ("issued", result["serial"], cf, datetime.datetime.now().isoformat(), result["expires_at"], cert_id))
+            "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
+            (
+                "issued",
+                result["serial"],
+                cf,
+                datetime.now(timezone.utc).isoformat(),
+                result["expires_at"],
+                cert_id,
+            ),
+        )
         conn2.commit()
         conn2.close()
-        return HTMLResponse(f'<span class="text-green-400">Issued! <a href="/api/certs/{cert_id}/pem" class="underline">PEM</a> | <a href="/api/certs/{cert_id}/pfx" class="underline">PFX</a> | <a href="/certs" class="underline">Refresh</a></span>')
+        return HTMLResponse(
+            f'<span class="text-green-400">Issued! '
+            f'<a href="/api/certs/{cert_id}/pem" class="underline">PEM</a> | '
+            f'<a href="/api/certs/{cert_id}/pfx" class="underline">PFX</a> | '
+            f'<a href="/certs" class="underline">Refresh</a></span>'
+        )
     except Exception as ex:
-        return HTMLResponse(f'<span class="text-red-400">Issue failed: {str(ex)}</span>')
+        logger.error("Signing failed: %s", sanitize_error(str(ex)))
-
+        return HTMLResponse("<span class='text-red-400'>Issue failed</span>")
 
 @app.get("/logout")
 async def logout():
@@ -253,174 +387,175 @@ async def logout():
     resp.delete_cookie("token", path="/")
     return resp
 
-# --- Web API (cookie auth) ---
 @app.post("/api/domains/web")
-async def create_domain_web(name: str = Form(...), description: str = Form(""), request: Request = None):
+async def create_domain_web(
+    name: str = Form(...),
+    description: str = Form(""),
+    csrf_token: str = Form(""),
+    request: Request = None,
+):
     user = get_user_from_cookie(request)
     if not user:
         return RedirectResponse("/login", status_code=302)
+    if not verify_csrf_token(user.get("sub", "anon"), csrf_token):
+        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
     conn = get_db()
     cur = conn.cursor()
-    cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1))
+    cur.execute(
+        "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
+        (name, description, get_user_id(conn, user.get("sub"))),
+    )
     conn.commit()
     conn.close()
-    return HTMLResponse('<span class="text-green-400">Domain registered! <a href="/domains" class="underline">Refresh</a></span>')
+    return HTMLResponse(
+        '<span class="text-green-400">Domain registered! '
+        '<a href="/domains" class="underline">Refresh</a></span>'
+    )
 
 @app.post("/api/certs/web/request")
-async def request_cert_web(cn: str = Form(...), sans: str = Form(""), days: int = Form(365), domain_id: int = Form(0), request: Request = None):
+async def request_cert_web(
+    cn: str = Form(...),
+    sans: str = Form(""),
+    days: int = Form(365),
+    domain_id: int = Form(0),
+    csrf_token: str = Form(""),
+    request: Request = None,
+):
     user = get_user_from_cookie(request)
     if not user:
         return RedirectResponse("/login", status_code=302)
+    if not verify_csrf_token(user.get("sub", "anon"), csrf_token):
+        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
     conn = get_db()
     cur = conn.cursor()
-    # Look up domain by CN if domain_id not provided
     if domain_id == 0:
         cur.execute("SELECT id FROM domains WHERE name=?", (cn,))
         row = cur.fetchone()
         domain_id = row[0] if row else None
-    cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", (domain_id, cn, sans, "pending", 1))
+    cur.execute(
+        "INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)",
+        (domain_id, cn, sans, "pending", get_user_id(conn, user.get("sub"))),
+    )
     conn.commit()
     conn.close()
-    return HTMLResponse('<span class="text-green-400">Certificate requested! Click Issue below. <a href="/certs" class="underline">Refresh</a></span>')
+    return HTMLResponse(
+        '<span class="text-green-400">Certificate requested! Click Issue below. '
+        '<a href="/certs" class="underline">Refresh</a></span>'
+    )
 
 @app.get("/domains", response_class=HTMLResponse)
 async def domains_page(request: Request):
     user = get_user_from_cookie(request)
-    if not user: return RedirectResponse("/login", status_code=302)
+    if not user:
+        return RedirectResponse("/login", status_code=302)
     conn = get_db()
     rows = conn.execute("SELECT * FROM domains ORDER BY created_at DESC").fetchall()
     conn.close()
-    return render("domains.html", {"request": request, "user": user, "domains": [dict(r) for r in rows]})
+    return render(
+        "domains.html",
+        {
+            "request": request,
+            "user": user,
+            "domains": [dict(r) for r in rows],
+            "csrf_token": get_csrf_token(user.get("sub", "anon")),
+        },
+    )
 
 @app.get("/certs", response_class=HTMLResponse)
 async def certs_page(request: Request):
     user = get_user_from_cookie(request)
-    if not user: return RedirectResponse("/login", status_code=302)
+    if not user:
+        return RedirectResponse("/login", status_code=302)
     conn = get_db()
-    rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall()
+    rows = conn.execute(
+        "SELECT c.*, d.name as domain_name FROM certificates c "
+        "LEFT JOIN domains d ON c.domain_id = d.id "
+        "ORDER BY c.created_at DESC"
+    ).fetchall()
     domains = conn.execute("SELECT * FROM domains").fetchall()
     conn.close()
-    return render("certs.html", {"request": request, "user": user, "certs": [dict(r) for r in rows],
+    return render(
-                                  "domains": [dict(r) for r in domains]})
+        "certs.html",
+        {
+            "request": request,
+            "user": user,
+            "certs": [dict(r) for r in rows],
+            "domains": [dict(r) for r in domains],
+            "csrf_token": get_csrf_token(user.get("sub", "anon")),
+        },
+    )
 
 @app.get("/history", response_class=HTMLResponse)
 async def history_page(request: Request):
     user = get_user_from_cookie(request)
-    if not user: return RedirectResponse("/login", status_code=302)
+    if not user:
+        return RedirectResponse("/login", status_code=302)
     conn = get_db()
-    rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall()
+    rows = conn.execute(
+        "SELECT c.*, d.name as domain_name FROM certificates c "
+        "LEFT JOIN domains d ON c.domain_id = d.id "
+        "ORDER BY c.created_at DESC"
+    ).fetchall()
     conn.close()
-    return render("history.html", {"request": request, "user": user, "certs": [dict(r) for r in rows]})
+    return render(
+        "history.html",
+        {
+            "request": request,
+            "user": user,
+            "certs": [dict(r) for r in rows],
+        },
+    )
 
 @app.get("/setup", response_class=HTMLResponse)
 async def setup_page(request: Request):
     user = get_user_from_cookie(request)
-    if not user: return RedirectResponse("/login", status_code=302)
+    if not user:
-    return render("setup.html", {"request": request, "user": user})
+        return RedirectResponse("/login", status_code=302)
+    return render(
+        "setup.html",
+        {
+            "request": request,
+            "user": user,
+        },
+    )
 
 @app.get("/setup.sh")
 async def setup_sh():
-    """One-liner bash setup script for Linux/macOS."""
     script = r'''#!/bin/bash
 set -e
-# CertAuth CA Chain Installer
-# Usage: curl -sL http://<certauth-ip>/setup.sh | bash
-#        curl -sL http://<certauth-ip>/setup.sh | sudo bash
-
-# Auto-detect CertAuth server IP
 DETECTED_IP=""
 if [[ -n "$1" ]]; then
     DETECTED_IP="$1"
 elif [[ -n "$CERTAUTH_IP" ]]; then
     DETECTED_IP="$CERTAUTH_IP"
 else
-    # Try Linux hostname -I first
     DETECTED_IP=$(hostname -I 2>/dev/null | awk '{print $1}') || true
-    # Fallback: ip route (Linux)
     [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(ip route get 1 2>/dev/null | awk '{print $7}' | head -1) || true
-    # Fallback: ifconfig (macOS/BSD)
-    [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(ifconfig 2>/dev/null | grep -E '^\s+(inet )' | awk '{print $2}' | grep -v '127.0.0.1' | head -1) || true
-    # Fallback: networksetup (macOS only)
-    [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(networksetup -getinfo $(networksetup -listallhardwareports 2>/dev/null | awk '/Hardware Port:/ {getline; gsub(/^[ \t]+/, ""); print}') 2>/dev/null | grep 'IP address:' | awk '{print $3}') || true
 fi
-
-# Prompt if auto-detection failed
 if [[ -z "$DETECTED_IP" ]]; then
-    read -r -p "Enter CertAuth server IP (e.g., 192.168.8.248): " DETECTED_IP
+    read -r -p "Enter CertAuth server IP: " DETECTED_IP
 fi
-
 CHAIN_URL="http://$DETECTED_IP/api/ca-chain"
-
 echo "Downloading CA chain..."
-curl -sLk "$CHAIN_URL" -o /tmp/ca-chain.crt || { echo "Failed to download CA chain from $CHAIN_URL"; exit 1; }
+curl -sLk "$CHAIN_URL" -o /tmp/ca-chain.crt || { echo "Failed"; exit 1; }
-
-# Detect OS and install
 if [[ -f /etc/os-release ]]; then
     . /etc/os-release
-    if [[ "$ID" == "debian" || "$ID" == "ubuntu" || "$ID" == "linuxmint" ]]; then
+    if [[ "$ID" == "debian" || "$ID" == "ubuntu" ]]; then
         sudo cp /tmp/ca-chain.crt /usr/local/share/ca-certificates/certauth.crt
         sudo update-ca-certificates
-        echo "✅ CA chain installed (Debian/Ubuntu)"
-    elif [[ "$ID" == "centos" || "$ID" == "rhel" || "$ID" == "fedora" ]]; then
-        # Install to system trust store (curl, openssl, etc.)
-        sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt
-        sudo update-ca-trust
-
-        # Install root CA to NSS database (Firefox, Thunderbird, etc.)
-        # The root CA is the self-signed cert (second cert in chain)
-        if command -v certutil &>/dev/null; then
-            sudo certutil -D -n "CertAuth Root CA" -d sql:/etc/pki/nssdb/ 2>/dev/null || true
-            python3 -c "
-import re, subprocess
-with open('/tmp/ca-chain.crt') as f:
-    content = f.read()
-certs = re.findall(r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----', content, re.DOTALL)
-# Find the self-signed root cert (subject == issuer)
-for cert in certs:
-    subj = subprocess.run(['openssl', 'x509', '-noout', '-subject'], input=cert, capture_output=True, text=True).stdout
-    iss = subprocess.run(['openssl', 'x509', '-noout', '-issuer'], input=cert, capture_output=True, text=True).stdout
-    if subj.replace('subject=', '') == iss.replace('issuer=', ''):
-        with open('/tmp/certauth-root.crt', 'w') as rf:
-            rf.write(cert + '\n')
-        break
-" 2>/dev/null
-            if [[ -f /tmp/certauth-root.crt ]]; then
-                sudo certutil -A -n "CertAuth Root CA" -t "CT,Cu,Tu" -d sql:/etc/pki/nssdb/ -i /tmp/certauth-root.crt 2>/dev/null
-                echo "✅ Root CA imported to NSS database"
-                rm -f /tmp/certauth-root.crt
-            fi
-        fi
-        echo "✅ CA chain installed (RHEL/CentOS/Fedora)"
-    elif [[ "$ID" == "arch" ]]; then
-        sudo cp /tmp/ca-chain.crt /etc/ca-certificates/trust-source/anchors/certauth.crt
-        sudo update-ca-trust
-        echo "✅ CA chain installed (Arch)"
     elif [[ "$ID" == "alpine" ]]; then
         sudo cp /tmp/ca-chain.crt /usr/local/share/ca-certificates/certauth.crt
         sudo update-ca-certificates
-        echo "✅ CA chain installed (Alpine)"
     else
-        echo "❌ Unsupported Linux distribution: $ID"
+        sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt
-        echo "   Download /tmp/ca-chain.crt and install manually"
+        sudo update-ca-trust 2>/dev/null || true
-        exit 1
     fi
 elif [[ "$(uname)" == "Darwin" ]]; then
     sudo security add-trusted-cert -d -r trustRoot -k /Library/Keychains/System.keychain /tmp/ca-chain.crt
-    echo "✅ CA chain installed (macOS)"
 else
-    echo "❌ Unsupported OS: $(uname -s)"
+    echo "Unsupported OS"
-    echo "   Download /tmp/ca-chain.crt and install manually"
     exit 1
 fi
-
-# Verify
-VERIFY_URL="${CHAIN_URL%/api/ca-chain}"
-if curl -sL "$VERIFY_URL" &>/dev/null; then
-    echo "🌐 Server at $VERIFY_URL is reachable"
-else
-    echo "⚠️  Server at $VERIFY_URL is not reachable (expected if not on same network)"
-fi
-
 rm -f /tmp/ca-chain.crt
 echo "Done!"
 '''
@@ -428,32 +563,14 @@ echo "Done!"
 
 @app.get("/setup.ps1")
 async def setup_ps1():
-    """PowerShell setup script for Windows."""
+    script = r'''
-    script = r'''# CertAuth CA Chain Installer for Windows
-# Usage: iex (New-Object Net.WebClient).DownloadString("http://<certauth-ip>/setup.ps1")
-#        iwr http://<certauth-ip>/setup.ps1 -UseBasicParsing | iex
-
 param([string]$CertAuthIP = "")
-
 if (-not $CertAuthIP) {
-    # Try to detect from environment or prompt
+    $CertAuthIP = Read-Host "Enter CertAuth server IP"
-    $CertAuthIP = Read-Host "Enter CertAuth server IP (e.g., 192.168.8.248)"
 }
-
 $ChainUrl = "http://$CertAuthIP/api/ca-chain"
 $ChainPath = "$env:TEMP\ca-chain.crt"
-
-Write-Host "Downloading CA chain from $ChainUrl ..." -ForegroundColor Cyan
-try {
 (New-Object Net.WebClient).DownloadFile($ChainUrl, $ChainPath)
-} catch {
-    Write-Host "Failed to download CA chain: $_" -ForegroundColor Red
-    exit 1
-}
-
-# Install to Local Machine Trusted Root store
-Write-Host "Installing to Trusted Root Certification Authorities..." -ForegroundColor Cyan
-try {
 $store = New-Object System.Security.Cryptography.X509Certificates.X509Store(
     [System.Security.Cryptography.X509Certificates.StoreName]::Root,
     [System.Security.Cryptography.X509Certificates.StoreLocation]::LocalMachine)
@@ -461,21 +578,6 @@ try {
 $cert = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2($ChainPath)
 $store.Add($cert)
 $store.Close()
-    Write-Host "CA chain installed successfully!" -ForegroundColor Green
-} catch {
-    Write-Host "Failed to install: $_" -ForegroundColor Red
-    Write-Host "Run as Administrator and try again." -ForegroundColor Yellow
-    exit 1
-}
-
-# Verify
-try {
-    $response = Invoke-WebRequest -Uri "http://$CertAuthIP/api/health" -TimeoutSec 3 -ErrorAction Stop
-    Write-Host "Server at http://$CertAuthIP is reachable." -ForegroundColor Green
-} catch {
-    Write-Host "Server at http://$CertAuthIP is not reachable (expected if not on same network)." -ForegroundColor Yellow
-}
-
 Remove-Item $ChainPath -Force -ErrorAction SilentlyContinue
 Write-Host "Done!" -ForegroundColor Green
 '''

api/models.py

@@ -1,5 +1,9 @@
-import sqlite3, datetime, secrets, bcrypt, os
+import sqlite3
-from config import DB_PATH, ADMIN_USERNAME
+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)
@@ -7,6 +11,7 @@ def get_db():
     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():
@@ -57,21 +62,30 @@ def init_db():
         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(b"CHANGE_ME_ADMIN_PASS", bcrypt.gensalt())
+        pw_hash = bcrypt.hashpw(ADMIN_PASSWORD.encode(), bcrypt.gensalt())
-        if isinstance(pw_hash, bytes): pw_hash = pw_hash.decode()
+        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()
+    if isinstance(hash_, str):
+        hash_ = hash_.encode()
     return bcrypt.checkpw(password.encode(), hash_)

api/signing.py

@@ -1,138 +1,227 @@
+"""Certificate signing via the YubiKey-held Intermediate CA key.
+
+The private key never leaves the YubiKey. Signing works in three steps:
+
+1. Build the TBS (to-be-signed) certificate bytes.
+2. Hand the TBS to ``pkcs11-tool`` (OpenSC PKCS#11) which signs it on the
+   YubiKey with ECDSA-SHA384.
+3. Reassemble the certificate DER from TBS + signature algorithm + YubiKey
+   signature, then verify it against the Intermediate CA public key before
+   trusting the result.
+"""
+
+import logging
+import os
+import subprocess
+import tempfile
+from datetime import datetime, timedelta, timezone
 
-import subprocess, datetime, os, hashlib, ipaddress, re
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption
 from cryptography.x509.oid import NameOID
-from config import *
 
-TMP_DIR = "/var/lib/certauth/tmp"
+from config import (
-os.makedirs(TMP_DIR, exist_ok=True)
+    CA_COUNTRY,
+    CA_ORG,
+    INT_CA_PATH,
+    PKCS11_MODULE,
+    PKCS11_TOKEN_LABEL,
+    YK_INT_PIN,
+)
+from models import get_db
 
-def get_root_pub_key():
+logger = logging.getLogger(__name__)
-    with open(YK_PUB_ROOT, "rb") as f:
-        return serialization.load_pem_public_key(f.read())
 
-def get_int_pub_key():
+# Scratch dir for pkcs11-tool input/output files (per-process, unlinked after).
-    with open(YK_PUB_INT, "rb") as f:
+PKCS11_TMP = tempfile.mkdtemp(prefix="certauth_")
-        return serialization.load_pem_public_key(f.read())
 
-def get_root_ca_cert():
+
-    with open(ROOT_CA_PATH, "rb") as f:
+def _der_read_len(data: bytes, i: int) -> tuple:
+    """Read a DER length field at offset i -> (length, offset_after_field)."""
+    b0 = data[i]
+    if b0 < 0x80:
+        return b0, i + 1
+    n = b0 & 0x7F
+    return int.from_bytes(data[i + 1:i + 1 + n], "big"), i + 1 + n
+
+
+def _der_seq(body: bytes) -> bytes:
+    if len(body) < 0x80:
+        return b"\x30" + bytes([len(body)]) + body
+    lb = len(body).to_bytes((len(body).bit_length() + 7) // 8, "big")
+    return b"\x30" + bytes([0x80 | len(lb)]) + lb + body
+
+
+def _split_cert_der(der: bytes):
+    """Split a certificate DER into (tbs, signature_algorithm, signature)."""
+    if der[0] != 0x30:
+        raise ValueError("not a DER certificate")
+    _, body_start = _der_read_len(der, 1)
+    # tbsCertificate SEQUENCE
+    if der[body_start] != 0x30:
+        raise ValueError("bad tbsCertificate tag")
+    tbs_len, tbs_len_end = _der_read_len(der, body_start + 1)
+    tbs_end = tbs_len_end + tbs_len
+    tbs = der[body_start:tbs_end]
+    # signatureAlgorithm SEQUENCE
+    if der[tbs_end] != 0x30:
+        raise ValueError("bad signatureAlgorithm tag")
+    alg_len, alg_len_end = _der_read_len(der, tbs_end + 1)
+    alg_end = alg_len_end + alg_len
+    alg = der[tbs_end:alg_end]
+    return tbs, alg, der[alg_end:]
+
+
+def get_int_ca_cert():
+    with open(INT_CA_PATH, "rb") as f:
         return x509.load_pem_x509_certificate(f.read())
 
-def der_len(n):
-    if n < 0x80: return bytes([n])
-    elif n < 0x100: return bytes([0x81, n])
-    return bytes([0x82, n>>8, n&0xff])
 
-def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
+def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label=None):
-    tbs_file = os.path.join(TMP_DIR, "tbs_sign.der")
+    """Sign DER bytes with the YubiKey SIGN key via pkcs11-tool.
-    sig_file = os.path.join(TMP_DIR, "sig_out.bin")
+
-    with open(tbs_file, "wb") as f:
+    Returns (signature_BIT_STRING, None) on success or (None, error).
-        f.write(tbs_bytes)
+    """
-    r = subprocess.run([
+    with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=PKCS11_TMP) as tbs_file:
+        tbs_file.write(tbs_bytes)
+        tbs_path = tbs_file.name
+    sig_path = os.path.join(PKCS11_TMP, f"sig_{os.path.basename(tbs_path)}")
+    cmd = [
         "sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
-        "--login", "--pin", yk_pin,
+        "--login", "--pin-source", "stdin",
         "--sign", "--mechanism", "ECDSA-SHA384",
-        "--token-label", token_label,
         "--label", "SIGN key",
-        "--input-file", tbs_file,
+        "--input-file", tbs_path,
-        "--output-file", sig_file
+        "--output-file", sig_path,
-    ], capture_output=True, text=True)
+    ]
+    label = token_label if token_label is not None else PKCS11_TOKEN_LABEL
+    if label:
+        cmd[3:3] = ["--token-label", label]
+    try:
+        r = subprocess.run(
+            cmd,
+            input=yk_pin.encode(),
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
         if r.returncode != 0:
-        return None, r.stderr
+            logger.error("YubiKey signing failed: %s", r.stderr[:200])
-    with open(sig_file, "rb") as f:
+            return None, "Signing failed"
+        with open(sig_path, "rb") as f:
             raw = f.read()
+        # pkcs11-tool returns raw r||s (48 bytes each for P-384); wrap in a
+        # BIT STRING carrying the DER ECDSA-Sig-Value.
         rb = raw[:48].lstrip(b"\x00") or b"\x00"
         sb = raw[48:].lstrip(b"\x00") or b"\x00"
-    if rb[0] & 0x80: rb = b"\x00" + rb
+        if rb[0] & 0x80:
-    if sb[0] & 0x80: sb = b"\x00" + sb
+            rb = b"\x00" + rb
+        if sb[0] & 0x80:
+            sb = b"\x00" + sb
         r_der = b"\x02" + bytes([len(rb)]) + rb
         s_der = b"\x02" + bytes([len(sb)]) + sb
         seq = b"\x30" + bytes([len(r_der + s_der)]) + r_der + s_der
         bs = b"\x00" + seq
         return b"\x03" + bytes([len(bs)]) + bs, None
+    finally:
+        for f in [tbs_path, sig_path]:
+            try:
+                os.unlink(f)
+            except OSError:
+                pass
+
 
 def build_leaf_cert(cn, sans, days=365):
-    root_cert = get_root_ca_cert()
+    """Issue a leaf certificate for cn, signed by the Intermediate CA.
-    int_pub = get_int_pub_key()
+
-    root_pub = get_root_pub_key()
+    Returns ({serial, cert_pem, key_pem, expires_at}, None) or (None, error).
+    The private key stays on the server (leaf keys are not hardware-backed);
+    only the CA signing key requires the YubiKey.
+    """
+    int_cert = get_int_ca_cert()
     leaf_key = ec.generate_private_key(ec.SECP384R1())
     subject = x509.Name([
-        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+        x509.NameAttribute(NameOID.COUNTRY_NAME, CA_COUNTRY),
-        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, CA_ORG),
         x509.NameAttribute(NameOID.COMMON_NAME, cn),
     ])
-    issuer = x509.Name([
+    builder = (
-        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+        x509.CertificateBuilder()
-        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
+        .subject_name(subject)
-        x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"),
+        .issuer_name(int_cert.subject)
-    ])
-    builder = (x509.CertificateBuilder()
-        .subject_name(subject).issuer_name(issuer)
         .public_key(leaf_key.public_key())
         .serial_number(x509.random_serial_number())
-        .not_valid_before(datetime.datetime.now(datetime.timezone.utc))
+        .not_valid_before(datetime.now(timezone.utc))
-        .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=days))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=days))
         .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
-        .add_extension(x509.KeyUsage(
+        .add_extension(
-            digital_signature=True, key_encipherment=True,
+            x509.KeyUsage(
-            key_cert_sign=False, crl_sign=False,
+                digital_signature=True,
-            content_commitment=False, data_encipherment=False,
+                key_encipherment=True,
-            key_agreement=False, encipher_only=False, decipher_only=False), critical=True)
+                content_commitment=False,
-        .add_extension(x509.ExtendedKeyUsage([
+                data_encipherment=False,
-            x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
+                key_agreement=False,
-            x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
+                key_cert_sign=False,
-        ]), critical=False)
+                crl_sign=False,
-        .add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False)
+                encipher_only=False,
-        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(int_pub), critical=False))
+                decipher_only=False,
+            ),
+            critical=True,
+        )
+        .add_extension(
+            x509.ExtendedKeyUsage([
+                x509.OID_SERVER_AUTH,
+                x509.OID_CLIENT_AUTH,
+            ]),
+            critical=False,
+        )
+    )
     if sans:
-        san_list = []
+        san_names = [x509.DNSName(s.strip()) for s in sans.split(",") if s.strip()]
-        for s in sans.split(","):
+        if san_names:
-            s = s.strip()
+            builder = builder.add_extension(
-            # Check if it's an IP address
+                x509.SubjectAlternativeName(san_names),
-            if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s):
+                critical=False,
-                san_list.append(x509.IPAddress(ipaddress.ip_address(s)))
+            )
-            else:
+
-                san_list.append(x509.DNSName(s))
+    # 1) TBS bytes: sign with a throwaway key of identical parameters —
-        builder = builder.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
+    #    the TBS block is independent of the signing key.
-    tmp = ec.generate_private_key(ec.SECP384R1())
+    dummy_key = ec.generate_private_key(ec.SECP384R1())
-    temp = builder.sign(tmp, hashes.SHA384())
+    tbs, sig_alg, _ = _split_cert_der(builder.sign(dummy_key, hashes.SHA384()))
-    td = temp.public_bytes(serialization.Encoding.DER)
+
-    o = 1
+    # 2) Real signature from the YubiKey.
-    if td[o] & 0x80: n = td[o] & 0x7f; o += 1 + n
+    yk_sig, err = sign_tbs_with_yk(tbs, YK_INT_PIN)
-    else: o += 1
+    if err:
-    tbs_start = o
+        return None, err
-    o += 1
+
-    if td[o] & 0x80: n = td[o] & 0x7f; tl = int.from_bytes(td[o+1:o+1+n], "big"); o += 1 + n
+    # 3) Reassemble and verify before trusting.
-    else: tl = td[o]; o += 1
+    cert = x509.load_der_x509_certificate(_der_seq(tbs + sig_alg + yk_sig))
-    tbs_end = o + tl
+    cert.verify(int_cert.public_key())
-    tbs_full = td[tbs_start:tbs_end]
+
-    alg_start = tbs_end
-    o2 = alg_start + 1
-    if td[o2] & 0x80: n = td[o2] & 0x7f; al = int.from_bytes(td[o2+1:o2+1+n], "big"); o2 += 1 + n
-    else: al = td[o2]; o2 += 1
-    alg_full = td[alg_start:o2+al]
-    new_sig, err = sign_tbs_with_yk(tbs_full, YK_INT_PIN)
-    if new_sig is None: return None, err
-    content = tbs_full + alg_full + new_sig
-    cl = len(content)
-    final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
-    der_file = os.path.join(TMP_DIR, "leaf.der")
-    pem_file = os.path.join(TMP_DIR, "leaf.pem")
-    with open(der_file, "wb") as f: f.write(final)
-    r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
-                         "-in", der_file, "-out", pem_file],
-                       capture_output=True, text=True)
-    if r.returncode != 0: return None, r.stderr
-    with open(pem_file) as f: leaf_pem = f.read()
     key_pem = leaf_key.private_bytes(
-        encoding=serialization.Encoding.PEM,
+        encoding=Encoding.PEM,
         format=serialization.PrivateFormat.PKCS8,
-        encryption_algorithm=serialization.NoEncryption()
+        encryption_algorithm=NoEncryption(),
     ).decode()
-    cert = x509.load_pem_x509_certificate(leaf_pem.encode())
+    return (
-    serial = format(cert.serial_number, 'x')
+        {
-    return {"cert_pem": leaf_pem, "key_pem": key_pem, "serial": serial,
+            "serial": hex(cert.serial_number),
-            "expires_at": cert.not_valid_after.isoformat()}, None
+            "cert_pem": cert.public_bytes(Encoding.PEM).decode(),
+            "key_pem": key_pem,
+            "expires_at": cert.not_valid_after_utc.isoformat(),
+        },
+        None,
+    )
+
+
+def revoke_certificate(serial_hex: str, reason: str = "key_compromise"):
+    conn = get_db()
+    conn.execute(
+        "INSERT OR REPLACE INTO crl (serial, revoked_at, reason) VALUES (?, ?, ?)",
+        (serial_hex, datetime.now(timezone.utc).isoformat(), reason),
+    )
+    conn.commit()
+    conn.close()
+    logger.info("Certificate %s revoked: %s", serial_hex, reason)

api/templates/login.html

@@ -14,6 +14,7 @@
         <div class="bg-red-900/50 border border-red-700 rounded p-3 mb-4 text-red-300 text-sm">{{ error }}</div>
         {% endif %}
         <form method="post" action="/login">
+            <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
             <div class="mb-4">
                 <label class="block text-sm text-gray-400 mb-1">Username</label>
                 <input type="text" name="username" required

certauth-api.service

@@ -8,6 +8,7 @@ User=certauth
 Group=certauth
 WorkingDirectory=/opt/certauth/api
 Environment=PYTHONUNBUFFERED=1
+EnvironmentFile=/etc/certauth/certauth.env
 ExecStart=/usr/bin/python3 -m uvicorn main:app --host 127.0.0.1 --port 8000
 Restart=on-failure
 RestartSec=5

pyproject.toml

@@ -0,0 +1,42 @@
+[project]
+name = "certauth"
+version = "1.0.0"
+description = "YubiKey-backed self-contained Certificate Authority: FastAPI + web UI, PKCS#11 signing, SQLite store"
+readme = "README.md"
+requires-python = ">=3.9"
+license = { text = "MIT" }
+authors = [{ name = "Jarian Cottingham", email = "jarianc@proton.me" }]
+keywords = ["security", "certificate-authority", "yubikey", "pkcs11", "fastapi"]
+classifiers = [
+    "License :: OSI Approved :: MIT License",
+    "Programming Language :: Python :: 3",
+]
+dependencies = [
+    "fastapi>=0.110",
+    "uvicorn[standard]>=0.29",
+    "jinja2>=3.1",
+    "python-multipart>=0.0.9",
+    "bcrypt>=4.1",
+    "python-jose[cryptography]>=3.3",
+    "ecdsa>=0.18",
+    "cryptography>=42.0",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=7.0", "ruff>=0.1.0"]
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.ruff]
+line-length = 120
+target-version = "py39"
+exclude = [".git", "venv"]
+
+[tool.ruff.lint]
+select = ["E", "F", "W"]
+ignore = ["E501", "E741"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

requirements.txt

@@ -0,0 +1,8 @@
+fastapi>=0.110
+uvicorn[standard]>=0.29
+jinja2>=3.1
+python-multipart>=0.0.9
+bcrypt>=4.1
+python-jose[cryptography]>=3.3
+ecdsa>=0.18
+cryptography>=42.0

tests/conftest.py

@@ -0,0 +1,31 @@
+"""Test setup: configure the environment before importing the app.
+
+The API fails closed on missing configuration (by design), so tests
+provide a complete, isolated environment pointing at a temp directory.
+"""
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+API_DIR = Path(__file__).resolve().parent.parent / "api"
+sys.path.insert(0, str(API_DIR))
+
+_TMP = tempfile.mkdtemp(prefix="certauth-test-")
+
+os.environ["YK_ROOT_SERIAL"] = "10000001"
+os.environ["YK_INT_SERIAL"] = "10000002"
+os.environ["YK_ROOT_PIN"] = "123456"
+os.environ["YK_INT_PIN"] = "234567"
+os.environ["JWT_SECRET"] = "test-secret-0123456789abcdef0123456789abcdef"
+os.environ["ADMIN_USERNAME"] = "certauth"
+os.environ["ADMIN_PASSWORD"] = "test-admin-pass-123"
+os.environ["CERTAUTH_DB_PATH"] = os.path.join(_TMP, "test.db")
+os.environ["CERTAUTH_TMP_DIR"] = os.path.join(_TMP, "tmp")
+os.environ["CERTAUTH_CA_BASE"] = os.path.join(_TMP, "ca")
+os.environ["CERTAUTH_ISSUED_DIR"] = os.path.join(_TMP, "ca", "issued")
+os.environ["PKCS11_MODULE"] = "/usr/lib/opensc-pkcs11.so"
+
+os.makedirs(os.environ["CERTAUTH_ISSUED_DIR"], exist_ok=True)
+os.makedirs(os.environ["CERTAUTH_TMP_DIR"], exist_ok=True)

tests/test_api.py

@@ -0,0 +1,115 @@
+"""API endpoint tests (no YubiKey required)."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from main import app
+
+ADMIN = {"username": "certauth", "password": "test-admin-pass-123"}
+
+
+@pytest.fixture(scope="module")
+def client():
+    with TestClient(app) as c:
+        yield c
+
+
+@pytest.fixture(scope="module")
+def auth(client):
+    r = client.post("/api/token", json=ADMIN)
+    assert r.status_code == 200
+    return {"Authorization": f"Bearer {r.json()['access_token']}"}
+
+
+def test_health(client):
+    r = client.get("/api/health")
+    assert r.status_code == 200
+    assert r.json() == {"status": "ok"}
+
+
+def test_token_wrong_password(client):
+    r = client.post("/api/token", json={"username": "certauth", "password": "wrong"})
+    assert r.status_code == 401
+
+
+def test_token_unknown_user(client):
+    r = client.post("/api/token", json={"username": "nobody", "password": "x"})
+    assert r.status_code == 401
+
+
+def test_me(client, auth):
+    r = client.get("/api/me", headers=auth)
+    assert r.status_code == 200
+    assert r.json() == {"username": "certauth"}
+
+
+def test_me_without_token(client):
+    assert client.get("/api/me").status_code == 401
+
+
+def test_login_page_renders_with_csrf(client):
+    r = client.get("/login")
+    assert r.status_code == 200
+    assert "csrf_token" in r.text
+
+
+def test_dashboard_redirects_when_unauthenticated(client):
+    r = client.get("/", follow_redirects=False)
+    assert r.status_code == 302
+    assert r.headers["location"] == "/login"
+
+
+def test_login_requires_valid_csrf(client):
+    r = client.post(
+        "/login",
+        data={"username": "certauth", "password": "test-admin-pass-123", "csrf_token": "bogus"},
+        follow_redirects=False,
+    )
+    assert r.status_code == 200
+    assert "Invalid request" in r.text
+
+
+def test_domain_create_and_list(client, auth):
+    r = client.post(
+        "/api/domains",
+        data={"name": "test.example.ms", "description": "unit test domain"},
+        headers=auth,
+    )
+    assert r.status_code == 200
+    r = client.get("/api/domains", headers=auth)
+    assert r.status_code == 200
+    names = [d["name"] for d in r.json()]
+    assert "test.example.ms" in names
+
+
+def test_domain_requires_auth(client):
+    r = client.post("/api/domains", data={"name": "x.ms"})
+    assert r.status_code == 401
+
+
+def test_cert_request_lifecycle(client, auth):
+    client.post(
+        "/api/domains",
+        data={"name": "certtest.example.ms"},
+        headers=auth,
+    )
+    r = client.post(
+        "/api/certs/request",
+        data={"cn": "certtest.example.ms", "sans": "alt.example.ms"},
+        headers=auth,
+    )
+    assert r.status_code == 200
+    cert_id = r.json()["id"]
+
+    r = client.get("/api/certs", headers=auth)
+    row = next(c for c in r.json() if c["id"] == cert_id)
+    assert row["status"] == "pending"
+    assert row["subject"] == "certtest.example.ms"
+
+    # Signing requires a physical YubiKey; must fail cleanly, not 500-crash.
+    r = client.post(f"/api/certs/{cert_id}/sign", headers=auth)
+    assert r.status_code == 500
+
+
+def test_sign_missing_cert(client, auth):
+    assert client.post("/api/certs/99999/sign", headers=auth).status_code == 400

tests/test_auth.py

@@ -0,0 +1,32 @@
+"""JWT auth unit tests."""
+
+from datetime import timedelta
+
+import pytest
+from fastapi import HTTPException
+
+from auth import create_access_token, get_current_user
+
+
+def test_token_roundtrip():
+    token = create_access_token({"sub": "certauth"})
+    assert get_current_user(token) == "certauth"
+
+
+def test_rejects_garbage_token():
+    with pytest.raises(HTTPException) as exc:
+        get_current_user("not-a-jwt")
+    assert exc.value.status_code == 401
+
+
+def test_rejects_expired_token():
+    token = create_access_token({"sub": "certauth"}, expires_delta=timedelta(minutes=-5))
+    with pytest.raises(HTTPException) as exc:
+        get_current_user(token)
+    assert exc.value.status_code == 401
+
+
+def test_rejects_token_without_subject():
+    token = create_access_token({"other": "claim"})
+    with pytest.raises(HTTPException):
+        get_current_user(token)

tests/test_models.py

@@ -0,0 +1,22 @@
+"""Password hashing tests."""
+
+from models import hash_password, verify_password
+
+
+def test_hash_and_verify():
+    h = hash_password("s3cure-Passw0rd!")
+    assert verify_password("s3cure-Passw0rd!", h)
+
+
+def test_wrong_password_rejected():
+    h = hash_password("s3cure-Passw0rd!")
+    assert not verify_password("wrong-password", h)
+
+
+def test_hashes_are_unique():
+    assert hash_password("same-pass") != hash_password("same-pass")
+
+
+def test_accepts_bytes_hash():
+    h = hash_password("s3cure-Passw0rd!")
+    assert verify_password("s3cure-Passw0rd!", h.encode())

tests/test_signing.py

@@ -0,0 +1,55 @@
+"""DER helpers and signing pipeline (no YubiKey required)."""
+
+from datetime import datetime, timedelta, timezone
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.x509.oid import NameOID
+
+from signing import _der_read_len, _der_seq, _split_cert_der
+
+
+from cryptography.hazmat.primitives.serialization import Encoding
+
+
+def _dummy_cert_der():
+    key = ec.generate_private_key(ec.SECP384R1())
+    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "dummy.test")])
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1))
+        .sign(key, hashes.SHA384())
+    )
+    return cert.public_bytes(Encoding.DER)
+
+
+def test_split_roundtrip():
+    der = _dummy_cert_der()
+    tbs, alg, sig = _split_cert_der(der)
+    assert alg[0] == 0x30
+    assert sig[0] == 0x03
+    # Reassembling the same parts reproduces the original certificate.
+    reassembled = _der_seq(tbs + alg + sig)
+    assert reassembled == der
+
+
+def test_der_read_len_short_and_long():
+    assert _der_read_len(b"\x05ABC", 0) == (5, 1)
+    long_len = b"\x81\x10"
+    assert _der_read_len(long_len + b"A" * 16, 0) == (16, 2)
+
+
+def test_build_leaf_cert_fails_cleanly_without_ca():
+    """Without the CA files the pipeline fails cleanly (no YubiKey available)."""
+    import pytest
+
+    from signing import build_leaf_cert
+
+    with pytest.raises(FileNotFoundError):
+        build_leaf_cert("nope.test", "", 365)