vacuum-wall/daemon/collectors/acme.py

"""ACME state collector."""

import logging
import os
from pathlib import Path
from typing import Any

from lib import schema
from lib.common import load_json
from lib.state import PROJECT_DIR, _now_iso, register_collector

logger = logging.getLogger(__name__)

_CA_NAME_MAP: dict[str, str] = {
    "letsencrypt": "Let's Encrypt",
    "zerossl": "ZeroSSL",
}


def _resolve_ca_name(ca_server: str) -> str:
    """Map a CA server identifier to its human-readable name.

    Uses prefix matching sorted by longest prefix first to avoid
    shorter prefixes winning (e.g. "letsencrypt" matching before
    "letsencrypt.org").

    Args:
        ca_server: Raw CA server string from acme.sh config.

    Returns:
        Human-readable name, or unchanged string if no match.
    """
    for prefix, name in sorted(
        _CA_NAME_MAP.items(), key=lambda x: len(x[0]), reverse=True
    ):
        if ca_server.startswith(prefix):
            return name
    return ca_server


def _parse_account_conf(acme_home: Path | None = None) -> dict[str, Any]:
    """Parse acme.sh account information and return account status dict.

    Checks three sources in order:
      1. Legacy ``.account.conf`` file (acme.sh v2.x format)
      2. Declarative ``config/acme/config.json`` (saved by the registration
         handler with ``email`` and ``ca`` fields)

    Args:
        acme_home: Optional override for ACME home directory. Falls back
            to ``ACME_HOME`` env var or ``PROJECT_DIR/data/acme``.

    Returns:
        Dict with ``registered``, ``email``, ``ca``, and
        ``key_length`` keys.  If no account is found, ``registered`` is
        ``False`` with empty / ``None`` values.
    """
    if acme_home is None:
        acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme"))
        acme_home = Path(acme_home_env)

    default = {
        "registered": False,
        "email": "",
        "ca": "",
        "key_length": None,
    }

    # 1. Legacy .account.conf (acme.sh v2.x)
    account_path = acme_home / ".account.conf"
    if account_path.is_file():
        try:
            text = account_path.read_text()
        except OSError:
            pass
        else:
            email = ""
            ca_raw = ""
            key_length = None
            for line in text.splitlines():
                if line.startswith("ACME_LEEMAIL="):
                    email = line.split("=", 1)[1].strip().strip("'\"")
                elif line.startswith("ACME_MCA="):
                    ca_raw = line.split("=", 1)[1].strip().strip("'\"")
                elif line.startswith("ACME_CERTKEYSIZE="):
                    raw_val = line.split("=", 1)[1].strip().strip("'\"")
                    key_length = int(raw_val) if raw_val.isdigit() else None
            if email and ca_raw:
                return {
                    "registered": True,
                    "email": email,
                    "ca": _resolve_ca_name(ca_raw),
                    "key_length": key_length,
                }

    # 2. Declarative config (saved by register_account / set_email handlers)
    #    Modern acme.sh (v3.x) stores account data in per-CA JSON files
    #    (ca/<server>/account.json) — we can't reliably parse those without
    #    walking the directory, so fall back to the declarative config
    #    which the handlers keep in sync.
    #    Derive project root from acme_home (acme_home is at <root>/data/acme).
    try:
        project_root = acme_home.parent.parent  # data/acme → data → project root
        acme_cfg = project_root / "config" / "acme" / "config.json"
        data = load_json(acme_cfg)
        email = (data.get("email") or "").strip()
        ca_raw = (data.get("ca") or "").strip()
        if email and ca_raw:
            return {
                "registered": True,
                "email": email,
                "ca": _resolve_ca_name(ca_raw),
                "key_length": None,
            }
    except (OSError, ValueError):
        pass

    return default


def _get_acme_email() -> str:
    """Read the ACME ``acme.sh`` email from the account config file.

    Falls back to the declarative ACME config (config/acme/config.json)
    if acme.sh account has not been registered yet.
    """
    from lib.acme import _read_acme_email

    return _read_acme_email()


def _collect_acme() -> schema.AcmeState:
    """Collect ACME certificate list and email.

    Returns:
        Dict containing certificate details and registered email.
    """
    email = _get_acme_email()

    # Non-fatal: a broken acme.sh (e.g. unreadable account.conf after an
    # ownership flip) must not blank the whole dashboard via a cleared
    # state store. Collect what we can and surface the failure in
    # `status.error` so the poll diff still detects recovery.
    cert_error: str | None = None
    try:
        from lib.acme import list_certs

        certs = list_certs()
    except Exception as exc:
        logger.warning("ACME state collection failed", exc_info=True)
        certs = []
        cert_error = str(exc)

    account = _parse_account_conf()

    return {
        "certs": certs,
        "email": email,
        "account": account,
        "status": {"error": cert_error},
        "timestamp": _now_iso(),
    }


register_collector("acme", _collect_acme)