diff --git a/.env.example b/.env.example index 4e0c247..5a5af9e 100644 --- a/.env.example +++ b/.env.example @@ -1,19 +1,50 @@ -# CertAuth Environment Variables -# Copy this to .env and fill in your real values +# CertAuth API environment variables. +# Installed to /etc/certauth/certauth.env (mode 600, owned by the API user) +# and loaded via EnvironmentFile= in certauth-api.service. +# +# The API fails closed on startup if any of the REQUIRED variables below +# are missing. -# Admin web login password -ADMIN_PASS=your-secure-admin-password +# ---- REQUIRED ------------------------------------------------------------- -# YubiKey 1 (Root CA) credentials -YK_ROOT_PIN=your-yk1-pin -YK_ROOT_PUK=your-yk1-puk +# YubiKey hardware assignments (serials are printed by `ykman list`; +# setup-certauth.sh auto-detects them at provision time) +YK_ROOT_SERIAL=your-root-yubikey-serial +YK_INT_SERIAL=your-intermediate-yubikey-serial -# YubiKey 2 (Intermediate CA) credentials -YK_INT_PIN=your-yk2-pin -YK_INT_PUK=your-yk2-puk +# YubiKey PINs (change from the defaults with: +# ykman piv access change-pin -P --new-pin ) +YK_ROOT_PIN=your-yk-root-pin +YK_INT_PIN=your-yk-intermediate-pin # JWT signing secret (generate with: python3 -c "import secrets; print(secrets.token_hex(32))") JWT_SECRET=your-64-char-hex-secret -# PFX download password +# Admin web/API login password +ADMIN_PASSWORD=your-secure-admin-password + +# ---- OPTIONAL (defaults shown) --------------------------------------------- + +# Admin username (default: certauth) +ADMIN_USERNAME=certauth + +# Filesystem layout (defaults match setup-certauth.sh production layout) +CERTAUTH_CA_BASE=/etc/ssl/ca +CERTAUTH_ROOT_CA=/etc/ssl/ca/root/root-ca.crt +CERTAUTH_INT_CA=/etc/ssl/ca/intermediate/intermediate-ca.crt +CERTAUTH_CA_CHAIN=/etc/ssl/ca/ca-chain.crt +CERTAUTH_ISSUED_DIR=/etc/ssl/ca/issued +CERTAUTH_TMP_DIR=/var/lib/certauth/tmp +CERTAUTH_DB_PATH=/var/lib/certauth/certauth.db + +# PKCS#11 (defaults work on aarch64 and x86_64 Ubuntu) +PKCS11_MODULE=/usr/lib/opensc-pkcs11.so +# pkcs11-tool --token-label; leave empty to match by key label only +PKCS11_TOKEN_LABEL= + +# Leaf certificate subject (matches setup-certauth.sh defaults) +CA_ORG=Home +CA_COUNTRY=US + +# PFX download default password (overridable per download) PFX_PASS=certauth diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..850a5a7 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025 Jarian Cottingham + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..5cdb420 --- /dev/null +++ b/README.md @@ -0,0 +1,107 @@ +# CertAuth Key Vault + +Self-contained Certificate Authority with YubiKey-backed signing. Two +YubiKey 5 Nano devices act as HSMs — one for the Root CA, one for the +Intermediate CA. Every certificate sign requires physical touch + PIN, so +compromising the server does not compromise the CA keys. + +``` +Root CA (YubiKey 1) → Intermediate CA (YubiKey 2) → Leaf certs + 25-year validity 15-year validity 1-year validity +``` + +## Components + +| Path | Purpose | +| --- | --- | +| `api/` | FastAPI service: REST API + Jinja2 web UI | +| `setup-certauth.sh` | Idempotent provisioner: fresh Ubuntu 24.04+ → full CA (aarch64/x86_64) | +| `certauth-api.service` | systemd unit for the API | +| `Caddyfile` | Caddy TLS-terminating reverse proxy in front of the API | +| `nginx-example.com` | Reference nginx config for the wider homelab vhost layout | +| `landing/` | Landing pages for the CA domain | +| `SKILL.md` | Operator/agent runbook: API usage, endpoints, workflows | +| `.env.example` | All environment variables documented | + +## Architecture + +- **Signing**: `pkcs11-tool` against the OpenSC PKCS#11 module; ECDSA-SHA384. + Private keys never leave the YubiKeys; only extracted public keys are + cached on disk. +- **Auth**: JWT bearer tokens for the REST API, session cookie for the web + UI, per-request CSRF tokens on all state-changing web endpoints. +- **Storage**: SQLite (WAL) for domains, certificate requests, audit log, + and CRL entries. +- **CRL**: revoked serials tracked in the DB; `scripts`-style daily CRL + refresh is documented in `SKILL.md`. + +## Quick start (provision a node) + +```bash +sudo bash setup-certauth.sh +``` + +Prerequisites: fresh Ubuntu 24.04+, two YubiKey 5 Nanos plugged in, root. +The script configures everything (CA hierarchy, API, Caddy, systemd) and +prints the generated PINs. Configuration is overridable via environment +(`CA_ORG`, `YK1_SERIAL`, `NETWORK_CIDR`, ...). + +## Running the API standalone + +```bash +pip install -r requirements.txt +export YK_ROOT_SERIAL=... YK_INT_SERIAL=... +export YK_ROOT_PIN=... YK_INT_PIN=... +export JWT_SECRET=$(python3 -c "import secrets; print(secrets.token_hex(32))") +export ADMIN_PASSWORD=... +python3 -m uvicorn main:app --host 127.0.0.1 --port 8000 # from api/ +``` + +All filesystem paths default to the production layout (`/etc/ssl/ca`, +`/var/lib/certauth`) and are overridable via `CERTAUTH_*` environment +variables — see `.env.example`. + +## API (summary) + +| Method | Path | Purpose | +| --- | --- | --- | +| POST | `/api/token` | Login → JWT | +| GET | `/api/me` | Validate token | +| GET/POST | `/api/domains` | List / register domain | +| GET | `/api/certs` | List certificate requests | +| POST | `/api/certs/request` | Create pending request (cn, sans) | +| POST | `/api/certs/{id}/sign` | Sign via YubiKey (touch + PIN) | +| GET | `/api/certs/{id}/pem` | Download leaf + chain | +| GET | `/api/certs/{id}/pfx` | Download PKCS#12 (password-protected) | +| GET | `/api/health` | Liveness | +| GET | `/api/ca-chain` | Download CA chain | + +Web UI: `/` dashboard, `/login`, `/domains`, `/certs`, `/history`, +`/setup` (provisioning helper pages). + +Full usage examples: see `SKILL.md`. + +## Tests + +```bash +pip install pytest +pytest tests/ -v +``` + +Unit tests cover auth (JWT round-trip, rejection of invalid tokens), +login/token endpoints, domain + certificate-request flows, and password +hashing. Signing paths that require a physical YubiKey are exercised at +the integration level on the provisioned node. + +## Security notes + +- YubiKey serials, PINs, JWT secret, and admin password are environment + driven and **fail closed** at import time — the service will not start + with missing configuration. +- Leaf keys are written `0600`, certs `0640`, under the issued directory. +- CSRF tokens use constant-time comparison (`secrets.compare_digest`). +- Error strings are HTML-escaped before rendering into pages. + +## License + +MIT — see [LICENSE](LICENSE). diff --git a/api/config.py b/api/config.py index 649365a..b99245e 100644 --- a/api/config.py +++ b/api/config.py @@ -1,29 +1,61 @@ -import os -import secrets +"""CertAuth configuration. -YK_ROOT_SERIAL = "35450561" -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") -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" -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 +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 + + +def _required(name: str) -> str: + value = os.environ.get(name) + if not value: + raise RuntimeError(f"{name} environment variable is required") + return value + + +# YubiKey hardware: serials must be supplied by the operator (auto-detected at +# setup time by setup-certauth.sh). Never hard-code device serials in source. +YK_ROOT_SERIAL = _required("YK_ROOT_SERIAL") +YK_INT_SERIAL = _required("YK_INT_SERIAL") +YK_ROOT_PIN = _required("YK_ROOT_PIN") +YK_INT_PIN = _required("YK_INT_PIN") + +# Filesystem layout (defaults match the production layout created by +# setup-certauth.sh; override via environment for testing or custom installs). +CA_BASE = os.environ.get("CERTAUTH_CA_BASE", "/etc/ssl/ca") +ROOT_CA_PATH = os.environ.get("CERTAUTH_ROOT_CA", f"{CA_BASE}/root/root-ca.crt") +INT_CA_PATH = os.environ.get("CERTAUTH_INT_CA", f"{CA_BASE}/intermediate/intermediate-ca.crt") +CA_CHAIN_PATH = os.environ.get("CERTAUTH_CA_CHAIN", f"{CA_BASE}/ca-chain.crt") +ISSUED_DIR = os.environ.get("CERTAUTH_ISSUED_DIR", f"{CA_BASE}/issued") +TMP_DIR = os.environ.get("CERTAUTH_TMP_DIR", "/var/lib/certauth/tmp") +DB_PATH = os.environ.get("CERTAUTH_DB_PATH", "/var/lib/certauth/certauth.db") + +# Extracted YubiKey public keys (written by setup-certauth.sh). +YK_PUB_ROOT = os.environ.get("YK_PUB_ROOT", "/tmp/yk1-root-pub.pem") +YK_PUB_INT = os.environ.get("YK_PUB_INT", "/tmp/yk2-int-pub.pem") + +# PKCS#11 module for pkcs11-tool. Default works on both aarch64 and x86_64 +# Ubuntu; override for nonstandard installs. +PKCS11_MODULE = os.environ.get("PKCS11_MODULE", "/usr/lib/opensc-pkcs11.so") +# Optional pkcs11-tool --token-label. Empty = match the setup-script +# invocation (key label only), which works with both YubiKeys attached. +PKCS11_TOKEN_LABEL = os.environ.get("PKCS11_TOKEN_LABEL", "") + +# Leaf certificate subject defaults (match setup-certauth.sh defaults) +CA_ORG = os.environ.get("CA_ORG", "Home") +CA_COUNTRY = os.environ.get("CA_COUNTRY", "US") + +# JWT / web session +JWT_SECRET = _required("JWT_SECRET") +SECRET_KEY = JWT_SECRET ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 + +# Admin bootstrap account 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" +ADMIN_PASSWORD = _required("ADMIN_PASSWORD") + +# PFX download default password (overridable per download) +PFX_DEFAULT_PASSWORD = os.environ.get("PFX_PASS", "certauth") diff --git a/api/main.py b/api/main.py index ee4aedd..0cc2871 100644 --- a/api/main.py +++ b/api/main.py @@ -1,28 +1,37 @@ 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 +from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, 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 -from config import * -from models import get_db, init_db, hash_password, verify_password +from config import ( + ALGORITHM, + CA_CHAIN_PATH, + INT_CA_PATH, + ISSUED_DIR, + PFX_DEFAULT_PASSWORD, + ROOT_CA_PATH, + SECRET_KEY, + TMP_DIR, +) +from models import get_db, init_db, verify_password from auth import create_access_token, get_current_user from signing import build_leaf_cert from cryptography.hazmat.primitives import serialization 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="/opt/certauth/api/static"), name="static") +app.mount("/static", StaticFiles(directory=STATIC_DIR, check_dir=False), name="static") _csrf_secrets = {} @@ -50,7 +59,7 @@ def get_user_from_cookie(request: Request): return None jinja_env = Environment( - loader=FileSystemLoader("/opt/certauth/api/templates"), + loader=FileSystemLoader(TEMPLATES_DIR), autoescape=select_autoescape(["html", "xml"]), ) @@ -67,6 +76,11 @@ def startup(): 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)) @@ -105,7 +119,7 @@ async def create_domain( cur = conn.cursor() cur.execute( "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", - (name, description, 1), + (name, description, get_user_id(conn, user)), ) conn.commit() conn.close() @@ -134,7 +148,7 @@ async def request_cert( cur = conn.cursor() cur.execute( "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 @@ -158,10 +172,12 @@ async def sign_cert(cert_id: int, user: str = Depends(get_current_user)): 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"]) + cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt" + kf = f"{ISSUED_DIR}/cert-{result['serial']}.key" + with open(cf, "w") as f: + f.write(result["cert_pem"]) + with open(kf, "w") as f: + f.write(result["key_pem"]) os.chmod(cf, 0o640) os.chmod(kf, 0o600) conn = get_db() @@ -185,7 +201,7 @@ async def download_pem(cert_id: int, request: Request = None): conn.close() if not row or row["status"] != "issued": 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: @@ -197,7 +213,7 @@ async def download_pem(cert_id: int, request: Request = None): @app.get("/api/certs/{cert_id}/pfx") async def download_pfx( cert_id: int, - password: str = "certauth", + password: str = PFX_DEFAULT_PASSWORD, request: Request = None, ): user = get_user_from_cookie(request) @@ -236,7 +252,7 @@ async def download_pfx( cas=chain_certs or None, encryption_algorithm=BestAvailableEncryption(password.encode()), ) - pfx_path = f"/var/lib/certauth/tmp/cert-{row['serial']}.pfx" + pfx_path = f"{TMP_DIR}/cert-{row['serial']}.pfx" with open(pfx_path, "wb") as f: f.write(pfx_data) return FileResponse(pfx_path, media_type="application/x-pkcs12", filename=f"cert-{row['serial']}.pfx") @@ -293,6 +309,11 @@ async def login_post( password: str = Form(...), csrf_token: str = Form(""), ): + if not verify_csrf_token("anon", csrf_token): + return render( + "login.html", + {"request": None, "error": "Invalid request", "csrf_token": get_csrf_token("anon")}, + ) conn = get_db() row = conn.execute( "SELECT * FROM users WHERE username = ?", (username,) @@ -327,11 +348,13 @@ async def sign_cert_web(cert_id: int, request: Request = None): try: result, err = build_leaf_cert(row["subject"], row["san"], 365) if 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"]) - open(kf, "w").write(result["key_pem"]) + return HTMLResponse("Issue failed") + cf = f"{ISSUED_DIR}/cert-{result['serial']}.crt" + kf = f"{ISSUED_DIR}/cert-{result['serial']}.key" + with open(cf, "w") as f: + f.write(result["cert_pem"]) + with open(kf, "w") as f: + f.write(result["key_pem"]) os.chmod(cf, 0o640) os.chmod(kf, 0o600) conn2 = get_db() @@ -380,7 +403,7 @@ async def create_domain_web( cur = conn.cursor() cur.execute( "INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", - (name, description, 1), + (name, description, get_user_id(conn, user.get("sub"))), ) conn.commit() conn.close() @@ -411,7 +434,7 @@ async def request_cert_web( 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), + (domain_id, cn, sans, "pending", get_user_id(conn, user.get("sub"))), ) conn.commit() conn.close() diff --git a/api/models.py b/api/models.py index a2de66b..5823f7b 100644 --- a/api/models.py +++ b/api/models.py @@ -33,7 +33,7 @@ def init_db(): ); CREATE TABLE IF NOT EXISTS certificates ( id INTEGER PRIMARY KEY AUTOINCREMENT, - domain_id INTEGER REFERENCES users(id), + domain_id INTEGER REFERENCES domains(id), subject TEXT NOT NULL, san TEXT, serial TEXT UNIQUE, diff --git a/api/signing.py b/api/signing.py index 8538869..ee6319e 100644 --- a/api/signing.py +++ b/api/signing.py @@ -1,48 +1,107 @@ -import subprocess -import os +"""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 + 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, BestAvailableEncryption -from cryptography.x509.oid import NameOID, ExtensionOID -from config import * +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption +from cryptography.x509.oid import NameOID + +from config import ( + CA_COUNTRY, + CA_ORG, + INT_CA_PATH, + PKCS11_MODULE, + PKCS11_TOKEN_LABEL, + YK_INT_PIN, +) +from models import get_db logger = logging.getLogger(__name__) -TMP_DIR = tempfile.mkdtemp(prefix="certauth_") -os.makedirs(TMP_DIR, exist_ok=True) +# Scratch dir for pkcs11-tool input/output files (per-process, unlinked after). +PKCS11_TMP = tempfile.mkdtemp(prefix="certauth_") -def get_root_pub_key(): - with open(YK_PUB_ROOT, "rb") as f: - return serialization.load_pem_public_key(f.read()) -def get_int_pub_key(): - with open(YK_PUB_INT, "rb") as f: - return serialization.load_pem_public_key(f.read()) +def _der_read_len(data: bytes, i: int) -> tuple: + """Read a DER length field at offset i -> (length, offset_after_field).""" + b0 = data[i] + if b0 < 0x80: + return b0, i + 1 + n = b0 & 0x7F + return int.from_bytes(data[i + 1:i + 1 + n], "big"), i + 1 + n -def get_root_ca_cert(): - with open(ROOT_CA_PATH, "rb") as f: + +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 sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"): - with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=TMP_DIR) as tbs_file: + +def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label=None): + """Sign DER bytes with the YubiKey SIGN key via pkcs11-tool. + + 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(TMP_DIR, f"sig_{os.path.basename(tbs_path)}") + 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( - [ - "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, - ], + cmd, input=yk_pin.encode(), capture_output=True, text=True, @@ -53,6 +112,8 @@ def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"): 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: @@ -71,25 +132,25 @@ def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"): except OSError: pass + def build_leaf_cert(cn, sans, days=365): - root_cert = get_root_ca_cert() - int_pub = get_int_pub_key() - root_pub = get_root_pub_key() + """Issue a leaf certificate for cn, signed by the Intermediate CA. + + Returns ({serial, cert_pem, key_pem, expires_at}, None) or (None, error). + The private key stays on the server (leaf keys are not hardware-backed); + only the CA signing key requires the YubiKey. + """ + int_cert = get_int_ca_cert() leaf_key = ec.generate_private_key(ec.SECP384R1()) subject = x509.Name([ - x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"), + x509.NameAttribute(NameOID.COUNTRY_NAME, CA_COUNTRY), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, CA_ORG), x509.NameAttribute(NameOID.COMMON_NAME, cn), ]) - issuer = x509.Name([ - x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"), - x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"), - ]) builder = ( x509.CertificateBuilder() .subject_name(subject) - .issuer_name(issuer) + .issuer_name(int_cert.subject) .public_key(leaf_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(datetime.now(timezone.utc)) @@ -124,9 +185,21 @@ def build_leaf_cert(cn, sans, days=365): 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 + + # 1) TBS bytes: sign with a throwaway key of identical parameters — + # the TBS block is independent of the signing key. + dummy_key = ec.generate_private_key(ec.SECP384R1()) + tbs, sig_alg, _ = _split_cert_der(builder.sign(dummy_key, hashes.SHA384())) + + # 2) Real signature from the YubiKey. + yk_sig, err = sign_tbs_with_yk(tbs, YK_INT_PIN) + if err: + return None, err + + # 3) Reassemble and verify before trusting. + cert = x509.load_der_x509_certificate(_der_seq(tbs + sig_alg + yk_sig)) + cert.verify(int_cert.public_key()) + key_pem = leaf_key.private_bytes( encoding=Encoding.PEM, format=serialization.PrivateFormat.PKCS8, @@ -134,14 +207,15 @@ def build_leaf_cert(cn, sans, days=365): ).decode() return ( { - "serial": hex(serial), - "cert_pem": cert_bytes.public_bytes(Encoding.PEM).decode(), + "serial": hex(cert.serial_number), + "cert_pem": cert.public_bytes(Encoding.PEM).decode(), "key_pem": key_pem, - "expires_at": (datetime.now(timezone.utc) + timedelta(days=days)).isoformat(), + "expires_at": cert.not_valid_after_utc.isoformat(), }, None, ) + def revoke_certificate(serial_hex: str, reason: str = "key_compromise"): conn = get_db() conn.execute( diff --git a/api/static/.gitkeep b/api/static/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/api/templates/login.html b/api/templates/login.html index cfa176a..29e8ec1 100644 --- a/api/templates/login.html +++ b/api/templates/login.html @@ -14,6 +14,7 @@
{{ error }}
{% endif %}
+
=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"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..248d242 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +fastapi>=0.110 +uvicorn[standard]>=0.29 +jinja2>=3.1 +python-multipart>=0.0.9 +bcrypt>=4.1 +python-jose[cryptography]>=3.3 +ecdsa>=0.18 +cryptography>=42.0 diff --git a/setup-certauth.sh b/setup-certauth.sh index 166ad49..3008051 100644 --- a/setup-certauth.sh +++ b/setup-certauth.sh @@ -487,1134 +487,38 @@ install_api() { mkdir -p /opt/certauth/{api/templates,api/static,logs} chown -R "$ADMIN_USER:$ADMIN_USER" /opt/certauth - # Write config - cat > /opt/certauth/api/config.py << CONFIGEOF -import os - -YK_ROOT_SERIAL = "$YK1_SERIAL" -YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "$YK1_PIN") -YK_INT_SERIAL = "$YK2_SERIAL" -YK_INT_PIN = os.environ.get("YK_INT_PIN", "$YK2_PIN") -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-$(tr -dc 'A-Za-z0-9' < /dev/urandom | head -c 32)") -ALGORITHM = "HS256" -ACCESS_TOKEN_EXPIRE_MINUTES = 60 -ADMIN_USERNAME = "$ADMIN_USER" -PKCS11_MODULE = "/usr/lib/$(uname -m)-linux-gnu/opensc-pkcs11.so" -YK_PUB_ROOT = "/tmp/yk1-root-pub.pem" -YK_PUB_INT = "/tmp/yk2-int-pub.pem" -CONFIGEOF - - # Embedded API files (heredocs) — always used for reproducibility - info "Deploying API files..." - - cat > /opt/certauth/api/models.py << 'MODELS_EOF' -import sqlite3, datetime, secrets, bcrypt, os -from config import DB_PATH, ADMIN_USERNAME - -def get_db(): - conn = sqlite3.connect(DB_PATH, timeout=30) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA busy_timeout=30000") - conn.execute("PRAGMA foreign_keys=ON") - return conn - -def init_db(): - conn = get_db() - conn.executescript(""" - CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT UNIQUE NOT NULL, - password_hash TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ); - CREATE TABLE IF NOT EXISTS domains ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT UNIQUE NOT NULL, - description TEXT, - status TEXT DEFAULT 'active', - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - created_by INTEGER REFERENCES users(id) - ); - CREATE TABLE IF NOT EXISTS certificates ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - domain_id INTEGER REFERENCES domains(id), - subject TEXT NOT NULL, - san TEXT, - serial TEXT UNIQUE, - status TEXT DEFAULT 'pending', - cert_path TEXT, - issued_at TIMESTAMP, - expires_at TIMESTAMP, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - created_by INTEGER REFERENCES users(id) - ); - CREATE TABLE IF NOT EXISTS api_keys ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - key_hash TEXT UNIQUE NOT NULL, - prefix TEXT NOT NULL, - permissions TEXT DEFAULT 'read', - active BOOLEAN DEFAULT 1, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - created_by INTEGER REFERENCES users(id) - ); - CREATE TABLE IF NOT EXISTS audit_log ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - action TEXT NOT NULL, - details TEXT, - user_id INTEGER REFERENCES users(id), - ip_address TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ); - """) - 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() - cur.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)", - (ADMIN_USERNAME, pw_hash)) - conn.commit() - conn.close() - -def hash_password(password): - h = bcrypt.hashpw(password.encode(), bcrypt.gensalt()) - return h.decode() if isinstance(h, bytes) else h - -def verify_password(password, hash_): - if isinstance(hash_, str): hash_ = hash_.encode() - return bcrypt.checkpw(password.encode(), hash_) -MODELS_EOF - - cat > /opt/certauth/api/auth.py << 'AUTH_EOF' -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)) - to_encode.update({"exp": expire}) - return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) - -def get_current_user(token: str = Depends(oauth2_scheme)): - credentials_exception = HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Could not validate credentials", - headers={"WWW-Authenticate": "Bearer"}, - ) - try: - payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) - username = payload.get("sub") - if username is None: - raise credentials_exception - except JWTError: - raise credentials_exception - return username -AUTH_EOF - - cat > /opt/certauth/api/signing.py << 'SIGNING_EOF' - -import subprocess, datetime, os, hashlib, ipaddress, re -from cryptography import x509 -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.x509.oid import NameOID -from config import * - -TMP_DIR = "/var/lib/certauth/tmp" -os.makedirs(TMP_DIR, exist_ok=True) - -def get_root_pub_key(): - with open(YK_PUB_ROOT, "rb") as f: - return serialization.load_pem_public_key(f.read()) - -def get_int_pub_key(): - with open(YK_PUB_INT, "rb") as f: - return serialization.load_pem_public_key(f.read()) - -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 - -def build_leaf_cert(cn, sans, days=365): - root_cert = get_root_ca_cert() - int_pub = get_int_pub_key() - root_pub = get_root_pub_key() - leaf_key = ec.generate_private_key(ec.SECP384R1()) - subject = x509.Name([ - x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"), - x509.NameAttribute(NameOID.COMMON_NAME, cn), - ]) - issuer = x509.Name([ - x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"), - 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.datetime.now(datetime.timezone.utc)) - .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(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)) - if sans: - san_list = [] - for s in sans.split(","): - s = s.strip() - if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s): - san_list.append(x509.IPAddress(ipaddress.IPv4Address(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() - key_pem = leaf_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.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 -SIGNING_EOF - cat > /opt/certauth/api/main.py << 'MAINPYEOF' -import os, sqlite3, datetime, secrets, hashlib, subprocess, json -from fastapi import FastAPI, Request, Depends, HTTPException, Form -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 models import get_db, init_db, hash_password, verify_password -from auth import create_access_token, get_current_user -from signing import build_leaf_cert -from cryptography.hazmat.primitives import serialization - -app = FastAPI(title="CertAuth Key Vault") -app.mount("/static", StaticFiles(directory="/opt/certauth/api/static"), name="static") - - -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 - -jinja_env = Environment( - loader=FileSystemLoader("/opt/certauth/api/templates"), - autoescape=select_autoescape(["html"]) -) - -@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 - -def render(name, ctx): - return HTMLResponse(jinja_env.get_template(name).render(**ctx)) - -class LoginRequest(BaseModel): - username: str - password: str - -@app.post("/api/token") -async def login(req: LoginRequest): - conn = get_db() - row = conn.execute("SELECT * FROM users WHERE username = ?", (req.username,)).fetchone() - conn.close() - if not row or not verify_password(req.password, row["password_hash"]): - raise HTTPException(401, "Invalid credentials") - token = create_access_token({"sub": req.username}) - return {"access_token": token, "token_type": "bearer"} - -@app.get("/api/me") -async def me(user: str = Depends(get_current_user)): - return {"username": user} - -@app.get("/api/domains") -async def list_domains(user: str = Depends(get_current_user)): - conn = get_db() - rows = conn.execute("SELECT * FROM domains ORDER BY created_at DESC").fetchall() - conn.close() - 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)): - conn = get_db() - cur = conn.cursor() - cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", - (name, description, 1)) - conn.commit() - conn.close() - return {"status": "ok"} - -@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() - 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)): - 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)) - conn.commit() - cid = cur.lastrowid - conn.close() - return {"status": "ok", "id": cid} - -@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() - 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}") - 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) - 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() - 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") - conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() - conn.close() - 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() - with open(CA_CHAIN_PATH) as f: - chain_pem = f.read() - with open(pem_path, "w") as f: - f.write(cert_pem.rstrip() + "\n" + chain_pem) - 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.""" - user = get_user_from_cookie(request) - if not user: raise HTTPException(401, "Login required") - conn = get_db() - row = conn.execute("SELECT * FROM certificates WHERE id = ?", (cert_id,)).fetchone() - conn.close() - 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) - from cryptography.hazmat.primitives.serialization import pkcs12, BestAvailableEncryption - from cryptography import x509 - with open(row["cert_path"], "rb") as f: - leaf = x509.load_pem_x509_certificate(f.read()) - with open(kf, "rb") as f: - key = serialization.load_pem_private_key(f.read(), password=None) - chain_certs = [] - with open(CA_CHAIN_PATH, "rb") as f: - 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-----")) - 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()) - ) - 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(): 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 - -@app.get("/", response_class=HTMLResponse) -async def dashboard(request: Request): - user = get_user_from_cookie(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() - 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"] - 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}) - -@app.get("/login", response_class=HTMLResponse) -async def login_page(request: Request): - return render("login.html", {"request": request, "error": None}) - -@app.post("/login") -async def login_post(username: str = Form(...), password: str = Form(...)): - conn = get_db() - 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"}) - token = create_access_token({"sub": username}) - resp = RedirectResponse("/", status_code=302) - resp.set_cookie("token", token, httponly=True, samesite="lax", 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) - conn = get_db() - 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") - conn.close() - try: - result, err = build_leaf_cert(row["subject"], row["san"], 365) - if err: - return HTMLResponse(f'Issue failed: {err}') - 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) - 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.commit() - conn2.close() - return HTMLResponse(f'Issued! PEM | PFX | Refresh') - except Exception as ex: - return HTMLResponse(f'Issue failed: {str(ex)}') - - -@app.get("/logout") -async def logout(): - resp = RedirectResponse("/login", status_code=302) - 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): - user = get_user_from_cookie(request) - if not user: - return RedirectResponse("/login", status_code=302) - conn = get_db() - cur = conn.cursor() - cur.execute("INSERT INTO domains (name, description, created_by) VALUES (?,?,?)", (name, description, 1)) - conn.commit() - conn.close() - 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): - user = get_user_from_cookie(request) - if not user: - return RedirectResponse("/login", status_code=302) - 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)) - conn.commit() - conn.close() - 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) - 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]}) - -@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) - 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() - 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]}) - -@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) - 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() - conn.close() - 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}) - -@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 -fi - -CHAIN_URL="http://$DETECTED_IP/api/ca-chain" - -echo "Downloading CA chain..." -curl -sL "$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" || "$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 - sudo cp /tmp/ca-chain.crt /etc/pki/ca-trust/source/anchors/certauth.crt - sudo update-ca-trust - 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 - 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" - 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!" -''' - return PlainTextResponse(script, media_type="text/x-shellscript") - -@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 - -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)" -} - -$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 -} - -Remove-Item $ChainPath -Force -ErrorAction SilentlyContinue -Write-Host "Done!" -ForegroundColor Green -''' - return PlainTextResponse(script, media_type="text/plain") - -MAINPYEOF - - cat > /opt/certauth/api/templates/base.html << 'TPL_BASE.HTML_EOF' - - - - - - CertAuth{% block title %}{% endblock %} - - - - - {% if user %} - - {% endif %} -
- {% block content %}{% endblock %} -
- - - -TPL_BASE.HTML_EOF - - cat > /opt/certauth/api/templates/certs.html << 'TPL_CERTS.HTML_EOF' -{% extends "base.html" %} -{% block title %} - Certificates{% endblock %} -{% block content %} -
-

