vacuum-wall/lib/acme.py

"""
ACME certificate manager for Vacuum Wall.

Wraps acme.sh to issue, renew, and manage SSL/TLS certificates
from Let's Encrypt (or other ACME providers).
"""

import logging
import os
import re
import shutil
import subprocess
from datetime import UTC, datetime
from pathlib import Path

logger = logging.getLogger(__name__)

_ACME_ENVIRON = {
    "HOME": str(Path.home()),
    "PATH": os.environ.get(
        "PATH", "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
    ),
}


def _find_acme() -> str:
    """Locate the acme.sh binary on the system.

    Checks:
      1. ~/.acme.sh/acme.sh
      2. /usr/local/bin/acme.sh

    Returns:
        Absolute path to the acme.sh binary.

    Raises:
        FileNotFoundError: If acme.sh cannot be found.
    """
    candidates = [
        Path.home() / ".acme.sh" / "acme.sh",
        Path("/usr/local/bin/acme.sh"),
    ]

    for path in candidates:
        if path.is_file() and os.access(path, os.X_OK):
            logger.info("Found acme.sh at %s", path)
            return str(path)

    acme = shutil.which("acme.sh")
    if acme:
        logger.info("Found acme.sh via PATH at %s", acme)
        return acme

    raise FileNotFoundError(
        "acme.sh not found in any standard location. "
        "Install it with: curl -sSL https://get.acme.sh | sh"
    )


def _run_acme(args: list[str]) -> str:
    """Execute acme.sh with the given arguments.

    Runs the command as root via sudo because standalone / webroot
    validation often requires binding to privileged ports (80/443).

    Args:
        args: List of arguments to pass to acme.sh.

    Returns:
        Combined stdout + stderr from the command, since acme.sh writes
        meaningful output to both streams.

    Raises:
        RuntimeError: If the acme.sh command exits with a non-zero code.
    """
    acme_bin = _find_acme()

    cmd: list[str] = [
        "sudo",
        acme_bin,
        "--home",
        str(Path.home() / ".acme.sh"),
        "--config-home",
        str(Path.home() / ".acme.sh"),
        *args,
    ]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=120,
            env={**os.environ, **_ACME_ENVIRON},
        )
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(
            f"acme.sh command timed out after 120s: {' '.join(cmd)}"
        ) from exc

    output = result.stdout
    if result.stderr:
        output = output + result.stderr if output else result.stderr

    if result.returncode != 0:
        logger.error("acme.sh failed (rc=%d): %s", result.returncode, output.strip())
        raise RuntimeError(
            f"acme.sh failed with exit code {result.returncode}: {output.strip()}"
        )

    return output


def set_email(email: str) -> None:
    """Configure the default ACME contact email.

    Registers or updates the ACME account with the given email address.

    Args:
        email: The contact email for the ACME account.
    """
    _run_acme(["--register-account", "-m", email])
    logger.info("ACME contact email set to %s", email)


def get_email() -> str:
    """Return the ACME contact email, or '' if none is configured."""
    try:
        account_conf = Path.home() / ".acme.sh" / "account.conf"
        if account_conf.is_file():
            text = account_conf.read_text()
            match = re.search(r"^ACME_LEEMAIL=(.+)$", text, re.MULTILINE)
            if match:
                return match.group(1).strip().strip("'\"")
    except OSError as exc:
        logger.warning("Could not read account.conf: %s", exc)
    return ""


def issue(domain: str, webroot: str | None = None, standalone: bool = False) -> dict:
    """Issue a new SSL certificate for a domain.

    Args:
        domain: The primary domain name.
        webroot: Path to the web root directory for HTTP-01 validation.
        standalone: If True, use standalone TCP validation (binds port 80).

    Returns:
        A dict with 'success', 'domain', 'message', 'output', and 'error'.
    """
    args: list[str] = ["--issue", "-d", domain]

    if webroot:
        args.extend(["--webroot", webroot])
    elif standalone:
        args.append("--standalone")

    email = get_email()
    if email:
        args.extend(["-m", email])
    args.append("--force")

    try:
        output = _run_acme(args)
        return {
            "success": True,
            "domain": domain,
            "message": f"Certificate for {domain} issued successfully",
            "output": output.strip(),
            "error": None,
        }
    except RuntimeError as exc:
        return {
            "success": False,
            "domain": domain,
            "message": f"Failed to issue certificate for {domain}",
            "output": "",
            "error": str(exc),
        }


