certmgr.py

#!/usr/bin/env python3
# Copyright (c) 2025-2026 Tim Riker
# SPDX-License-Identifier: MIT
"""
Command-line orchestration for obtaining and deploying certificates.

Usage:
    ./certmgr.py [options]

Options:
    --config PATH                Path to config file (default: config.yaml)
    --credentials PATH           Path to credentials file (default: credentials.yaml)
    --account-key PATH           Path to ACME account key (default: account.key)
    --days N                     Renew if cert expires within N days (default: 30)
    --force                      Force renewal regardless of expiration
    --staging                    Use Let's Encrypt staging directory (safe for testing)
    --dry-run                    Do not perform network calls; print planned actions
    --verbose                    Enable verbose (debug) logging output
    --prepopulate                Create/overwrite TXT records for all _acme-challenge.<domain> names listed in config without requesting a certificate
    --deploy                     Deploy existing local certificates to F5 targets without requesting new certificates
    --list                       List existing local certificates with their domains and expiration dates
    --certs NAMES                Comma-delimited list of certificate names to process (e.g. dicm.org,example.com)
    --dns-wait-seconds N         Seconds to wait for DNS propagation (default: 5)

Default behavior (no options):
    - Shows list of all certificates with expiration status
    - Automatically renews certificates that expire within 30 days (or --days threshold)
    - Renews certificates that have domain changes in config
    - Deploys renewed certificates to configured `f5_ltm` and `f5_httpd` targets
    - Shows summary of actions taken

IMPORTANT: When adding or changing CLI options, update this comment block, the README, and the argparse help strings to keep documentation in sync.
"""
import argparse
import logging
import os
from acme.messages import Error
import yaml
from datetime import datetime, timezone
from cryptography import x509
import certifi
import sys
from typing import List, Optional, Tuple

# Make the script runnable from any current working directory by ensuring the
# script directory is on sys.path and by using config/credentials paths
# relative to the script directory by default.
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if SCRIPT_DIR not in sys.path:
    sys.path.insert(0, SCRIPT_DIR)

try:
    from .acme_client import AcmeClient
    from .dns_rfc2136 import resolve_cname_target, update_txt_record, discover_zone_for_name
    from .f5_ltm import F5LTM
    from .f5_httpd import F5HTTPD
except Exception:
    # Allow running this file directly (not via -m) for quick testing by
    # falling back to module-level imports.
    from acme_client import AcmeClient
    from dns_rfc2136 import resolve_cname_target, update_txt_record, discover_zone_for_name
    from f5_ltm import F5LTM
    from f5_httpd import F5HTTPD

log = logging.getLogger(__name__)


def load_yaml(path):
    with open(path, 'r') as f:
        return yaml.safe_load(f)


def ensure_dir(path):
    if not os.path.exists(path):
        os.makedirs(path)


def days_until_expiry(pem_data: bytes) -> int:
    try:
        cert = x509.load_pem_x509_certificate(pem_data)
        not_after = cert.not_valid_after
        # Ensure not_after is timezone-aware
        if not_after.tzinfo is None:
            not_after = not_after.replace(tzinfo=timezone.utc)
        delta = not_after - datetime.now(timezone.utc)
        return delta.days
    except ValueError:
        # PEM is invalid or missing
        return -9999


def expand_domains(domains: list) -> list:
    """Expand domain list to include bare domains for wildcards.

    If a wildcard like *.example.com is present, automatically include
    example.com as well (unless already present).
    """
    expanded = list(domains)  # Copy the original list
    wildcards = [d for d in domains if d.startswith('*.')]
    bare_domains_to_add = set()
    for wc in wildcards:
        bare = wc[2:]
        # Only add the bare domain if it is not covered by another broader wildcard
        covered = False
        for other_wc in wildcards:
            if other_wc == wc:
                continue
            # If this wildcard is a subdomain of another wildcard, skip adding its bare domain
            if bare.endswith('.' + other_wc[2:]):
                covered = True
                break
        if not covered and bare not in expanded:
            bare_domains_to_add.add(bare)
    expanded.extend(sorted(bare_domains_to_add))
    return expanded


