650 lines
20 KiB
Python
650 lines
20 KiB
Python
"""ACME certificate daemon handler."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
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
|
|
|
|
from daemon.iface import (
|
|
DELETE_ACME_REMOVE,
|
|
GET_ACME_EMAIL,
|
|
GET_ACME_INFO,
|
|
GET_ACME_ISSUE_STATUS,
|
|
GET_ACME_LIST,
|
|
GET_ACME_PATHS,
|
|
POST_ACME_EMAIL,
|
|
POST_ACME_ISSUE,
|
|
POST_ACME_RENEW,
|
|
POST_ACME_SELF_SIGNED,
|
|
POST_ACME_VALIDATE,
|
|
)
|
|
from daemon.server import NotFoundError, refresh_state, registry
|
|
from lib.state import _run_acme
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parent.parent.parent
|
|
_ACME_HOME = PROJECT_DIR / "data" / "acme"
|
|
_DEPLOY_HOOK = str(PROJECT_DIR / "system" / "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
|
|
|
|
|
|
@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.state 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, ""
|
|
|
|
|
|
def _check_dns_resolves(domain: str) -> tuple[bool, str]:
|
|
"""Check that domain resolves to this machine's IP via A record."""
|
|
try:
|
|
results = socket.getaddrinfo(domain, 80, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
|
if not results:
|
|
return False, "Domain does not resolve to any address"
|
|
|
|
local_ips = set()
|
|
hostname = socket.gethostname()
|
|
with suppress(OSError):
|
|
local_ips.add(socket.gethostbyname(hostname))
|
|
# Also collect all interface IPs
|
|
try:
|
|
import ipaddress
|
|
from fcntl import ioctl
|
|
|
|
def get_interfaces():
|
|
import struct
|
|
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
names = b"\x00" * 4096
|
|
raw = ioctl(s.fileno(), 0x8912, names)
|
|
s.close()
|
|
ifaces = []
|
|
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]
|
|
ifaces.append(str(ipaddress.IPv4Address(addr)))
|
|
return ifaces
|
|
|
|
local_ips.update(get_interfaces())
|
|
except Exception:
|
|
pass
|
|
|
|
resolved = False
|
|
for _, _, _, _, addr in results:
|
|
if addr in local_ips:
|
|
resolved = True
|
|
break
|
|
|
|
if resolved:
|
|
return True, "DNS resolves correctly"
|
|
return (
|
|
False,
|
|
f"Domain resolves to {results[0][4][0]}, not this server",
|
|
)
|
|
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, "No ACME contact email configured"
|
|
|
|
|
|
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:
|
|
from lib.acme import days_until_expiry
|
|
|
|
days = days_until_expiry(domain)
|
|
if days is not None and days > 0:
|
|
return True, f"Valid certificate exists ({days} days remaining)"
|
|
except (ValueError, RuntimeError, FileNotFoundError):
|
|
pass
|
|
return True, ""
|
|
|
|
|
|
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 = [
|
|
("acme_installed", _check_acme_installed, True),
|
|
("email_configured", _check_email_configured, True),
|
|
("webroot_ready", _check_webroot, True),
|
|
("challenge_configured", _check_challenge_config, True),
|
|
("domain_format", lambda: _check_domain_format(domain), True),
|
|
("dns_resolves", lambda: _check_dns_resolves(domain), True),
|
|
("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,
|
|
}
|
|
)
|
|
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)
|
|
for c in certs:
|
|
if c["domain"] == domain or domain in c.get("san_domains", []):
|
|
return c
|
|
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.
|
|
RuntimeError: When pre-flight checks fail.
|
|
"""
|
|
if not body:
|
|
raise ValueError("Request body required")
|
|
domain = body.get("domain", "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
email = body.get("email", "").strip() or None
|
|
webroot = body.get("webroot")
|
|
|
|
_clean_expired_issuances()
|
|
|
|
# Dedup: if domain already has an active request, return existing ID
|
|
for existing in _ISSUANCES.values():
|
|
if existing.domain == domain and existing.status == "running":
|
|
return {
|
|
"request_id": existing.request_id,
|
|
"status": "existing",
|
|
"domain": domain,
|
|
"message": "Issuance already in progress for this domain",
|
|
}
|
|
|
|
# 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,
|
|
email=email,
|
|
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)])
|
|
contact = req.email
|
|
if not contact:
|
|
contact = _get_acme_email()
|
|
if contact:
|
|
args.extend(["-m", contact])
|
|
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"]
|
|
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
|
|
acme_home = str(Path(acme_home_env) / domain)
|
|
return {
|
|
"cert": f"{acme_home}/{domain}.cert",
|
|
"key": f"{acme_home}/{domain}.key",
|
|
"ca": f"{acme_home}/ca.cer",
|
|
"fullchain": f"{acme_home}/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,
|
|
}
|