Fix dashboard template bugs, acme date parsing, wireguard sudoers match, and stale docs

- dashboard.html: Fix zones, leases, wg_status, cert key names, add services var
- server.py: Pass services to dashboard template via _get_service_status()
- lib/acme.py: Fix dead third date format (%Y%m%d%H%M%z) using astimezone(UTC)
- lib/wireguard.py: Add -- separator to cp command to match sudoers rule
- lib/nginx.py: Replace shallow dict.copy() with {**...} for DEFAULT_SSL
- AGENTS.md: Update test count 149 -> 154
- docs/api.md: Rename cert field expiry -> expires_at
This commit is contained in:
2026-05-08 19:11:54 +00:00
parent e2f56b8cc8
commit 65741644a3
26 changed files with 384 additions and 200 deletions
+40 -61
View File
@@ -2,7 +2,8 @@
ACME certificate manager for Vacuum Wall.
Wraps acme.sh to issue, renew, and manage SSL/TLS certificates
from Let's Encrypt (or other ACME providers).
from Let's Encrypt (or other ACME providers). acme.sh runs as the
vacuum-wall system user; nginx is reloaded via a deploy hook script.
"""
import logging
@@ -22,6 +23,9 @@ _ACME_ENVIRON = {
),
}
PROJECT_DIR = Path("/home/wall/vacuum-wall")
_DEPLOY_HOOK = str(PROJECT_DIR / "system" / "acme-deploy.sh")
def _find_acme() -> str:
"""Locate the acme.sh binary on the system.
@@ -58,10 +62,12 @@ def _find_acme() -> str:
def _run_acme(args: list[str]) -> str:
"""Execute acme.sh with the given arguments.
"""Execute acme.sh with the given arguments (as the current user).
Runs the command as root via sudo because standalone / webroot
validation often requires binding to privileged ports (80/443).
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.
@@ -76,7 +82,6 @@ def _run_acme(args: list[str]) -> str:
acme_bin = _find_acme()
cmd: list[str] = [
"sudo",
acme_bin,
"--home",
str(Path.home() / ".acme.sh"),
@@ -137,13 +142,12 @@ def get_email() -> str:
return ""
def issue(domain: str, webroot: str | None = None, standalone: bool = False) -> dict:
def issue(domain: str, webroot: str | None = None) -> dict:
"""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.
standalone: If True, use standalone TCP validation (binds port 80).
Returns:
A dict with 'success', 'domain', 'message', 'output', and 'error'.
@@ -152,8 +156,6 @@ def issue(domain: str, webroot: str | None = None, standalone: bool = False) ->
if webroot:
args.extend(["--webroot", webroot])
elif standalone:
args.append("--standalone")
email = get_email()
if email:
@@ -162,6 +164,7 @@ def issue(domain: str, webroot: str | None = None, standalone: bool = False) ->
try:
output = _run_acme(args)
deploy(domain)
return {
"success": True,
"domain": domain,
@@ -399,37 +402,27 @@ def get_cert_paths(domain: str) -> dict:
}
def setup_nginx_install(domain: str) -> None:
"""Configure acme.sh to automatically install certs for nginx.
def deploy(domain: str) -> None:
"""Register the deploy hook for a domain.
Sets up a post-hook so that nginx-specific files are copied to
/etc/ssl/certs and /etc/ssl/private after each (re)issue, followed
by an nginx reload.
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.
"""
cert_dest = f"/etc/ssl/certs/{domain}"
key_dest = f"/etc/ssl/private/{domain}.key"
args: list[str] = [
"--install-cert",
"-d",
domain,
"--cert-file",
cert_dest,
"--key-file",
key_dest,
"--ca-file",
f"/etc/ssl/certs/{domain}-ca.crt",
"--fullchain-file",
f"/etc/ssl/certs/{domain}-fullchain.crt",
"--reloadcmd",
"sudo nginx -t && sudo systemctl reload nginx",
]
_run_acme(args)
logger.info("nginx auto-install configured for %s", domain)
_run_acme(
[
"--deploy",
"-d",
domain,
"--deploy-hook",
_DEPLOY_HOOK,
]
)
logger.info("Deploy hook registered for %s", domain)
# ------------------------------------------------------------------
@@ -468,43 +461,29 @@ 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", "%Y%m%d%H%M%z"):
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 a scheduled cron renewal.
"""Check whether a domain has automatic renewal configured.
The README states the cron entry format is:
0 0 * * * "~/.acme.sh"/acme.sh --cron --home "~/.acme.sh" > /dev/null
A per-domain ``{domain}.conf`` file existing under ``~/.acme.sh/``
indicates the domain is being tracked by the cron job.
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 = Path.home() / ".acme.sh"
# The cron job iterates all domains tracked in ~/.acme.sh/; if the
# per-domain config exists, the cron will pick it up.
domain_conf = acme_home / f"{domain}.conf"
if domain_conf.is_file():
return True
# Fallback: check crontab -l for the domain.
try:
result = subprocess.run(
["sudo", "crontab", "-l"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0 and "--cron" in result.stdout:
return True
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
return False
return bool(domain_conf.is_file())
+35 -7
View File
@@ -491,13 +491,32 @@ def remove_forward_port(
# ---------------------------------------------------------------------------
def _parse_forward_ports(value: str) -> list[str]:
"""Parse the 'forward-ports' line into individual forward-port specifiers.
def _parse_forward_port(raw: str) -> dict[str, Any]:
"""Parse a single forward-port specifier into a structured dict.
Multiple entries are space-separated; each looks like
``port=443/proto=tcp/toaddr=192.168.1.5/toport=8080``.
Input: ``port=443/proto=tcp/toaddr=192.168.1.5/toport=8080``
"""
return value.split() if value else []
result: dict[str, Any] = {}
for piece in raw.split("/"):
if "=" not in piece:
continue
key, _, val = piece.partition("=")
if key == "port":
result["port"] = int(val)
elif key == "proto":
result["proto"] = val
elif key == "toaddr":
result["toaddr"] = val
elif key == "toport":
result["toport"] = int(val)
return result
def _parse_forward_ports(value: str) -> list[dict[str, Any]]:
"""Parse the 'forward-ports' line into a list of structured dicts."""
if not value:
return []
return [_parse_forward_port(raw) for raw in value.split()]
# ---------------------------------------------------------------------------
@@ -594,14 +613,23 @@ def restore_backup(state: dict[str, Any]) -> None:
if zinfo.get("masquerade"):
set_masquerade(zone_name, True)
# Forward ports (stored as raw strings in zinfo)
# Forward ports (stored as dicts, or raw strings from old backups)
for fp in zinfo.get("forward-ports", []):
if isinstance(fp, str):
fp_str = fp
else:
parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
if "toaddr" in fp:
parts.append(f"toaddr={fp['toaddr']}")
if "toport" in fp:
parts.append(f"toport={fp['toport']}")
fp_str = "/".join(parts)
_run(
[
"sudo",
"firewall-cmd",
f"--zone={zone_name}",
f"--add-forward-port={fp}",
f"--add-forward-port={fp_str}",
"--permanent",
],
check=False,
+1 -1
View File
@@ -43,7 +43,7 @@ DEFAULT_SSL = {
DEFAULT_CONFIG = {
"domains": {},
"management": None,
"ssl": DEFAULT_SSL.copy(),
"ssl": {**DEFAULT_SSL},
}
+1 -1
View File
@@ -147,7 +147,7 @@ def apply() -> None:
with open(local_tmp, "w") as f:
f.write(conf_text)
os.chmod(local_tmp, 0o600)
_run(["cp", str(local_tmp), WG_CONF_PATH])
_run(["cp", "--", str(local_tmp), WG_CONF_PATH])
_run(["chown", "root:root", WG_CONF_PATH], check=False)
local_tmp.unlink(missing_ok=True)