#!/bin/bash ############################################################################### # CertAuth Setup Script # Provisions a fresh Ubuntu machine as a Certificate Authority # # Prerequisites: # - Fresh Ubuntu 24.04+ Server (aarch64/x86_64) # - Two YubiKey 5 Nano devices plugged in # - Root or sudo access # - Network access to Ubuntu repositories # # Usage: # sudo bash setup-certauth.sh # # Configuration: Edit the variables below or pass as environment variables ############################################################################### set -euo pipefail # ===================== CONFIGURATION ===================== CA_ORG="${CA_ORG:-Home}" CA_COUNTRY="${CA_COUNTRY:-US}" ROOT_CA_CN="${ROOT_CA_CN:-certauth Root CA}" INT_CA_CN="${INT_CA_CN:-certauth Intermediate CA}" ROOT_VALID_DAYS="${ROOT_VALID_DAYS:-9125}" # 25 years INT_VALID_DAYS="${INT_VALID_DAYS:-5475}" # 15 years LEAF_VALID_DAYS="${LEAF_VALID_DAYS:-365}" # 1 year per cert NETWORK_CIDR="${NETWORK_CIDR:-192.168.8.0/24}" ADMIN_USER="${ADMIN_USER:-certauth}" ADMIN_PASS="${ADMIN_PASS:-CHANGE_ME_ADMIN_PASS}" # YubiKey assignment (will be detected automatically if not set) YK1_SERIAL="${YK1_SERIAL:-}" # Root CA YubiKey serial YK2_SERIAL="${YK2_SERIAL:-}" # Intermediate CA YubiKey serial # PINs (will be generated if not set) YK1_PIN="${YK1_PIN:-}" YK1_PUK="${YK1_PUK:-}" YK2_PIN="${YK2_PIN:-}" YK2_PUK="${YK2_PUK:-}" # ===================== COLORS ===================== RED='\033[0;31m' GREEN='\033[0;32m' YELLOW='\033[1;33m' BLUE='\033[0;34m' NC='\033[0m' info() { echo -e "${BLUE}[INFO]${NC} $*"; } success() { echo -e "${GREEN}[OK]${NC} $*"; } warn() { echo -e "${YELLOW}[WARN]${NC} $*"; } error() { echo -e "${RED}[ERROR]${NC} $*"; exit 1; } # ===================== PRE-FLIGHT CHECKS ===================== check_prerequisites() { info "Running pre-flight checks..." # Must be root if [[ $EUID -ne 0 ]]; then error "This script must be run as root (use sudo)" fi # Check Ubuntu version if ! grep -q "Ubuntu" /etc/os-release; then error "This script requires Ubuntu" fi UBUNTU_VERSION=$(grep VERSION_ID /etc/os-release | cut -d'"' -f2) info "Ubuntu version: $UBUNTU_VERSION" # Check architecture ARCH=$(uname -m) info "Architecture: $ARCH" # Check YubiKeys YK_COUNT=$(lsusb 2>/dev/null | grep -c "Yubico" || true) if [[ $YK_COUNT -lt 2 ]]; then error "Need at least 2 YubiKeys connected (found: $YK_COUNT)" fi success "Found $YK_COUNT YubiKeys" # Check disk space AVAIL=$(df -BG / | awk 'NR==2 {print $4}' | tr -d 'G') if [[ $AVAIL -lt 5 ]]; then error "Need at least 5GB free disk space (have: ${AVAIL}GB)" fi success "Disk space: ${AVAIL}GB available" } # ===================== STEP 1: SYSTEM SETUP ===================== setup_system() { info "=== Step 1: System Setup ===" # Update system info "Updating package lists..." apt-get update -qq # Install base packages info "Installing packages..." DEBIAN_FRONTEND=noninteractive apt-get install -y -qq \ ufw fail2ban opensc pcscd yubikey-manager ykcs11 \ python3-pip python3-venv caddy sqlite3 libsqlite3-dev \ haveged apparmor-utils curl wget \ 2>/dev/null || true # Install Python packages info "Installing Python packages..." pip3 install --break-system-packages --ignore-installed \ typing_extensions 2>/dev/null || true pip3 install --break-system-packages \ fastapi "uvicorn[standard]" jinja2 python-multipart \ bcrypt "python-jose[cryptography]" ecdsa cryptography \ 2>/dev/null || true success "Packages installed" } # ===================== STEP 2: CREATE USER ===================== setup_user() { info "=== Step 2: Create $ADMIN_USER user ===" if ! id "$ADMIN_USER" &>/dev/null; then useradd -m -s /bin/bash "$ADMIN_USER" usermod -aG sudo,adm,plugdev,pcscd "$ADMIN_USER" echo "$ADMIN_USER:$ADMIN_PASS" | chpasswd echo "$ADMIN_USER ALL=(ALL) NOPASSWD: ALL" > /etc/sudoers.d/certauth chmod 0440 /etc/sudoers.d/certauth success "User $ADMIN_USER created" else warn "User $ADMIN_USER already exists" fi } # ===================== STEP 3: FIREWALL ===================== setup_firewall() { info "=== Step 3: Configure Firewall ===" ufw --force reset 2>/dev/null || true ufw default deny incoming ufw default deny outgoing ufw allow in on lo ufw allow out on lo ufw allow out 53 ufw allow out 123 ufw allow out 80/tcp ufw allow out 443/tcp ufw allow from "$NETWORK_CIDR" port 22 proto tcp ufw allow from "$NETWORK_CIDR" port 80 proto tcp ufw allow from "$NETWORK_CIDR" port 443 proto tcp echo "y" | ufw enable # Create swap if [[ ! -f /swapfile ]]; then fallocate -l 4G /swapfile chmod 600 /swapfile mkswap /swapfile swapon /swapfile echo "/swapfile none swap sw 0 0" >> /etc/fstab success "Swap created (4GB)" fi success "Firewall configured" } # ===================== STEP 4: DETECT YUBIKEYS ===================== detect_yubikeys() { info "=== Step 4: Detect YubiKeys ===" # Start pcscd systemctl enable pcscd systemctl start pcscd sleep 2 # Fix polkit for pcscd access mkdir -p /etc/polkit-1/rules.d cat > /etc/polkit-1/rules.d/45-access-pcsc.rules << 'POLKIT' polkit.addRule(function(action, subject) { if (action.id == "org.debian.pcsc-lite.access_pcsc") { return polkit.Result.YES; } }); POLKIT systemctl restart polkit systemctl restart pcscd sleep 2 # List YubiKeys info "Connected YubiKeys:" sudo -u "$ADMIN_USER" ykman list 2>/dev/null || ykman list # Auto-detect serials if not set SERIALS=$(ykman list 2>/dev/null | grep -oP 'Serial: \K\d+' || true) if [[ -z "$YK1_SERIAL" ]]; then YK1_SERIAL=$(echo "$SERIALS" | head -1) fi if [[ -z "$YK2_SERIAL" ]]; then YK2_SERIAL=$(echo "$SERIALS" | tail -1) fi info "YK1 (Root CA): $YK1_SERIAL" info "YK2 (Intermediate): $YK2_SERIAL" # Generate PINs if not set if [[ -z "$YK1_PIN" ]]; then YK1_PIN=$(tr -dc 'A-Za-z0-9' < /dev/urandom | head -c 8) YK1_PUK=$(tr -dc 'A-Za-z0-9' < /dev/urandom | head -c 8) fi if [[ -z "$YK2_PIN" ]]; then YK2_PIN=$(tr -dc 'A-Za-z0-9' < /dev/urandom | head -c 8) YK2_PUK=$(tr -dc 'A-Za-z0-9' < /dev/urandom | head -c 8) fi echo "" echo "============================================================" echo " YUBIKEY CREDENTIALS - RECORD THESE SECURELY" echo "============================================================" echo "" echo "YK1 (Root CA) - Serial $YK1_SERIAL:" echo " PIN: $YK1_PIN" echo " PUK: $YK1_PUK" echo "" echo "YK2 (Intermediate) - Serial $YK2_SERIAL:" echo " PIN: $YK2_PIN" echo " PUK: $YK2_PUK" echo "" echo "============================================================" echo "" } # ===================== STEP 5: CONFIGURE YUBIKEYS ===================== configure_yubikeys() { info "=== Step 5: Configure YubiKeys ===" DEFAULT_MGMT="010203040506070801020304050607080102030405060708" # --- YK1: Root CA --- info "Configuring YK1 (Root CA)..." echo "y" | ykman -d "$YK1_SERIAL" piv reset ykman -d "$YK1_SERIAL" piv keys generate --algorithm ECCP384 \ -m "$DEFAULT_MGMT" 9c /tmp/yk1-root-pub.pem ykman -d "$YK1_SERIAL" piv access change-pin -P 123456 --new-pin "$YK1_PIN" ykman -d "$YK1_SERIAL" piv access change-puk -p 12345678 --new-puk "$YK1_PUK" echo "" | ykman -d "$YK1_SERIAL" piv certificates generate \ -P "$YK1_PIN" -m "$DEFAULT_MGMT" \ --subject "CN=$ROOT_CA_CN,O=$CA_ORG,C=$CA_COUNTRY" \ --valid-days "$ROOT_VALID_DAYS" --hash-algorithm SHA384 \ 9c /tmp/yk1-root-pub.pem ykman -d "$YK1_SERIAL" piv certificates export 9c /tmp/root-ca-temp.crt success "YK1 configured" # --- YK2: Intermediate CA --- info "Configuring YK2 (Intermediate CA)..." echo "y" | ykman -d "$YK2_SERIAL" piv reset ykman -d "$YK2_SERIAL" piv keys generate --algorithm ECCP384 \ -m "$DEFAULT_MGMT" 9c /tmp/yk2-int-pub.pem ykman -d "$YK2_SERIAL" piv access change-pin -P 123456 --new-pin "$YK2_PIN" ykman -d "$YK2_SERIAL" piv access change-puk -p 12345678 --new-puk "$YK2_PUK" echo "" | ykman -d "$YK2_SERIAL" piv certificates request \ -P "$YK2_PIN" \ --subject "CN=$INT_CA_CN,O=$CA_ORG,C=$CA_COUNTRY" \ 9c /tmp/yk2-int-pub.pem /tmp/yk2-intermediate.csr success "YK2 configured" } # ===================== STEP 6: GENERATE CA CERTS ===================== generate_ca_certs() { info "=== Step 6: Generate CA Certificates ===" # Create directories mkdir -p /etc/ssl/ca/{root,intermediate,issued,crl} mkdir -p /var/lib/certauth/tmp mkdir -p /var/log/certauth chown -R "$ADMIN_USER:$ADMIN_USER" /etc/ssl/ca /var/lib/certauth /var/log/certauth chmod -R 700 /etc/ssl/ca /var/lib/certauth /var/log/certauth # Generate Root CA cert with proper extensions info "Generating Root CA certificate..." python3 << ROOTPY import datetime, subprocess, base64 from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import NameOID with open("/tmp/yk1-root-pub.pem", "rb") as f: root_pub = serialization.load_pem_public_key(f.read()) subj = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "$CA_COUNTRY"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "$CA_ORG"), x509.NameAttribute(NameOID.COMMON_NAME, "$ROOT_CA_CN"), ]) builder = (x509.CertificateBuilder() .subject_name(subj).issuer_name(subj) .public_key(root_pub).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=$ROOT_VALID_DAYS)) .add_extension(x509.BasicConstraints(ca=True, path_length=1), critical=True) .add_extension(x509.KeyUsage(digital_signature=True, key_cert_sign=True, crl_sign=True, key_encipherment=False, content_commitment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False), critical=True) .add_extension(x509.SubjectKeyIdentifier.from_public_key(root_pub), critical=False) .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(root_pub), critical=False)) tmp = ec.generate_private_key(ec.SECP384R1()) temp = builder.sign(tmp, hashes.SHA384()) td = temp.public_bytes(serialization.Encoding.DER) # Parse TBS 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] # Sign with YubiKey with open("/tmp/tbs.der", "wb") as f: f.write(tbs_full) r = subprocess.run(["pkcs11-tool", "--login", "--pin", "$YK1_PIN", "--sign", "--mechanism", "ECDSA-SHA384", "--label", "SIGN key", "--input-file", "/tmp/tbs.der", "--output-file", "/tmp/sig.bin"], capture_output=True, text=True) if r.returncode != 0: print(f"Sign failed: {r.stderr}") exit(1) with open("/tmp/sig.bin", "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 new_sig = b"\x03" + bytes([len(bs)]) + bs content = tbs_full + alg_full + new_sig cl = len(content) final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content with open("/tmp/root.der", "wb") as f: f.write(final) r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM", "-in", "/tmp/root.der", "-out", "/tmp/root.pem"], capture_output=True, text=True) if r.returncode != 0: print(f"DER->PEM failed: {r.stderr}") exit(1) with open("/tmp/root.pem") as f: pem = f.read() with open("/etc/ssl/ca/root/root-ca.crt", "w") as f: f.write(pem) # Verify r = subprocess.run(["openssl", "verify", "-CAfile", "/etc/ssl/ca/root/root-ca.crt", "/etc/ssl/ca/root/root-ca.crt"], capture_output=True, text=True) print(f"Root CA self-verify: {r.stdout.strip() or r.stderr.strip()}") # Import to YK1 subprocess.run(["echo", "", "|", "ykman", "-d", "$YK1_SERIAL", "piv", "certificates", "import", "9c", "/etc/ssl/ca/root/root-ca.crt"], shell=True) print("Root CA imported to YK1") ROOTPY # Generate Intermediate CA cert signed by Root info "Generating Intermediate CA certificate..." python3 << INTPY import datetime, subprocess, base64 from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import NameOID with open("/tmp/yk1-root-pub.pem", "rb") as f: root_pub = serialization.load_pem_public_key(f.read()) with open("/tmp/yk2-int-pub.pem", "rb") as f: int_pub = serialization.load_pem_public_key(f.read()) root_subj = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "$CA_COUNTRY"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "$CA_ORG"), x509.NameAttribute(NameOID.COMMON_NAME, "$ROOT_CA_CN"), ]) int_subj = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "$CA_COUNTRY"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "$CA_ORG"), x509.NameAttribute(NameOID.COMMON_NAME, "$INT_CA_CN"), ]) builder = (x509.CertificateBuilder() .subject_name(int_subj).issuer_name(root_subj) .public_key(int_pub).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=$INT_VALID_DAYS)) .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) .add_extension(x509.KeyUsage(digital_signature=True, key_cert_sign=True, crl_sign=True, key_encipherment=False, content_commitment=False, data_encipherment=False, key_agreement=False, encipher_only=False, decipher_only=False), critical=True) .add_extension(x509.SubjectKeyIdentifier.from_public_key(int_pub), critical=False) .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(root_pub), 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] with open("/tmp/tbs.der", "wb") as f: f.write(tbs_full) r = subprocess.run(["pkcs11-tool", "--login", "--pin", "$YK1_PIN", "--sign", "--mechanism", "ECDSA-SHA384", "--label", "SIGN key", "--input-file", "/tmp/tbs.der", "--output-file", "/tmp/sig.bin"], capture_output=True, text=True) if r.returncode != 0: print(f"Sign failed: {r.stderr}") exit(1) with open("/tmp/sig.bin", "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 new_sig = b"\x03" + bytes([len(bs)]) + bs content = tbs_full + alg_full + new_sig cl = len(content) final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content with open("/tmp/int.der", "wb") as f: f.write(final) r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM", "-in", "/tmp/int.der", "-out", "/tmp/int.pem"], capture_output=True, text=True) if r.returncode != 0: print(f"DER->PEM failed: {r.stderr}") exit(1) with open("/tmp/int.pem") as f: pem = f.read() with open("/etc/ssl/ca/intermediate/intermediate-ca.crt", "w") as f: f.write(pem) # Verify chain r = subprocess.run(["openssl", "verify", "-CAfile", "/etc/ssl/ca/root/root-ca.crt", "/etc/ssl/ca/intermediate/intermediate-ca.crt"], capture_output=True, text=True) print(f"Chain verify: {r.stdout.strip() or r.stderr.strip()}") # Import to YK2 subprocess.run(["echo", "", "|", "ykman", "-d", "$YK2_SERIAL", "piv", "certificates", "import", "9c", "/etc/ssl/ca/intermediate/intermediate-ca.crt"], shell=True) print("Intermediate CA imported to YK2") INTPY # Create CA chain file cat /etc/ssl/ca/intermediate/intermediate-ca.crt /etc/ssl/ca/root/root-ca.crt \ > /etc/ssl/ca/ca-chain.crt success "CA certificates generated and verified" } # ===================== STEP 7: INSTALL API ===================== install_api() { info "=== Step 7: Install Certificate API ===" # Create API directory 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 %}
Subject Domain Status Expires Actions
{{ 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 %}
Subject Domain Status Expires Actions
{{ 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 %}
Domain Description Status Created
{{ 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 %}
Serial Subject Domain SANs Status Issued Expires Actions
{{ 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 success "API installed" } # ===================== STEP 8: SERVICES ===================== setup_services() { info "=== Step 8: Configure Services ===" # Caddy cat > /etc/caddy/Caddyfile << 'CADDYEOF' :80 { encode gzip reverse_proxy 127.0.0.1:8000 { header_up Host {host} header_up X-Real-IP {remote} } } CADDYEOF # Systemd service for API cat > /etc/systemd/system/certauth-api.service << SVCCEOF [Unit] Description=CertAuth Certificate Management API After=network.target pcscd.service [Service] Type=simple User=$ADMIN_USER Group=$ADMIN_USER WorkingDirectory=/opt/certauth/api Environment=PYTHONUNBUFFERED=1 ExecStart=/usr/bin/python3 -m uvicorn main:app --host 127.0.0.1 --port 8000 Restart=on-failure RestartSec=5 [Install] WantedBy=multi-user.target SVCCEOF systemctl daemon-reload systemctl enable caddy certauth-api pcscd systemctl start caddy certauth-api success "Services configured and started" } # ===================== STEP 9: VERIFY ===================== verify_setup() { info "=== Step 9: Verification ===" sleep 3 # Check services for svc in certauth-api caddy pcscd; do STATUS=$(systemctl is-active "$svc") if [[ "$STATUS" == "active" ]]; then success "$svc: active" else warn "$svc: $STATUS" fi done # Test API HEALTH=$(curl -s http://localhost:8000/api/health 2>/dev/null || echo "FAILED") if echo "$HEALTH" | grep -q "ok"; then success "API health check: OK" else warn "API health check: $HEALTH" fi # Verify CA chain openssl verify -CAfile /etc/ssl/ca/root/root-ca.crt \ /etc/ssl/ca/intermediate/intermediate-ca.crt 2>&1 | while read line; do if echo "$line" | grep -q "OK"; then success "CA chain verification: OK" else warn "CA chain: $line" fi done echo "" echo "============================================================" echo " SETUP COMPLETE" echo "============================================================" echo "" echo " Access the Key Vault UI at:" echo " http://$(hostname -I | awk '{print $1}')/" echo "" echo " Admin login:" echo " Username: $ADMIN_USER" echo " Password: $ADMIN_PASS" echo "" echo " API endpoints:" echo " POST /api/token - Login (returns JWT)" echo " GET /api/domains - List domains" echo " POST /api/domains - Register domain" echo " GET /api/certs - List certificates" echo " POST /api/certs/request - Request certificate" echo " POST /api/certs/{id}/sign - Sign certificate (requires YK2)" echo " GET /api/certs/{id}/download - Download cert" echo " GET /api/certs/{id}/key - Download private key" echo " GET /api/ca-chain - Download CA chain" echo "" echo " IMPORTANT: Change the admin password after first login!" echo "============================================================" } # ===================== MAIN ===================== SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" echo "" echo "============================================================" echo " CertAuth Setup" echo "============================================================" echo "" check_prerequisites setup_system setup_user setup_firewall detect_yubikeys configure_yubikeys generate_ca_certs install_api setup_services verify_setup echo "" success "All done!"