def renew(domain: str, force: bool = False) -> dict:
    """Renew an existing SSL certificate.

    Args:
        domain: The domain whose certificate should be renewed.
        force: If True, renew even if the certificate isn't close to expiry.

    Returns:
        A dict with 'success', 'domain', 'message', 'output', and 'error'.
    """
    args: list[str] = ["--renew", "-d", domain]
    if force:
        args.append("--force")

    try:
        output = _run_acme(args)
        return {
            "success": True,
            "domain": domain,
            "message": f"Certificate for {domain} renewed successfully",
            "output": output.strip(),
            "error": None,
        }
    except RuntimeError as exc:
        return {
            "success": False,
            "domain": domain,
            "message": f"Failed to renew certificate for {domain}",
            "output": "",
            "error": str(exc),
        }


def remove(domain: str) -> str:
    """Stop auto-renewal for a domain.

    Runs ``acme.sh --remove`` which stops the cron job from renewing
    the certificate.  Per the acme.sh README the cert/key files are
    **not** deleted from disk after ``--remove``; remove them with
    the ``--ecc`` flag if needed, or delete the ``~/.acme.sh/{domain}``
    directory manually.

    Args:
        domain: The domain to remove from the renewal list.

    Returns:
        The combined stdout from the acme.sh command.
    """
    output = _run_acme(["--remove", "-d", domain])
    logger.info("Certificate for %s removed", domain)
    return output


def list_certs() -> list[dict]:
    """List all managed certificates with expiry information.

    Returns:
        A list of dicts, one per certificate, with keys matching
        the cert-info schema (domain, ca, cert_path, etc.).
    """
    raw = _run_acme(["--list"])
    certs: list[dict] = []

    entries = _parse_list_output(raw)
    acme_home = Path.home() / ".acme.sh"

    for entry in entries:
        main = entry["main_domain"]
        if not main:
            continue

        san_domains = [
            d.strip() for d in entry.get("san_domain", "").split(",") if d.strip()
        ]

        cert_dir = acme_home / main
        cert_path = str(cert_dir / "fullchain.cer")
        key_path = str(cert_dir / f"{main}.key")
        ca_path = str(cert_dir / "ca.cer")

        days = _days_until(entry.get("certificate_expires", ""))
        auto = _has_auto_renew(main)

        certs.append(
            {
                "domain": main,
                "ca": entry.get("CA", ""),
                "cert_path": cert_path,
                "key_path": key_path,
                "ca_path": ca_path,
                "issued_at": entry.get("certificate_date", ""),
                "expires_at": entry.get("certificate_expires", ""),
                "days_until_expiry": days,
                "auto_renew": auto,
                "san_domains": san_domains,
            }
        )

    return certs


def get_cert_info(domain: str) -> dict:
    """Return detailed information about a certificate.

    Args:
        domain: The domain name.

    Returns:
        A dict matching the cert-info schema.

    Raises:
        ValueError: If no certificate is found for the domain.
    """
    certs = list_certs()
    for c in certs:
        if c["domain"] == domain or domain in c["san_domains"]:
            return c

    raise ValueError(f"No certificate found for domain: {domain}")


def get_expiry(domain: str) -> str | None:
    """Return the certificate expiry date as an ISO string, or None.

    Args:
        domain: The domain name.

    Returns:
        Expiry date string (e.g. '2026-04-15') or None.
    """
    try:
        info = get_cert_info(domain)
        return info.get("expires_at")
    except ValueError:
        return None


def is_expired(domain: str) -> bool:
    """Check whether a certificate has expired.

    Args:
        domain: The domain name.

    Returns:
        True if the certificate is expired or not found, False otherwise.
    """
    days = days_until_expiry(domain)
    if days is None:
        return True
    return days < 0


def days_until_expiry(domain: str) -> int | None:
    """Calculate the number of days until a certificate expires.

    Args:
        domain: The domain name.

    Returns:
        Integer days remaining (negative if expired), or None if cert not found.
    """
    try:
        info = get_cert_info(domain)
        return _days_until(info.get("expires_at", ""))
    except ValueError:
        return None


