"""Write out every part of the issued certificate: key, cert, chain, PEM bundle, PFX, info.""" from __future__ import annotations import datetime import json import logging from pathlib import Path from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, rsa from cryptography.hazmat.primitives.serialization import pkcs12 from cryptography.x509.oid import NameOID from .config import Config log = logging.getLogger(__name__) # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def not_before(cert: x509.Certificate) -> datetime.datetime: try: return cert.not_valid_before_utc except AttributeError: # pragma: no cover - cryptography < 42 return cert.not_valid_before.replace(tzinfo=datetime.timezone.utc) def not_after(cert: x509.Certificate) -> datetime.datetime: try: return cert.not_valid_after_utc except AttributeError: # pragma: no cover - cryptography < 42 return cert.not_valid_after.replace(tzinfo=datetime.timezone.utc) def san_list(cert: x509.Certificate) -> list[str]: try: ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName) except x509.ExtensionNotFound: return [] return list(ext.value.get_values_for_type(x509.DNSName)) def days_remaining(cert: x509.Certificate) -> int: delta = not_after(cert) - datetime.datetime.now(datetime.timezone.utc) return delta.days def _write(path: Path, data: bytes | str, secret: bool = False) -> Path: mode = "wb" if isinstance(data, bytes) else "w" with open(path, mode) as fh: fh.write(data) path.chmod(0o600 if secret else 0o644) return path def _key_description(key) -> dict[str, object]: if isinstance(key, rsa.RSAPrivateKey) or isinstance(key, rsa.RSAPublicKey): return {"algorithm": "RSA", "size_bits": key.key_size} if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)): return {"algorithm": "EC", "curve": key.curve.name, "size_bits": key.key_size} return {"algorithm": type(key).__name__} # -------------------------------------------------------------------------- # output # -------------------------------------------------------------------------- def write_certificate_files( cfg: Config, fqdn: str, private_key, csr_pem: bytes, chain: list[x509.Certificate], domains: list[str], ) -> dict[str, Path]: """Write all certificate artefacts for *fqdn* and return {label: path}.""" out_dir = cfg.cert_output_dir / fqdn out_dir.mkdir(parents=True, exist_ok=True) leaf, intermediates = chain[0], chain[1:] password = cfg.cert_password.encode() paths: dict[str, Path] = {} # -- private key, in the usual flavours -------------------------------- paths["private key (PKCS#8)"] = _write( out_dir / "privkey.pem", private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ), secret=True, ) paths["private key (traditional)"] = _write( out_dir / "privkey-traditional.pem", private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption(), ), secret=True, ) paths["private key (encrypted)"] = _write( out_dir / "privkey-encrypted.pem", private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.BestAvailableEncryption(password), ), secret=True, ) paths["public key"] = _write( out_dir / "pubkey.pem", private_key.public_key().public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo, ), ) # -- CSR --------------------------------------------------------------- paths["CSR"] = _write(out_dir / "csr.pem", csr_pem) # -- certificate parts ------------------------------------------------- leaf_pem = leaf.public_bytes(serialization.Encoding.PEM) chain_pem = b"".join(c.public_bytes(serialization.Encoding.PEM) for c in intermediates) fullchain_pem = leaf_pem + chain_pem paths["certificate (leaf)"] = _write(out_dir / "cert.pem", leaf_pem) paths["certificate (.crt copy)"] = _write(out_dir / "cert.crt", leaf_pem) paths["certificate (DER/.cer)"] = _write( out_dir / "cert.der", leaf.public_bytes(serialization.Encoding.DER) ) paths["CA chain"] = _write(out_dir / "chain.pem", chain_pem) paths["fullchain"] = _write(out_dir / "fullchain.pem", fullchain_pem) for index, ca in enumerate(intermediates, start=1): cn = _common_name(ca) or f"ca-{index}" paths[f"chain part {index} ({cn})"] = _write( out_dir / f"chain-{index:02d}.pem", ca.public_bytes(serialization.Encoding.PEM) ) # -- combined PEM (key + certificate + chain) -------------------------- bundle = ( private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) + fullchain_pem ) paths["PEM bundle (key + fullchain)"] = _write(out_dir / "bundle.pem", bundle, secret=True) # -- PKCS#12 / PFX ----------------------------------------------------- paths["PKCS#12 (PFX)"] = _write( out_dir / "cert.pfx", _build_pfx(cfg, fqdn, private_key, leaf, intermediates, password), secret=True, ) # -- human/machine readable summary ------------------------------------ info = certificate_info(chain, private_key, domains) paths["info (JSON)"] = _write(out_dir / "cert-info.json", json.dumps(info, indent=2) + "\n") paths["info (text)"] = _write(out_dir / "cert-info.txt", format_info(info) + "\n") return paths def _build_pfx(cfg: Config, fqdn: str, private_key, leaf, intermediates, password: bytes) -> bytes: if cfg.pfx_legacy_compat: try: encryption = ( serialization.PrivateFormat.PKCS12.encryption_builder() .key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC) .hmac_hash(hashes.SHA1()) .build(password) ) log.info("Building PFX in legacy format (SHA1 / 3DES)") except Exception as exc: # noqa: BLE001 - depends on the OpenSSL build log.warning("Legacy PFX encryption unavailable (%s), falling back to AES-256.", exc) encryption = serialization.BestAvailableEncryption(password) else: encryption = serialization.BestAvailableEncryption(password) return pkcs12.serialize_key_and_certificates( name=fqdn.encode(), key=private_key, cert=leaf, cas=intermediates or None, encryption_algorithm=encryption, ) # -------------------------------------------------------------------------- # certificate information # -------------------------------------------------------------------------- def _common_name(cert: x509.Certificate) -> str: try: return cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value except (IndexError, ValueError): return "" def certificate_info(chain: list[x509.Certificate], private_key, domains: list[str]) -> dict: leaf = chain[0] return { "requested_domains": domains, "key": _key_description(private_key), "leaf": _cert_info(leaf), "chain": [_cert_info(c) for c in chain[1:]], "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), } def _cert_info(cert: x509.Certificate) -> dict: return { "subject": cert.subject.rfc4514_string(), "common_name": _common_name(cert), "issuer": cert.issuer.rfc4514_string(), "serial_number": format(cert.serial_number, "x"), "not_before": not_before(cert).isoformat(timespec="seconds"), "not_after": not_after(cert).isoformat(timespec="seconds"), "days_remaining": days_remaining(cert), "subject_alternative_names": san_list(cert), "signature_algorithm": getattr( cert.signature_algorithm_oid, "_name", cert.signature_algorithm_oid.dotted_string ), "public_key": _key_description(cert.public_key()), "fingerprint_sha256": cert.fingerprint(hashes.SHA256()).hex(":"), "fingerprint_sha1": cert.fingerprint(hashes.SHA1()).hex(":"), } def format_info(info: dict) -> str: lines = ["Certificate details", "=" * 60] leaf = info["leaf"] lines += [ f"Common name : {leaf['common_name']}", f"SANs : {', '.join(leaf['subject_alternative_names'])}", f"Issuer : {leaf['issuer']}", f"Serial : {leaf['serial_number']}", f"Valid from : {leaf['not_before']}", f"Valid until : {leaf['not_after']} ({leaf['days_remaining']} days)", f"Key : {info['key'].get('algorithm')} " f"{info['key'].get('size_bits', '')} {info['key'].get('curve', '')}".rstrip(), f"Signature : {leaf['signature_algorithm']}", f"SHA-256 : {leaf['fingerprint_sha256']}", f"SHA-1 : {leaf['fingerprint_sha1']}", ] if info["chain"]: lines += ["", "Chain:"] for index, ca in enumerate(info["chain"], start=1): lines.append(f" {index}. {ca['common_name'] or ca['subject']} (until {ca['not_after']})") return "\n".join(lines) # -------------------------------------------------------------------------- # renewal check # -------------------------------------------------------------------------- def load_existing_certificate(cfg: Config, fqdn: str) -> x509.Certificate | None: path = cfg.cert_output_dir / fqdn / "cert.pem" if not path.is_file(): return None try: return x509.load_pem_x509_certificate(path.read_bytes()) except ValueError as exc: log.warning("Existing certificate %s is unreadable (%s) - issuing a new one.", path, exc) return None def renewal_needed(cfg: Config, fqdn: str, domains: list[str]) -> tuple[bool, str]: cert = load_existing_certificate(cfg, fqdn) if cert is None: return True, "no certificate found yet" have = {d.lower() for d in san_list(cert)} want = {d.lower() for d in domains} if not want.issubset(have): return True, f"missing names in existing certificate: {', '.join(sorted(want - have))}" left = days_remaining(cert) if left <= cfg.renew_days_before_expiry: return True, f"expires in {left} days" return False, f"still valid for {left} days"