Certificates

-
- -
-

Request New Certificate

- - - - - - -
-
- -
- - - - - - - - - - - - {% for c in certs %} - - - - - - - - {% endfor %} - {% if not certs %} - - {% endif %} - -
SubjectDomainStatusExpiresActions
{{ c.subject }}{{ c.domain_name or '-' }} - - {{ c.status }} - - {{ c.expires_at[:10] if c.expires_at else '-' }} - {% if c.status == 'issued' %} - PEM - PFX - {% elif c.status == 'pending' %} - - - {% endif %} -
No certificates yet
-
-{% endblock %} - -TPL_CERTS.HTML_EOF - - cat > /opt/certauth/api/templates/dashboard.html << 'TPL_DASHBOARD.HTML_EOF' -{% extends "base.html" %} -{% block title %} - Dashboard{% endblock %} -{% block content %} -

Dashboard

- -
-
-
Issued Certificates
-
{{ issued }}
-
-
-
Pending Issue
-
{{ pending }}
-
-
-
Registered Domains
-
{{ domains|length }}
-
-
- -

Recent Certificates

-
- - - - - - - - - - - - {% for c in certs %} - - - - - - - - {% endfor %} - {% if not certs %} - - {% endif %} - -
SubjectDomainStatusExpiresActions
{{ c.subject }}{{ c.domain_name or '-' }} - - {{ c.status }} - - {{ c.expires_at[:10] if c.expires_at else '-' }} - {% if c.status == 'issued' %} - PEM - PFX - {% elif c.status == 'pending' %} - - - {% endif %} -
No certificates yet
-
-{% endblock %} - -TPL_DASHBOARD.HTML_EOF - - cat > /opt/certauth/api/templates/domains.html << 'TPL_DOMAINS.HTML_EOF' -{% extends "base.html" %} -{% block title %} - Domains{% endblock %} -{% block content %} -
-

