vacuum-wall/lib/acme.py

"""
ACME certificate manager for Vacuum Wall.

Wraps acme.sh to issue, renew, and manage SSL/TLS certificates
from ACME providers such as ZeroSSL or Let's Encrypt.  acme.sh runs as the
vacuum-wall system user; nginx is reloaded via a deploy hook script.
"""

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

logger = logging.getLogger(__name__)

PROJECT_DIR = Path(__file__).resolve().parent.parent
_ACME_HOME = PROJECT_DIR / "data" / "acme"
_DEPLOY_HOOK = str(PROJECT_DIR / "system" / "acme-deploy.sh")

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

_WEBROOT = PROJECT_DIR / "data" / "acme" / "www"


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 = [
        _ACME_HOME / "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 (as the current user).

    acme.sh does not need root for most operations. Only standalone
    and TLS-ALPN validation modes require binding to privileged ports,
    which are not used by Vacuum Wall (webroot validation is used
    instead).

    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()

    # Check for ACME_HOME env var (set by systemd in production)
    acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))

    cmd: list[str] = [
        acme_bin,
        "--home",
        acme_home_env,
        "--config-home",
        acme_home_env,
        *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:
        acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
        account_conf = acme_home / "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, email: str | None = None) -> str:
    """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.
        email: Contact email. Falls back to configured ACME email if not given.

    Returns:
        Combined stdout from the acme.sh command.

    Raises:
        RuntimeError: If issuance fails.
    """
    args: list[str] = ["--issue", "-d", domain]
    args.extend(["--webroot", webroot or str(_WEBROOT)])
    contact = email or get_email()
    if contact:
        args.extend(["-m", contact])
    args.append("--force")
    output = _run_acme(args)
    deploy(domain)
    logger.info("Certificate for %s issued successfully", domain)
    return output.strip()


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

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

    Returns:
        Combined stdout from the acme.sh command.

    Raises:
        RuntimeError: If renewal fails.
    """
    args: list[str] = ["--renew", "-d", domain]
    if force:
        args.append("--force")
    output = _run_acme(args)
    deploy(domain)
    logger.info("Certificate for %s renewed successfully", domain)
    return output.strip()


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_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
    acme_home = Path(acme_home_env)

    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,
                "issuer": entry.get("CA", ""),
                "expiry": entry.get("certificate_expires", ""),
                "days_remaining": days,
                "expired": days is not None and days <= 0,
                "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_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
    acme_home = str(Path(acme_home_env) / 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 deploy(domain: str) -> None:
    """Register the deploy hook for a domain.

    Tells acme.sh to run the Vacuum Wall deploy script after every
    successful issue or renewal.  The hook fires automatically on
    future renewals as well, so this only needs to be called once per
    domain.

    Args:
        domain: The domain name.
    """
    _run_acme(
        [
            "--deploy",
            "-d",
            domain,
            "--deploy-hook",
            _DEPLOY_HOOK,
        ]
    )
    logger.info("Deploy hook registered 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"):
        try:
            dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC)
            delta = dt - datetime.now(UTC)
            return delta.days
        except ValueError:
            continue
    try:
        dt = datetime.strptime(date_str, "%Y%m%d%H%M%z").astimezone(UTC)
        delta = dt - datetime.now(UTC)
        return delta.days
    except ValueError:
        pass
    return None


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

    acme.sh tracks certificates in per-domain ``{domain}.conf`` files
    under ``~/.acme.sh/``; existence of this file means the systemd
    timer's ``--cron`` run will pick it up.
    """
    acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
    domain_conf = Path(acme_home_env) / f"{domain}.conf"
    return bool(domain_conf.is_file())


__all__ = [
    "copy_cert",
    "days_until_expiry",
    "deploy",
    "get_cert_info",
    "get_cert_paths",
    "get_email",
    "get_expiry",
    "is_expired",
    "issue",
    "list_certs",
    "remove",
    "renew",
    "set_email",
]