Files
vacuum-wall/lib/acme.py
T
mteehan 78fcb01877 fix: install.sh loop abort, ACME poll sudo gate, /static/ sub-paths
- install.sh: the traversal-chmod loop assigned _d but looped over the
  never-set $d; under set -u every fresh install aborted with
  "d: unbound variable" at that line. Loop over $_d.
- acme collector: the self-heal normalize (sudo chmod g+rwX) now runs
  only when a no-sudo group-read-bit probe detects a lost bit — acme.sh
  re-hardens the tree 600 on every run, so the steady-state poll makes
  no sudo call. The group bit (not daemon readability) is what the
  two-user model keeps for the WebUI user.
- lib.acme: new get_acme_home() accessor (ACME_HOME env, default
  data/acme), reused by _run_acme; _summarize_acme_output preserves a
  "Permission denied" line even when it is not among the final two, so
  the collector's actionable-error matcher keeps firing.
- nginx template: emit location /static/ for any is_management path
  (not only '/'); the SPA references /static/... at the domain root
  regardless of the management backend path.
- tests: probe, summarizer, and nginx-subpath cases in
  test_state.py, test_acme.py, test_nginx.py.
2026-09-05 00:38:34 +00:00

677 lines
20 KiB
Python

"""
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 get_acme_home() -> Path:
"""Resolve the ACME home directory (``ACME_HOME`` env, default ``data/acme``)."""
return Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
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 = str(get_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. The log file is passed explicitly (never
# as a bare trailing --log): a valueless trailing --log makes
# acme.sh's arg loop double-shift under dash (the --log branch
# shifts once, then the loop's trailing `shift 1` runs with zero
# positional params) and fails with "shift: can't shift that many"
# (exit 2). The explicit path keeps the same default destination
# ($LE_CONFIG_HOME/acme.sh.log) and can never swallow a real arg.
"--log",
str(Path(acme_home_env) / "acme.sh.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}: "
f"{_summarize_acme_output(output)}"
)
return output
def _summarize_acme_output(output: str) -> str:
"""Reduce raw acme.sh output to a concise, human-readable summary.
acme.sh prints timestamped transcript lines; the failure reason is
in the final lines (e.g. "The retryafter=86400 value is too large
(> 600), will not retry anymore."). Strips per-line timestamps and
the "Please check log file" pointer so the summary stays toast-
sized. A "Permission denied" diagnostic is preserved even when it
is not among the final lines — the actionable-error matcher in
daemon/collectors/acme.py keys off it. The full transcript remains
in the log and acme.sh.log.
"""
lines = [line.strip() for line in output.strip().splitlines() if line.strip()]
lines = [re.sub(r"^\[[^\]]*\] ", "", line) for line in lines]
lines = [line for line in lines if not line.startswith("Please check log file")]
if not lines:
return "(no output)"
tail = list(lines[-2:])
for line in reversed(lines):
if "Permission denied" in line and line not in tail:
tail.insert(0, line)
break
return "; ".join(tail)
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_acme_home",
"get_cert_info",
"get_cert_paths",
"get_email",
"get_expiry",
"is_expired",
"issue",
"list_certs",
"remove",
"renew",
"set_email",
]