def copy_cert(domain: str, dest_dir: str) -> dict:
    """Copy certificate files to a target directory.

    Copies the fullchain, key, and CA certificate files.

    Args:
        domain: The domain name.
        dest_dir: Destination directory path.

    Returns:
        A dict with paths to the copied files.
    """
    paths = get_cert_paths(domain)
    target = Path(dest_dir)
    target.mkdir(parents=True, exist_ok=True)

    copied = {}
    for label, src in paths.items():
        src_path = Path(src)
        if src_path.is_file():
            dst = target / src_path.name
            shutil.copy2(str(src_path), str(dst))
            copied[label] = str(dst)
        else:
            logger.warning("Source %s (%s) not found, skipping", label, src)

    return {
        "domain": domain,
        "dest_dir": str(target),
        "copied": copied,
        "failed": [k for k in paths if k not in copied],
    }


def get_cert_paths(domain: str) -> dict:
    """Return the file paths for all certificate components.

    Args:
        domain: The domain name.

    Returns:
        Dict with keys 'cert', 'key', 'ca', 'fullchain' mapped to paths.
    """
    acme_home = str(Path.home() / ".acme.sh" / domain)
    return {
        "cert": f"{acme_home}/{domain}.cert",
        "key": f"{acme_home}/{domain}.key",
        "ca": f"{acme_home}/ca.cer",
        "fullchain": f"{acme_home}/fullchain.cer",
    }


def setup_nginx_install(domain: str) -> None:
    """Configure acme.sh to automatically install certs for nginx.

    Sets up a post-hook so that nginx-specific files are copied to
    /etc/ssl/certs and /etc/ssl/private after each (re)issue, followed
    by an nginx reload.

    Args:
        domain: The domain name.
    """
    cert_dest = f"/etc/ssl/certs/{domain}"
    key_dest = f"/etc/ssl/private/{domain}.key"

    args: list[str] = [
        "--install-cert",
        "-d",
        domain,
        "--cert-file",
        cert_dest,
        "--key-file",
        key_dest,
        "--ca-file",
        f"/etc/ssl/certs/{domain}-ca.crt",
        "--fullchain-file",
        f"/etc/ssl/certs/{domain}-fullchain.crt",
        "--reloadcmd",
        "sudo nginx -t && sudo systemctl reload nginx",
    ]

    _run_acme(args)
    logger.info("nginx auto-install configured for %s", domain)


# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------


def _parse_list_output(raw: str) -> list[dict]:
    """Parse the text output from ``acme.sh --list`` into a list of dicts.

    Each line in the output contains ``Key:Value`` tokens separated by
    whitespace, e.g.::

        Main_Domain:example.com SAN_Domain:www.example.com CA:Let's
        Encrypt Certificate_Date:2026-04-01 Certificate_Expired:No

    Keys are converted to lowercase in the returned dicts.
    """
    entries: list[dict] = []
    for line in raw.strip().splitlines():
        line = line.strip()
        if not line:
            continue
        entry: dict[str, str] = {}
        for token in line.split():
            if ":" not in token:
                continue
            key, _, value = token.partition(":")
            entry[key.lower()] = value
        if entry:
            entries.append(entry)
    return entries


def _days_until(date_str: str) -> int | None:
    """Parse an ISO date string and return days until that date from now."""
    if not date_str:
        return None
    for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S", "%Y%m%d%H%M%z"):
        try:
            dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC)
            delta = dt - datetime.now(UTC)
            return delta.days
        except ValueError:
            continue
    return None


def _has_auto_renew(domain: str) -> bool:
    """Check whether a domain has a scheduled cron renewal.

    The README states the cron entry format is:
      0 0 * * * "~/.acme.sh"/acme.sh --cron --home "~/.acme.sh" > /dev/null
    A per-domain ``{domain}.conf`` file existing under ``~/.acme.sh/``
    indicates the domain is being tracked by the cron job.
    """
    acme_home = Path.home() / ".acme.sh"

    # The cron job iterates all domains tracked in ~/.acme.sh/; if the
    # per-domain config exists, the cron will pick it up.
    domain_conf = acme_home / f"{domain}.conf"
    if domain_conf.is_file():
        return True

    # Fallback: check crontab -l for the domain.
    try:
        result = subprocess.run(
            ["sudo", "crontab", "-l"],
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode == 0 and "--cron" in result.stdout:
            return True
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass

    return False