Domains

-
- -
-

Register New Domain

-
- - - -
-
-
- -
- - - - - - - - - - - {% for d in domains %} - - - - - - - {% endfor %} - {% if not domains %} - - {% endif %} - -
DomainDescriptionStatusCreated
{{ d.name }}{{ d.description or '-' }}{{ d.status }}{{ d.created_at[:10] }}
No domains registered
-
-{% endblock %} - -TPL_DOMAINS.HTML_EOF - - cat > /opt/certauth/api/templates/history.html << 'TPL_HISTORY.HTML_EOF' -{% extends "base.html" %} -{% block title %} - History{% endblock %} -{% block content %} -
-

Certificate History

-
- -
- - - - - - - - - - - - - - - {% for c in certs %} - - - - - - - - - - - {% endfor %} - {% if not certs %} - - {% endif %} - -
SerialSubjectDomainSANsStatusIssuedExpiresActions
{{ c.serial or '-' }}{{ c.subject }}{{ c.domain_name or '-' }}{{ c.san or '-' }} - - {{ c.status }} - - {{ c.issued_at[:10] if c.issued_at else '-' }}{{ c.expires_at[:10] if c.expires_at else '-' }} - {% if c.status == 'issued' %} - PEM - PFX - {% elif c.status == 'pending' %} - - - {% endif %} -
No certificates yet
-
-{% endblock %} - -TPL_HISTORY.HTML_EOF - - cat > /opt/certauth/api/templates/login.html << 'TPL_LOGIN.HTML_EOF' - - - - - - CertAuth - Login - - - -
-

