feat: pre-computed state store and async ACME issuance (fixes timeout mismatch)
- Add lib/state.py: in-memory state store with subsystem collectors (firewall, dnsmasq, nginx, acme, wireguard) - Refactor all handlers: read from state on GET, call refresh_state() after mutations instead of invoking subprocesses per request - daemon/server.py: add refresh_state(), /status/all, /status/refresh; populate state at startup - webui/api/certs.py: async step-by-step ACME issuance (validate, issue with request_id, poll status) replacing blocking endpoint - webui/server.py: render pages from state instead of direct lib calls - Update templates, JS for async cert issuance with polling UI - Update tests for state-based mocking; add test_state.py - Fix SIM105 lint issue (contextlib.suppress) - Add TODO.md with certificate issuance issue tracking Resolves: WebUI 30s timeout freeze during cert issuance (Problem 1)
This commit is contained in:
+374
-107
@@ -1,15 +1,19 @@
|
||||
"""ACME certificate daemon handler."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
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.server import NotFoundError, registry
|
||||
from daemon.server import NotFoundError, refresh_state, registry
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -26,21 +30,56 @@ _ACME_ENVIRON = {
|
||||
|
||||
_WEBROOT = PROJECT_DIR / "data" / "acme" / "www"
|
||||
|
||||
_ACME_TAGS = {"acme"}
|
||||
# In-memory store for active issuance requests.
|
||||
_ISSUANCES: dict[str, "IssueRequest"] = {}
|
||||
_ISSUANCE_TTL = 300 # seconds to keep completed requests
|
||||
|
||||
|
||||
def _find_acme() -> str:
|
||||
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):
|
||||
return str(path)
|
||||
acme = shutil.which("acme.sh")
|
||||
if acme:
|
||||
return acme
|
||||
raise FileNotFoundError("acme.sh not found")
|
||||
@dataclass
|
||||
class IssueStep:
|
||||
name: str
|
||||
label: str
|
||||
status: str = "pending"
|
||||
message: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class IssueRequest:
|
||||
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]:
|
||||
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 _run_acme(args: list[str]) -> str:
|
||||
from lib.state import _find_acme
|
||||
|
||||
acme_bin = _find_acme()
|
||||
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
|
||||
cmd = [acme_bin, "--home", acme_home_env, "--config-home", acme_home_env, *args]
|
||||
@@ -62,126 +101,348 @@ def _run_acme(args: list[str]) -> str:
|
||||
return output
|
||||
|
||||
|
||||
def _days_until(date_str: str) -> int | None:
|
||||
if not date_str:
|
||||
return None
|
||||
for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"):
|
||||
try:
|
||||
dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC)
|
||||
return (dt - datetime.now(UTC)).days
|
||||
except ValueError:
|
||||
continue
|
||||
def _find_acme_bin() -> str:
|
||||
from lib.state import _find_acme
|
||||
|
||||
return _find_acme()
|
||||
|
||||
|
||||
def _get_acme_email() -> str:
|
||||
try:
|
||||
dt = datetime.strptime(date_str, "%Y%m%d%H%M%z").astimezone(UTC)
|
||||
return (dt - datetime.now(UTC)).days
|
||||
except ValueError:
|
||||
acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
|
||||
account_conf = acme_home / "account.conf"
|
||||
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:
|
||||
pass
|
||||
return None
|
||||
return ""
|
||||
|
||||
|
||||
def _parse_list_output(raw: str) -> list[dict]:
|
||||
entries: list[dict] = []
|
||||
for line in raw.strip().splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
entry: dict[str, str] = {}
|
||||
for token in line.split():
|
||||
if ":" not in token:
|
||||
continue
|
||||
key, _, value = token.partition(":")
|
||||
entry[key.lower()] = value
|
||||
if entry:
|
||||
entries.append(entry)
|
||||
return entries
|
||||
def _get_state() -> dict[str, Any] | None:
|
||||
from lib.state import state as state_store
|
||||
|
||||
return state_store.get("acme")
|
||||
|
||||
|
||||
def _has_auto_renew(domain: str) -> bool:
|
||||
acme_home_env = os.environ.get("ACME_HOME", str(_ACME_HOME))
|
||||
return bool(Path(acme_home_env) / f"{domain}.conf")
|
||||
def _get_acme_state() -> dict[str, Any]:
|
||||
ac = _get_state()
|
||||
if ac is None:
|
||||
return {}
|
||||
return ac
|
||||
|
||||
|
||||
@registry.register("GET", "/acme/list", cache_tags=_ACME_TAGS)
|
||||
def list_certs(_request: Any, _body: Any) -> list[dict]:
|
||||
raw = _run_acme(["--list"])
|
||||
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_domain", "").split(",") if d.strip()
|
||||
]
|
||||
cert_dir = acme_home / main
|
||||
days = _days_until(entry.get("certificate_expires", ""))
|
||||
certs.append(
|
||||
{
|
||||
"domain": main,
|
||||
"issuer": entry.get("CA", ""),
|
||||
"expiry": entry.get("certificate_expires", ""),
|
||||
"days_remaining": days,
|
||||
"expired": days is not None and days <= 0,
|
||||
"cert_path": str(cert_dir / "fullchain.cer"),
|
||||
"key_path": str(cert_dir / f"{main}.key"),
|
||||
"ca_path": str(cert_dir / "ca.cer"),
|
||||
"issued_at": entry.get("certificate_date", ""),
|
||||
"expires_at": entry.get("certificate_expires", ""),
|
||||
"days_until_expiry": days,
|
||||
"auto_renew": _has_auto_renew(main),
|
||||
"san_domains": san_domains,
|
||||
}
|
||||
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",
|
||||
)
|
||||
return certs
|
||||
except socket.gaierror:
|
||||
return False, "Domain does not resolve (NXDOMAIN or timeout)"
|
||||
|
||||
|
||||
@registry.register("GET", "/acme/info", cache_tags=_ACME_TAGS)
|
||||
def _check_acme_installed() -> tuple[bool, str]:
|
||||
try:
|
||||
_find_acme_bin()
|
||||
return True, "acme.sh found"
|
||||
except FileNotFoundError:
|
||||
return False, "acme.sh not installed"
|
||||
|
||||
|
||||
def _check_email_configured() -> tuple[bool, str]:
|
||||
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]:
|
||||
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]:
|
||||
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]:
|
||||
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:
|
||||
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["san_domains"]:
|
||||
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/issue", invalidate=_ACME_TAGS | {"nginx"})
|
||||
def issue_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
@registry.register("POST", "/acme/validate")
|
||||
def validate_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
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]:
|
||||
if not body:
|
||||
raise ValueError("Request body required")
|
||||
domain = body.get("domain", "").strip()
|
||||
if not domain:
|
||||
raise ValueError("'domain' is required")
|
||||
webroot = body.get("webroot")
|
||||
email = body.get("email", "").strip() or None
|
||||
args: list[str] = ["--issue", "-d", domain]
|
||||
args.extend(["--webroot", webroot or str(_WEBROOT)])
|
||||
contact = email
|
||||
if not contact:
|
||||
try:
|
||||
acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
|
||||
account_conf = acme_home / "account.conf"
|
||||
if account_conf.is_file():
|
||||
text = account_conf.read_text()
|
||||
match = re.search(r"^ACME_LEEMAIL=(.+)$", text, re.MULTILINE)
|
||||
if match:
|
||||
contact = match.group(1).strip().strip("'\"")
|
||||
except OSError:
|
||||
pass
|
||||
if contact:
|
||||
args.extend(["-m", contact])
|
||||
args.append("--force")
|
||||
output = _run_acme(args)
|
||||
_run_acme(["--deploy", "-d", domain, "--deploy-hook", _DEPLOY_HOOK])
|
||||
logger.info("Certificate for %s issued", domain)
|
||||
return {"domain": domain, "output": output.strip()}
|
||||
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("POST", "/acme/renew", invalidate=_ACME_TAGS | {"nginx"})
|
||||
@registry.register("GET", "/acme/issue/status")
|
||||
def get_issuance_status(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
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]:
|
||||
if not body:
|
||||
raise ValueError("Request body required")
|
||||
@@ -195,10 +456,11 @@ def renew_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
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", invalidate=_ACME_TAGS)
|
||||
@registry.register("DELETE", "/acme/remove")
|
||||
def remove_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
if not body:
|
||||
raise ValueError("Request body required")
|
||||
@@ -207,10 +469,11 @@ def remove_cert(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
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", invalidate=_ACME_TAGS)
|
||||
@registry.register("POST", "/acme/email")
|
||||
def set_email(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
if not body:
|
||||
raise ValueError("Request body required")
|
||||
@@ -219,11 +482,15 @@ def set_email(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
|
||||
raise ValueError("'email' is required")
|
||||
_run_acme(["--register-account", "-m", email])
|
||||
logger.info("ACME email set to %s", email)
|
||||
refresh_state(["acme"])
|
||||
return {"email": email}
|
||||
|
||||
|
||||
@registry.register("GET", "/acme/email", cache_tags=_ACME_TAGS)
|
||||
@registry.register("GET", "/acme/email")
|
||||
def get_email(_request: Any, _body: Any) -> dict[str, Any]:
|
||||
ac = _get_acme_state()
|
||||
if ac:
|
||||
return {"email": ac.get("email", "")}
|
||||
try:
|
||||
acme_home = Path(os.environ.get("ACME_HOME", str(_ACME_HOME)))
|
||||
account_conf = acme_home / "account.conf"
|
||||
@@ -237,7 +504,7 @@ def get_email(_request: Any, _body: Any) -> dict[str, Any]:
|
||||
return {"email": ""}
|
||||
|
||||
|
||||
@registry.register("GET", "/acme/paths", cache_tags=_ACME_TAGS)
|
||||
@registry.register("GET", "/acme/paths")
|
||||
def get_cert_paths(_request: Any, body: dict[str, Any] | None) -> dict[str, str]:
|
||||
if not body or "domain" not in body:
|
||||
raise ValueError("'domain' is required")
|
||||
|
||||
Reference in New Issue
Block a user