#!/usr/bin/env python3 """ Usage: python create-cert.py mydomain.internal python create-cert.py app.internal --san api.internal --san admin.internal The CA material is expected under secrets/ca/ relative to the repo root: secrets/ca/ rootCA.key # encrypted with sops rootCA.crt # plaintext (public) certs/ # output directory, created if missing """ from __future__ import annotations import argparse import shutil import subprocess import sys import tempfile from pathlib import Path from typing import List def run(cmd: List[str], **kwargs) -> subprocess.CompletedProcess: """Execute a command, check for errors.""" try: return subprocess.run(cmd, check=True, text=True, capture_output=True, **kwargs) except subprocess.CalledProcessError as e: print(f"Command failed: {' '.join(cmd)}", file=sys.stderr) print(e.stderr, file=sys.stderr) raise def check_dependencies() -> None: """Verify that git, openssl, and sops are available on $PATH.""" required = ["git", "openssl", "sops"] missing = [tool for tool in required if shutil.which(tool) is None] if missing: sys.exit(f"Missing required tools: {', '.join(missing)}") def get_repo_root() -> Path: """Return the root directory of the git repository.""" result = run(["git", "rev-parse", "--show-toplevel"]) return Path(result.stdout.strip()) def build_csr_config(domain: str, extra_sans: List[str]) -> str: """Return an OpenSSL config for a CSR, including SANs.""" sans = [domain, f"*.{domain}"] + extra_sans san_entries = "\n".join(f"DNS.{i+1} = {s}" for i, s in enumerate(sans)) return f"""\ [ req ] default_bits = 2048 distinguished_name = req_distinguished_name req_extensions = req_ext string_mask = utf8only prompt = no [ req_distinguished_name ] organizationName = Internal commonName = {domain} [ req_ext ] subjectAltName = @alt_names [ alt_names ] {san_entries} """ def build_ext_config(domain: str, extra_sans: List[str]) -> str: """Return an OpenSSL extension config for signing a server certificate.""" sans = [domain, f"*.{domain}"] + extra_sans san_entries = "\n".join(f"DNS.{i+1} = {s}" for i, s in enumerate(sans)) return f"""\ [ v3_server ] subjectKeyIdentifier = hash authorityKeyIdentifier = keyid,issuer basicConstraints = critical, CA:FALSE keyUsage = critical, digitalSignature, keyEncipherment extendedKeyUsage = serverAuth, clientAuth subjectAltName = @alt_names [ alt_names ] {san_entries} """ def provision_cert( domain: str, extra_sans: List[str], ca_key_path: Path, ca_crt_path: Path, out_dir: Path, validity_days: int = 1095, ) -> None: """ Generate key, CSR, sign with CA, then encrypt artifacts. All temporary work is done inside a temporary directory; the final key and certificate are moved to `out_dir` only after successful signing. Existing files are never overwritten. """ key_path = out_dir / f"{domain}.key" crt_path = out_dir / f"{domain}.crt" out_dir.mkdir(parents=True, exist_ok=True) if key_path.exists(): raise FileExistsError(f"File already exists: {key_path}") if crt_path.exists(): raise FileExistsError(f"File already exists: {crt_path}") with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) key_tmp = tmp / f"{domain}.key" crt_tmp = tmp / f"{domain}.crt" csr_path = tmp / f"{domain}.csr" csr_conf = tmp / "csr.cnf" ext_conf = tmp / "ext.cnf" # Generate private key inside temporary directory subprocess.run( ["openssl", "genrsa", "-out", str(key_tmp), "2048"], check=True, text=True, capture_output=True, ) # CSR configuration csr_conf.write_text(build_csr_config(domain, extra_sans)) subprocess.run( [ "openssl", "req", "-new", "-key", str(key_tmp), "-config", str(csr_conf), "-out", str(csr_path), ], check=True, text=True, capture_output=True, ) # Signing configuration ext_conf.write_text(build_ext_config(domain, extra_sans)) # Sign the CSR with the decrypted CA key sops_decrypt = subprocess.Popen( ["sops", "decrypt", str(ca_key_path)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) openssl_sign = subprocess.Popen( [ "openssl", "x509", "-req", "-in", str(csr_path), "-CA", str(ca_crt_path), "-CAkey", "/dev/stdin", "-CAcreateserial", "-days", str(validity_days), "-sha256", "-extfile", str(ext_conf), "-extensions", "v3_server", "-out", str(crt_tmp), ], stdin=sops_decrypt.stdout, stdout=subprocess.DEVNULL, ) sops_decrypt.stdout.close() sops_return = sops_decrypt.wait() sign_return = openssl_sign.wait() if sops_return != 0: raise RuntimeError("sops decrypt failed") if sign_return != 0: raise RuntimeError("openssl signing failed") # Verify the certificate chain before moving anything subprocess.run( ["openssl", "verify", "-CAfile", str(ca_crt_path), str(crt_tmp)], check=True, text=True, capture_output=True, ) # Move the final artefacts to the output directory shutil.move(str(key_tmp), str(key_path)) shutil.move(str(crt_tmp), str(crt_path)) # Encrypt both files in place (they are now in out_dir) subprocess.run( ["sops", "encrypt", "--in-place", str(key_path)], check=True, text=True, capture_output=True, ) subprocess.run( ["sops", "encrypt", "--in-place", str(crt_path)], check=True, text=True, capture_output=True, ) print("✔ Certificate provisioned and encrypted:") print(f" {key_path}") print(f" {crt_path}") def main(): parser = argparse.ArgumentParser( description="Provision an internal TLS certificate using a sops‑encrypted Root CA." ) parser.add_argument("domain", help="Primary domain name (e.g., myapp.internal)") parser.add_argument( "--san", action="append", default=[], help="Additional Subject Alternative Names (can be used multiple times)", ) args = parser.parse_args() check_dependencies() repo_root = get_repo_root() ca_dir = repo_root / "secrets/ca" ca_key_path = ca_dir / "rootCA.key" ca_crt_path = repo_root / "etc/certs/rootCA.crt" out_dir = ca_dir / "certs" if not ca_key_path.exists(): sys.exit(f"CA private key not found at {ca_key_path}") if not ca_crt_path.exists(): sys.exit(f"CA certificate not found at {ca_crt_path}") try: provision_cert(args.domain, args.san, ca_key_path, ca_crt_path, out_dir) except Exception as e: sys.exit(f"Error: {e}") if __name__ == "__main__": main()