CertAuth

-

Certificate Authority Management

- {% if error %} -
{{ error }}
- {% endif %} -
-
- - -
-
- - -
- -
-
- - - -TPL_LOGIN.HTML_EOF - - cat > /opt/certauth/api/templates/setup.html << 'TPL_SETUP.HTML_EOF' -{% extends "base.html" %} -{% block title %} - Setup{% endblock %} -{% block content %} -

Setup

-

Install the CA chain on client machines to trust certificates from this authority.

- - -
-

Quick Install

-

Run one command on any machine to download and install the CA chain automatically.

- -
-
- Linux / macOS - 
curl -sL http://192.168.8.248/setup.sh | sudo bash
-
-
- Windows (PowerShell) - 
iwr http://192.168.8.248/setup.ps1 -UseBasicParsing | iex
-
-
-
- - -
-

Manual Download

-

Contains the Intermediate + Root CA certificates.

- Download ca-chain.crt -
- - -
-

Manual Installation

- - -
-

Linux (Debian/Ubuntu)

- 
sudo cp ca-chain.crt /usr/local/share/ca-certificates/certauth.crt
-sudo update-ca-certificates
-
- - -
-

macOS

- 
sudo security add-trusted-cert -d -r trustRoot -k /Library/Keychains/System.keychain ca-chain.crt
-
- - -
-

