add TLS cert bootstrap script with 22 tests
dev/scripts/bootstrap_certs.py generates CA, wildcard cert, distributes to Docker certs.d, and regenerates CoreDNS hosts. Fully tested with tmpdir isolation — CA generation, signing, SAN verification, chain validation, file permissions, skip/force semantics, and distribution.
This commit is contained in:
kert
278
dev/scripts/bootstrap_certs.py
Normal file
278
dev/scripts/bootstrap_certs.py
Normal file
@@ -0,0 +1,278 @@
|
||||
"""Bootstrap TLS certificates for the homelab stack.
|
||||
|
||||
Generates a self-signed CA and wildcard certificate for *.homelab.fhirworx.io,
|
||||
then distributes the CA cert to Docker's certs.d for registry trust and
|
||||
regenerates the CoreDNS hosts file from HOST_IP.
|
||||
|
||||
Usage:
|
||||
uv run python dev/scripts/bootstrap_certs.py # defaults
|
||||
uv run python dev/scripts/bootstrap_certs.py --force # overwrite existing
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
CERT_DIR = REPO_ROOT / "traefik" / "certs"
|
||||
COREDNS_DIR = REPO_ROOT / "coredns"
|
||||
DOCKER_CERTS_DIR = CERT_DIR / "docker-certs.d"
|
||||
|
||||
CA_KEY = CERT_DIR / "ca.key"
|
||||
CA_CRT = CERT_DIR / "ca.crt"
|
||||
TLS_KEY = CERT_DIR / "homelab.key"
|
||||
TLS_CRT = CERT_DIR / "homelab.crt"
|
||||
|
||||
CA_SUBJECT = "/CN=Homelab CA/O=fhirworx"
|
||||
CA_DAYS = 3650
|
||||
TLS_DAYS = 3650
|
||||
|
||||
DOMAIN = os.environ.get("DOMAIN", "homelab.fhirworx.io")
|
||||
HOST_IP = os.environ.get("HOST_IP", "192.168.1.192")
|
||||
|
||||
HOSTS_SUBDOMAINS = [
|
||||
"",
|
||||
"gitea",
|
||||
"ci",
|
||||
"docs",
|
||||
"notebooks",
|
||||
"api",
|
||||
"s3",
|
||||
"s3console",
|
||||
"grafana",
|
||||
"traefik",
|
||||
"zotero",
|
||||
"webdav",
|
||||
"prometheus",
|
||||
"jaeger",
|
||||
"loki",
|
||||
"nessie",
|
||||
"trino",
|
||||
"polaris",
|
||||
]
|
||||
|
||||
|
||||
def _run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(cmd, check=True, capture_output=True, text=True, **kwargs)
|
||||
|
||||
|
||||
def generate_ca(*, force: bool = False) -> tuple[Path, Path]:
|
||||
"""Generate a self-signed CA key and certificate."""
|
||||
if CA_KEY.exists() and CA_CRT.exists() and not force:
|
||||
return CA_KEY, CA_CRT
|
||||
|
||||
CERT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
_run(
|
||||
[
|
||||
"openssl",
|
||||
"req",
|
||||
"-x509",
|
||||
"-new",
|
||||
"-nodes",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-keyout",
|
||||
str(CA_KEY),
|
||||
"-out",
|
||||
str(CA_CRT),
|
||||
"-days",
|
||||
str(CA_DAYS),
|
||||
"-subj",
|
||||
CA_SUBJECT,
|
||||
]
|
||||
)
|
||||
CA_KEY.chmod(0o600)
|
||||
return CA_KEY, CA_CRT
|
||||
|
||||
|
||||
def generate_wildcard(
|
||||
domain: str = DOMAIN, *, force: bool = False
|
||||
) -> tuple[Path, Path]:
|
||||
"""Generate a wildcard TLS cert signed by the CA."""
|
||||
if TLS_KEY.exists() and TLS_CRT.exists() and not force:
|
||||
return TLS_KEY, TLS_CRT
|
||||
|
||||
if not CA_KEY.exists() or not CA_CRT.exists():
|
||||
raise FileNotFoundError("CA cert/key not found — run generate_ca first")
|
||||
|
||||
san = f"DNS:*.{domain},DNS:{domain}"
|
||||
|
||||
_run(
|
||||
[
|
||||
"openssl",
|
||||
"req",
|
||||
"-new",
|
||||
"-nodes",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-keyout",
|
||||
str(TLS_KEY),
|
||||
"-out",
|
||||
str(TLS_KEY.with_suffix(".csr")),
|
||||
"-subj",
|
||||
f"/CN=*.{domain}",
|
||||
]
|
||||
)
|
||||
|
||||
_run(
|
||||
[
|
||||
"openssl",
|
||||
"x509",
|
||||
"-req",
|
||||
"-in",
|
||||
str(TLS_KEY.with_suffix(".csr")),
|
||||
"-CA",
|
||||
str(CA_CRT),
|
||||
"-CAkey",
|
||||
str(CA_KEY),
|
||||
"-CAcreateserial",
|
||||
"-out",
|
||||
str(TLS_CRT),
|
||||
"-days",
|
||||
str(TLS_DAYS),
|
||||
"-extfile",
|
||||
"/dev/stdin",
|
||||
],
|
||||
input=f"subjectAltName={san}\n",
|
||||
)
|
||||
|
||||
TLS_KEY.chmod(0o600)
|
||||
TLS_KEY.with_suffix(".csr").unlink(missing_ok=True)
|
||||
CA_CRT.with_suffix(".srl").unlink(missing_ok=True)
|
||||
return TLS_KEY, TLS_CRT
|
||||
|
||||
|
||||
def distribute_ca(
|
||||
registry: str = f"gitea.{DOMAIN}",
|
||||
) -> Path:
|
||||
"""Copy CA cert to Docker certs.d for registry trust."""
|
||||
dest_dir = DOCKER_CERTS_DIR / registry
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
dest = dest_dir / "ca.crt"
|
||||
shutil.copy2(CA_CRT, dest)
|
||||
|
||||
# Also install for rootless Docker daemon
|
||||
user_certs = Path.home() / ".config" / "docker" / "certs.d" / registry
|
||||
user_certs.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(CA_CRT, user_certs / "ca.crt")
|
||||
|
||||
return dest
|
||||
|
||||
|
||||
def generate_coredns_hosts(
|
||||
host_ip: str = HOST_IP,
|
||||
domain: str = DOMAIN,
|
||||
) -> Path:
|
||||
"""Generate the CoreDNS hosts file from HOST_IP and DOMAIN."""
|
||||
COREDNS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
hosts_path = COREDNS_DIR / "hosts"
|
||||
lines = []
|
||||
for sub in HOSTS_SUBDOMAINS:
|
||||
fqdn = f"{sub}.{domain}" if sub else domain
|
||||
lines.append(f"{host_ip} {fqdn}")
|
||||
hosts_path.write_text("\n".join(lines) + "\n")
|
||||
return hosts_path
|
||||
|
||||
|
||||
def verify_ca() -> dict[str, str]:
|
||||
"""Read CA cert metadata. Returns dict with subject, issuer, dates."""
|
||||
result = _run(
|
||||
[
|
||||
"openssl",
|
||||
"x509",
|
||||
"-in",
|
||||
str(CA_CRT),
|
||||
"-noout",
|
||||
"-subject",
|
||||
"-issuer",
|
||||
"-startdate",
|
||||
"-enddate",
|
||||
]
|
||||
)
|
||||
info = {}
|
||||
for line in result.stdout.strip().splitlines():
|
||||
key, _, val = line.partition("=")
|
||||
info[key.strip()] = val.strip()
|
||||
return info
|
||||
|
||||
|
||||
def verify_wildcard(domain: str = DOMAIN) -> dict[str, str]:
|
||||
"""Read wildcard cert metadata and verify CA signature."""
|
||||
result = _run(
|
||||
[
|
||||
"openssl",
|
||||
"x509",
|
||||
"-in",
|
||||
str(TLS_CRT),
|
||||
"-noout",
|
||||
"-subject",
|
||||
"-issuer",
|
||||
"-startdate",
|
||||
"-enddate",
|
||||
]
|
||||
)
|
||||
info = {}
|
||||
for line in result.stdout.strip().splitlines():
|
||||
key, _, val = line.partition("=")
|
||||
info[key.strip()] = val.strip()
|
||||
|
||||
# Verify signature chain
|
||||
_run(["openssl", "verify", "-CAfile", str(CA_CRT), str(TLS_CRT)])
|
||||
info["verified"] = "true"
|
||||
|
||||
# Check SAN
|
||||
san_result = _run(
|
||||
[
|
||||
"openssl",
|
||||
"x509",
|
||||
"-in",
|
||||
str(TLS_CRT),
|
||||
"-noout",
|
||||
"-ext",
|
||||
"subjectAltName",
|
||||
]
|
||||
)
|
||||
info["san"] = san_result.stdout.strip()
|
||||
return info
|
||||
|
||||
|
||||
def bootstrap(*, force: bool = False) -> None:
|
||||
"""Run the full bootstrap: CA → wildcard → distribute → hosts."""
|
||||
print(f"Domain: {DOMAIN}")
|
||||
print(f"Host IP: {HOST_IP}")
|
||||
print()
|
||||
|
||||
print("1. Generating CA...")
|
||||
ca_key, ca_crt = generate_ca(force=force)
|
||||
print(f" {ca_key}")
|
||||
print(f" {ca_crt}")
|
||||
|
||||
print("2. Generating wildcard cert...")
|
||||
tls_key, tls_crt = generate_wildcard(force=force)
|
||||
print(f" {tls_key}")
|
||||
print(f" {tls_crt}")
|
||||
|
||||
print("3. Distributing CA to Docker certs.d...")
|
||||
dest = distribute_ca()
|
||||
print(f" {dest}")
|
||||
|
||||
print("4. Generating CoreDNS hosts...")
|
||||
hosts = generate_coredns_hosts()
|
||||
print(f" {hosts}")
|
||||
|
||||
print()
|
||||
print("Done. Run dev/scripts/install_certs.sh to install in browser trust stores.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--force", action="store_true", help="overwrite existing certs")
|
||||
args = parser.parse_args()
|
||||
bootstrap(force=args.force)
|
||||
226
tests/test_bootstrap_certs.py
Normal file
226
tests/test_bootstrap_certs.py
Normal file
@@ -0,0 +1,226 @@
|
||||
"""Tests for dev/scripts/bootstrap_certs.py — TLS cert bootstrap."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from dev.scripts.bootstrap_certs import (
|
||||
HOSTS_SUBDOMAINS,
|
||||
distribute_ca,
|
||||
generate_ca,
|
||||
generate_coredns_hosts,
|
||||
generate_wildcard,
|
||||
verify_ca,
|
||||
verify_wildcard,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def cert_dir(tmp_path: Path) -> Path:
|
||||
"""Provide a temporary cert directory."""
|
||||
d = tmp_path / "certs"
|
||||
d.mkdir()
|
||||
return d
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def _patch_dirs(tmp_path: Path, cert_dir: Path):
|
||||
"""Patch all module-level paths to use tmp_path."""
|
||||
coredns = tmp_path / "coredns"
|
||||
coredns.mkdir()
|
||||
docker_certs = cert_dir / "docker-certs.d"
|
||||
docker_certs.mkdir()
|
||||
user_certs = tmp_path / "user-docker"
|
||||
with (
|
||||
patch("dev.scripts.bootstrap_certs.CERT_DIR", cert_dir),
|
||||
patch("dev.scripts.bootstrap_certs.CA_KEY", cert_dir / "ca.key"),
|
||||
patch("dev.scripts.bootstrap_certs.CA_CRT", cert_dir / "ca.crt"),
|
||||
patch("dev.scripts.bootstrap_certs.TLS_KEY", cert_dir / "homelab.key"),
|
||||
patch("dev.scripts.bootstrap_certs.TLS_CRT", cert_dir / "homelab.crt"),
|
||||
patch("dev.scripts.bootstrap_certs.COREDNS_DIR", coredns),
|
||||
patch("dev.scripts.bootstrap_certs.DOCKER_CERTS_DIR", docker_certs),
|
||||
patch("dev.scripts.bootstrap_certs.Path.home", return_value=user_certs),
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
class TestGenerateCA:
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_creates_key_and_cert(self, cert_dir: Path) -> None:
|
||||
key, crt = generate_ca(force=True)
|
||||
assert key.exists()
|
||||
assert crt.exists()
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_key_is_private(self, cert_dir: Path) -> None:
|
||||
key, _ = generate_ca(force=True)
|
||||
assert oct(key.stat().st_mode)[-3:] == "600"
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_cert_has_correct_subject(self, cert_dir: Path) -> None:
|
||||
_, crt = generate_ca(force=True)
|
||||
result = subprocess.run(
|
||||
["openssl", "x509", "-in", str(crt), "-noout", "-subject"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
assert "Homelab CA" in result.stdout
|
||||
assert "fhirworx" in result.stdout
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_skips_if_exists(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
mtime_key = (cert_dir / "ca.key").stat().st_mtime
|
||||
generate_ca(force=False)
|
||||
assert (cert_dir / "ca.key").stat().st_mtime == mtime_key
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_force_overwrites(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
content_before = (cert_dir / "ca.crt").read_bytes()
|
||||
generate_ca(force=True)
|
||||
content_after = (cert_dir / "ca.crt").read_bytes()
|
||||
assert content_before != content_after
|
||||
|
||||
|
||||
class TestGenerateWildcard:
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_creates_key_and_cert(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
key, crt = generate_wildcard(force=True)
|
||||
assert key.exists()
|
||||
assert crt.exists()
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_signed_by_ca(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
_, crt = generate_wildcard(force=True)
|
||||
ca_crt = cert_dir / "ca.crt"
|
||||
result = subprocess.run(
|
||||
["openssl", "verify", "-CAfile", str(ca_crt), str(crt)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
assert "OK" in result.stdout
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_has_wildcard_san(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
_, crt = generate_wildcard(domain="test.example.com", force=True)
|
||||
result = subprocess.run(
|
||||
["openssl", "x509", "-in", str(crt), "-noout", "-ext", "subjectAltName"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
assert "*.test.example.com" in result.stdout
|
||||
assert "test.example.com" in result.stdout
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_raises_without_ca(self, cert_dir: Path) -> None:
|
||||
with pytest.raises(FileNotFoundError, match="CA cert/key not found"):
|
||||
generate_wildcard(force=True)
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_no_csr_leftover(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
generate_wildcard(force=True)
|
||||
csrs = list(cert_dir.glob("*.csr"))
|
||||
srls = list(cert_dir.glob("*.srl"))
|
||||
assert csrs == []
|
||||
assert srls == []
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_key_is_private(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
key, _ = generate_wildcard(force=True)
|
||||
assert oct(key.stat().st_mode)[-3:] == "600"
|
||||
|
||||
|
||||
class TestDistributeCA:
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_copies_to_docker_certs_d(self, cert_dir: Path) -> None:
|
||||
generate_ca(force=True)
|
||||
dest = distribute_ca(registry="reg.example.com")
|
||||
assert dest.exists()
|
||||
assert dest.parent.name == "reg.example.com"
|
||||
assert dest.name == "ca.crt"
|
||||
assert dest.read_bytes() == (cert_dir / "ca.crt").read_bytes()
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_copies_to_user_docker_config(
|
||||
self,
|
||||
cert_dir: Path,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
generate_ca(force=True)
|
||||
distribute_ca(registry="reg.example.com")
|
||||
user_cert = (
|
||||
tmp_path
|
||||
/ "user-docker"
|
||||
/ ".config"
|
||||
/ "docker"
|
||||
/ "certs.d"
|
||||
/ "reg.example.com"
|
||||
/ "ca.crt"
|
||||
)
|
||||
assert user_cert.exists()
|
||||
|
||||
|
||||
class TestGenerateCoreDNSHosts:
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_creates_hosts_file(self, tmp_path: Path) -> None:
|
||||
hosts = generate_coredns_hosts(host_ip="10.0.0.1", domain="x.io")
|
||||
assert hosts.exists()
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_contains_all_subdomains(self, tmp_path: Path) -> None:
|
||||
hosts = generate_coredns_hosts(host_ip="10.0.0.1", domain="x.io")
|
||||
content = hosts.read_text()
|
||||
for sub in HOSTS_SUBDOMAINS:
|
||||
fqdn = f"{sub}.x.io" if sub else "x.io"
|
||||
assert f"10.0.0.1 {fqdn}" in content
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_uses_correct_ip(self, tmp_path: Path) -> None:
|
||||
hosts = generate_coredns_hosts(host_ip="172.16.5.99", domain="d.io")
|
||||
content = hosts.read_text()
|
||||
assert all(
|
||||
line.startswith("172.16.5.99 ") for line in content.strip().splitlines()
|
||||
)
|
||||
|
||||
@pytest.mark.usefixtures("_patch_dirs")
|
||||
def test_line_count_matches_subdomains(self, tmp_path: Path) -> None:
|
||||
hosts = generate_coredns_hosts(host_ip="10.0.0.1", domain="x.io")
|
||||
lines = [l for l in hosts.read_text().strip().splitlines() if l.strip()]
|
||||
assert len(lines) == len(HOSTS_SUBDOMAINS)
|
||||
|
||||
|
||||
class TestVerify:
|
||||
"""Test verify functions against the actual repo certs."""
|
||||
|
||||
def test_verify_ca_returns_subject(self) -> None:
|
||||
info = verify_ca()
|
||||
assert "Homelab CA" in info.get("subject", "")
|
||||
|
||||
def test_verify_ca_returns_issuer(self) -> None:
|
||||
info = verify_ca()
|
||||
assert "fhirworx" in info.get("issuer", "")
|
||||
|
||||
def test_verify_wildcard_chain(self) -> None:
|
||||
info = verify_wildcard()
|
||||
assert info["verified"] == "true"
|
||||
|
||||
def test_verify_wildcard_san(self) -> None:
|
||||
info = verify_wildcard()
|
||||
assert "*.homelab.fhirworx.io" in info.get("san", "")
|
||||
|
||||
def test_verify_wildcard_issuer_is_ca(self) -> None:
|
||||
info = verify_wildcard()
|
||||
assert "Homelab CA" in info.get("issuer", "")
|
||||
Reference in New Issue
Block a user