add python cert helper

1 files changed, 244 insertions, 0 deletions

diff --git a/etc/bin/cert.py b/etc/bin/cert.py
new file mode 100755
index 0000000..92baa7e
--- /dev/null
+++ b/etc/bin/cert.py
@@ -0,0 +1,244 @@
+#!/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()