fix: CSRF protection, cookie security flags, rate limiting, XSS (#23,#24,#25,#28,#33)

Add CSRF tokens to all cookie-based POST endpoints. Set Secure and SameSite=Strict on auth cookie. Rate limit login to 5 attempts per 15min per IP. Escape HTML in signing error messages (XSS fix). Remove duplicate get_user_from_cookie definition.
fix: pass YubiKey PIN via stdin instead of CLI arg (#18 , #31 )
2026-07-04 04:51:56 +00:00 · 2026-07-04 04:49:56 +00:00 · 2026-07-04 04:48:30 +00:00 · 2026-07-03 01:14:04 +00:00
29 changed files with 1878 additions and 1249 deletions
--- a/.env.example
+++ b/.env.example
@ -1,50 +1,19 @@
-# CertAuth API environment variables.
+# CertAuth Environment Variables
-# Installed to /etc/certauth/certauth.env (mode 600, owned by the API user)
+# Copy this to .env and fill in your real values
 # and loaded via EnvironmentFile= in certauth-api.service.
 #
 # The API fails closed on startup if any of the REQUIRED variables below
 # are missing.
-# ---- REQUIRED -------------------------------------------------------------
+# Admin web login password
 ADMIN_PASS=your-secure-admin-password
-# YubiKey hardware assignments (serials are printed by `ykman list`;
+# YubiKey 1 (Root CA) credentials
-# setup-certauth.sh auto-detects them at provision time)
+YK_ROOT_PIN=your-yk1-pin
-YK_ROOT_SERIAL=your-root-yubikey-serial
+YK_ROOT_PUK=your-yk1-puk
 YK_INT_SERIAL=your-intermediate-yubikey-serial
-# YubiKey PINs (change from the defaults with:
+# YubiKey 2 (Intermediate CA) credentials
-#   ykman piv access change-pin -P <old> --new-pin <new>)
+YK_INT_PIN=your-yk2-pin
-YK_ROOT_PIN=your-yk-root-pin
+YK_INT_PUK=your-yk2-puk
 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
-# Admin web/API login password
+# PFX download 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
--- a/.gitea/workflows/ci.yml
+++ b/.gitea/workflows/ci.yml
@ -1,143 +0,0 @@
 name: CI
 on:
  push:
    branches: [main, master]
  pull_request:
    branches: [main, master]
 env:
  GITEA_URL: https://git.example.com
 jobs:
  lint:
    runs-on: ubuntu-latest
    container:
      image: gitea-job-image
    steps:
      - name: Clone repo
        run: |
          rm -rf $GITHUB_WORKSPACE/*
          git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
          git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
      - name: Run ruff (Python lint)
        if: always()
        run: |
          if [[ -f pyproject.toml ]]; then
            pip3 install ruff
            ruff check .
          else
            echo "No Python project detected, skipping ruff"
          fi
      - name: Run npm lint (JS/TS)
        if: always()
        run: |
          if [[ -f package.json ]]; then
            npm ci
            npm run lint --if-present || true
          else
            echo "No Node.js project detected, skipping npm lint"
          fi
  test:
    runs-on: ubuntu-latest
    container:
      image: gitea-job-image
    steps:
      - name: Clone repo
        run: |
          rm -rf $GITHUB_WORKSPACE/*
          git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
          git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
      - name: Run pytest (Python)
        if: always()
        run: |
          if [[ -f pyproject.toml ]]; then
            python3 -m pip install --upgrade pip
            pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true
            pip3 install pytest
            pytest tests/ -v --tb=short 2>/dev/null || true
          else
            echo "No Python project detected, skipping pytest"
          fi
      - name: Run npm test (JS/TS)
        if: always()
        run: |
          if [[ -f package.json ]]; then
            npm ci
            npm run test --if-present || true
          else
            echo "No Node.js project detected, skipping npm test"
          fi
      - name: Run Go tests
        if: always()
        run: |
          if [[ -f go.mod ]]; then
            go test ./...
          else
            echo "No Go project detected, skipping go test"
          fi
  docker-build:
    runs-on: ubuntu-latest
    steps:
      - name: Clone repo
        run: |
          rm -rf $GITHUB_WORKSPACE/*
          git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
          git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
      - name: Build Docker image
        if: always()
        run: |
          if [[ -f Dockerfile ]]; then
            docker build -t $GITHUB_REPOSITORY:test .
          else
            echo "No Dockerfile found, skipping docker build"
          fi
  security:
    runs-on: ubuntu-latest
    container:
      image: gitea-job-image
    steps:
      - name: Clone repo
        run: |
          rm -rf $GITHUB_WORKSPACE/*
          git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
          git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
      - name: Run bandit (Python SAST)
        if: always()
        run: |
          if [[ -f pyproject.toml ]]; then
            pip3 install bandit
            bandit -r . --severity-level high --confidence-level high --exclude tests/,test_*
          else
            echo "No Python project detected, skipping bandit"
          fi
      - name: Run npm audit (JS/TS)
        if: always()
        run: |
          if [[ -f package.json ]]; then
            npm ci
            npm audit --audit-level=high 2>/dev/null || echo "npm audit: vulnerabilities found (non-blocking)"
          else
            echo "No Node.js project detected, skipping npm audit"
          fi
  build-result:
    needs: [lint, test, docker-build, security]
    runs-on: ubuntu-latest
    container:
      image: gitea-job-image
    if: always()
    steps:
      - name: Summary
        run: echo "All CI checks completed"
--- a/.gitignore
+++ b/.gitignore
@ -1,11 +1,21 @@
-__pycache__/
+# Environment variables containing real secrets
 *.pyc
 .env
 .env.local
 .env.*.local
 # Python
 __pycache__/
 *.py[cod]
 *.egg-info/
 # OS
 .DS_Store
 Thumbs.db
 # Local overrides
 *.local
 # Secrets
 .password
-*.pem
+ssl/
-*.key
+ssl-home/
 *.p12
 *.pfx
 *.db
 *.sqlite
 *.log
--- a/Caddyfile-playground.ms
+++ b/Caddyfile-playground.ms
@ -0,0 +1,15 @@
 {
    admin off
 }
 example.com, www.example.com {
    encode gzip
    tls /etc/ssl/certs/example.com.pem /etc/ssl/private/example.com.key
    root * /var/www/example.com
    file_server browse
 }
 :80 {
    redir https://{host}{uri} permanent
 }
--- a/21
+++ b/21
@ -1,21 +0,0 @@
 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.
--- a/docs/PLAN.md
+++ b/docs/PLAN.md
--- a/README.md
+++ b/README.md
@ -1,107 +0,0 @@
 # 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).
--- a/api/auth.py
+++ b/api/auth.py
@ -1,14 +1,15 @@
 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.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
+    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
--- a/api/config.py
+++ b/api/config.py
@ -1,61 +1,63 @@
 """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
 import secrets
 _YK_ROOT_SERIAL = "35450561"
 _YK_INT_SERIAL = "33930436"
 _ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
 _INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
 _CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
 _ISSUED_DIR = "/etc/ssl/ca/issued"
 _DB_PATH = "/var/lib/certauth/certauth.db"
 _JWT_SECRET_FILE = "/var/lib/certauth/.jwt_secret"
 _ALGORITHM = "HS256"
 _ACCESS_TOKEN_EXPIRE_MINUTES = 60
 _ADMIN_USERNAME = "certauth"
 _PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
 _YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
 _YK_PUB_INT = "/tmp/yk2-int-pub.pem"
 _TMP_DIR = "/var/lib/certauth/tmp"
-def _required(name: str) -> str:
+def _get_jwt_secret() -> str:
-    value = os.environ.get(name)
+    """Return JWT secret from env, persisted file, or generate new one."""
-    if not value:
+    env_secret = os.environ.get("JWT_SECRET")
-        raise RuntimeError(f"{name} environment variable is required")
+    if env_secret:
-    return value
+        return env_secret
    if os.path.exists(_JWT_SECRET_FILE):
        with open(_JWT_SECRET_FILE) as f:
            return f.read().strip()
    secret = secrets.token_hex(32)
    os.makedirs(os.path.dirname(_JWT_SECRET_FILE), exist_ok=True)
    fd = os.open(_JWT_SECRET_FILE, os.O_WRONLY | os.O_CREAT, 0o600)
    with os.fdopen(fd, "w") as f:
        f.write(secret)
    return secret
-# YubiKey hardware: serials must be supplied by the operator (auto-detected at
+def _get_pin(env_var: str) -> str:
-# setup time by setup-certauth.sh). Never hard-code device serials in source.
+    """Require YubiKey PIN from environment — no default allowed."""
-YK_ROOT_SERIAL = _required("YK_ROOT_SERIAL")
+    pin = os.environ.get(env_var)
-YK_INT_SERIAL = _required("YK_INT_SERIAL")
+    if not pin:
-YK_ROOT_PIN = _required("YK_ROOT_PIN")
+        raise RuntimeError(f"Missing required environment variable: {env_var}")
-YK_INT_PIN = _required("YK_INT_PIN")
+    return 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_ROOT_SERIAL = _YK_ROOT_SERIAL
-YK_PUB_ROOT = os.environ.get("YK_PUB_ROOT", "/tmp/yk1-root-pub.pem")
+YK_INT_SERIAL = _YK_INT_SERIAL
-YK_PUB_INT = os.environ.get("YK_PUB_INT", "/tmp/yk2-int-pub.pem")
+ROOT_CA_PATH = _ROOT_CA_PATH
 INT_CA_PATH = _INT_CA_PATH
 CA_CHAIN_PATH = _CA_CHAIN_PATH
 ISSUED_DIR = _ISSUED_DIR
 DB_PATH = _DB_PATH
 SECRET_KEY = _get_jwt_secret()
 ALGORITHM = _ALGORITHM
 ACCESS_TOKEN_EXPIRE_MINUTES = _ACCESS_TOKEN_EXPIRE_MINUTES
 ADMIN_USERNAME = _ADMIN_USERNAME
 PKCS11_MODULE = _PKCS11_MODULE
 YK_PUB_ROOT = _YK_PUB_ROOT
 YK_PUB_INT = _YK_PUB_INT
 TMP_DIR = _TMP_DIR
-# PKCS#11 module for pkcs11-tool. Default works on both aarch64 and x86_64
+YK_ROOT_PIN = _get_pin("YK_ROOT_PIN")
-# Ubuntu; override for nonstandard installs.
+YK_INT_PIN = _get_pin("YK_INT_PIN")
 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 bootstrap account
 ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "certauth")
 ADMIN_PASSWORD = _required("ADMIN_PASSWORD")
 # PFX download default password (overridable per download)
 PFX_DEFAULT_PASSWORD = os.environ.get("PFX_PASS", "certauth")
--- a/api/main.py
+++ b/api/main.py
@ -1,85 +1,69 @@
-import os
+import os, sqlite3, datetime, secrets, hashlib, subprocess, json, time, functools
 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, PlainTextResponse
+from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse
 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 *
-    ALGORITHM,
+from models import get_db, init_db, hash_password, verify_password
    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
 logger = logging.getLogger(__name__)
 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")
+app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static")
-_csrf_secrets = {}
+_login_attempts = {}
 _LOGIN_MAX_ATTEMPTS = 5
 _LOGIN_WINDOW_SECONDS = 900
-def get_csrf_token(session_id: str) -> str:
+_csrf_secret = secrets.token_hex(32)
    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)
+def _check_rate_limit(client_ip: str) -> bool:
-    if not stored:
+    now = time.time()
    if client_ip not in _login_attempts:
        _login_attempts[client_ip] = []
    _login_attempts[client_ip] = [
        t for t in _login_attempts[client_ip] if now - t < _LOGIN_WINDOW_SECONDS
    ]
    if len(_login_attempts[client_ip]) >= _LOGIN_MAX_ATTEMPTS:
        return False
    _login_attempts[client_ip].append(now)
    return True
 def _generate_csrf_token() -> str:
    return secrets.token_hex(32)
 def _verify_csrf_token(request: Request, token: str) -> bool:
    stored = request.cookies.get("csrf_token")
    if not stored or not token:
        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):
+def _set_csrf_cookie(resp):
-    token = request.cookies.get("token")
+    token = _generate_csrf_token()
-    if not token:
+    resp.set_cookie("csrf_token", token, httponly=False, samesite="strict", path="/")
-        return None
+    return token
-    try:
+
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except Exception:
        return None
 jinja_env = Environment(
-    loader=FileSystemLoader(TEMPLATES_DIR),
+    loader=FileSystemLoader("/opt/certauth/api/templates"),
-    autoescape=select_autoescape(["html", "xml"]),
+    autoescape=select_autoescape(["html"])
 )
@app.on_event("startup")
 def startup():
    init_db()
    try:
-        with open(ROOT_CA_PATH) as f:
+        with open(ROOT_CA_PATH) as f: root = f.read()
-            root = f.read()
+        with open(INT_CA_PATH) as f: inter = f.read()
-        with open(INT_CA_PATH) as f:
+        with open(CA_CHAIN_PATH, "w") as f: f.write(inter + "\n" + root)
-            inter = f.read()
+    except: pass
        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))
@ -89,7 +73,10 @@ class LoginRequest(BaseModel):
    password: str
@app.post("/api/token")
-async def login(req: LoginRequest):
+async def login(req: LoginRequest, request: Request):
    client_ip = request.client.host
    if not _check_rate_limit(client_ip):
        raise HTTPException(429, "Too many login attempts. Try again later.")
    conn = get_db()
    row = conn.execute("SELECT * FROM users WHERE username = ?", (req.username,)).fetchone()
    conn.close()
@ -110,17 +97,12 @@ 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(
+async def create_domain(name: str = Form(...), description: str = Form(""),
-    name: str = Form(...),
+                        user: str = Depends(get_current_user)):
    description: str = Form(""),
    user: str = Depends(get_current_user),
 ):
    conn = get_db()
    cur = conn.cursor()
-    cur.execute(
+    cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
-        "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
+                 (name, description, 1))
        (name, description, get_user_id(conn, user)),
    )
    conn.commit()
    conn.close()
    return {"status": "ok"}
@ -128,28 +110,18 @@ async def create_domain(
@app.get("/api/certs")
 async def list_certs(user: str = Depends(get_current_user)):
    conn = get_db()
-    rows = conn.execute(
+    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()
        "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(
+async def request_cert(cn: str = Form(...), sans: str = Form(""),
-    cn: str = Form(...),
+                       days: int = Form(365), domain_id: int = Form(0),
-    sans: str = Form(""),
+                       user: str = Depends(get_current_user)):
    days: int = Form(365),
    domain_id: int = Form(0),
    user: str = Depends(get_current_user),
 ):
    conn = get_db()
    cur = conn.cursor()
-    cur.execute(
+    cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)",
-        "INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)",
+                 (domain_id, cn, sans, "pending", 1))
        (domain_id or None, cn, sans, "pending", get_user_id(conn, user)),
    )
    conn.commit()
    cid = cur.lastrowid
    conn.close()
@ -158,50 +130,36 @@ async def request_cert(
@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(
+    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
        "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)
        result, err = build_leaf_cert(row["subject"], row["san"], 365)
    except Exception as e:
        logger.error("Signing failed: %s", sanitize_error(str(e)))
        raise HTTPException(500, "Signing failed")
    if err:
-        raise HTTPException(500, "Signing failed")
+        import html as h
-    cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt"
+        raise HTTPException(500, f"Signing failed: {h.escape(str(err))}")
-    kf = f"{ISSUED_DIR}/cert-{result['serial']}.key"
+    cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
-    with open(cf, "w") as f:
+    kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
-        f.write(result["cert_pem"])
+    open(cf, "w").write(result["cert_pem"])
-    with open(kf, "w") as f:
+    open(kf, "w").write(result["key_pem"])
-        f.write(result["key_pem"])
+    os.chmod(cf, 0o640); os.chmod(kf, 0o600)
    os.chmod(cf, 0o640)
    os.chmod(kf, 0o600)
    conn = get_db()
-    conn.execute(
+    conn.execute("UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
-        "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
+                 ("issued", result["serial"], cf, datetime.datetime.now().isoformat(), result["expires_at"], cert_id))
-        ("issued", result["serial"], cf, datetime.now(timezone.utc).isoformat(), result["expires_at"], cert_id),
+    conn.commit(); conn.close()
    )
    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:
+    if not user: raise HTTPException(401, "Login required")
        raise HTTPException(401, "Login required")
    conn = get_db()
-    row = conn.execute(
+    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
    ).fetchone()
    conn.close()
-    if not row or row["status"] != "issued":
+    if not row or row["status"] != "issued": raise HTTPException(404)
-        raise HTTPException(404)
+    pem_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pem"
    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:
@ -211,24 +169,16 @@ 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(
+async def download_pfx(cert_id: int, password: str = "certauth", request: Request = None):
-    cert_id: int,
+    """Download cert + key + chain as PKCS12/PFX."""
    password: str = PFX_DEFAULT_PASSWORD,
    request: Request = None,
 ):
    user = get_user_from_cookie(request)
-    if not user:
+    if not user: raise HTTPException(401, "Login required")
        raise HTTPException(401, "Login required")
    conn = get_db()
-    row = conn.execute(
+    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
    ).fetchone()
    conn.close()
-    if not row or row["status"] != "issued":
+    if not row or row["status"] != "issued": raise HTTPException(404)
        raise HTTPException(404)
    kf = row["cert_path"].replace(".crt", ".key")
-    if not os.path.exists(kf):
+    if not os.path.exists(kf): raise HTTPException(404)
        raise HTTPException(404)
    from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption
    from cryptography import x509
    with open(row["cert_path"], "rb") as f:
@ -240,30 +190,33 @@ async def download_pfx(
        for cert_pem in f.read().split(b"-----END CERTIFICATE-----"):
            cert_pem = cert_pem.strip()
            if cert_pem:
-                chain_certs.append(
+                chain_certs.append(x509.load_pem_x509_certificate(cert_pem + b"\n-----END CERTIFICATE-----"))
                    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"{TMP_DIR}/cert-{row['serial']}.pfx"
+    pfx_path = f"/var/lib/certauth/tmp/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():
+async def health(): return {"status": "ok"}
    return {"status": "ok"}
@app.get("/api/ca-chain")
-async def ca_chain():
+async def ca_chain(): return FileResponse(CA_CHAIN_PATH, filename="ca-chain.crt")
-    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):
@ -271,115 +224,73 @@ async def dashboard(request: Request):
    if not user:
        return RedirectResponse("/login", status_code=302)
    conn = get_db()
-    certs = conn.execute(
+    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()
        "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(
+    p = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("pending",)).fetchone()["c"]
-        "SELECT COUNT(*) as c FROM certificates WHERE status = ?",
+    i = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("issued",)).fetchone()["c"]
        ("pending",),
    ).fetchone()["c"]
    i = conn.execute(
        "SELECT COUNT(*) as c FROM certificates WHERE status = ?",
        ("issued",),
    ).fetchone()["c"]
    conn.close()
-    return render(
+    return render("dashboard.html", {"request": request, "user": user,
-        "dashboard.html",
+        "certs": [dict(r) for r in certs], "domains": [dict(r) for r in domains],
-        {
+        "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, "csrf_token": get_csrf_token("anon")})
+    return render("login.html", {"request": request, "error": None})
@app.post("/login")
-async def login_post(
+async def login_post(username: str = Form(...), password: str = Form(...),
-    username: str = Form(...),
+                     request: Request = None):
-    password: str = Form(...),
+    client_ip = request.client.host if request else "unknown"
-    csrf_token: str = Form(""),
+    if not _check_rate_limit(client_ip):
-):
+        return render("login.html", {"request": None, "error": "Too many login attempts. Try again later."})
    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(
+    row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
        "SELECT * FROM users WHERE username = ?", (username,)
    ).fetchone()
    conn.close()
    if not row or not verify_password(password, row["password_hash"]):
-        return render(
+        return render("login.html", {"request": None, "error": "Invalid credentials"})
            "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", secure=True, path="/")
+    resp.set_cookie("token", token, httponly=True, samesite="strict", secure=True, path="/")
    _set_csrf_cookie(resp)
    return resp
@app.post("/api/certs/{cert_id}/sign/web")
-async def sign_cert_web(cert_id: int, request: Request = None):
+async def sign_cert_web(cert_id: int, csrf_token: str = Form(""), 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(request, csrf_token):
-    if not verify_csrf_token(user.get("sub", "anon"), csrf):
+        raise HTTPException(403, "Invalid CSRF token")
        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
    conn = get_db()
-    row = conn.execute(
+    row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone()
        "SELECT * FROM certificates WHERE id = ?", (cert_id,)
    ).fetchone()
    if not row or row["status"] != "pending":
        conn.close()
-        return HTMLResponse("<span class='text-red-400'>Not found or already issued</span>", status_code=400)
+        raise HTTPException(400, "Not found or already issued")
    conn.close()
    import html as html_lib
    try:
        result, err = build_leaf_cert(row["subject"], row["san"], 365)
        if err:
-            return HTMLResponse("<span class='text-red-400'>Issue failed</span>")
+            safe_err = html_lib.escape(str(err))
-        cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt"
+            return HTMLResponse(f'<span class="text-red-400">Issue failed: {safe_err}</span>')
-        kf = f"{ISSUED_DIR}/cert-{result['serial']}.key"
+        cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt"
-        with open(cf, "w") as f:
+        kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key"
-            f.write(result["cert_pem"])
+        open(cf, "w").write(result["cert_pem"])
-        with open(kf, "w") as f:
+        open(kf, "w").write(result["key_pem"])
            f.write(result["key_pem"])
        os.chmod(cf, 0o640)
        os.chmod(kf, 0o600)
        conn2 = get_db()
-        conn2.execute(
+        conn2.execute("UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
-            "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?",
+                      ("issued", result["serial"], cf, datetime.datetime.now().isoformat(), result["expires_at"], cert_id))
            (
                "issued",
                result["serial"],
                cf,
                datetime.now(timezone.utc).isoformat(),
                result["expires_at"],
                cert_id,
            ),
        )
        conn2.commit()
        conn2.close()
-        return HTMLResponse(
+        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>')
            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:
-        logger.error("Signing failed: %s", sanitize_error(str(ex)))
+        safe_ex = html_lib.escape(str(ex))
-        return HTMLResponse("<span class='text-red-400'>Issue failed</span>")
+        return HTMLResponse(f'<span class="text-red-400">Issue failed: {safe_ex}</span>')
@app.get("/logout")
 async def logout():
@ -387,175 +298,180 @@ async def logout():
    resp.delete_cookie("token", path="/")
    return resp
 # --- Web API (cookie auth) ---
@app.post("/api/domains/web")
-async def create_domain_web(
+async def create_domain_web(name: str = Form(...), description: str = Form(""),
-    name: str = Form(...),
+                            csrf_token: str = Form(""), request: Request = None):
    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):
+    if not _verify_csrf_token(request, csrf_token):
-        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
+        raise HTTPException(403, "Invalid CSRF token")
    conn = get_db()
    cur = conn.cursor()
-    cur.execute(
+    cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1))
        "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)",
        (name, description, get_user_id(conn, user.get("sub"))),
    )
    conn.commit()
    conn.close()
-    return HTMLResponse(
+    return HTMLResponse('<span class="text-green-400">Domain registered! <a href="/domains" class="underline">Refresh</a></span>')
        '<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(
+async def request_cert_web(cn: str = Form(...), sans: str = Form(""), days: int = Form(365),
-    cn: str = Form(...),
+                           domain_id: int = Form(0), csrf_token: str = Form(""),
-    sans: str = Form(""),
+                           request: Request = None):
    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):
+    if not _verify_csrf_token(request, csrf_token):
-        return HTMLResponse("<span class='text-red-400'>Invalid request</span>", status_code=403)
+        raise HTTPException(403, "Invalid CSRF token")
    conn = get_db()
    cur = conn.cursor()
    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(
+    cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", (domain_id, cn, sans, "pending", 1))
        "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(
+    return HTMLResponse('<span class="text-green-400">Certificate requested! Click Issue below. <a href="/certs" class="underline">Refresh</a></span>')
        '<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:
+    if not user: return RedirectResponse("/login", status_code=302)
        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(
+    return render("domains.html", {"request": request, "user": user, "domains": [dict(r) for r in rows]})
        "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:
+    if not user: return RedirectResponse("/login", status_code=302)
        return RedirectResponse("/login", status_code=302)
    conn = get_db()
-    rows = conn.execute(
+    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()
        "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(
+    return render("certs.html", {"request": request, "user": user, "certs": [dict(r) for r in rows],
-        "certs.html",
+                                  "domains": [dict(r) for r in domains]})
        {
            "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:
+    if not user: return RedirectResponse("/login", status_code=302)
        return RedirectResponse("/login", status_code=302)
    conn = get_db()
-    rows = conn.execute(
+    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()
        "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(
+    return render("history.html", {"request": request, "user": user, "certs": [dict(r) for r in rows]})
        "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:
+    if not user: return RedirectResponse("/login", status_code=302)
-        return RedirectResponse("/login", status_code=302)
+    return render("setup.html", {"request": request, "user": user})
    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: " DETECTED_IP
+    read -r -p "Enter CertAuth server IP (e.g., 192.168.8.248): " 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"; exit 1; }
+curl -sLk "$CHAIN_URL" -o /tmp/ca-chain.crt || { echo "Failed to download CA chain from $CHAIN_URL"; exit 1; }
 # Detect OS and install
 if [[ -f /etc/os-release ]]; then
    . /etc/os-release
-    if [[ "$ID" == "debian" || "$ID" == "ubuntu" ]]; then
+    if [[ "$ID" == "debian" || "$ID" == "ubuntu" || "$ID" == "linuxmint" ]]; 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
-        sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt
+        echo "❌ Unsupported Linux distribution: $ID"
-        sudo update-ca-trust 2>/dev/null || true
+        echo "   Download /tmp/ca-chain.crt and install manually"
        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"
+    echo "❌ Unsupported OS: $(uname -s)"
    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!"
 '''
@ -563,21 +479,54 @@ echo "Done!"
@app.get("/setup.ps1")
 async def setup_ps1():
-    script = r'''
+    """PowerShell setup script for Windows."""
    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) {
-    $CertAuthIP = Read-Host "Enter CertAuth server IP"
+    # Try to detect from environment or prompt
    $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"
-(New-Object Net.WebClient).DownloadFile($ChainUrl, $ChainPath)
+
-$store = New-Object System.Security.Cryptography.X509Certificates.X509Store(
+Write-Host "Downloading CA chain from $ChainUrl ..." -ForegroundColor Cyan
-    [System.Security.Cryptography.X509Certificates.StoreName]::Root,
+try {
-    [System.Security.Cryptography.X509Certificates.StoreLocation]::LocalMachine)
+    (New-Object Net.WebClient).DownloadFile($ChainUrl, $ChainPath)
-$store.Open([System.Security.Cryptography.X509Certificates.OpenFlags]::ReadWrite)
+} catch {
-$cert = New-Object System.Security.Cryptography.X509Certificates.X509Certificate2($ChainPath)
+    Write-Host "Failed to download CA chain: $_" -ForegroundColor Red
-$store.Add($cert)
+    exit 1
-$store.Close()
+}
 # 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)
    $store.Open([System.Security.Cryptography.X509Certificates.OpenFlags]::ReadWrite)
    $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
 '''
--- a/api/models.py
+++ b/api/models.py
@ -1,9 +1,5 @@
-import sqlite3
+import sqlite3, datetime, secrets, bcrypt, os
-import bcrypt
+from config import DB_PATH, ADMIN_USERNAME
 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)
@ -11,7 +7,6 @@ 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():
@ -62,30 +57,21 @@ 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(ADMIN_PASSWORD.encode(), bcrypt.gensalt())
+        pw_hash = bcrypt.hashpw(b"CHANGE_ME_ADMIN_PASS", bcrypt.gensalt())
-        if isinstance(pw_hash, bytes):
+        if isinstance(pw_hash, bytes): pw_hash = pw_hash.decode()
            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):
+    if isinstance(hash_, str): hash_ = hash_.encode()
        hash_ = hash_.encode()
    return bcrypt.checkpw(password.encode(), hash_)
--- a/api/signing.py
+++ b/api/signing.py
@ -1,227 +1,158 @@
 """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, tempfile
 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 *
-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
-logger = logging.getLogger(__name__)
+def get_root_pub_key():
    with open(YK_PUB_ROOT, "rb") as f:
        return serialization.load_pem_public_key(f.read())
-# Scratch dir for pkcs11-tool input/output files (per-process, unlinked after).
+def get_int_pub_key():
-PKCS11_TMP = tempfile.mkdtemp(prefix="certauth_")
+    with open(YK_PUB_INT, "rb") as f:
        return serialization.load_pem_public_key(f.read())
-
+def get_root_ca_cert():
-def _der_read_len(data: bytes, i: int) -> tuple:
+    with open(ROOT_CA_PATH, "rb") as f:
    """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=None):
+def _make_temp_file(prefix: str, data: bytes = None):
-    """Sign DER bytes with the YubiKey SIGN key via pkcs11-tool.
+    """Create temp file with unpredictable name in TMP_DIR."""
-
+    fd, path = tempfile.mkstemp(prefix=prefix, dir=TMP_DIR)
    Returns (signature_BIT_STRING, None) on success or (None, error).
    """
    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-source", "stdin",
        "--sign", "--mechanism", "ECDSA-SHA384",
        "--label", "SIGN key",
        "--input-file", tbs_path,
        "--output-file", sig_path,
    ]
    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(
+        if data is not None:
-            cmd,
+            os.write(fd, data)
            input=yk_pin.encode(),
            capture_output=True,
            text=True,
            timeout=30,
        )
        if r.returncode != 0:
            logger.error("YubiKey signing failed: %s", r.stderr[:200])
            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 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]:
+        os.close(fd)
-            try:
+    os.chmod(path, 0o600)
-                os.unlink(f)
+    return path
            except OSError:
                pass
 def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
    tbs_file = _make_temp_file("tbs_", tbs_bytes)
    sig_file = _make_temp_file("sig_")
    os.unlink(sig_file)
    try:
        r = subprocess.run([
            "sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
            "--login", "--pin-source", "stdin",
            "--sign", "--mechanism", "ECDSA-SHA384",
            "--token-label", token_label,
            "--label", "SIGN key",
            "--input-file", tbs_file,
            "--output-file", sig_file
        ], input=yk_pin, capture_output=True, text=True)
        if r.returncode != 0:
            return None, r.stderr
        with open(sig_file, "rb") as f:
            raw = f.read()
    finally:
        for f in (tbs_file, sig_file):
            if os.path.exists(f):
                os.unlink(f)
    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 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
 def build_leaf_cert(cn, sans, days=365):
-    """Issue a leaf certificate for cn, signed by the Intermediate CA.
+    root_cert = get_root_ca_cert()
-
+    int_pub = get_int_pub_key()
-    Returns ({serial, cert_pem, key_pem, expires_at}, None) or (None, error).
+    root_pub = get_root_pub_key()
    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, CA_COUNTRY),
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-        x509.NameAttribute(NameOID.ORGANIZATION_NAME, CA_ORG),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
        x509.NameAttribute(NameOID.COMMON_NAME, cn),
    ])
-    builder = (
+    issuer = x509.Name([
-        x509.CertificateBuilder()
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
-        .subject_name(subject)
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
-        .issuer_name(int_cert.subject)
+        x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"),
    ])
    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.now(timezone.utc))
+        .not_valid_before(datetime.datetime.now(datetime.timezone.utc))
-        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=days))
+        .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=days))
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
-        .add_extension(
+        .add_extension(x509.KeyUsage(
-            x509.KeyUsage(
+            digital_signature=True, key_encipherment=True,
-                digital_signature=True,
+            key_cert_sign=False, crl_sign=False,
-                key_encipherment=True,
+            content_commitment=False, data_encipherment=False,
-                content_commitment=False,
+            key_agreement=False, encipher_only=False, decipher_only=False), critical=True)
-                data_encipherment=False,
+        .add_extension(x509.ExtendedKeyUsage([
-                key_agreement=False,
+            x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
-                key_cert_sign=False,
+            x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
-                crl_sign=False,
+        ]), critical=False)
-                encipher_only=False,
+        .add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False)
-                decipher_only=False,
+        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(int_pub), critical=False))
            ),
            critical=True,
        )
        .add_extension(
            x509.ExtendedKeyUsage([
                x509.OID_SERVER_AUTH,
                x509.OID_CLIENT_AUTH,
            ]),
            critical=False,
        )
    )
    if sans:
-        san_names = [x509.DNSName(s.strip()) for s in sans.split(",") if s.strip()]
+        san_list = []
-        if san_names:
+        for s in sans.split(","):
-            builder = builder.add_extension(
+            s = s.strip()
-                x509.SubjectAlternativeName(san_names),
+            # Check if it's an IP address
-                critical=False,
+            if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s):
-            )
+                san_list.append(x509.IPAddress(ipaddress.ip_address(s)))
-
+            else:
-    # 1) TBS bytes: sign with a throwaway key of identical parameters —
+                san_list.append(x509.DNSName(s))
-    #    the TBS block is independent of the signing key.
+        builder = builder.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
-    dummy_key = ec.generate_private_key(ec.SECP384R1())
+    tmp = ec.generate_private_key(ec.SECP384R1())
-    tbs, sig_alg, _ = _split_cert_der(builder.sign(dummy_key, hashes.SHA384()))
+    temp = builder.sign(tmp, hashes.SHA384())
-
+    td = temp.public_bytes(serialization.Encoding.DER)
-    # 2) Real signature from the YubiKey.
+    o = 1
-    yk_sig, err = sign_tbs_with_yk(tbs, YK_INT_PIN)
+    if td[o] & 0x80: n = td[o] & 0x7f; o += 1 + n
-    if err:
+    else: o += 1
-        return None, err
+    tbs_start = o
-
+    o += 1
-    # 3) Reassemble and verify before trusting.
+    if td[o] & 0x80: n = td[o] & 0x7f; tl = int.from_bytes(td[o+1:o+1+n], "big"); o += 1 + n
-    cert = x509.load_der_x509_certificate(_der_seq(tbs + sig_alg + yk_sig))
+    else: tl = td[o]; o += 1
-    cert.verify(int_cert.public_key())
+    tbs_end = o + tl
-
+    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 = _make_temp_file("leaf_", final)
    pem_file = _make_temp_file("leaf_pem_")
    os.unlink(pem_file)
    try:
        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()
    finally:
        for f in (der_file, pem_file):
            if os.path.exists(f):
                os.unlink(f)
    key_pem = leaf_key.private_bytes(
-        encoding=Encoding.PEM,
+        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
-        encryption_algorithm=NoEncryption(),
+        encryption_algorithm=serialization.NoEncryption()
    ).decode()
-    return (
+    cert = x509.load_pem_x509_certificate(leaf_pem.encode())
-        {
+    serial = format(cert.serial_number, 'x')
-            "serial": hex(cert.serial_number),
+    return {"cert_pem": leaf_pem, "key_pem": key_pem, "serial": serial,
-            "cert_pem": cert.public_bytes(Encoding.PEM).decode(),
+            "expires_at": cert.not_valid_after.isoformat()}, None
            "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)
--- a/api/static/.gitkeep
+++ b/api/static/.gitkeep
--- a/api/templates/login.html
+++ b/api/templates/login.html
@ -14,7 +14,6 @@
        <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
--- a/certauth-api.service
+++ b/certauth-api.service
@ -8,7 +8,6 @@ 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
--- a/landing/playground.ms.html
+++ b/landing/playground.ms.html
@ -3,7 +3,7 @@
 <head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
-    <title>Playground</title>
+    <title>Playground Network</title>
    <style>
        :root { --bg: #0f172a; --card: #1e293b; --text: #f8fafc; --accent: #38bdf8; }
        body { font-family: system-ui, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }
@ -19,15 +19,13 @@
 </head>
 <body>
    <div class="container">
-        <h1>Playground</h1>
+        <h1>Playground Network</h1>
        <div class="grid">
            <a href="https://pinvault.example.com" class="card"><div class="icon">🔐</div><h2>PinVault</h2><p>Password manager</p></a>
-            <a href="https://youtube.example.com" class="card"><div class="icon">📺</div><h2>YouTube</h2><p>YouTube CLI</p></a>
+            <a href="https://nextcloud.example.com" class="card"><div class="icon">☁️</div><h2>Nextcloud</h2><p>Cloud storage</p></a>
-            <a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
+            <a href="https://opencloud.example.com" class="card"><div class="icon">🌐</div><h2>OpenCloud</h2><p>OpenCloud service</p></a>
            <a href="https://vote.example.com" class="card"><div class="icon">🗳️</div><h2>Vote</h2><p>Voting app</p></a>
            <a href="https://search.example.com" class="card"><div class="icon">🔍</div><h2>Search</h2><p>Privacy metasearch</p></a>
            <a href="https://paste.example.com" class="card"><div class="icon">📋</div><h2>Paste</h2><p>Pastebin service</p></a>
-            <a href="https://ai.example.com" class="card"><div class="icon">🤖</div><h2>AI</h2><p>Local LLM</p></a>
+            <a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
        </div>
    </div>
 </body>
--- a/landing/playground.ms.key
+++ b/landing/playground.ms.key
@ -0,0 +1,5 @@
 -----BEGIN PRIVATE KEY-----
 MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQggL95oQbMsPLm3Kd0
 ihyuirXpBdKFE2IAqZyVZwqtAWKhRANCAAR+r1V3/0xyeoa50fq+XdZ90U3di7wQ
 q2mQVmwMFTCNBrNkOjg6AruiPP1aURCKbT3C+TlRFTFE7iLKFCkGeihp
 -----END PRIVATE KEY-----
--- a/landing/playground.ms.pem
+++ b/landing/playground.ms.pem
@ -0,0 +1,12 @@
 -----BEGIN CERTIFICATE-----
 MIIBsjCCAVmgAwIBAgIUJpVaTpjyW1GLONCx3rXwKnc4OwAwCgYIKoZIzj0EAwIw
 GDEWMBQGA1UEAwwNcGxheWdyb3VuZC5tczAeFw0yNjA3MDIxOTQyMThaFw0yNzA3
 MDIxOTQyMThaMBgxFjAUBgNVBAMMDXBsYXlncm91bmQubXMwWTATBgcqhkjOPQIB
 BggqhkjOPQMBBwNCAAR+r1V3/0xyeoa50fq+XdZ90U3di7wQq2mQVmwMFTCNBrNk
 Ojg6AruiPP1aURCKbT3C+TlRFTFE7iLKFCkGeihpo4GAMH4wHQYDVR0OBBYEFA19
 0spDELs37Zq3rci0GxkppbGnMB8GA1UdIwQYMBaAFA190spDELs37Zq3rci0Gxkp
 pbGnMA8GA1UdEwEB/wQFMAMBAf8wKwYDVR0RBCQwIoINcGxheWdyb3VuZC5tc4IR
 d3d3LnBsYXlncm91bmQubXMwCgYIKoZIzj0EAwIDRwAwRAIgfSL/SDyLeCmg4l3S
 8U9KvBEQkJe1yaId8QmFhhUV+gICIE5KFmPJ3GrEbjzJyB1DFjHTDNUocJBgut0w
 09Y9Lz5F
 -----END CERTIFICATE-----
--- a/landing/www/playground.ms/index.html
+++ b/landing/www/playground.ms/index.html
@ -0,0 +1,32 @@
 <!DOCTYPE html>
 <html lang="en">
 <head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Playground Network</title>
    <style>
        :root { --bg: #0f172a; --card: #1e293b; --text: #f8fafc; --accent: #38bdf8; }
        body { font-family: system-ui, sans-serif; background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }
        .container { max-width: 1200px; margin: 0 auto; }
        h1 { text-align: center; margin-bottom: 3rem; font-size: 2.5rem; background: linear-gradient(135deg, var(--accent), #818cf8); -webkit-background-clip: text; -webkit-text-fill-color: transparent; }
        .grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 1.5rem; }
        .card { background: var(--card); border-radius: 12px; padding: 1.5rem; text-decoration: none; color: var(--text); transition: all 0.2s; border: 1px solid rgba(255,255,255,0.1); }
        .card:hover { transform: translateY(-2px); border-color: var(--accent); box-shadow: 0 4px 12px rgba(56, 189, 248, 0.1); }
        .card h2 { margin: 0 0 0.5rem; font-size: 1.25rem; }
        .card p { margin: 0; color: #94a3b8; font-size: 0.9rem; }
        .icon { font-size: 2rem; margin-bottom: 1rem; }
    </style>
 </head>
 <body>
    <div class="container">
        <h1>Playground Network</h1>
        <div class="grid">
            <a href="https://pinvault.example.com" class="card"><div class="icon">🔐</div><h2>PinVault</h2><p>Password manager</p></a>
            <a href="https://nextcloud.example.com" class="card"><div class="icon">☁️</div><h2>Nextcloud</h2><p>Cloud storage</p></a>
            <a href="https://opencloud.example.com" class="card"><div class="icon">🌐</div><h2>OpenCloud</h2><p>OpenCloud service</p></a>
            <a href="https://paste.example.com" class="card"><div class="icon">📋</div><h2>Paste</h2><p>Pastebin service</p></a>
            <a href="https://archive.example.com" class="card"><div class="icon">🏛️</div><h2>Archive</h2><p>Website archiving</p></a>
        </div>
    </div>
 </body>
 </html>
--- a/nginx-playground.ms
+++ b/nginx-playground.ms
@ -0,0 +1,26 @@
 # ============================================================
 # example.com - Main landing page (HTTPS)
 # ============================================================
 server {
    listen 443 ssl;
    server_name example.com www.example.com;
    ssl_certificate     /etc/ssl/certs/example.com.pem;
    ssl_certificate_key /etc/ssl/private/example.com.key;
    ssl_protocols       TLSv1.2 TLSv1.3;
    ssl_ciphers         HIGH:!aNULL:!MD5;
    root /var/www/example.com;
    index index.html index.htm;
    location / {
        try_files $uri $uri/ =404;
    }
 }
 # HTTP -> HTTPS redirect for example.com
 server {
    listen 80;
    server_name example.com www.example.com;
    return 301 https://$host$request_uri;
 }
--- a/pyproject.toml
+++ b/pyproject.toml
@ -1,42 +0,0 @@
 [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"]
--- a/requirements.txt
+++ b/requirements.txt
@ -1,8 +0,0 @@
 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
--- a/setup-certauth.sh
+++ b/setup-certauth.sh
--- a/tests/conftest.py
+++ b/tests/conftest.py
@ -1,31 +0,0 @@
 """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)
--- a/tests/test_api.py
+++ b/tests/test_api.py
@ -1,115 +0,0 @@
 """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
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@ -1,32 +1,59 @@
-"""JWT auth unit tests."""
+import os
 import sys
 import unittest
 from unittest import mock
 from httpx import AsyncClient, ASGITransport
-from datetime import timedelta
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))
 import pytest
 from fastapi import HTTPException
 from auth import create_access_token, get_current_user
-def test_token_roundtrip():
+class TestAuthSecurity(unittest.TestCase):
-    token = create_access_token({"sub": "certauth"})
+
-    assert get_current_user(token) == "certauth"
+    def test_rate_limiting_exists(self):
        import main
        self.assertTrue(hasattr(main, '_check_rate_limit'))
        self.assertEqual(main._LOGIN_MAX_ATTEMPTS, 5)
        self.assertEqual(main._LOGIN_WINDOW_SECONDS, 900)
    def test_csrf_token_generation(self):
        import main
        t1 = main._generate_csrf_token()
        t2 = main._generate_csrf_token()
        self.assertNotEqual(t1, t2)
        self.assertEqual(len(t1), 64)
    def test_csrf_verify_rejects_empty(self):
        import main
        req = mock.MagicMock()
        req.cookies.get.return_value = None
        self.assertFalse(main._verify_csrf_token(req, "any-token"))
    def test_csrf_verify_rejects_mismatch(self):
        import main
        req = mock.MagicMock()
        req.cookies.get.return_value = "stored-token"
        self.assertFalse(main._verify_csrf_token(req, "different-token"))
    def test_csrf_verify_accepts_match(self):
        import main
        req = mock.MagicMock()
        req.cookies.get.return_value = "matching-token"
        self.assertTrue(main._verify_csrf_token(req, "matching-token"))
    def test_no_duplicate_get_user_from_cookie(self):
        import main, inspect
        sources = inspect.getsourcelines(main)[0]
        count = sum(1 for line in sources if 'def get_user_from_cookie' in line)
        self.assertEqual(count, 1, "get_user_from_cookie must be defined exactly once")
    def test_xss_escaped_in_sign_error(self):
        """Check that HTML in error messages gets escaped."""
        import html as h
        err = '<script>alert("xss")</script>'
        escaped = h.escape(err)
        self.assertNotIn("<script>", escaped)
        self.assertIn("&lt;script&gt;", escaped)
-def test_rejects_garbage_token():
+if __name__ == "__main__":
-    with pytest.raises(HTTPException) as exc:
+    unittest.main()
        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)
--- a/tests/test_config.py
+++ b/tests/test_config.py
@ -0,0 +1,66 @@
 import os
 import sys
 import tempfile
 import unittest
 class TestConfigSecurity(unittest.TestCase):
    def test_jwt_secret_generates_new(self):
        with tempfile.NamedTemporaryFile(suffix=".jwt_secret", delete=False) as f:
            secret_path = f.name
        os.unlink(secret_path)
        try:
            os.environ.pop("JWT_SECRET", None)
            sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
            import importlib
            if "config" in sys.modules:
                del sys.modules["config"]
            os.environ["_JWT_SECRET_FILE"] = secret_path
            cfg = __import__("config")
            self.assertIsNotNone(cfg.SECRET_KEY)
            self.assertEqual(len(cfg.SECRET_KEY), 64)
            with open(secret_path) as sf:
                self.assertEqual(sf.read().strip(), cfg.SECRET_KEY)
        finally:
            os.environ.pop("JWT_SECRET", None)
            os.environ.pop("_JWT_SECRET_FILE", None)
            if os.path.exists(secret_path):
                os.unlink(secret_path)
            if "config" in sys.modules:
                del sys.modules["config"]
    def test_jwt_secret_env_override(self):
        os.environ["JWT_SECRET"] = "test-secret-from-env"
        sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
        if "config" in sys.modules:
            del sys.modules["config"]
        cfg = __import__("config")
        self.assertEqual(cfg.SECRET_KEY, "test-secret-from-env")
        os.environ.pop("JWT_SECRET", None)
        if "config" in sys.modules:
            del sys.modules["config"]
    def test_yk_pin_requires_env(self):
        os.environ.pop("YK_ROOT_PIN", None)
        os.environ.pop("YK_INT_PIN", None)
        sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
        if "config" in sys.modules:
            del sys.modules["config"]
        with self.assertRaises(RuntimeError):
            __import__("config")
        os.environ["YK_ROOT_PIN"] = "test1234"
        os.environ["YK_INT_PIN"] = "test5678"
        if "config" in sys.modules:
            del sys.modules["config"]
        cfg = __import__("config")
        self.assertEqual(cfg.YK_ROOT_PIN, "test1234")
        self.assertEqual(cfg.YK_INT_PIN, "test5678")
        os.environ.pop("YK_ROOT_PIN", None)
        os.environ.pop("YK_INT_PIN", None)
        if "config" in sys.modules:
            del sys.modules["config"]
 if __name__ == "__main__":
    unittest.main()
--- a/tests/test_models.py
+++ b/tests/test_models.py
@ -1,22 +0,0 @@
 """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())
--- a/tests/test_signing.py
+++ b/tests/test_signing.py
@ -1,55 +1,51 @@
-"""DER helpers and signing pipeline (no YubiKey required)."""
+import os
 import sys
 import unittest
 from unittest import mock
-from datetime import datetime, timedelta, timezone
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "api"))
 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
+class TestSigningSecurity(unittest.TestCase):
    @mock.patch("signing.subprocess.run")
    @mock.patch("signing._make_temp_file")
    def test_pin_not_in_cli_args(self, mock_mkstemp, mock_run):
        """PIN must NOT appear in subprocess command-line args."""
        mock_mkstemp.return_value = "/tmp/test_file"
        mock_run.return_value = mock.MagicMock(returncode=0)
        import signing
        signing.sign_tbs_with_yk(b"\x00" * 100, "test-pin-123")
        cmd = mock_run.call_args[0][0]
        self.assertNotIn("test-pin-123", cmd, "PIN must not appear in command args")
        self.assertIn("--pin-source", cmd)
        self.assertIn("stdin", cmd)
        self.assertNotIn("--pin", cmd) or cmd.index("--pin-source") < cmd.index("--pin")
    @mock.patch("signing.subprocess.run")
    def test_temp_files_use_mkstemp(self, mock_run):
        """Temp files must use mkstemp, not predictable names."""
        mock_run.return_value = mock.MagicMock(returncode=0)
        with mock.patch("signing.tempfile.mkstemp") as mock_mkstemp:
            mock_mkstemp.return_value = (0, "/tmp/unpredictable_name")
            import signing
            try:
                signing.sign_tbs_with_yk(b"\x00" * 100, "pin")
            except Exception:
                pass
            calls = [c[0][1] for c in mock_mkstemp.call_args_list]
            self.assertTrue(all("/tmp/unpredictable_name" in c for c in calls),
                           "All temp files should use mkstemp")
    def test_no_hardcoded_pin_default(self):
        """YK PIN must fail if env var not set."""
        os.environ.pop("YK_ROOT_PIN", None)
        os.environ.pop("YK_INT_PIN", None)
        if "config" in sys.modules:
            del sys.modules["config"]
        with self.assertRaises(RuntimeError):
            __import__("config")
-def _dummy_cert_der():
+if __name__ == "__main__":
-    key = ec.generate_private_key(ec.SECP384R1())
+    unittest.main()
    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)
Author	SHA1	Message	Date
Jarian Cottingham	df82068aeb	fix: CSRF protection, cookie security flags, rate limiting, XSS (#23,#24,#25,#28,#33) Add CSRF tokens to all cookie-based POST endpoints. Set Secure and SameSite=Strict on auth cookie. Rate limit login to 5 attempts per 15min per IP. Escape HTML in signing error messages (XSS fix). Remove duplicate get_user_from_cookie definition.	2026-07-04 04:51:56 +00:00
Jarian Cottingham	c759597ad0	fix: pass YubiKey PIN via stdin instead of CLI arg (#18 , #31 ) --pin-source stdin prevents PIN visibility in `ps` output. Use tempfile.mkstemp for all temp files (unpredictable names, 0600 perms). Clean up temp files in finally block. Add tests for PIN not in args and mkstemp usage.	2026-07-04 04:49:56 +00:00
Jarian Cottingham	279a28515f	fix: generate JWT secret on first boot instead of hardcoded default (#19 ) Generate random 256-bit secret stored in /var/lib/certauth/.jwt_secret (mode 0600) ENV JWT_SECRET takes precedence. Remove CHANGE_ME_JWT_SECRET default. Add tests for secret generation and env override	2026-07-04 04:48:30 +00:00
Jarian Cottingham	c4d952430e	update local changes	2026-07-03 01:14:04 +00:00