def split_pem_certificates(pem_data: bytes) -> List[bytes]:
    blocks = []
    current = []
    inside = False
    for line in pem_data.splitlines(keepends=True):
        if b'-----BEGIN CERTIFICATE-----' in line:
            current = [line]
            inside = True
        elif inside:
            current.append(line)
            if b'-----END CERTIFICATE-----' in line:
                blocks.append(b''.join(current))
                current = []
                inside = False
    return blocks


def normalize_pem_bundle(pem_data: bytes) -> bytes:
    blocks = [block.strip() for block in split_pem_certificates(pem_data)]
    if not blocks:
        return pem_data
    return b"\n\n".join(blocks) + b"\n"


def load_trust_store_certificates() -> List[Tuple[x509.Certificate, bytes]]:
    with open(certifi.where(), 'rb') as f:
        pem_data = f.read()
    certs = []
    for pem_block in split_pem_certificates(pem_data):
        certs.append((x509.load_pem_x509_certificate(pem_block), pem_block))
    return certs


def build_with_root_pem(cert_pem: bytes) -> Optional[bytes]:
    cert_blocks = split_pem_certificates(cert_pem)
    if not cert_blocks:
        return None

    chain = [x509.load_pem_x509_certificate(block) for block in cert_blocks]
    last_cert = chain[-1]

    for trust_cert, trust_pem in load_trust_store_certificates():
        if trust_cert.subject != last_cert.issuer:
            continue
        if trust_cert.subject != trust_cert.issuer:
            continue
        return normalize_pem_bundle(cert_pem + b"\n" + trust_pem)

    return None


def get_f5_ltm_targets(cert: dict, config: dict) -> list:
    """Return the traffic-certificate deployment targets for a certificate."""
    return cert.get('f5_ltm') or config.get('f5_ltm') or []


def get_f5_httpd_targets(cert: dict, config: dict) -> list:
    """Return the management HTTPD deployment targets for a certificate."""
    return cert.get('f5_httpd') or config.get('f5_httpd') or []


