mteehan
8feb56faf6
- Add find_cert_dir() to resolve both RSA and ECC (domain_ecc/) cert dirs - Copy acme deploy hook to /deploy/ where acme.sh resolves it - _parse_account_conf checks both legacy .account.conf and declarative config - Skip public DNS check when all local IPs are private (NAT) - Improve check message strings for validity and expiry status - Support timezone-aware date formats in _days_until parsing - Filter out "no" SAN domains in cert listing - Bump frontend asset version cache keys - Fix DOMContentLoaded race condition in app.js boot - Fix spread operator in certs.js modal template
1094 lines
35 KiB
Python
1094 lines
35 KiB
Python
"""ACME certificate daemon handler."""
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import urllib.request
|
|
from contextlib import suppress
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
import lib.acme
|
|
import lib.common as lib_common
|
|
from daemon.iface import (
|
|
DELETE_ACME_ACCOUNT_DEACTIVATE,
|
|
DELETE_ACME_REMOVE,
|
|
GET_ACME_ACCOUNT,
|
|
GET_ACME_EMAIL,
|
|
GET_ACME_INFO,
|
|
GET_ACME_ISSUE_STATUS,
|
|
GET_ACME_LIST,
|
|
GET_ACME_PATHS,
|
|
POST_ACME_ACCOUNT_REGISTER,
|
|
POST_ACME_EMAIL,
|
|
POST_ACME_ISSUE,
|
|
POST_ACME_RENEW,
|
|
POST_ACME_SELF_SIGNED,
|
|
POST_ACME_VALIDATE,
|
|
)
|
|
from daemon.server import ConflictError, NotFoundError, refresh_state, registry
|
|
from lib.acme import _run_acme
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parent.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"
|
|
|
|
# In-memory store for active issuance requests.
|
|
_ISSUANCES: dict[str, "IssueRequest"] = {}
|
|
_ISSUANCE_TTL = 300 # seconds to keep completed requests
|
|
|
|
|
|
def _find_issuance(domain: str) -> "IssueRequest | None":
|
|
"""Find an active (running) issuance request by domain."""
|
|
for req in _ISSUANCES.values():
|
|
if req.domain == domain and req.status == "running":
|
|
return req
|
|
return None
|
|
|
|
|
|
@dataclass
|
|
class IssueStep:
|
|
"""Single step in a certificate issuance workflow.
|
|
|
|
Attributes:
|
|
name: Machine-readable step identifier (e.g. "issue").
|
|
label: Human-readable description shown to the user.
|
|
status: Current state: "pending", "running", "done", or "error".
|
|
message: Optional detail or error message for the step.
|
|
"""
|
|
|
|
name: str
|
|
label: str
|
|
status: str = "pending"
|
|
message: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class IssueRequest:
|
|
"""Tracked certificate issuance request.
|
|
|
|
Attributes:
|
|
request_id: Unique hex identifier for polling.
|
|
domain: Target domain for the certificate.
|
|
email: Optional ACME contact email.
|
|
webroot: Optional custom webroot path.
|
|
steps: Ordered list of issuance steps.
|
|
status: Overall status: "running", "completed", or "failed".
|
|
created_at: Unix timestamp when request was created.
|
|
expires_at: Unix timestamp when entry expires from store.
|
|
"""
|
|
|
|
request_id: str
|
|
domain: str
|
|
email: str | None = None
|
|
webroot: str | None = None
|
|
steps: list[IssueStep] = field(default_factory=list)
|
|
status: str = "running"
|
|
created_at: float = field(default_factory=lambda: datetime.now(UTC).timestamp())
|
|
expires_at: float | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Serialize request to a JSON-compatible dictionary."""
|
|
return {
|
|
"request_id": self.request_id,
|
|
"domain": self.domain,
|
|
"status": self.status,
|
|
"steps": [
|
|
{
|
|
"name": s.name,
|
|
"label": s.label,
|
|
"status": s.status,
|
|
"message": s.message,
|
|
}
|
|
for s in self.steps
|
|
],
|
|
"created_at": self.created_at,
|
|
"expires_at": self.expires_at,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
|
|
|
|
def _find_acme_bin() -> str:
|
|
"""Return the path to the acme.sh binary."""
|
|
from lib.acme import _find_acme
|
|
|
|
return _find_acme()
|
|
|
|
|
|
def _get_acme_email() -> str:
|
|
"""Read registered contact email from ACME account config."""
|
|
from lib.acme import _read_acme_email
|
|
|
|
return _read_acme_email()
|
|
|
|
|
|
def _get_state() -> dict[str, Any] | None:
|
|
"""Return the raw ACME entry from the shared state store."""
|
|
from lib.state import state as state_store
|
|
|
|
return state_store.get("acme")
|
|
|
|
|
|
def _get_acme_state() -> dict[str, Any]:
|
|
"""Return the ACME state or empty dict when missing."""
|
|
ac = _get_state()
|
|
if ac is None:
|
|
return {}
|
|
return ac
|
|
|
|
|
|
def _clean_expired_issuances() -> None:
|
|
"""Remove completed requests older than TTL."""
|
|
now = datetime.now(UTC).timestamp()
|
|
expired = [
|
|
rid
|
|
for rid, req in _ISSUANCES.items()
|
|
if req.expires_at and now > req.expires_at
|
|
]
|
|
for rid in expired:
|
|
del _ISSUANCES[rid]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation helpers
|
|
|
|
|
|
def _check_domain_format(domain: str) -> tuple[bool, str]:
|
|
"""Validate basic domain name format."""
|
|
import re as _re
|
|
|
|
pattern = r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*$"
|
|
if not _re.match(pattern, domain):
|
|
return False, "Invalid domain name format"
|
|
return True, "Domain format is valid"
|
|
|
|
|
|
def _get_local_ips() -> set[str]:
|
|
"""Return the set of all non-loopback IPv4 addresses on this host."""
|
|
import struct
|
|
from fcntl import ioctl
|
|
|
|
ips: set[str] = set()
|
|
with suppress(OSError):
|
|
ips.add(socket.gethostbyname(socket.gethostname()))
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
names = b"\x00" * 4096
|
|
raw = ioctl(s.fileno(), 0x8912, names)
|
|
s.close()
|
|
for i in range(0, 4096, 32):
|
|
name = raw[i : i + 16].split(b"\x00")[0].decode()
|
|
if name == "lo":
|
|
continue
|
|
addr = struct.unpack("<I", raw[i + 16 : i + 20])[0]
|
|
ips.add(str(ipaddress.IPv4Address(addr)))
|
|
except Exception:
|
|
pass
|
|
return ips
|
|
|
|
|
|
def _get_external_ip(timeout: int = 5) -> str | None:
|
|
"""Fetch the server's public IP address from external services.
|
|
|
|
Returns None on error or timeout. Honours VACUUM_WALL_EXTERNAL_IP_URL
|
|
env var for testing or custom providers.
|
|
"""
|
|
urls: list[str] = []
|
|
|
|
custom_url = os.environ.get("VACUUM_WALL_EXTERNAL_IP_URL")
|
|
if custom_url:
|
|
urls.append(custom_url)
|
|
else:
|
|
urls.append("https://api.ipify.org")
|
|
urls.append("https://checkip.amazonaws.com")
|
|
|
|
for url in urls:
|
|
try:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "vacuum-wall/1.0"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return resp.read().decode().strip()
|
|
except Exception:
|
|
continue
|
|
|
|
return None
|
|
|
|
|
|
def _is_private_ip(ip_str: str) -> bool:
|
|
"""Return True if the IP address is not globally routable."""
|
|
try:
|
|
return not ipaddress.ip_address(ip_str).is_global
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _check_dns_resolves(domain: str) -> tuple[bool, str]:
|
|
"""Check that domain resolves to this machine's IP (NAT-aware).
|
|
|
|
1. Match against local interface IPs — pass immediately.
|
|
2. If no local match, compare against external IP for NAT scenarios.
|
|
3. If ext IP lookup fails, downgrade to non-blocking warning.
|
|
4. Private-range resolved IP always fails.
|
|
"""
|
|
try:
|
|
results = socket.getaddrinfo(domain, 80, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
|
if not results:
|
|
return False, "Domain does not resolve to any address"
|
|
|
|
resolved_ips = [addr[4][0] for addr in results]
|
|
local_ips = _get_local_ips()
|
|
|
|
# Step 1: direct local match
|
|
for rip in resolved_ips:
|
|
if rip in local_ips:
|
|
return True, "DNS resolves correctly"
|
|
|
|
# Step 2: NAT — compare against external IP
|
|
external_ip = _get_external_ip()
|
|
if external_ip:
|
|
for rip in resolved_ips:
|
|
if rip == external_ip:
|
|
return (
|
|
True,
|
|
"DNS resolves correctly (matches external IP — server is behind NAT)",
|
|
)
|
|
|
|
# Resolved IP is public but doesn't match external IP
|
|
for rip in resolved_ips:
|
|
if not _is_private_ip(rip):
|
|
return (
|
|
False,
|
|
f"Domain resolves to {rip} but external IP is {external_ip}. "
|
|
f"Check your DNS A record points to this server's public IP.",
|
|
)
|
|
|
|
# Step 3: external IP unavailable — check for private range first, then warn
|
|
for rip in resolved_ips:
|
|
if _is_private_ip(rip):
|
|
return (
|
|
False,
|
|
f"Domain resolves to private IP {rip}. "
|
|
f"Ensure public DNS points to this server's public IP.",
|
|
)
|
|
|
|
return (
|
|
False,
|
|
f"Domain resolves to {resolved_ips[0]}, not a local interface IP. "
|
|
f"Cannot verify via external IP (lookup failed).",
|
|
)
|
|
|
|
except socket.gaierror:
|
|
return False, "Domain does not resolve (NXDOMAIN or timeout)"
|
|
|
|
|
|
def _check_acme_installed() -> tuple[bool, str]:
|
|
"""Verify acme.sh binary is installed and executable."""
|
|
try:
|
|
_find_acme_bin()
|
|
return True, "acme.sh found"
|
|
except FileNotFoundError:
|
|
return False, "acme.sh not installed"
|
|
|
|
|
|
def _check_email_configured() -> tuple[bool, str]:
|
|
"""Check whether an ACME contact email has been configured."""
|
|
email = _get_acme_email() or ""
|
|
if email:
|
|
return True, f"Contact email configured: {email}"
|
|
return (
|
|
False,
|
|
"Contact email not set — configure in Account Settings for renewal notifications",
|
|
)
|
|
|
|
|
|
def _check_webroot() -> tuple[bool, str]:
|
|
"""Verify the ACME webroot directory exists and is writable."""
|
|
if _WEBROOT.is_dir() and os.access(str(_WEBROOT), os.W_OK):
|
|
return True, "ACME webroot ready"
|
|
return False, "ACME webroot not ready or not writable"
|
|
|
|
|
|
def _check_challenge_config() -> tuple[bool, str]:
|
|
"""Check for the ACME HTTP-01 challenge nginx config file."""
|
|
from lib.nginx import SITES_DIR
|
|
|
|
site_conf = SITES_DIR / "_acme-challenge.conf" if SITES_DIR else None
|
|
if site_conf and site_conf.is_file():
|
|
return True, "ACME challenge nginx config present"
|
|
return False, "ACME challenge nginx config missing"
|
|
|
|
|
|
def _check_existing_cert(domain: str) -> tuple[bool, str]:
|
|
"""Warn if a valid cert already exists (not blocking)."""
|
|
try:
|
|
days = lib.acme.days_until_expiry(domain)
|
|
except (RuntimeError, FileNotFoundError):
|
|
return True, "No existing certificate found"
|
|
if days is None:
|
|
return True, "No existing certificate found"
|
|
if days > 0:
|
|
return True, f"Valid certificate exists ({days} days remaining)"
|
|
return True, f"Certificate expired ({abs(days)} days ago)"
|
|
|
|
|
|
def _check_nginx_running() -> tuple[bool, str]:
|
|
"""Check whether the nginx process is currently running."""
|
|
try:
|
|
result = lib_common.run_proc(
|
|
["systemctl", "is-active", "nginx"], sudo=True, check=False, timeout=10
|
|
)
|
|
if result.stdout.strip() == "active":
|
|
return True, "nginx is running"
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
try:
|
|
pid_file = Path("/var/run/nginx.pid")
|
|
if pid_file.is_file():
|
|
pid = int(pid_file.read_text().strip())
|
|
proc_status = Path(f"/proc/{pid}/status")
|
|
if proc_status.is_file():
|
|
return True, "nginx is running"
|
|
except (ValueError, OSError):
|
|
pass
|
|
|
|
return False, "nginx is not running — start it before issuing certificates"
|
|
|
|
|
|
def _check_nginx_config() -> tuple[bool, str]:
|
|
"""Test nginx configuration syntax via ``nginx -t``."""
|
|
from lib.nginx import test_config
|
|
|
|
ok, msg = test_config()
|
|
if ok:
|
|
return True, "nginx configuration is valid"
|
|
return False, f"nginx configuration test failed: {msg}"
|
|
|
|
|
|
def _check_firewall_port_80() -> tuple[bool, str]:
|
|
"""Check that port 80/tcp is open in firewalld across all active zones."""
|
|
try:
|
|
proc = lib_common.run_proc(
|
|
["firewall-cmd", "--get-active-zones"], sudo=True, check=False, timeout=10
|
|
)
|
|
if proc.returncode != 0:
|
|
return True, "firewalld not detected, skipping port check"
|
|
|
|
zone_lines = proc.stdout.strip()
|
|
if not zone_lines:
|
|
return True, "firewalld not detected, skipping port check"
|
|
|
|
zones = _parse_active_zones(zone_lines)
|
|
port_open = False
|
|
|
|
for zone in zones:
|
|
proc = lib_common.run_proc(
|
|
["firewall-cmd", f"--zone={zone}", "--list-services"],
|
|
sudo=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
if "http" in (proc.stdout or "").split():
|
|
port_open = True
|
|
break
|
|
|
|
proc = lib_common.run_proc(
|
|
["firewall-cmd", f"--zone={zone}", "--list-ports"],
|
|
sudo=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
for item in (proc.stdout or "").split():
|
|
if "80" in item.split("/"):
|
|
port_open = True
|
|
break
|
|
|
|
if port_open:
|
|
break
|
|
|
|
if port_open:
|
|
return True, "Port 80 is open in firewall"
|
|
return (
|
|
False,
|
|
"Port 80 blocked by firewall — allow with: firewall-cmd --add-service=http --permanent && firewall-cmd --reload",
|
|
)
|
|
except Exception:
|
|
return True, "firewalld check unavailable, skipping"
|
|
|
|
|
|
def _parse_active_zones(output: str) -> list[str]:
|
|
"""Parse ``firewall-cmd --get-active-zones`` output into zone names."""
|
|
zones = []
|
|
for line in output.splitlines():
|
|
stripped = line.strip()
|
|
if stripped and not stripped.startswith(" "):
|
|
zones.append(stripped.removesuffix(" (default)"))
|
|
return zones
|
|
|
|
|
|
def _check_acme_home_writable() -> tuple[bool, str]:
|
|
"""Verify data/acme/ is writable with a temporary file probe."""
|
|
if not _ACME_HOME.is_dir():
|
|
return False, "ACME home directory does not exist"
|
|
if not os.access(str(_ACME_HOME), os.W_OK):
|
|
return False, "ACME home directory is not writable"
|
|
try:
|
|
probe = _ACME_HOME / ".write-probe"
|
|
probe.write_text("ok")
|
|
probe.unlink()
|
|
return True, "ACME home directory is writable"
|
|
except OSError:
|
|
return False, "ACME home directory is not writable"
|
|
|
|
|
|
def _check_openssl_available() -> tuple[bool, str]:
|
|
"""Verify openssl binary is available and functional."""
|
|
openssl_path = shutil.which("openssl")
|
|
if not openssl_path:
|
|
return False, "openssl binary not found"
|
|
try:
|
|
result = subprocess.run(
|
|
["openssl", "version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
if result.returncode == 0:
|
|
ver = result.stdout.strip()
|
|
return True, f"openssl available ({ver})"
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
return False, "openssl is not working"
|
|
|
|
|
|
def _check_port_80_listening() -> tuple[bool, str]:
|
|
"""Check that something is listening on port 80 (IPv4 or IPv6)."""
|
|
# Check localhost first
|
|
for af, host in [(socket.AF_INET, "127.0.0.1"), (socket.AF_INET6, "::1")]:
|
|
with suppress(OSError), socket.socket(af, socket.SOCK_STREAM) as s:
|
|
s.settimeout(2)
|
|
if s.connect_ex((host, 80)) == 0:
|
|
return True, "Port 80 is listening"
|
|
# Also check all interface IPs in case nginx only binds on a public interface
|
|
for ip in _get_local_ips():
|
|
with suppress(OSError), socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.settimeout(2)
|
|
if s.connect_ex((ip, 80)) == 0:
|
|
return True, "Port 80 is listening"
|
|
return False, "Nothing listening on port 80 — needed for ACME HTTP-01 challenge"
|
|
|
|
|
|
def _check_acme_account() -> tuple[bool, str]:
|
|
"""Non-blocking: check acme.sh account is configured.
|
|
|
|
Tries acme.sh --info first, then falls back to parsed account state
|
|
(handles both legacy .account.conf and modern declarative config).
|
|
"""
|
|
try:
|
|
acme_bin = _find_acme_bin()
|
|
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
|
|
result = subprocess.run(
|
|
[
|
|
acme_bin,
|
|
"--home",
|
|
acme_home_env,
|
|
"--config-home",
|
|
acme_home_env,
|
|
"--info",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
env={**os.environ, **_ACME_ENVIRON},
|
|
)
|
|
if result.returncode == 0:
|
|
return True, "ACME account is configured"
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
from lib.state import _parse_account_conf
|
|
|
|
info = _parse_account_conf(_ACME_HOME)
|
|
if info.get("registered"):
|
|
return True, "ACME account is configured"
|
|
|
|
return (
|
|
False,
|
|
"ACME account may need re-registration — check email is set before issuance",
|
|
)
|
|
|
|
|
|
def _check_account_registered() -> tuple[bool, str]:
|
|
"""Blocking check: verify an ACME account is registered.
|
|
|
|
Delegates to ``lib.state._parse_account_conf()`` which checks both
|
|
the legacy .account.conf and the declarative config/acme/config.json
|
|
used by modern acme.sh (v3.x).
|
|
"""
|
|
from lib.state import _parse_account_conf
|
|
|
|
info = _parse_account_conf(_ACME_HOME)
|
|
if info.get("registered"):
|
|
return True, "ACME account is registered"
|
|
return False, "Register an ACME account before issuing certificates"
|
|
|
|
|
|
def _get_account_info() -> dict[str, Any]:
|
|
"""Read and return the ACME account info dict.
|
|
|
|
Delegates to ``lib.state._parse_account_conf()`` for a single
|
|
source of truth.
|
|
"""
|
|
from lib.state import _parse_account_conf
|
|
|
|
return _parse_account_conf(_ACME_HOME)
|
|
|
|
|
|
def _check_dns_public(domain: str) -> tuple[bool, str]:
|
|
"""Non-blocking: verify public DNS resolves domain to this server."""
|
|
local_ips = _get_local_ips()
|
|
|
|
if not local_ips:
|
|
return True, "Public DNS check skipped (no local IPs detected)"
|
|
|
|
# Behind NAT: public DNS can never match local (private) IPs.
|
|
# dns_resolves already verified the domain correctly via external IP.
|
|
if all(_is_private_ip(ip) for ip in local_ips):
|
|
return True, "Public DNS check skipped (NAT detected — dns_resolves verified)"
|
|
|
|
for dns_server in ("8.8.8.8", "1.1.1.1"):
|
|
try:
|
|
result = subprocess.run(
|
|
["host", domain, dns_server],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
output = result.stdout or ""
|
|
for ip in local_ips:
|
|
for line in output.splitlines():
|
|
if ip in line.split():
|
|
return True, "Public DNS resolves correctly"
|
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
|
continue
|
|
|
|
return (
|
|
False,
|
|
"Public DNS may not resolve to this server — allow a few minutes for propagation",
|
|
)
|
|
|
|
|
|
def _validate(domain: str) -> dict[str, Any]:
|
|
"""Run all pre-checks for a domain. Returns structured results."""
|
|
checks: list[dict[str, Any]] = []
|
|
ready = True
|
|
|
|
check_fns = [
|
|
# Environment — must be present before anything else
|
|
("acme_installed", _check_acme_installed, True),
|
|
("openssl_available", _check_openssl_available, True),
|
|
("acme_home_writable", _check_acme_home_writable, True),
|
|
("account_registered", _check_account_registered, True),
|
|
("email_configured", _check_email_configured, False),
|
|
("acme_account_valid", _check_acme_account, False),
|
|
("webroot_ready", _check_webroot, True),
|
|
# Nginx stack — must serve challenges
|
|
("nginx_running", _check_nginx_running, True),
|
|
("nginx_config_valid", _check_nginx_config, True),
|
|
("challenge_configured", _check_challenge_config, True),
|
|
("port_80_listening", _check_port_80_listening, True),
|
|
("firewall_open", _check_firewall_port_80, True),
|
|
# Domain — must be reachable
|
|
("domain_format", lambda: _check_domain_format(domain), True),
|
|
("dns_resolves", lambda: _check_dns_resolves(domain), True),
|
|
("dns_public", lambda: _check_dns_public(domain), False),
|
|
# Existing cert — informational
|
|
("existing_cert", lambda: _check_existing_cert(domain), False),
|
|
]
|
|
|
|
for name, fn, blocking in check_fns:
|
|
try:
|
|
passed, msg = fn()
|
|
checks.append(
|
|
{"name": name, "passed": passed, "message": msg, "blocking": blocking}
|
|
)
|
|
if not passed and blocking:
|
|
ready = False
|
|
except Exception as exc:
|
|
checks.append(
|
|
{
|
|
"name": name,
|
|
"passed": False,
|
|
"message": str(exc),
|
|
"blocking": blocking,
|
|
}
|
|
)
|
|
if blocking:
|
|
ready = False
|
|
|
|
return {"domain": domain, "checks": checks, "ready": ready}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Routes — status reads from state, mutations call refresh_state
|
|
|
|
|
|
@registry.register(GET_ACME_LIST)
|
|
def list_certs(_request: Any, _body: Any) -> list[dict]:
|
|
"""GET /acme/list — return managed certificates."""
|
|
ac = _get_acme_state()
|
|
return ac.get("certs", [])
|
|
|
|
|
|
@registry.register(GET_ACME_INFO)
|
|
def get_cert_info(_request: Any, body: dict[str, Any] | None) -> dict:
|
|
"""GET /acme/info — return details for a single domain certificate.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
NotFoundError: When no certificate exists for domain.
|
|
"""
|
|
if not body or "domain" not in body:
|
|
raise ValueError("'domain' is required")
|
|
domain = body["domain"]
|
|
certs = list_certs(None, None)
|
|
req = _find_issuance(domain)
|
|
|
|
for c in certs:
|
|
if c["domain"] == domain or domain in c.get("san_domains", []):
|
|
result = dict(c)
|
|
if req:
|
|
result["issuance"] = req.to_dict()
|
|
return result
|
|
|
|
# No cert found — check if there's an in-progress issuance
|
|
if req:
|
|
return {
|
|
"domain": domain,
|
|
"status": "issuing",
|
|
"issuance": req.to_dict(),
|
|
}
|
|
|
|
raise NotFoundError(f"No certificate found for domain: {domain}")
|
|
|
|
|
|
@registry.register(POST_ACME_VALIDATE)
|
|
def validate_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/validate — run pre-flight checks for a domain.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = body.get("domain", "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
return _validate(domain)
|
|
|
|
|
|
@registry.register(POST_ACME_ISSUE)
|
|
async def issue_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/issue — create a new certificate issuance request.
|
|
|
|
Deduplicates in-progress requests. Spawns background task for actual issuance.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = (body.get("domain") or "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
# email is kept for backward API compatibility but ignored —
|
|
# _run_issue() uses the registered account's email instead
|
|
provided_email = (body.get("email") or "").strip()
|
|
if provided_email:
|
|
logger.warning(
|
|
"email field in issue/start is ignored, using registered account's email"
|
|
)
|
|
webroot = body.get("webroot")
|
|
|
|
_clean_expired_issuances()
|
|
|
|
# Dedup: if domain already has an active request, return it
|
|
for existing in _ISSUANCES.values():
|
|
if existing.domain == domain and existing.status == "running":
|
|
return {
|
|
"request_id": existing.request_id,
|
|
"domain": domain,
|
|
"status": "existing",
|
|
}
|
|
|
|
# Check if cert already exists — call acme.sh directly, not via state
|
|
try:
|
|
certs = lib.acme.list_certs()
|
|
except RuntimeError as exc:
|
|
raise RuntimeError(f"Cannot check existing certificates: {exc}") from exc
|
|
for c in certs:
|
|
if c["domain"] == domain or domain in c.get("san_domains", []):
|
|
days = c.get("days_until_expiry")
|
|
if days is not None and days >= 0:
|
|
raise ConflictError(
|
|
f"Certificate already exists for {domain} ({days} day{'s' if days != 1 else ''} remaining). Renew instead."
|
|
)
|
|
|
|
# Run pre-flight checks
|
|
_validate_checks = _validate(domain)
|
|
if not _validate_checks["ready"]:
|
|
failed = [
|
|
c["name"]
|
|
for c in _validate_checks["checks"]
|
|
if not c["passed"] and c["blocking"]
|
|
]
|
|
raise RuntimeError(f"Pre-flight checks failed: {', '.join(failed)}")
|
|
|
|
# Create tracked request
|
|
request_id = uuid4().hex[:12]
|
|
steps = [
|
|
IssueStep(name="issue", label="Issuing certificate"),
|
|
IssueStep(name="deploy", label="Registering deploy hook"),
|
|
IssueStep(name="refresh", label="Refreshing certificate state"),
|
|
]
|
|
|
|
req = IssueRequest(
|
|
request_id=request_id,
|
|
domain=domain,
|
|
webroot=webroot,
|
|
steps=steps,
|
|
)
|
|
_ISSUANCES[request_id] = req
|
|
|
|
# Spawn background task
|
|
_task = asyncio.create_task(_run_issue(req)) # noqa: RUF006 — task runs to completion on its own
|
|
|
|
return {"request_id": request_id, "domain": domain}
|
|
|
|
|
|
@registry.register(GET_ACME_ISSUE_STATUS)
|
|
def get_issuance_status(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""GET /acme/issue/status — poll status of an issuance request.
|
|
|
|
Raises:
|
|
ValueError: When id is missing.
|
|
NotFoundError: When request_id is unknown.
|
|
"""
|
|
request_id = (body or {}).get("id", "").strip()
|
|
if not request_id:
|
|
raise ValueError("'id' is required")
|
|
|
|
req = _ISSUANCES.get(request_id)
|
|
if not req:
|
|
raise NotFoundError(f"Issuance request {request_id} not found")
|
|
|
|
return req.to_dict()
|
|
|
|
|
|
async def _run_issue(req: IssueRequest) -> None:
|
|
"""Background task: run acme.sh steps, update step status."""
|
|
try:
|
|
# Step 1: issue
|
|
req.steps[0].status = "running"
|
|
args: list[str] = ["--issue", "-d", req.domain]
|
|
args.extend(["--webroot", req.webroot or str(_WEBROOT)])
|
|
account_email = _get_acme_email()
|
|
if account_email:
|
|
args.extend(["-m", account_email])
|
|
args.append("--force")
|
|
output = _run_acme(args)
|
|
req.steps[0].status = "done"
|
|
req.steps[0].message = output.strip()[:200]
|
|
|
|
# Step 2: deploy
|
|
req.steps[1].status = "running"
|
|
_run_acme(["--deploy", "-d", req.domain, "--deploy-hook", _DEPLOY_HOOK])
|
|
req.steps[1].status = "done"
|
|
req.steps[1].message = "Deploy hook registered"
|
|
|
|
# Step 3: refresh state
|
|
req.steps[2].status = "running"
|
|
refresh_state(["acme"])
|
|
req.steps[2].status = "done"
|
|
req.steps[2].message = "State refreshed"
|
|
|
|
req.status = "completed"
|
|
req.expires_at = datetime.now(UTC).timestamp() + _ISSUANCE_TTL
|
|
logger.info(
|
|
"Certificate for %s issued (request %s)", req.domain, req.request_id
|
|
)
|
|
except Exception as exc:
|
|
# Mark current running step as error, overall as failed
|
|
for step in req.steps:
|
|
if step.status == "running":
|
|
step.status = "error"
|
|
step.message = str(exc)
|
|
break
|
|
else:
|
|
req.steps.append(
|
|
IssueStep(name="error", label="Error", status="error", message=str(exc))
|
|
)
|
|
req.status = "failed"
|
|
req.expires_at = datetime.now(UTC).timestamp() + _ISSUANCE_TTL
|
|
logger.error("Cert issuance for %s failed: %s", req.domain, exc)
|
|
|
|
|
|
@registry.register(POST_ACME_RENEW)
|
|
def renew_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/renew — renew a certificate for the given domain.
|
|
|
|
Args:
|
|
force: Force renewal regardless of expiry.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = body.get("domain", "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
force = body.get("force", False)
|
|
args: list[str] = ["--renew", "-d", domain]
|
|
if force:
|
|
args.append("--force")
|
|
output = _run_acme(args)
|
|
_run_acme(["--deploy", "-d", domain, "--deploy-hook", _DEPLOY_HOOK])
|
|
logger.info("Certificate for %s renewed", domain)
|
|
refresh_state(["acme"])
|
|
return {"domain": domain, "output": output.strip()}
|
|
|
|
|
|
@registry.register(DELETE_ACME_REMOVE)
|
|
def remove_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""DELETE /acme/remove — remove a certificate from ACME management.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = body.get("domain", "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
_run_acme(["--remove", "-d", domain])
|
|
logger.info("Certificate for %s removed", domain)
|
|
refresh_state(["acme"])
|
|
return {"domain": domain}
|
|
|
|
|
|
@registry.register(POST_ACME_EMAIL)
|
|
def set_email(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/email — set the ACME contact email via account registration.
|
|
|
|
Raises:
|
|
ValueError: When email is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
email = body.get("email", "").strip()
|
|
if not email:
|
|
raise ValueError("'email' is required")
|
|
_run_acme(["--register-account", "-m", email])
|
|
# Persist to declarative ACME config
|
|
acme_cfg = PROJECT_DIR / "config" / "acme" / "config.json"
|
|
acme_cfg.parent.mkdir(parents=True, exist_ok=True)
|
|
import json as _json
|
|
|
|
_acme_data: dict[str, str] = {}
|
|
if acme_cfg.is_file():
|
|
_acme_data = _json.loads(acme_cfg.read_text())
|
|
_acme_data["email"] = email
|
|
acme_cfg.write_text(_json.dumps(_acme_data, indent=4) + "\n")
|
|
logger.info("ACME email set to %s", email)
|
|
refresh_state(["acme"])
|
|
return {"email": email}
|
|
|
|
|
|
@registry.register(GET_ACME_EMAIL)
|
|
def get_email(_request: Any, _body: Any) -> dict[str, Any]:
|
|
"""GET /acme/email — return the currently configured ACME contact email."""
|
|
ac = _get_acme_state()
|
|
email = ""
|
|
if ac:
|
|
email = ac.get("email", "")
|
|
if not email:
|
|
email = _get_acme_email()
|
|
return {"email": email}
|
|
|
|
|
|
@registry.register(GET_ACME_PATHS)
|
|
def get_cert_paths(_request: Any, body: dict[str, Any] | None) -> dict[str, str]:
|
|
"""GET /acme/paths — return filesystem paths for a domain's certificate files.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body or "domain" not in body:
|
|
raise ValueError("'domain' is required")
|
|
domain = body["domain"]
|
|
from lib.acme import find_cert_dir
|
|
|
|
cert_dir = str(find_cert_dir(domain, _ACME_HOME))
|
|
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",
|
|
}
|
|
|
|
|
|
@registry.register(POST_ACME_SELF_SIGNED)
|
|
def generate_self_signed(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/self-signed — generate a self-signed certificate for a domain.
|
|
|
|
Idempotent: skips generation if cert and key already exist.
|
|
|
|
Raises:
|
|
ValueError: When domain is missing.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = body.get("domain", "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
days = body.get("days", 365)
|
|
|
|
cert_dir = _ACME_HOME / domain
|
|
cert_dir.mkdir(parents=True, exist_ok=True)
|
|
cert_file = cert_dir / "fullchain.cer"
|
|
key_file = cert_dir / f"{domain}.key"
|
|
|
|
if cert_file.is_file() and key_file.is_file():
|
|
logger.info("Self-signed cert for %s already exists, skipping", domain)
|
|
return {
|
|
"domain": domain,
|
|
"cert": str(cert_file),
|
|
"key": str(key_file),
|
|
"generated": False,
|
|
}
|
|
|
|
subprocess.run(
|
|
[
|
|
"openssl",
|
|
"req",
|
|
"-x509",
|
|
"-newkey",
|
|
"rsa:2048",
|
|
"-keyout",
|
|
str(key_file),
|
|
"-out",
|
|
str(cert_file),
|
|
"-days",
|
|
str(days),
|
|
"-nodes",
|
|
"-subj",
|
|
f"/CN={domain}",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
timeout=30,
|
|
)
|
|
|
|
cert_file.chmod(0o644)
|
|
key_file.chmod(0o600)
|
|
|
|
logger.info("Self-signed cert for %s generated (%d days)", domain, days)
|
|
return {
|
|
"domain": domain,
|
|
"cert": str(cert_file),
|
|
"key": str(key_file),
|
|
"generated": True,
|
|
}
|
|
|
|
|
|
@registry.register(GET_ACME_ACCOUNT)
|
|
def get_account(_request: Any, _body: Any) -> dict[str, Any]:
|
|
"""GET /acme/account — return ACME account information."""
|
|
return _get_account_info()
|
|
|
|
|
|
@registry.register(POST_ACME_ACCOUNT_REGISTER)
|
|
def register_account(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""POST /acme/account/register — register a new ACME account.
|
|
|
|
Raises:
|
|
ValueError: When email is missing or invalid.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
email = (body.get("email") or "").strip()
|
|
if not email:
|
|
raise ValueError("'email' is required")
|
|
import re as _re
|
|
|
|
if not _re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email):
|
|
raise ValueError("Invalid email format")
|
|
server = (body.get("server") or "letsencrypt").strip()
|
|
|
|
_run_acme(["--register-account", "-m", email, "--server", server])
|
|
|
|
acme_cfg = PROJECT_DIR / "config" / "acme" / "config.json"
|
|
acme_cfg.parent.mkdir(parents=True, exist_ok=True)
|
|
from lib.common import load_json, save_json
|
|
|
|
acme_data = load_json(acme_cfg)
|
|
acme_data["email"] = email
|
|
acme_data["ca"] = server
|
|
save_json(acme_cfg, acme_data)
|
|
|
|
logger.info("ACME account registered: %s (%s)", email, server)
|
|
refresh_state(["acme"])
|
|
return {"registered": True, "email": email, "ca": server}
|
|
|
|
|
|
@registry.register(DELETE_ACME_ACCOUNT_DEACTIVATE)
|
|
def deactivate_account(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""DELETE /acme/account/deactivate — deactivate the ACME account."""
|
|
try:
|
|
_run_acme(["--deactivate-account"])
|
|
except RuntimeError as exc:
|
|
logger.warning("acme.sh deactivate failed: %s", exc)
|
|
acme_cfg = PROJECT_DIR / "config" / "acme" / "config.json"
|
|
if acme_cfg.is_file():
|
|
from lib.common import load_json, save_json
|
|
|
|
acme_data = load_json(acme_cfg)
|
|
acme_data.pop("email", None)
|
|
acme_data.pop("ca", None)
|
|
save_json(acme_cfg, acme_data)
|
|
|
|
account_conf = _ACME_HOME / ".account.conf"
|
|
if account_conf.is_file():
|
|
account_conf.unlink()
|
|
account_conf_no_dot = _ACME_HOME / "account.conf"
|
|
if account_conf_no_dot.is_file():
|
|
account_conf_no_dot.unlink()
|
|
|
|
logger.info("ACME account deactivated")
|
|
refresh_state(["acme"])
|
|
return {"email": ""}
|