Merge pull request 'fix: YubiKey CA signing, remove inline API copies, env-driven config' (#42) from improve/v1 into master
This commit is contained in:
commit
96f1102fb1
53
.env.example
53
.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 <old> --new-pin <new>)
|
||||
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
|
||||
|
||||
21
LICENSE
Normal file
21
LICENSE
Normal file
@ -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.
|
||||
107
README.md
Normal file
107
README.md
Normal file
@ -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).
|
||||
@ -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")
|
||||
|
||||
73
api/main.py
73
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"<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"])
|
||||
open(kf, "w").write(result["key_pem"])
|
||||
return HTMLResponse("<span class='text-red-400'>Issue failed</span>")
|
||||
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()
|
||||
|
||||
@ -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,
|
||||
|
||||
154
api/signing.py
154
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)}")
|
||||
try:
|
||||
r = subprocess.run(
|
||||
[
|
||||
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",
|
||||
"--token-label", token_label,
|
||||
"--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(
|
||||
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(
|
||||
|
||||
0
api/static/.gitkeep
Normal file
0
api/static/.gitkeep
Normal file
@ -14,6 +14,7 @@
|
||||
<div class="bg-red-900/50 border border-red-700 rounded p-3 mb-4 text-red-300 text-sm">{{ error }}</div>
|
||||
{% endif %}
|
||||
<form method="post" action="/login">
|
||||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
|
||||
<div class="mb-4">
|
||||
<label class="block text-sm text-gray-400 mb-1">Username</label>
|
||||
<input type="text" name="username" required
|
||||
|
||||
@ -8,6 +8,7 @@ User=certauth
|
||||
Group=certauth
|
||||
WorkingDirectory=/opt/certauth/api
|
||||
Environment=PYTHONUNBUFFERED=1
|
||||
EnvironmentFile=/etc/certauth/certauth.env
|
||||
ExecStart=/usr/bin/python3 -m uvicorn main:app --host 127.0.0.1 --port 8000
|
||||
Restart=on-failure
|
||||
RestartSec=5
|
||||
|
||||
42
pyproject.toml
Normal file
42
pyproject.toml
Normal file
@ -0,0 +1,42 @@
|
||||
[project]
|
||||
name = "certauth"
|
||||
version = "1.0.0"
|
||||
description = "YubiKey-backed self-contained Certificate Authority: FastAPI + web UI, PKCS#11 signing, SQLite store"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "Jarian Cottingham", email = "jarianc@proton.me" }]
|
||||
keywords = ["security", "certificate-authority", "yubikey", "pkcs11", "fastapi"]
|
||||
classifiers = [
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Programming Language :: Python :: 3",
|
||||
]
|
||||
dependencies = [
|
||||
"fastapi>=0.110",
|
||||
"uvicorn[standard]>=0.29",
|
||||
"jinja2>=3.1",
|
||||
"python-multipart>=0.0.9",
|
||||
"bcrypt>=4.1",
|
||||
"python-jose[cryptography]>=3.3",
|
||||
"ecdsa>=0.18",
|
||||
"cryptography>=42.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = ["pytest>=7.0", "ruff>=0.1.0"]
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 120
|
||||
target-version = "py39"
|
||||
exclude = [".git", "venv"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "W"]
|
||||
ignore = ["E501", "E741"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
8
requirements.txt
Normal file
8
requirements.txt
Normal file
@ -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
|
||||
1155
setup-certauth.sh
1155
setup-certauth.sh
File diff suppressed because it is too large
Load Diff
31
tests/conftest.py
Normal file
31
tests/conftest.py
Normal file
@ -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)
|
||||
115
tests/test_api.py
Normal file
115
tests/test_api.py
Normal file
@ -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
|
||||
32
tests/test_auth.py
Normal file
32
tests/test_auth.py
Normal file
@ -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)
|
||||
22
tests/test_models.py
Normal file
22
tests/test_models.py
Normal file
@ -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())
|
||||
55
tests/test_signing.py
Normal file
55
tests/test_signing.py
Normal file
@ -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)
|
||||
Loading…
x
Reference in New Issue
Block a user