def get_acme_profile(cert: dict, config: dict) -> Optional[str]:
    """Return the ACME profile to request for a certificate, if configured."""
    return cert.get('acme_profile') or config.get('acme_profile')


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--config', default=os.path.join(SCRIPT_DIR, 'config.yaml'))
    parser.add_argument('--credentials', default=os.path.join(SCRIPT_DIR, 'credentials.yaml'))
    parser.add_argument('--account-key', default='account.key')
    parser.add_argument('--days', type=int, default=30, help='Renew if cert expires within DAYS')
    parser.add_argument('--force', action='store_true')
    parser.add_argument('--staging', action='store_true', help='Use Let\'s Encrypt staging directory (safe for testing)')
    parser.add_argument('--dry-run', action='store_true', help='Do not perform network calls; print planned actions')
    parser.add_argument('--verbose', action='store_true', help='Enable verbose (debug) logging output')
    parser.add_argument('--prepopulate', action='store_true',
                        help='Create (or overwrite) TXT records for all _acme-challenge.<domain> names listed in config without requesting a certificate')
    parser.add_argument('--deploy', action='store_true',
                        help='Deploy existing local certificates to F5 targets without requesting new certificates')
    parser.add_argument('--list', action='store_true',
                        help='List existing local certificates with their domains and expiration dates')
    parser.add_argument('--certs', type=str, help='Comma-delimited list of certificate names to process (e.g. dicm.org,example.com)')
    parser.add_argument('--dns-wait-seconds', type=int, default=5, help='Seconds to wait for DNS propagation (default: 5)')
    args = parser.parse_args()

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)

    config = load_yaml(args.config)
    creds = load_yaml(args.credentials)

    certificates = config.get('certificates', [])
    if args.certs:
        cert_names = [n.strip() for n in args.certs.split(',') if n.strip()]
        certificates = [c for c in certificates if c['name'] in cert_names]

    certs_dir = os.path.join(SCRIPT_DIR, 'certs')
    if not args.dry_run:
        ensure_dir(certs_dir)

    # Choose ACME directory: production by default, staging if requested
    directory_url = "https://acme-v02.api.letsencrypt.org/directory"
    if args.staging:
        directory_url = "https://acme-staging-v02.api.letsencrypt.org/directory"
        log.info("Using Let's Encrypt STAGING directory: %s", directory_url)
    acme = AcmeClient(directory_url=directory_url, dns_wait_seconds=args.dns_wait_seconds)

    # We'll create per-certificate publish/remove closures inside the loop so
    # that each certificate can carry its own rfc2136/zone overrides. If a
    # certificate doesn't provide rfc2136 settings, we fall back to
    # credentials.yaml rfc2136 block.

    # Handle --list option (or default with no action options)
    show_list = args.list or not (args.prepopulate or args.deploy)

    if show_list:
        print(f"{'Certificate':<30} {'Expires In':<15} {'Status':<15} {'Domains'}")
        print("-" * 120)
        for cert in certificates:
            name = cert['name']
            domains = expand_domains(cert['domains'])
            cert_path = os.path.join('certs', f"{name}.pem")

            if os.path.exists(cert_path):
                try:
                    with open(cert_path, 'rb') as f:
                        pem = f.read()
                    days = days_until_expiry(pem)
                    domains_str = ', '.join(domains)
                    if days < 0:
                        status = "EXPIRED"
                    elif days <= 30:
                        status = "RENEW SOON"
                    else:
                        status = "Valid"
                    print(f"{name:<30} {days:>3} days        {status:<15} {domains_str}")
                except Exception as e:
                    domains_str = ', '.join(domains)
                    print(f"{name:<30} {'ERROR':<15} {str(e)[:20]:<15} {domains_str}")
            else:
                domains_str = ', '.join(domains)
                print(f"{name:<30} {'NOT FOUND':<15} {'Missing':<15} {domains_str}")
        print()

    # If --list only, exit after showing list
    if args.list:
        return

    # Track summary of operations when running default mode
    summary = {
        'renewed': [],
        'reordered': [],
        'ltm_deployed': [],
        'httpd_deployed': [],
        'errors': []
    }

    # Iterate certificates
    for cert in certificates:
        name = cert['name']
        domains = expand_domains(cert['domains'])
        cert_path = os.path.join('certs', f"{name}.pem")
        with_root_path = os.path.join('certs', f"{name}.with-root.pem")
        key_path = os.path.join('certs', f"{name}.key")
        acme_profile = get_acme_profile(cert, config)

        if args.deploy:
            # Deploy existing certificates without requesting new ones
            if not os.path.exists(cert_path):
                log.warning("Certificate %s not found at %s, skipping", name, cert_path)
                continue
            if not os.path.exists(key_path):
                log.warning("Key for %s not found at %s, skipping", name, key_path)
                continue
            with open(cert_path, 'rb') as f:
                cert_pem = f.read()
            if not os.path.exists(with_root_path):
                log.warning("Certificate with root for %s not found at %s, skipping HTTPD deployment", name, with_root_path)
                with_root_pem = None
            else:
                with open(with_root_path, 'rb') as f:
                    with_root_pem = f.read()
            with open(key_path, 'rb') as f:
                key_pem = f.read()

            f5_ltm = get_f5_ltm_targets(cert, config)
            f5_httpd = get_f5_httpd_targets(cert, config)
            f5_creds = creds.get('f5', {})
            ltm_deployer = F5LTM(f5_creds.get('username'), f5_creds.get('password'), verify_ssl=f5_creds.get('verify_ssl', False))
            httpd_deployer = F5HTTPD(f5_creds.get('username'), f5_creds.get('password'), verify_ssl=f5_creds.get('verify_ssl', False))

            for host in f5_ltm:
                try:
                    ltm_deployer.deploy(host, cert_pem, key_pem, name)
                    log.info("Deployed %s to %s", name, host)
                    summary['ltm_deployed'].append(f"{name} -> {host}")
                except Exception as e:
                    log.exception("Failed to deploy to %s: %s", host, e)
                    summary['errors'].append(f"Deploy {name} to {host}: {str(e)}")

            if with_root_pem:
                for cluster_host in f5_httpd:
                    try:
                        members = httpd_deployer.discover_cluster_members(cluster_host)
                        for member in members:
                            result = httpd_deployer.deploy_member(member, name, with_root_pem, key_pem)
                            log.info(
                                "Deployed HTTPD certificate for %s to %s (%s), presented subject %s",
                                name,
                                result['member']['device_name'],
                                result['member']['management_ip'],
                                result['verification']['subject'],
                            )
                            summary['httpd_deployed'].append(f"{name} -> {result['member']['device_name']}")
                    except Exception as e:
                        log.exception("Failed HTTPD deployment for %s on %s: %s", name, cluster_host, e)
                        summary['errors'].append(f"HTTPD deploy {name} on {cluster_host}: {str(e)}")
            continue

        if args.prepopulate:
            # For each domain create its ACME challenge name and publish a placeholder TXT.
            # Wildcard domains (*.example.com) map to base domain for dns-01.
            log.info("Prepopulating TXT records for certificate %s", name)
            timestamp = int(datetime.now(timezone.utc).timestamp())
            for d in domains:
                is_wildcard = d.startswith('*.')
                base = d[2:] if is_wildcard else d
                challenge_name = f"_acme-challenge.{base}"
                # Make placeholder unique for wildcard vs bare
                if is_wildcard:
                    placeholder = f"prepopulate-wildcard-{timestamp}-{base}"
                else:
                    placeholder = f"prepopulate-bare-{timestamp}-{base}"
                try:
                    r = creds.get('rfc2136')
                    if not r:
                        raise RuntimeError("No rfc2136 block in credentials.yaml for prepopulate")
                    server = r.get('server')
                    port = r.get('port', 53)
                    key_name = r.get('key_name')
                    key = r.get('key')
                    algorithm = r.get('algorithm')
                    target = resolve_cname_target(challenge_name, server, key_name, key, algorithm)
                    zname = discover_zone_for_name(target, server, key_name, key, algorithm)
                    log.info("Publishing placeholder TXT for %s (target %s zone %s): %s", challenge_name, target, zname, placeholder)
                    update_txt_record(server, port, key_name, key, algorithm, zname, target, placeholder)
                except Exception as e:
                    log.error("Failed to prepopulate %s: %s", challenge_name, e)
            # Skip renewal logic when prepopulate is requested
            continue

        need = args.force
        reorder = False  # Track if domains have changed
        if not need and os.path.exists(cert_path):
            with open(cert_path, 'rb') as f:
                pem = f.read()
                days = days_until_expiry(pem)
                if days <= args.days:
                    log.info("Certificate %s expires in %d days (<= %d), will renew", name, days, args.days)
                    need = True
                else:
                    log.info("Certificate %s is valid for %d more days, skipping", name, days)

                # Check if domains have changed
                try:
                    cert_obj = x509.load_pem_x509_certificate(pem)
                    # Extract Subject Alternative Names
                    san_ext = cert_obj.extensions.get_extension_for_class(x509.SubjectAlternativeName)
                    cert_domains = set()
                    for san in san_ext.value:
                        if isinstance(san, x509.DNSName):
                            cert_domains.add(san.value)
                    config_domains = set(domains)
                    if cert_domains != config_domains:
                        log.info("Certificate %s has domain changes (cert: %s, config: %s), will reorder", name, cert_domains, config_domains)
                        need = True
                        reorder = True
                except Exception as e:
                    log.warning("Failed to check domains for %s: %s", name, e)
        else:
            need = True

        if need:
            log.info("Requesting certificate for %s (%s)", name, domains)

            if args.dry_run:
                # Dry-run: print the planned actions and skip network interactions.
                log.info("DRY RUN: would request ACME certificate for %s with domains %s", name, domains)
                if acme_profile:
                    log.info("DRY RUN: would request ACME profile %s for %s", acme_profile, name)
                f5_ltm = get_f5_ltm_targets(cert, config)
                for host in f5_ltm:
                    log.info("DRY RUN: would deploy %s.crt and %s.key to %s", name, name, host)
                f5_httpd = get_f5_httpd_targets(cert, config)
                for cluster_host in f5_httpd:
                    log.info("DRY RUN: would deploy %s.with-root.pem and %s.key to HTTPD cluster %s", name, name, cluster_host)
                # continue to next certificate without making network calls
                continue

            # Publish/remove helpers use only the RFC2136 settings from
            # credentials.yaml. We follow CNAMEs for the challenge and if the
            # zone name is not explicit we derive the last two labels as a
            # best-effort zone name.
            def publish(fqdn, txt):
                r = creds.get('rfc2136')
                if not r:
                    raise RuntimeError(f"No RFC2136 configuration in credentials.yaml required for publishing challenge for {fqdn}")
                server = r.get('server')
                port = r.get('port', 53)
                key_name = r.get('key_name')
                key = r.get('key')
                algorithm = r.get('algorithm')
                # Always use base domain for DNS-01 challenge
                base_fqdn = fqdn
                if fqdn.startswith("_acme-challenge.*."):
                    base_fqdn = "_acme-challenge." + fqdn[len("_acme-challenge.*."):]
                elif fqdn.startswith("*." ):
                    base_fqdn = fqdn[2:]
                target = resolve_cname_target(base_fqdn, server, key_name, key, algorithm)
                zname = discover_zone_for_name(target, server, key_name, key, algorithm)
                log.info("Publishing TXT record %s as %s (zone %s): %s", fqdn, target, zname, txt)
                update_txt_record(server, port, key_name, key, algorithm, zname, target, txt)

            def remove(fqdn):
                r = creds.get('rfc2136')
                if not r:
                    log.warning("No rfc2136 in credentials.yaml for %s, skipping removal", fqdn)
                    return
                server = r.get('server')
                port = r.get('port', 53)
                key_name = r.get('key_name')
                key = r.get('key')
                algorithm = r.get('algorithm')
                # Always use base domain for DNS-01 challenge
                base_fqdn = fqdn
                if fqdn.startswith("_acme-challenge.*."):
                    base_fqdn = "_acme-challenge." + fqdn[len("_acme-challenge.*."):]
                elif fqdn.startswith("*." ):
                    base_fqdn = fqdn[2:]
                target = resolve_cname_target(base_fqdn, server, key_name, key, algorithm)
                zname = discover_zone_for_name(target, server, key_name, key, algorithm)
                log.info("Removing TXT record %s as %s (zone %s)", fqdn, target, zname)
                try:
                    update_txt_record(server, port, key_name, key, algorithm, zname, target, "")
                except Exception as e:
                    log.warning("Failed to remove TXT record for %s: %s", target, e)

            try:
                cert_pem, _, key_pem = acme.obtain_certificate(
                    domains,
                    publish,
                    remove,
                    account_key_path=args.account_key,
                    profile=acme_profile,
                )
            except Error as e:
                err_msg = f"ACME error for {name}: {e}"
                print(err_msg)
                summary['errors'].append(err_msg)
                continue
            # cert_pem is fullchain; save key and cert
            if cert_pem is not None and cert_pem.startswith(b'ACME_VALIDATION_ERROR:'):
                error_blob = cert_pem[len(b'ACME_VALIDATION_ERROR:'):].decode(errors='replace')
                for err_line in error_blob.split('; '):
                    err_msg = f"Certificate {name}: {err_line}"
                    print(err_msg)
                    summary['errors'].append(err_msg)
                continue
            if cert_pem is None or cert_pem.startswith(b'ACME error:'):
                warn_msg = f"Certificate request failed for {name}; no certificate issued."
                log.warning(warn_msg)
                summary['errors'].append(warn_msg)
                continue
            cert_pem = normalize_pem_bundle(cert_pem)
            with open(cert_path, 'wb') as f:
                f.write(cert_pem)
            with_root_pem = build_with_root_pem(cert_pem)
            with_root_path = os.path.join('certs', f"{name}.with-root.pem")
            if with_root_pem:
                with open(with_root_path, 'wb') as f:
                    f.write(with_root_pem)
                log.info("Saved certificate with root to %s", with_root_path)
            else:
                log.warning("Could not find a matching root certificate for %s; %s was not written", name, with_root_path)
            # save private key if returned
            if key_pem:
                with open(key_path, 'wb') as kf:
                    kf.write(key_pem)
                log.info("Saved private key to %s", key_path)
            log.info("Saved certificate to %s", cert_path)

            # Track what was done
            if reorder:
                summary['reordered'].append(name)
            else:
                summary['renewed'].append(name)

            # Deploy to F5 targets
            # Determine f5 targets: per-cert override, per-zone or global
            f5_ltm = get_f5_ltm_targets(cert, config)
            f5_httpd = get_f5_httpd_targets(cert, config)
            f5_creds = creds.get('f5', {})
            ltm_deployer = F5LTM(f5_creds.get('username'), f5_creds.get('password'), verify_ssl=f5_creds.get('verify_ssl', False))
            httpd_deployer = F5HTTPD(f5_creds.get('username'), f5_creds.get('password'), verify_ssl=f5_creds.get('verify_ssl', False))
            # We currently do not have the private key saved separately; if the
            # ACME client returns it we should save and deploy it. This example
            # expects the CSR-generation key to be made available; for now we
            # store cert only and attempt to deploy cert (some F5s accept cert-only)
            for host in f5_ltm:
                try:
                    # pass empty key for now if not present
                    key_pem = b""
                    if os.path.exists(key_path):
                        with open(key_path, 'rb') as kf:
                            key_pem = kf.read()
                    # use certificate base name; on the F5 the objects will be
                    # named <name>.crt and <name>.key
                    ltm_deployer.deploy(host, cert_pem, key_pem, name)
                    log.info("Deployed %s to %s", name, host)
                    summary['ltm_deployed'].append(f"{name} -> {host}")
                except Exception as e:
                    log.exception("Failed to deploy to %s: %s", host, e)
                    summary['errors'].append(f"Deploy {name} to {host}: {str(e)}")
            if with_root_pem:
                for cluster_host in f5_httpd:
                    try:
                        members = httpd_deployer.discover_cluster_members(cluster_host)
                        for member in members:
                            result = httpd_deployer.deploy_member(member, name, with_root_pem, key_pem)
                            log.info(
                                "Deployed HTTPD certificate for %s to %s (%s), presented subject %s",
                                name,
                                result['member']['device_name'],
                                result['member']['management_ip'],
                                result['verification']['subject'],
                            )
                            summary['httpd_deployed'].append(f"{name} -> {result['member']['device_name']}")
                    except Exception as e:
                        log.exception("Failed HTTPD deployment for %s on %s: %s", name, cluster_host, e)
                        summary['errors'].append(f"HTTPD deploy {name} on {cluster_host}: {str(e)}")

    # Print summary for all action modes except --prepopulate and --list.
    if not (args.prepopulate or args.list):
        print()
        print("=" * 80)
        print("SUMMARY")
        print("=" * 80)

        if summary['renewed']:
            print(f"Renewed certificates: {', '.join(summary['renewed'])}")
        if summary['reordered']:
            print(f"Reordered certificates (domain changes): {', '.join(summary['reordered'])}")
        if summary['ltm_deployed']:
            ltm_unique = sorted(set(summary['ltm_deployed']))
            print("LTM deployments:")
            for item in ltm_unique:
                print(f"  - {item}")
        if summary['httpd_deployed']:
            httpd_unique = sorted(set(summary['httpd_deployed']))
            print("HTTPD deployments:")
            for item in httpd_unique:
                print(f"  - {item}")
        if not (summary['renewed'] or summary['reordered'] or summary['ltm_deployed'] or summary['httpd_deployed']):
            if summary['errors']:
                print("No certificates were renewed or deployed due to errors.")
            else:
                print("No certificates needed renewal. All certificates are up to date.")
        # Always print errors at the end of the summary
        if summary['errors']:
            print(f"Errors encountered: {len(summary['errors'])}")
            for err in summary['errors']:
                print(f"  - {err}")

if __name__ == '__main__':
    main()