path: root/etc/bin/cert.py

#!/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()