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.

    Checks account.conf first (acme.sh registered account), then falls
    back to the declarative acme config.
    """
    return _read_acme_email()


def _read_acme_email() -> str:
    """Read ACME email from account.conf, falling back to declarative config."""
    try:
        acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
        for conf_name in (".account.conf", "account.conf"):
            account_conf = acme_home / conf_name
            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 config: %s", exc)
    # Fallback: read from declarative ACME config
    try:
        from lib.common import load_json

        acme_cfg = PROJECT_DIR / "config" / "acme" / "config.json"
        conf = load_json(acme_cfg)
        if conf and "email" in conf:
            return conf["email"]
    except (OSError, ValueError, KeyError):
        pass
    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_domains", "").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("renew", ""))
        auto = _has_auto_renew(main)

        certs.append(
            {
                "domain": main,
                "issuer": entry.get("ca", ""),
                "expiry": entry.get("renew", ""),
                "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("created", ""),
                "expires_at": entry.get("renew", ""),
                "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 _split_line(line: str, separator: str | None) -> list[str]:
    """Split a line by *separator*, falling back to whitespace for column output."""
    if separator in line:
        return line.split(separator)
    return line.split()


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

    Handles three formats depending on system capabilities:
    - Raw pipe-separated output (``|``)
    - Tab-separated output (when ``column`` is unavailable)
    - Column-aligned output (when ``column`` is available)

    All formats share the same header: Main_Domain, KeyLength, SAN_Domains,
    Profile, CA, Created, Renew.
    """
    lines = raw.strip().splitlines()

    if len(lines) < 2:
        return []

    header_line = lines[0]

    # Detect separator from header: pipe, tab, or whitespace
    if "|" in header_line:
        headers = _split_line(header_line, "|")
        separator = "|"
    elif "\t" in header_line:
        headers = _split_line(header_line, "\t")
        separator = "\t"
    else:
        headers = _split_line(header_line, None)  # whitespace
        separator = None

    if "Main_Domain" not in headers:
        raise ValueError(
            f"acme.sh --list output is not in expected format: {header_line!r}"
        )

    entries: list[dict] = []
    for line in lines[1:]:
        line = line.strip()
        if not line:
            continue
        fields = _split_line(line, separator)
        entry: dict[str, str] = {}
        for i, h in enumerate(headers):
            if i < len(fields):
                entry[h.lower()] = fields[i].strip().strip('"')
        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",
]