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"
# acme.sh resolves deploy hooks from $ACME_HOME/deploy/ -- _findHook
# only searches the deploy subdirectory, never accepts absolute paths.
_DEPLOY_HOOK = "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,
        # Append the full transcript to $ACME_HOME/acme.sh.log so manual
        # runs (whose stdout is captured below) leave a persistent record
        # of the raw CA exchange. Last on purpose: acme.sh treats the next
        # token after --log as its optional file argument, so a trailing
        # --log defaults the log to $LE_CONFIG_HOME/acme.sh.log and can
        # never swallow a real argument.
        "--log",
    ]

    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
    # Derive project root from acme_home (acme_home is at <root>/data/acme).
    try:
        from lib.common import load_json

        project_root = acme_home.parent.parent
        acme_cfg = project_root / "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", "--listraw"])
    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() and d.strip().lower() != "no"
        ]

        cert_dir = find_cert_dir(main, acme_home)
        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 find_cert_dir(domain: str, acme_home: Path | None = None) -> Path:
    """Find the certificate directory for a domain.

    acme.sh may name the directory {domain}/ (RSA) or {domain}_ecc/ (ECC).
    Checks both and returns whichever exists. Falls back to {domain}/ if
    neither exists (preserves original behaviour for forward compatibility).

    Args:
        domain: The domain name.
        acme_home: Override ACME home directory. Defaults to _ACME_HOME.

    Returns:
        Path to the directory containing the cert files.
    """
    if acme_home is None:
        acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
        acme_home = Path(acme_home_env)

    ecc_dir = acme_home / f"{domain}_ecc"
    rsa_dir = acme_home / domain

    if ecc_dir.is_dir():
        return ecc_dir
    if rsa_dir.is_dir():
        return rsa_dir
    return rsa_dir


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.
    """
    cert_dir = str(find_cert_dir(domain))
    return {
        "cert": f"{cert_dir}/{domain}.cert",
        "key": f"{cert_dir}/{domain}.key",
        "ca": f"{cert_dir}/ca.cer",
        "fullchain": f"{cert_dir}/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 is not None and 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:
        # Column-aligned: use position-based parsing via helper
        return _parse_column_aligned(header_line, lines[1:])

    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 _find_header_positions(header_line: str):
    """Find start positions of each header word in a column-aligned header."""
    names: list[str] = []
    starts: list[int] = []
    i = 0
    while i < len(header_line):
        while i < len(header_line) and header_line[i] == " ":
            i += 1
        j = i
        while j < len(header_line) and header_line[j] != " ":
            j += 1
        if j > i:
            names.append(header_line[i:j])
            starts.append(i)
        i = j
    return names, starts


def _parse_column_aligned(header_line: str, data_lines: list[str]) -> list[dict]:
    """Parse column-aligned output using header positions to locate fields.

    Unlike simple whitespace splitting, this preserves empty fields by using
    character positions rather than token counts. Empty columns (e.g. missing
    Profile or SAN_Domains) are correctly handled.
    """
    names, starts = _find_header_positions(header_line)

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

    # Column ends: midpoint before next header starts (or end of line for last)
    ends: list[int] = len(starts) * [len(header_line)]
    for i in range(len(starts) - 1):
        ends[i] = (starts[i] + starts[i + 1]) // 2

    entries: list[dict] = []
    for line in data_lines:
        line = line.rstrip()
        if not line.strip():
            continue
        entry: dict[str, str] = {}
        for k, name in enumerate(names):
            s, e = starts[k], ends[k]
            cell = line[s:e] if len(line) > s else ""
            entry[name.lower()] = cell.strip().strip('"')
        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:%SZ",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%dT%H:%M:%S%z",
    ):
        try:
            dt = datetime.strptime(date_str, fmt)
            dt = dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt.astimezone(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",
    "find_cert_dir",
    "get_cert_info",
    "get_cert_paths",
    "get_email",
    "get_expiry",
    "is_expired",
    "issue",
    "list_certs",
    "remove",
    "renew",
    "set_email",
]