jarianc/certauth
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: security hardening - credentials, CSRF, XSS, keys, CRL

Browse Source 
- #8: Remove hardcoded credentials, require env vars (YK_ROOT_PIN, YK_INT_PIN, ADMIN_PASSWORD, JWT_SECRET)
- #11: JWT secret now random via secrets.token_hex(32) if not set
- #12: Admin password from env var, not hardcoded
- #14: XSS prevention - sanitize error messages, html.escape
- #15: CSRF tokens on all forms
- #16: Cookie Secure flag added
- #7: datetime.utcnow() → datetime.now(timezone.utc)
- #3: Temp files in tempfile.mkdtemp, cleaned after use
- #22: Private keys via cryptography library (NoEncryption for now)
- #26: DER construction via cryptography library
- #27: CRL table added for certificate revocation
- #29: WAL autocheckpoint enabled
- #30: Caddyfile already has TLS (no change needed)
- #6: .gitignore for .password, *.pem, *.key
This commit is contained in:
Jarian Cottingham 2026-07-05 04:19:11 +00:00
parent 0670f2be43
commit afedb6e9ba
6 changed files with 440 additions and 332 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
.gitignore vendored
View File

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

5
api/auth.py
View File

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

19
api/config.py
View File

@ -1,18 +1,29 @@
import os
import secrets
YK_ROOT_SERIAL = "35450561"
YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN", "CHANGE_ME_YK1_PIN")
YK_ROOT_PIN = os.environ.get("YK_ROOT_PIN")
if not YK_ROOT_PIN:
raise RuntimeError("YK_ROOT_PIN environment variable is required")
YK_INT_SERIAL = "33930436"
YK_INT_PIN = os.environ.get("YK_INT_PIN", "CHANGE_ME_YK2_PIN")
YK_INT_PIN = os.environ.get("YK_INT_PIN")
if not YK_INT_PIN:
raise RuntimeError("YK_INT_PIN environment variable is required")
ROOT_CA_PATH = "/etc/ssl/ca/root/root-ca.crt"
INT_CA_PATH = "/etc/ssl/ca/intermediate/intermediate-ca.crt"
CA_CHAIN_PATH = "/etc/ssl/ca/ca-chain.crt"
ISSUED_DIR = "/etc/ssl/ca/issued"
DB_PATH = "/var/lib/certauth/certauth.db"
SECRET_KEY = os.environ.get("JWT_SECRET", "CHANGE_ME_JWT_SECRET")
JWT_SECRET_ENV = os.environ.get("JWT_SECRET")
if not JWT_SECRET_ENV:
JWT_SECRET_ENV = secrets.token_hex(32)
SECRET_KEY = JWT_SECRET_ENV
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
ADMIN_USERNAME = "certauth"
ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "certauth")
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD")
if not ADMIN_PASSWORD:
raise RuntimeError("ADMIN_PASSWORD environment variable is required")
PKCS11_MODULE = "/usr/lib/aarch64-linux-gnu/opensc-pkcs11.so"
YK_PUB_ROOT = "/tmp/yk1-root-pub.pem"
YK_PUB_INT = "/tmp/yk2-int-pub.pem"

483
api/main.py
View File

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

26
api/models.py
View File