Windows

-

Double-click ca-chain.crt, then:

- 
1. Click "Install Certificate"
-2. Select "Local Machine" → Next
-3. Select "Place all certificates in the following store"
-4. Browse → "Trusted Root Certification Authorities"
-5. OK → Next → Finish
-
- - -
-

Docker

- 
COPY ca-chain.crt /usr/local/share/ca-certificates/certauth.crt
-RUN update-ca-certificates
-
-
-{% endblock %} -TPL_SETUP.HTML_EOF - - - chown -R "$ADMIN_USER:$ADMIN_USER" /opt/certauth - - # Initialize database - cd /opt/certauth/api - sudo -u "$ADMIN_USER" python3 -c "from models import init_db; init_db()" 2>/dev/null || true + # Deploy API from the repository (single source of truth: the api/ + # directory next to this script). Never inline a second copy here. + SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + for f in config.py models.py auth.py signing.py main.py; do + if [ ! -f "$SCRIPT_DIR/api/$f" ]; then + error "api/$f not found next to setup-certauth.sh (run from the repository checkout)" + fi + done + cp "$SCRIPT_DIR/api/config.py" "$SCRIPT_DIR/api/models.py" "$SCRIPT_DIR/api/auth.py" \ + "$SCRIPT_DIR/api/signing.py" "$SCRIPT_DIR/api/main.py" /opt/certauth/api/ + cp -r "$SCRIPT_DIR/api/templates/." /opt/certauth/api/templates/ + mkdir -p "$SCRIPT_DIR/api/static" + cp -r "$SCRIPT_DIR/api/static/." /opt/certauth/api/static/ + + # Runtime configuration. The API fails closed on startup if any variable + # is missing, so this file must be complete and private. + JWT_SECRET_GENERATED="$(python3 -c 'import secrets; print(secrets.token_hex(32))')" + install -d -m 750 -o "$ADMIN_USER" -g "$ADMIN_USER" /etc/certauth + cat > /etc/certauth/certauth.env << ENV_EOF +YK_ROOT_SERIAL=$YK1_SERIAL +YK_INT_SERIAL=$YK2_SERIAL +YK_ROOT_PIN=$YK1_PIN +YK_INT_PIN=$YK2_PIN +JWT_SECRET=$JWT_SECRET_GENERATED +ADMIN_USERNAME=$ADMIN_USER +ADMIN_PASSWORD=$ADMIN_PASS +ENV_EOF + chown "$ADMIN_USER:$ADMIN_USER" /etc/certauth/certauth.env + chmod 600 /etc/certauth/certauth.env + + # Initialize database (loads the private env file as the API user) + sudo -u "$ADMIN_USER" bash -c 'set -a; . /etc/certauth/certauth.env; set +a; cd /opt/certauth/api; python3 -c "from models import init_db; init_db()"' success "API installed" } @@ -1646,6 +550,7 @@ User=$ADMIN_USER Group=$ADMIN_USER 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 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..9f0b044 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,31 @@ +"""Test setup: configure the environment before importing the app. + +The API fails closed on missing configuration (by design), so tests +provide a complete, isolated environment pointing at a temp directory. +""" + +import os +import sys +import tempfile +from pathlib import Path + +API_DIR = Path(__file__).resolve().parent.parent / "api" +sys.path.insert(0, str(API_DIR)) + +_TMP = tempfile.mkdtemp(prefix="certauth-test-") + +os.environ["YK_ROOT_SERIAL"] = "10000001" +os.environ["YK_INT_SERIAL"] = "10000002" +os.environ["YK_ROOT_PIN"] = "123456" +os.environ["YK_INT_PIN"] = "234567" +os.environ["JWT_SECRET"] = "test-secret-0123456789abcdef0123456789abcdef" +os.environ["ADMIN_USERNAME"] = "certauth" +os.environ["ADMIN_PASSWORD"] = "test-admin-pass-123" +os.environ["CERTAUTH_DB_PATH"] = os.path.join(_TMP, "test.db") +os.environ["CERTAUTH_TMP_DIR"] = os.path.join(_TMP, "tmp") +os.environ["CERTAUTH_CA_BASE"] = os.path.join(_TMP, "ca") +os.environ["CERTAUTH_ISSUED_DIR"] = os.path.join(_TMP, "ca", "issued") +os.environ["PKCS11_MODULE"] = "/usr/lib/opensc-pkcs11.so" + +os.makedirs(os.environ["CERTAUTH_ISSUED_DIR"], exist_ok=True) +os.makedirs(os.environ["CERTAUTH_TMP_DIR"], exist_ok=True) diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..e28da1e --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,115 @@ +"""API endpoint tests (no YubiKey required).""" + +import pytest +from fastapi.testclient import TestClient + +from main import app + +ADMIN = {"username": "certauth", "password": "test-admin-pass-123"} + + +@pytest.fixture(scope="module") +def client(): + with TestClient(app) as c: + yield c + + +@pytest.fixture(scope="module") +def auth(client): + r = client.post("/api/token", json=ADMIN) + assert r.status_code == 200 + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +def test_health(client): + r = client.get("/api/health") + assert r.status_code == 200 + assert r.json() == {"status": "ok"} + + +def test_token_wrong_password(client): + r = client.post("/api/token", json={"username": "certauth", "password": "wrong"}) + assert r.status_code == 401 + + +def test_token_unknown_user(client): + r = client.post("/api/token", json={"username": "nobody", "password": "x"}) + assert r.status_code == 401 + + +def test_me(client, auth): + r = client.get("/api/me", headers=auth) + assert r.status_code == 200 + assert r.json() == {"username": "certauth"} + + +def test_me_without_token(client): + assert client.get("/api/me").status_code == 401 + + +def test_login_page_renders_with_csrf(client): + r = client.get("/login") + assert r.status_code == 200 + assert "csrf_token" in r.text + + +def test_dashboard_redirects_when_unauthenticated(client): + r = client.get("/", follow_redirects=False) + assert r.status_code == 302 + assert r.headers["location"] == "/login" + + +def test_login_requires_valid_csrf(client): + r = client.post( + "/login", + data={"username": "certauth", "password": "test-admin-pass-123", "csrf_token": "bogus"}, + follow_redirects=False, + ) + assert r.status_code == 200 + assert "Invalid request" in r.text + + +def test_domain_create_and_list(client, auth): + r = client.post( + "/api/domains", + data={"name": "test.example.ms", "description": "unit test domain"}, + headers=auth, + ) + assert r.status_code == 200 + r = client.get("/api/domains", headers=auth) + assert r.status_code == 200 + names = [d["name"] for d in r.json()] + assert "test.example.ms" in names + + +def test_domain_requires_auth(client): + r = client.post("/api/domains", data={"name": "x.ms"}) + assert r.status_code == 401 + + +def test_cert_request_lifecycle(client, auth): + client.post( + "/api/domains", + data={"name": "certtest.example.ms"}, + headers=auth, + ) + r = client.post( + "/api/certs/request", + data={"cn": "certtest.example.ms", "sans": "alt.example.ms"}, + headers=auth, + ) + assert r.status_code == 200 + cert_id = r.json()["id"] + + r = client.get("/api/certs", headers=auth) + row = next(c for c in r.json() if c["id"] == cert_id) + assert row["status"] == "pending" + assert row["subject"] == "certtest.example.ms" + + # Signing requires a physical YubiKey; must fail cleanly, not 500-crash. + r = client.post(f"/api/certs/{cert_id}/sign", headers=auth) + assert r.status_code == 500 + + +def test_sign_missing_cert(client, auth): + assert client.post("/api/certs/99999/sign", headers=auth).status_code == 400 diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..d683a7e --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,32 @@ +"""JWT auth unit tests.""" + +from datetime import timedelta + +import pytest +from fastapi import HTTPException + +from auth import create_access_token, get_current_user + + +def test_token_roundtrip(): + token = create_access_token({"sub": "certauth"}) + assert get_current_user(token) == "certauth" + + +def test_rejects_garbage_token(): + with pytest.raises(HTTPException) as exc: + get_current_user("not-a-jwt") + assert exc.value.status_code == 401 + + +def test_rejects_expired_token(): + token = create_access_token({"sub": "certauth"}, expires_delta=timedelta(minutes=-5)) + with pytest.raises(HTTPException) as exc: + get_current_user(token) + assert exc.value.status_code == 401 + + +def test_rejects_token_without_subject(): + token = create_access_token({"other": "claim"}) + with pytest.raises(HTTPException): + get_current_user(token) diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..ce90df1 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,22 @@ +"""Password hashing tests.""" + +from models import hash_password, verify_password + + +def test_hash_and_verify(): + h = hash_password("s3cure-Passw0rd!") + assert verify_password("s3cure-Passw0rd!", h) + + +def test_wrong_password_rejected(): + h = hash_password("s3cure-Passw0rd!") + assert not verify_password("wrong-password", h) + + +def test_hashes_are_unique(): + assert hash_password("same-pass") != hash_password("same-pass") + + +def test_accepts_bytes_hash(): + h = hash_password("s3cure-Passw0rd!") + assert verify_password("s3cure-Passw0rd!", h.encode()) diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..89a0ed1 --- /dev/null +++ b/tests/test_signing.py @@ -0,0 +1,55 @@ +"""DER helpers and signing pipeline (no YubiKey required).""" + +from datetime import datetime, timedelta, timezone + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID + +from signing import _der_read_len, _der_seq, _split_cert_der + + +from cryptography.hazmat.primitives.serialization import Encoding + + +def _dummy_cert_der(): + key = ec.generate_private_key(ec.SECP384R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "dummy.test")]) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.now(timezone.utc)) + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1)) + .sign(key, hashes.SHA384()) + ) + return cert.public_bytes(Encoding.DER) + + +def test_split_roundtrip(): + der = _dummy_cert_der() + tbs, alg, sig = _split_cert_der(der) + assert alg[0] == 0x30 + assert sig[0] == 0x03 + # Reassembling the same parts reproduces the original certificate. + reassembled = _der_seq(tbs + alg + sig) + assert reassembled == der + + +def test_der_read_len_short_and_long(): + assert _der_read_len(b"\x05ABC", 0) == (5, 1) + long_len = b"\x81\x10" + assert _der_read_len(long_len + b"A" * 16, 0) == (16, 2) + + +def test_build_leaf_cert_fails_cleanly_without_ca(): + """Without the CA files the pipeline fails cleanly (no YubiKey available).""" + import pytest + + from signing import build_leaf_cert + + with pytest.raises(FileNotFoundError): + build_leaf_cert("nope.test", "", 365)