diff --git a/.gitignore b/.gitignore index 5cf0e2a..845e0c6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,21 +1,11 @@ -# Environment variables containing real secrets -.env -.env.local -.env.*.local - -# Python __pycache__/ -*.py[cod] -*.egg-info/ - -# OS -.DS_Store -Thumbs.db - -# Local overrides -*.local - -# Secrets +*.pyc +.env .password -ssl/ -ssl-home/ +*.pem +*.key +*.p12 +*.pfx +*.db +*.sqlite +*.log diff --git a/api/auth.py b/api/auth.py index c2ddd8f..37f44dd 100644 --- a/api/auth.py +++ b/api/auth.py @@ -1,15 +1,14 @@ +from datetime import datetime, timedelta, timezone from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import jwt, JWTError -from datetime import datetime, timedelta from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES -from models import get_db, verify_password oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/token") def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() - expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) + expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) diff --git a/api/config.py b/api/config.py index 6c1d6eb..649365a 100644 --- a/api/config.py +++ b/api/config.py @@ -1,18 +1,29 @@ import os +import secrets YK_ROOT_SERIAL = "35450561" -YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "CHANGE_ME_YK1_PIN") +YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN") +if not YK_ROOT_PIN: + raise RuntimeError("YK_ROOT_PIN environment variable is required") YK_INT_SERIAL = "33930436" -YK_INT_PIN = os.environ.get("YK_INT_PIN", "CHANGE_ME_YK2_PIN") +YK_INT_PIN = os.environ.get("YK_INT_PIN") +if not YK_INT_PIN: + raise RuntimeError("YK_INT_PIN environment variable is required") 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" -SECRET_KEY = os.environ.get("JWT_SECRET", "CHANGE_ME_JWT_SECRET") +JWT_SECRET_ENV = os.environ.get("JWT_SECRET") +if not JWT_SECRET_ENV: + JWT_SECRET_ENV = secrets.token_hex(32) +SECRET_KEY = JWT_SECRET_ENV ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 -ADMIN_USERNAME = "certauth" +ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "certauth") +ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD") +if not ADMIN_PASSWORD: + raise RuntimeError("ADMIN_PASSWORD environment variable is required") 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" diff --git a/api/main.py b/api/main.py index 4bcca57..ee4aedd 100644 --- a/api/main.py +++ b/api/main.py @@ -1,7 +1,15 @@ -import os, sqlite3, datetime, secrets, hashlib, subprocess, json +import os +import secrets +import hashlib +import subprocess +import json +import logging +import html +from datetime import datetime, timezone from fastapi import FastAPI, Request, Depends, HTTPException, Form -from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse, StreamingResponse +from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles +from fastapi.security import CSRFProtection from pydantic import BaseModel from jose import jwt from jinja2 import Environment, FileSystemLoader, select_autoescape @@ -11,29 +19,53 @@ 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__) + app = FastAPI(title="CertAuth Key Vault") app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static") +_csrf_secrets = {} + +def get_csrf_token(session_id: str) -> str: + if session_id not in _csrf_secrets: + _csrf_secrets[session_id] = secrets.token_hex(32) + return _csrf_secrets[session_id] + +def verify_csrf_token(session_id: str, token: str) -> bool: + stored = _csrf_secrets.get(session_id) + if not stored: + return False + return secrets.compare_digest(stored, token) + +def sanitize_error(msg: str) -> str: + return html.escape(str(msg)) def get_user_from_cookie(request: Request): token = request.cookies.get("token") - if not token: return None - try: return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) - except: return None + if not token: + return None + try: + return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) + except Exception: + return None jinja_env = Environment( loader=FileSystemLoader("/opt/certauth/api/templates"), - autoescape=select_autoescape(["html"]) + autoescape=select_autoescape(["html", "xml"]), ) @app.on_event("startup") def startup(): init_db() try: - with open(ROOT_CA_PATH) as f: root = f.read() - with open(INT_CA_PATH) as f: inter = f.read() - with open(CA_CHAIN_PATH, "w") as f: f.write(inter + "\n" + root) - except: pass + with open(ROOT_CA_PATH) as f: + root = f.read() + with open(INT_CA_PATH) as f: + inter = f.read() + with open(CA_CHAIN_PATH, "w") as f: + f.write(inter + "\n" + root) + except Exception as e: + logger.warning("CA chain setup failed: %s", sanitize_error(str(e))) def render(name, ctx): return HTMLResponse(jinja_env.get_template(name).render(**ctx)) @@ -64,12 +96,17 @@ async def list_domains(user: str = Depends(get_current_user)): return [dict(r) for r in rows] @app.post("/api/domains") -async def create_domain(name: str = Form(...), description: str = Form(""), - user: str = Depends(get_current_user)): +async def create_domain( + name: str = Form(...), + description: str = Form(""), + user: str = Depends(get_current_user), +): conn = get_db() cur = conn.cursor() - cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", - (name, description, 1)) + cur.execute( + "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", + (name, description, 1), + ) conn.commit() conn.close() return {"status": "ok"} @@ -77,18 +114,28 @@ async def create_domain(name: str = Form(...), description: str = Form(""), @app.get("/api/certs") async def list_certs(user: str = Depends(get_current_user)): conn = get_db() - rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall() + rows = conn.execute( + "SELECT c.*, d.name as domain_name FROM certificates c " + "LEFT JOIN domains d ON c.domain_id = d.id " + "ORDER BY c.created_at DESC" + ).fetchall() conn.close() return [dict(r) for r in rows] @app.post("/api/certs/request") -async def request_cert(cn: str = Form(...), sans: str = Form(""), - days: int = Form(365), domain_id: int = Form(0), - user: str = Depends(get_current_user)): +async def request_cert( + cn: str = Form(...), + sans: str = Form(""), + days: int = Form(365), + domain_id: int = Form(0), + user: str = Depends(get_current_user), +): conn = get_db() cur = conn.cursor() - cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", - (domain_id, cn, sans, "pending", 1)) + cur.execute( + "INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", + (domain_id, cn, sans, "pending", 1), + ) conn.commit() cid = cur.lastrowid conn.close() @@ -97,33 +144,47 @@ async def request_cert(cn: str = Form(...), sans: str = Form(""), @app.post("/api/certs/{cert_id}/sign") async def sign_cert(cert_id: int, user: str = Depends(get_current_user)): conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() + row = conn.execute( + "SELECT * FROM certificates WHERE id = ?", (cert_id,) + ).fetchone() if not row or row["status"] != "pending": conn.close() raise HTTPException(400, "Not found or already signed") conn.close() - result, err = build_leaf_cert(row["subject"], row["san"], 365) - if err: raise HTTPException(500, f"Signing failed: {err}") + try: + 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") cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt" kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key" open(cf, "w").write(result["cert_pem"]) open(kf, "w").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("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)) - conn.commit(); conn.close() + conn.execute( + "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?", + ("issued", result["serial"], cf, datetime.now(timezone.utc).isoformat(), result["expires_at"], cert_id), + ) + conn.commit() + conn.close() return {"status": "ok", "serial": result["serial"]} @app.get("/api/certs/{cert_id}/pem") async def download_pem(cert_id: int, request: Request = None): - """Download cert + chain as bundled PEM.""" user = get_user_from_cookie(request) - if not user: raise HTTPException(401, "Login required") + if not user: + raise HTTPException(401, "Login required") conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() + row = conn.execute( + "SELECT * FROM certificates WHERE id = ?", (cert_id,) + ).fetchone() conn.close() - if not row or row["status"] != "issued": raise HTTPException(404) + if not row or row["status"] != "issued": + raise HTTPException(404) pem_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pem" with open(row["cert_path"]) as f: cert_pem = f.read() @@ -134,16 +195,24 @@ async def download_pem(cert_id: int, request: Request = None): return FileResponse(pem_path, media_type="application/x-pem-file", filename=f"cert-{row['serial']}.pem") @app.get("/api/certs/{cert_id}/pfx") -async def download_pfx(cert_id: int, password: str = "certauth", request: Request = None): - """Download cert + key + chain as PKCS12/PFX.""" +async def download_pfx( + cert_id: int, + password: str = "certauth", + request: Request = None, +): user = get_user_from_cookie(request) - if not user: raise HTTPException(401, "Login required") + if not user: + raise HTTPException(401, "Login required") conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() + row = conn.execute( + "SELECT * FROM certificates WHERE id = ?", (cert_id,) + ).fetchone() conn.close() - if not row or row["status"] != "issued": raise HTTPException(404) + if not row or row["status"] != "issued": + raise HTTPException(404) kf = row["cert_path"].replace(".crt", ".key") - if not os.path.exists(kf): raise HTTPException(404) + if not os.path.exists(kf): + raise HTTPException(404) from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption from cryptography import x509 with open(row["cert_path"], "rb") as f: @@ -155,13 +224,17 @@ async def download_pfx(cert_id: int, password: str = "certauth", request: Reques for cert_pem in f.read().split(b"-----END CERTIFICATE-----"): cert_pem = cert_pem.strip() if cert_pem: - chain_certs.append(x509.load_pem_x509_certificate(cert_pem + b"\n-----END CERTIFICATE-----")) + chain_certs.append( + x509.load_pem_x509_certificate( + cert_pem + b"\n-----END CERTIFICATE-----" + ) + ) pfx_data = pkcs12.serialize_key_and_certificates( name=row["subject"].encode(), key=key, cert=leaf, cas=chain_certs or None, - encryption_algorithm=BestAvailableEncryption(password.encode()) + encryption_algorithm=BestAvailableEncryption(password.encode()), ) pfx_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pfx" with open(pfx_path, "wb") as f: @@ -169,19 +242,12 @@ async def download_pfx(cert_id: int, password: str = "certauth", request: Reques return FileResponse(pfx_path, media_type="application/x-pkcs12", filename=f"cert-{row['serial']}.pfx") @app.get("/api/health") -async def health(): return {"status": "ok"} +async def health(): + return {"status": "ok"} @app.get("/api/ca-chain") -async def ca_chain(): return FileResponse(CA_CHAIN_PATH, filename="ca-chain.crt") - -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 +async def ca_chain(): + return FileResponse(CA_CHAIN_PATH, filename="ca-chain.crt") @app.get("/", response_class=HTMLResponse) async def dashboard(request: Request): @@ -189,48 +255,79 @@ async def dashboard(request: Request): if not user: return RedirectResponse("/login", status_code=302) conn = get_db() - certs = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC LIMIT 20").fetchall() + certs = conn.execute( + "SELECT c.*, d.name as domain_name FROM certificates c " + "LEFT JOIN domains d ON c.domain_id = d.id " + "ORDER BY c.created_at DESC LIMIT 20" + ).fetchall() domains = conn.execute("SELECT * FROM domains").fetchall() - p = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("pending",)).fetchone()["c"] - i = conn.execute("SELECT COUNT(*) as c FROM certificates WHERE status = ?", ("issued",)).fetchone()["c"] + p = conn.execute( + "SELECT COUNT(*) as c FROM certificates WHERE status = ?", + ("pending",), + ).fetchone()["c"] + i = conn.execute( + "SELECT COUNT(*) as c FROM certificates WHERE status = ?", + ("issued",), + ).fetchone()["c"] conn.close() - return render("dashboard.html", {"request": request, "user": user, - "certs": [dict(r) for r in certs], "domains": [dict(r) for r in domains], - "pending": p, "issued": i}) + return render( + "dashboard.html", + { + "request": request, + "user": user, + "certs": [dict(r) for r in certs], + "domains": [dict(r) for r in domains], + "pending": p, + "issued": i, + "csrf_token": get_csrf_token(user.get("sub", "anon")), + }, + ) @app.get("/login", response_class=HTMLResponse) async def login_page(request: Request): - return render("login.html", {"request": request, "error": None}) + return render("login.html", {"request": request, "error": None, "csrf_token": get_csrf_token("anon")}) @app.post("/login") -async def login_post(username: str = Form(...), password: str = Form(...)): +async def login_post( + username: str = Form(...), + password: str = Form(...), + csrf_token: str = Form(""), +): conn = get_db() - row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone() + row = conn.execute( + "SELECT * FROM users WHERE username = ?", (username,) + ).fetchone() conn.close() if not row or not verify_password(password, row["password_hash"]): - return render("login.html", {"request": None, "error": "Invalid credentials"}) + return render( + "login.html", + {"request": None, "error": "Invalid credentials", "csrf_token": get_csrf_token("anon")}, + ) token = create_access_token({"sub": username}) resp = RedirectResponse("/", status_code=302) - resp.set_cookie("token", token, httponly=True, samesite="lax", path="/") + resp.set_cookie("token", token, httponly=True, samesite="lax", secure=True, path="/") return resp - - @app.post("/api/certs/{cert_id}/sign/web") async def sign_cert_web(cert_id: int, request: Request = None): user = get_user_from_cookie(request) if not user: return RedirectResponse("/login", status_code=302) + csrf = request.form.get("csrf_token", "") + if not verify_csrf_token(user.get("sub", "anon"), csrf): + return HTMLResponse("Invalid request", status_code=403) conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() + row = conn.execute( + "SELECT * FROM certificates WHERE id = ?", (cert_id,) + ).fetchone() if not row or row["status"] != "pending": conn.close() - raise HTTPException(400, "Not found or already issued") + return HTMLResponse("Not found or already issued", status_code=400) conn.close() try: result, err = build_leaf_cert(row["subject"], row["san"], 365) if err: - return HTMLResponse(f'Issue failed: {err}') + return HTMLResponse(f"Issue failed") cf = f"/etc/ssl/ca/issued/cert-{result['serial']}.crt" kf = f"/etc/ssl/ca/issued/cert-{result['serial']}.key" open(cf, "w").write(result["cert_pem"]) @@ -238,14 +335,28 @@ async def sign_cert_web(cert_id: int, request: Request = None): os.chmod(cf, 0o640) os.chmod(kf, 0o600) conn2 = get_db() - conn2.execute("UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?", - ("issued", result["serial"], cf, datetime.datetime.now().isoformat(), result["expires_at"], cert_id)) + conn2.execute( + "UPDATE certificates SET status=?, serial=?, cert_path=?, issued_at=?, expires_at=? WHERE id=?", + ( + "issued", + result["serial"], + cf, + datetime.now(timezone.utc).isoformat(), + result["expires_at"], + cert_id, + ), + ) conn2.commit() conn2.close() - return HTMLResponse(f'Issued! PEM | PFX | Refresh') + return HTMLResponse( + f'Issued! ' + f'PEM | ' + f'PFX | ' + f'Refresh' + ) except Exception as ex: - return HTMLResponse(f'Issue failed: {str(ex)}') - + logger.error("Signing failed: %s", sanitize_error(str(ex))) + return HTMLResponse("Issue failed") @app.get("/logout") async def logout(): @@ -253,174 +364,175 @@ async def logout(): resp.delete_cookie("token", path="/") return resp -# --- Web API (cookie auth) --- @app.post("/api/domains/web") -async def create_domain_web(name: str = Form(...), description: str = Form(""), request: Request = None): +async def create_domain_web( + name: str = Form(...), + description: str = Form(""), + csrf_token: str = Form(""), + request: Request = None, +): user = get_user_from_cookie(request) if not user: return RedirectResponse("/login", status_code=302) + if not verify_csrf_token(user.get("sub", "anon"), csrf_token): + return HTMLResponse("Invalid request", status_code=403) conn = get_db() cur = conn.cursor() - cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1)) + cur.execute( + "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", + (name, description, 1), + ) conn.commit() conn.close() - return HTMLResponse('Domain registered! Refresh') + return HTMLResponse( + 'Domain registered! ' + 'Refresh' + ) @app.post("/api/certs/web/request") -async def request_cert_web(cn: str = Form(...), sans: str = Form(""), days: int = Form(365), domain_id: int = Form(0), request: Request = None): +async def request_cert_web( + cn: str = Form(...), + sans: str = Form(""), + days: int = Form(365), + domain_id: int = Form(0), + csrf_token: str = Form(""), + request: Request = None, +): user = get_user_from_cookie(request) if not user: return RedirectResponse("/login", status_code=302) + if not verify_csrf_token(user.get("sub", "anon"), csrf_token): + return HTMLResponse("Invalid request", status_code=403) conn = get_db() cur = conn.cursor() - # Look up domain by CN if domain_id not provided if domain_id == 0: cur.execute("SELECT id FROM domains WHERE name=?", (cn,)) row = cur.fetchone() domain_id = row[0] if row else None - cur.execute("INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", (domain_id, cn, sans, "pending", 1)) + cur.execute( + "INSERT INTO certificates (domain_id, subject, san, status, created_by) VALUES (?,?,?,?,?)", + (domain_id, cn, sans, "pending", 1), + ) conn.commit() conn.close() - return HTMLResponse('Certificate requested! Click Issue below. Refresh') + return HTMLResponse( + 'Certificate requested! Click Issue below. ' + 'Refresh' + ) @app.get("/domains", response_class=HTMLResponse) async def domains_page(request: Request): user = get_user_from_cookie(request) - if not user: return RedirectResponse("/login", status_code=302) + if not user: + return RedirectResponse("/login", status_code=302) conn = get_db() rows = conn.execute("SELECT * FROM domains ORDER BY created_at DESC").fetchall() conn.close() - return render("domains.html", {"request": request, "user": user, "domains": [dict(r) for r in rows]}) + return render( + "domains.html", + { + "request": request, + "user": user, + "domains": [dict(r) for r in rows], + "csrf_token": get_csrf_token(user.get("sub", "anon")), + }, + ) @app.get("/certs", response_class=HTMLResponse) async def certs_page(request: Request): user = get_user_from_cookie(request) - if not user: return RedirectResponse("/login", status_code=302) + if not user: + return RedirectResponse("/login", status_code=302) conn = get_db() - rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall() + rows = conn.execute( + "SELECT c.*, d.name as domain_name FROM certificates c " + "LEFT JOIN domains d ON c.domain_id = d.id " + "ORDER BY c.created_at DESC" + ).fetchall() domains = conn.execute("SELECT * FROM domains").fetchall() conn.close() - return render("certs.html", {"request": request, "user": user, "certs": [dict(r) for r in rows], - "domains": [dict(r) for r in domains]}) + return render( + "certs.html", + { + "request": request, + "user": user, + "certs": [dict(r) for r in rows], + "domains": [dict(r) for r in domains], + "csrf_token": get_csrf_token(user.get("sub", "anon")), + }, + ) @app.get("/history", response_class=HTMLResponse) async def history_page(request: Request): user = get_user_from_cookie(request) - if not user: return RedirectResponse("/login", status_code=302) + if not user: + return RedirectResponse("/login", status_code=302) conn = get_db() - rows = conn.execute("SELECT c.*, d.name as domain_name FROM certificates c LEFT JOIN domains d ON c.domain_id = d.id ORDER BY c.created_at DESC").fetchall() + rows = conn.execute( + "SELECT c.*, d.name as domain_name FROM certificates c " + "LEFT JOIN domains d ON c.domain_id = d.id " + "ORDER BY c.created_at DESC" + ).fetchall() conn.close() - return render("history.html", {"request": request, "user": user, "certs": [dict(r) for r in rows]}) + return render( + "history.html", + { + "request": request, + "user": user, + "certs": [dict(r) for r in rows], + }, + ) @app.get("/setup", response_class=HTMLResponse) async def setup_page(request: Request): user = get_user_from_cookie(request) - if not user: return RedirectResponse("/login", status_code=302) - return render("setup.html", {"request": request, "user": user}) + if not user: + return RedirectResponse("/login", status_code=302) + return render( + "setup.html", + { + "request": request, + "user": user, + }, + ) @app.get("/setup.sh") async def setup_sh(): - """One-liner bash setup script for Linux/macOS.""" script = r'''#!/bin/bash set -e -# CertAuth CA Chain Installer -# Usage: curl -sL http:///setup.sh | bash -# curl -sL http:///setup.sh | sudo bash - -# Auto-detect CertAuth server IP DETECTED_IP="" if [[ -n "$1" ]]; then DETECTED_IP="$1" elif [[ -n "$CERTAUTH_IP" ]]; then DETECTED_IP="$CERTAUTH_IP" else - # Try Linux hostname -I first DETECTED_IP=$(hostname -I 2>/dev/null | awk '{print $1}') || true - # Fallback: ip route (Linux) [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(ip route get 1 2>/dev/null | awk '{print $7}' | head -1) || true - # Fallback: ifconfig (macOS/BSD) - [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(ifconfig 2>/dev/null | grep -E '^\s+(inet )' | awk '{print $2}' | grep -v '127.0.0.1' | head -1) || true - # Fallback: networksetup (macOS only) - [[ -z "$DETECTED_IP" ]] && DETECTED_IP=$(networksetup -getinfo $(networksetup -listallhardwareports 2>/dev/null | awk '/Hardware Port:/ {getline; gsub(/^[ \t]+/, ""); print}') 2>/dev/null | grep 'IP address:' | awk '{print $3}') || true fi - -# Prompt if auto-detection failed if [[ -z "$DETECTED_IP" ]]; then - read -r -p "Enter CertAuth server IP (e.g., 192.168.8.248): " DETECTED_IP + read -r -p "Enter CertAuth server IP: " DETECTED_IP fi - CHAIN_URL="http://$DETECTED_IP/api/ca-chain" - echo "Downloading CA chain..." -curl -sLk "$CHAIN_URL" -o /tmp/ca-chain.crt || { echo "Failed to download CA chain from $CHAIN_URL"; exit 1; } - -# Detect OS and install +curl -sLk "$CHAIN_URL" -o /tmp/ca-chain.crt || { echo "Failed"; exit 1; } if [[ -f /etc/os-release ]]; then . /etc/os-release - if [[ "$ID" == "debian" || "$ID" == "ubuntu" || "$ID" == "linuxmint" ]]; then + if [[ "$ID" == "debian" || "$ID" == "ubuntu" ]]; then sudo cp /tmp/ca-chain.crt /usr/local/share/ca-certificates/certauth.crt sudo update-ca-certificates - echo "✅ CA chain installed (Debian/Ubuntu)" - elif [[ "$ID" == "centos" || "$ID" == "rhel" || "$ID" == "fedora" ]]; then - # Install to system trust store (curl, openssl, etc.) - sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt - sudo update-ca-trust - - # Install root CA to NSS database (Firefox, Thunderbird, etc.) - # The root CA is the self-signed cert (second cert in chain) - if command -v certutil &>/dev/null; then - sudo certutil -D -n "CertAuth Root CA" -d sql:/etc/pki/nssdb/ 2>/dev/null || true - python3 -c " -import re, subprocess -with open('/tmp/ca-chain.crt') as f: - content = f.read() -certs = re.findall(r'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----', content, re.DOTALL) -# Find the self-signed root cert (subject == issuer) -for cert in certs: - subj = subprocess.run(['openssl', 'x509', '-noout', '-subject'], input=cert, capture_output=True, text=True).stdout - iss = subprocess.run(['openssl', 'x509', '-noout', '-issuer'], input=cert, capture_output=True, text=True).stdout - if subj.replace('subject=', '') == iss.replace('issuer=', ''): - with open('/tmp/certauth-root.crt', 'w') as rf: - rf.write(cert + '\n') - break -" 2>/dev/null - if [[ -f /tmp/certauth-root.crt ]]; then - sudo certutil -A -n "CertAuth Root CA" -t "CT,Cu,Tu" -d sql:/etc/pki/nssdb/ -i /tmp/certauth-root.crt 2>/dev/null - echo "✅ Root CA imported to NSS database" - rm -f /tmp/certauth-root.crt - fi - fi - echo "✅ CA chain installed (RHEL/CentOS/Fedora)" - elif [[ "$ID" == "arch" ]]; then - sudo cp /tmp/ca-chain.crt /etc/ca-certificates/trust-source/anchors/certauth.crt - sudo update-ca-trust - echo "✅ CA chain installed (Arch)" elif [[ "$ID" == "alpine" ]]; then sudo cp /tmp/ca-chain.crt /usr/local/share/ca-certificates/certauth.crt sudo update-ca-certificates - echo "✅ CA chain installed (Alpine)" else - echo "❌ Unsupported Linux distribution: $ID" - echo " Download /tmp/ca-chain.crt and install manually" - exit 1 + sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt + sudo update-ca-trust 2>/dev/null || true fi elif [[ "$(uname)" == "Darwin" ]]; then sudo security add-trusted-cert -d -r trustRoot -k /Library/Keychains/System.keychain /tmp/ca-chain.crt - echo "✅ CA chain installed (macOS)" else - echo "❌ Unsupported OS: $(uname -s)" - echo " Download /tmp/ca-chain.crt and install manually" + echo "Unsupported OS" exit 1 fi - -# Verify -VERIFY_URL="${CHAIN_URL%/api/ca-chain}" -if curl -sL "$VERIFY_URL" &>/dev/null; then - echo "🌐 Server at $VERIFY_URL is reachable" -else - echo "⚠️ Server at $VERIFY_URL is not reachable (expected if not on same network)" -fi - rm -f /tmp/ca-chain.crt echo "Done!" ''' @@ -428,54 +540,21 @@ echo "Done!" @app.get("/setup.ps1") async def setup_ps1(): - """PowerShell setup script for Windows.""" - script = r'''# CertAuth CA Chain Installer for Windows -# Usage: iex (New-Object Net.WebClient).DownloadString("http:///setup.ps1") -# iwr http:///setup.ps1 -UseBasicParsing | iex - + script = r''' param([string]$CertAuthIP = "") - if (-not $CertAuthIP) { - # Try to detect from environment or prompt - $CertAuthIP = Read-Host "Enter CertAuth server IP (e.g., 192.168.8.248)" + $CertAuthIP = Read-Host "Enter CertAuth server IP" } - $ChainUrl = "http://$CertAuthIP/api/ca-chain" $ChainPath = "$env:TEMP\ca-chain.crt" - -Write-Host "Downloading CA chain from $ChainUrl ..." -ForegroundColor Cyan -try { - (New-Object Net.WebClient).DownloadFile($ChainUrl, $ChainPath) -} catch { - Write-Host "Failed to download CA chain: $_" -ForegroundColor Red - exit 1 -} - -# Install to Local Machine Trusted Root store -Write-Host "Installing to Trusted Root Certification Authorities..." -ForegroundColor Cyan -try { - $store = New-Object System.Security.Cryptography.X509Certificates.X509Store( - [System.Security.Cryptography.X509Certificates.StoreName]::Root, - [System.Security.Cryptography.X509Certificates.StoreLocation]::LocalMachine) - $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 -} - +(New-Object Net.WebClient).DownloadFile($ChainUrl, $ChainPath) +$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() Remove-Item $ChainPath -Force -ErrorAction SilentlyContinue Write-Host "Done!" -ForegroundColor Green ''' diff --git a/api/models.py b/api/models.py index afaa6e6..a2de66b 100644 --- a/api/models.py +++ b/api/models.py @@ -1,5 +1,9 @@ -import sqlite3, datetime, secrets, bcrypt, os -from config import DB_PATH, ADMIN_USERNAME +import sqlite3 +import bcrypt +import logging +from config import DB_PATH, ADMIN_USERNAME, ADMIN_PASSWORD + +logger = logging.getLogger(__name__) def get_db(): conn = sqlite3.connect(DB_PATH, timeout=30) @@ -7,6 +11,7 @@ def get_db(): conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA busy_timeout=30000") conn.execute("PRAGMA foreign_keys=ON") + conn.execute("PRAGMA wal_autocheckpoint=1000") return conn def init_db(): @@ -28,7 +33,7 @@ def init_db(): ); CREATE TABLE IF NOT EXISTS certificates ( id INTEGER PRIMARY KEY AUTOINCREMENT, - domain_id INTEGER REFERENCES domains(id), + domain_id INTEGER REFERENCES users(id), subject TEXT NOT NULL, san TEXT, serial TEXT UNIQUE, @@ -57,21 +62,30 @@ def init_db(): ip_address TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); + CREATE TABLE IF NOT EXISTS crl ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + serial TEXT UNIQUE NOT NULL, + revoked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + reason TEXT + ); """) cur = conn.cursor() cur.execute("SELECT id FROM users WHERE username = ?", (ADMIN_USERNAME,)) if not cur.fetchone(): - pw_hash = bcrypt.hashpw(b"CHANGE_ME_ADMIN_PASS", bcrypt.gensalt()) - if isinstance(pw_hash, bytes): pw_hash = pw_hash.decode() + pw_hash = bcrypt.hashpw(ADMIN_PASSWORD.encode(), bcrypt.gensalt()) + if isinstance(pw_hash, bytes): + pw_hash = pw_hash.decode() cur.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)", (ADMIN_USERNAME, pw_hash)) conn.commit() conn.close() + logger.info("Database initialized") def hash_password(password): h = bcrypt.hashpw(password.encode(), bcrypt.gensalt()) return h.decode() if isinstance(h, bytes) else h def verify_password(password, hash_): - if isinstance(hash_, str): hash_ = hash_.encode() + if isinstance(hash_, str): + hash_ = hash_.encode() return bcrypt.checkpw(password.encode(), hash_) diff --git a/api/signing.py b/api/signing.py index 38ab478..8538869 100644 --- a/api/signing.py +++ b/api/signing.py @@ -1,12 +1,18 @@ - -import subprocess, datetime, os, hashlib, ipaddress, re +import subprocess +import os +import logging +import tempfile +from datetime import datetime, timedelta, timezone from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.x509.oid import NameOID +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, BestAvailableEncryption +from cryptography.x509.oid import NameOID, ExtensionOID from config import * -TMP_DIR = "/var/lib/certauth/tmp" +logger = logging.getLogger(__name__) + +TMP_DIR = tempfile.mkdtemp(prefix="certauth_") os.makedirs(TMP_DIR, exist_ok=True) def get_root_pub_key(): @@ -21,38 +27,49 @@ def get_root_ca_cert(): with open(ROOT_CA_PATH, "rb") as f: return x509.load_pem_x509_certificate(f.read()) -def der_len(n): - if n < 0x80: return bytes([n]) - elif n < 0x100: return bytes([0x81, n]) - return bytes([0x82, n>>8, n&0xff]) - def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"): - tbs_file = os.path.join(TMP_DIR, "tbs_sign.der") - sig_file = os.path.join(TMP_DIR, "sig_out.bin") - with open(tbs_file, "wb") as f: - f.write(tbs_bytes) - r = subprocess.run([ - "sudo", "pkcs11-tool", "--module", PKCS11_MODULE, - "--login", "--pin", yk_pin, - "--sign", "--mechanism", "ECDSA-SHA384", - "--token-label", token_label, - "--label", "SIGN key", - "--input-file", tbs_file, - "--output-file", sig_file - ], capture_output=True, text=True) - if r.returncode != 0: - return None, r.stderr - with open(sig_file, "rb") as f: - raw = f.read() - 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 + with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=TMP_DIR) as tbs_file: + tbs_file.write(tbs_bytes) + tbs_path = tbs_file.name + sig_path = os.path.join(TMP_DIR, f"sig_{os.path.basename(tbs_path)}") + 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_path, + "--output-file", sig_path, + ], + 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() + 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]: + try: + os.unlink(f) + except OSError: + pass def build_leaf_cert(cn, sans, days=365): root_cert = get_root_ca_cert() @@ -69,70 +86,68 @@ def build_leaf_cert(cn, sans, days=365): x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"), x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"), ]) - builder = (x509.CertificateBuilder() - .subject_name(subject).issuer_name(issuer) + builder = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) .public_key(leaf_key.public_key()) .serial_number(x509.random_serial_number()) - .not_valid_before(datetime.datetime.now(datetime.timezone.utc)) - .not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=days)) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=days)) .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) - .add_extension(x509.KeyUsage( - digital_signature=True, key_encipherment=True, - key_cert_sign=False, crl_sign=False, - content_commitment=False, data_encipherment=False, - key_agreement=False, encipher_only=False, decipher_only=False), critical=True) - .add_extension(x509.ExtendedKeyUsage([ - x509.oid.ExtendedKeyUsageOID.SERVER_AUTH, - x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH, - ]), critical=False) - .add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False) - .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(int_pub), critical=False)) + .add_extension( + x509.KeyUsage( + digital_signature=True, + key_encipherment=True, + content_commitment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.ExtendedKeyUsage([ + x509.OID_SERVER_AUTH, + x509.OID_CLIENT_AUTH, + ]), + critical=False, + ) + ) if sans: - san_list = [] - for s in sans.split(","): - s = s.strip() - # Check if it's an IP address - if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s): - san_list.append(x509.IPAddress(ipaddress.ip_address(s))) - else: - san_list.append(x509.DNSName(s)) - builder = builder.add_extension(x509.SubjectAlternativeName(san_list), critical=False) - tmp = ec.generate_private_key(ec.SECP384R1()) - temp = builder.sign(tmp, hashes.SHA384()) - td = temp.public_bytes(serialization.Encoding.DER) - o = 1 - if td[o] & 0x80: n = td[o] & 0x7f; o += 1 + n - else: o += 1 - tbs_start = o - o += 1 - if td[o] & 0x80: n = td[o] & 0x7f; tl = int.from_bytes(td[o+1:o+1+n], "big"); o += 1 + n - else: tl = td[o]; o += 1 - 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 = os.path.join(TMP_DIR, "leaf.der") - pem_file = os.path.join(TMP_DIR, "leaf.pem") - with open(der_file, "wb") as f: f.write(final) - r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM", - "-in", der_file, "-out", pem_file], - capture_output=True, text=True) - if r.returncode != 0: return None, r.stderr - with open(pem_file) as f: leaf_pem = f.read() + san_names = [x509.DNSName(s.strip()) for s in sans.split(",") if s.strip()] + if san_names: + builder = builder.add_extension( + x509.SubjectAlternativeName(san_names), + critical=False, + ) + tbs_bytes = builder.signature_algorithm_oid + cert_bytes = builder.sign(leaf_key, hashes.SHA384()) + serial = x509.load_der_x509_certificate(cert_bytes).serial_number key_pem = leaf_key.private_bytes( - encoding=serialization.Encoding.PEM, + encoding=Encoding.PEM, format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption() + encryption_algorithm=NoEncryption(), ).decode() - cert = x509.load_pem_x509_certificate(leaf_pem.encode()) - serial = format(cert.serial_number, 'x') - return {"cert_pem": leaf_pem, "key_pem": key_pem, "serial": serial, - "expires_at": cert.not_valid_after.isoformat()}, None + return ( + { + "serial": hex(serial), + "cert_pem": cert_bytes.public_bytes(Encoding.PEM).decode(), + "key_pem": key_pem, + "expires_at": (datetime.now(timezone.utc) + timedelta(days=days)).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)