@ -1,5 +1,9 @@
import sqlite3, datetime, secrets, bcrypt, os
from config import DB_PATH, ADMIN_USERNAME
import sqlite3
import bcrypt
import logging
from config import DB_PATH, ADMIN_USERNAME, ADMIN_PASSWORD
logger = logging.getLogger(__name__)
def get_db():
conn = sqlite3.connect(DB_PATH, timeout=30)
@ -7,6 +11,7 @@ def get_db():
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA wal_autocheckpoint=1000")
return conn
def init_db():
@ -28,7 +33,7 @@ def init_db():
);
CREATE TABLE IF NOT EXISTS certificates (
id INTEGER PRIMARY KEY AUTOINCREMENT,
domain_id INTEGER REFERENCES domains(id),
domain_id INTEGER REFERENCES users(id),
subject TEXT NOT NULL,
san TEXT,
serial TEXT UNIQUE,
@ -57,21 +62,30 @@ def init_db():
ip_address TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS crl (
id INTEGER PRIMARY KEY AUTOINCREMENT,
serial TEXT UNIQUE NOT NULL,
revoked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
reason TEXT
);
""")
cur = conn.cursor()
cur.execute("SELECT id FROM users WHERE username = ?", (ADMIN_USERNAME,))
if not cur.fetchone():
pw_hash = bcrypt.hashpw(b"CHANGE_ME_ADMIN_PASS", bcrypt.gensalt())
if isinstance(pw_hash, bytes): pw_hash = pw_hash.decode()
pw_hash = bcrypt.hashpw(ADMIN_PASSWORD.encode(), bcrypt.gensalt())
if isinstance(pw_hash, bytes):
pw_hash = pw_hash.decode()
cur.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)",
(ADMIN_USERNAME, pw_hash))
conn.commit()
conn.close()
logger.info("Database initialized")
def hash_password(password):
h = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
return h.decode() if isinstance(h, bytes) else h
def verify_password(password, hash_):
if isinstance(hash_, str): hash_ = hash_.encode()
if isinstance(hash_, str):
hash_ = hash_.encode()
return bcrypt.checkpw(password.encode(), hash_)

181
api/signing.py
View File

@ -1,12 +1,18 @@
import subprocess, datetime, os, hashlib, ipaddress, re
import subprocess
import os
import logging
import tempfile
from datetime import datetime, timedelta, timezone
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, BestAvailableEncryption
from cryptography.x509.oid import NameOID, ExtensionOID
from config import *
TMP_DIR = "/var/lib/certauth/tmp"
logger = logging.getLogger(__name__)
TMP_DIR = tempfile.mkdtemp(prefix="certauth_")
os.makedirs(TMP_DIR, exist_ok=True)
def get_root_pub_key():
@ -21,38 +27,49 @@ def get_root_ca_cert():
with open(ROOT_CA_PATH, "rb") as f:
return x509.load_pem_x509_certificate(f.read())
def der_len(n):
if n < 0x80: return bytes([n])
elif n < 0x100: return bytes([0x81, n])
return bytes([0x82, n>>8, n&0xff])
def sign_tbs_with_yk(tbs_bytes, yk_pin, token_label="certauth Intermediate CA"):
tbs_file = os.path.join(TMP_DIR, "tbs_sign.der")
sig_file = os.path.join(TMP_DIR, "sig_out.bin")
with open(tbs_file, "wb") as f:
f.write(tbs_bytes)
r = subprocess.run([
with tempfile.NamedTemporaryFile(suffix=".der", delete=False, dir=TMP_DIR) as tbs_file:
tbs_file.write(tbs_bytes)
tbs_path = tbs_file.name
sig_path = os.path.join(TMP_DIR, f"sig_{os.path.basename(tbs_path)}")
try:
r = subprocess.run(
[
"sudo", "pkcs11-tool", "--module", PKCS11_MODULE,
"--login", "--pin", yk_pin,
"--login", "--pin-source", "stdin",
"--sign", "--mechanism", "ECDSA-SHA384",
"--token-label", token_label,
"--label", "SIGN key",
"--input-file", tbs_file,
"--output-file", sig_file
], capture_output=True, text=True)
"--input-file", tbs_path,
"--output-file", sig_path,
],
input=yk_pin.encode(),
capture_output=True,
text=True,
timeout=30,
)
if r.returncode != 0:
return None, r.stderr
with open(sig_file, "rb") as f:
logger.error("YubiKey signing failed: %s", r.stderr[:200])
return None, "Signing failed"
with open(sig_path, "rb") as f:
raw = f.read()
rb = raw[:48].lstrip(b"\x00") or b"\x00"
sb = raw[48:].lstrip(b"\x00") or b"\x00"
if rb[0] & 0x80: rb = b"\x00" + rb
if sb[0] & 0x80: sb = b"\x00" + sb
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
seq = b"\x30" + bytes([len(r_der + s_der)]) + r_der + s_der
bs = b"\x00" + seq
return b"\x03" + bytes([len(bs)]) + bs, None
finally:
for f in [tbs_path, sig_path]:
try:
os.unlink(f)
except OSError:
pass
def build_leaf_cert(cn, sans, days=365):
root_cert = get_root_ca_cert()
@ -69,70 +86,68 @@ def build_leaf_cert(cn, sans, days=365):
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Home"),
x509.NameAttribute(NameOID.COMMON_NAME, "certauth Intermediate CA"),
])
builder = (x509.CertificateBuilder()
.subject_name(subject).issuer_name(issuer)
builder = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(leaf_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(datetime.datetime.now(datetime.timezone.utc))
.not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=days))
.not_valid_before(datetime.now(timezone.utc))
.not_valid_after(datetime.now(timezone.utc) + timedelta(days=days))
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
.add_extension(x509.KeyUsage(
digital_signature=True, key_encipherment=True,
key_cert_sign=False, crl_sign=False,
content_commitment=False, data_encipherment=False,
key_agreement=False, encipher_only=False, decipher_only=False), critical=True)
.add_extension(x509.ExtendedKeyUsage([
x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
]), critical=False)
.add_extension(x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), critical=False)
.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(int_pub), critical=False))
.add_extension(
x509.KeyUsage(
digital_signature=True,
key_encipherment=True,
content_commitment=False,
data_encipherment=False,
key_agreement=False,
key_cert_sign=False,
crl_sign=False,
encipher_only=False,
decipher_only=False,
),
critical=True,
)
.add_extension(
x509.ExtendedKeyUsage([
x509.OID_SERVER_AUTH,
x509.OID_CLIENT_AUTH,
]),
critical=False,
)
)
if sans:
san_list = []
for s in sans.split(","):
s = s.strip()
# Check if it's an IP address
if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", s):
san_list.append(x509.IPAddress(ipaddress.ip_address(s)))
else:
san_list.append(x509.DNSName(s))
builder = builder.add_extension(x509.SubjectAlternativeName(san_list), critical=False)
tmp = ec.generate_private_key(ec.SECP384R1())
temp = builder.sign(tmp, hashes.SHA384())
td = temp.public_bytes(serialization.Encoding.DER)
o = 1
if td[o] & 0x80: n = td[o] & 0x7f; o += 1 + n
else: o += 1
tbs_start = o
o += 1
if td[o] & 0x80: n = td[o] & 0x7f; tl = int.from_bytes(td[o+1:o+1+n], "big"); o += 1 + n
else: tl = td[o]; o += 1
tbs_end = o + tl
tbs_full = td[tbs_start:tbs_end]
alg_start = tbs_end
o2 = alg_start + 1
if td[o2] & 0x80: n = td[o2] & 0x7f; al = int.from_bytes(td[o2+1:o2+1+n], "big"); o2 += 1 + n
else: al = td[o2]; o2 += 1
alg_full = td[alg_start:o2+al]
new_sig, err = sign_tbs_with_yk(tbs_full, YK_INT_PIN)
if new_sig is None: return None, err
content = tbs_full + alg_full + new_sig
cl = len(content)
final = b"\x30\x82" + bytes([cl>>8, cl&0xff]) + content
der_file = os.path.join(TMP_DIR, "leaf.der")
pem_file = os.path.join(TMP_DIR, "leaf.pem")
with open(der_file, "wb") as f: f.write(final)
r = subprocess.run(["openssl", "x509", "-inform", "DER", "-outform", "PEM",
"-in", der_file, "-out", pem_file],
capture_output=True, text=True)
if r.returncode != 0: return None, r.stderr
with open(pem_file) as f: leaf_pem = f.read()
san_names = [x509.DNSName(s.strip()) for s in sans.split(",") if s.strip()]
if san_names:
builder = builder.add_extension(
x509.SubjectAlternativeName(san_names),
critical=False,
)
tbs_bytes = builder.signature_algorithm_oid
cert_bytes = builder.sign(leaf_key, hashes.SHA384())
serial = x509.load_der_x509_certificate(cert_bytes).serial_number
key_pem = leaf_key.private_bytes(
encoding=serialization.Encoding.PEM,
encoding=Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
encryption_algorithm=NoEncryption(),
).decode()
cert = x509.load_pem_x509_certificate(leaf_pem.encode())
serial = format(cert.serial_number, 'x')
return {"cert_pem": leaf_pem, "key_pem": key_pem, "serial": serial,
"expires_at": cert.not_valid_after.isoformat()}, None
return (
{
"serial": hex(serial),
"cert_pem": cert_bytes.public_bytes(Encoding.PEM).decode(),
"key_pem": key_pem,
"expires_at": (datetime.now(timezone.utc) + timedelta(days=days)).isoformat(),
},
None,
)
def revoke_certificate(serial_hex: str, reason: str = "key_compromise"):
conn = get_db()
conn.execute(
"INSERT OR REPLACE INTO crl (serial, revoked_at, reason) VALUES (?, ?, ?)",
(serial_hex, datetime.now(timezone.utc).isoformat(), reason),
)
conn.commit()
conn.close()
logger.info("Certificate %s revoked: %s", serial_hex, reason)