mteehan
faa076370d
- move state collectors from lib/state.py to daemon/collectors/ (7 modules, registration side-effect; daemon/server.py imports the package before the first populate()) - webui/api: new daemon_route() decorator factory in common.py collapses the try/except daemon-proxy boilerplate in all 8 blueprints (rules/params/body/transform keep responses identical) - firewall: interface-coverage invariant — config is the source of truth for zone interfaces (absent key = empty, no hands-off zones); pure validate_coverage() enforced at save (400) and apply (409, force: true overrides), top-level `unmanaged` exemption - lib: get_config() reads are now pure (no dir creation or writes); new lib/bootstrap.py creates runtime dirs and persists the one-shot nginx legacy migration at daemon start, after system_import (lib.nginx.migrate_config_file) - lib/common: compute_pending() apply-bookkeeping helper - daemon: emit_and_refresh() handler helper; refresh_state(bump=) so /status/refresh no longer bumps versions (poll/mutation only) - acme: move --log last so acme.sh never treats a real arg as the log-file argument - docs: AGENTS.md, config.md, state-model.md, api.md updated; HARDEN.md dropped (plan implemented); apply-confirm force wording Tests: 917 passed; ruff check + format clean.
127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
"""ACME certificate management API blueprint.
|
|
|
|
Exposed at /api/certs/* and delegates to vacuum-walld.
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from flask import Blueprint
|
|
|
|
from daemon.client import delete, get, post # noqa: F401 (resolved via module globals)
|
|
from daemon.iface import (
|
|
DELETE_ACME_ACCOUNT_DEACTIVATE,
|
|
DELETE_ACME_REMOVE,
|
|
GET_ACME_ACCOUNT,
|
|
GET_ACME_INFO,
|
|
GET_ACME_ISSUE_STATUS,
|
|
GET_ACME_LIST,
|
|
GET_ACME_RENEW_STATUS,
|
|
POST_ACME_ACCOUNT_REGISTER,
|
|
POST_ACME_EMAIL,
|
|
POST_ACME_ISSUE,
|
|
POST_ACME_RENEW,
|
|
POST_ACME_VALIDATE,
|
|
)
|
|
from webui.api.common import NO_BODY, daemon_route, void_transform
|
|
|
|
bp = Blueprint("certs", __name__)
|
|
|
|
|
|
def _validate_body(request: Any, _va: Any) -> dict[str, Any]:
|
|
domain = ((request.get_json(silent=True) or {}).get("domain") or "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
return {"domain": domain}
|
|
|
|
|
|
def _issue_body(request: Any, _va: Any) -> dict[str, Any]:
|
|
body = request.get_json(silent=True) or {}
|
|
domain = (body.get("domain") or "").strip()
|
|
if not domain:
|
|
raise ValueError("'domain' is required")
|
|
email = (body.get("email") or "").strip() or None
|
|
return {"domain": domain, "webroot": body.get("webroot"), "email": email}
|
|
|
|
|
|
def _email_body(request: Any, _va: Any) -> dict[str, Any]:
|
|
email = ((request.get_json(silent=True) or {}).get("email") or "").strip()
|
|
if not email:
|
|
raise ValueError("'email' is required")
|
|
return {"email": email}
|
|
|
|
|
|
def _register_body(request: Any, _va: Any) -> dict[str, Any]:
|
|
body = request.get_json(silent=True) or {}
|
|
email = (body.get("email") or "").strip()
|
|
if not email:
|
|
raise ValueError("'email' is required")
|
|
return {"email": email, "server": (body.get("server") or "").strip()}
|
|
|
|
|
|
def _email_echo(_data: Any, _va: Any, sent: Any) -> Any:
|
|
return {"email": sent["email"]}
|
|
|
|
|
|
@daemon_route(GET_ACME_LIST, bp)
|
|
def list_certs_bp():
|
|
"""GET /api/certs/list — List all managed ACME certificates."""
|
|
|
|
|
|
@daemon_route(GET_ACME_INFO, bp, rule="/<domain>")
|
|
def cert_details():
|
|
"""GET /api/certs/<domain> — Get details for a specific certificate."""
|
|
|
|
|
|
@daemon_route(POST_ACME_VALIDATE, bp, body=_validate_body)
|
|
def validate():
|
|
"""POST /api/certs/validate — Run pre-flight checks for issuance."""
|
|
|
|
|
|
@daemon_route(POST_ACME_ISSUE, bp, rule="/issue/start", body=_issue_body)
|
|
def issue_start():
|
|
"""POST /api/certs/issue/start — Create a new certificate issuance request."""
|
|
|
|
|
|
@daemon_route(
|
|
GET_ACME_ISSUE_STATUS, bp, rule="/issue/<request_id>", params={"id": "request_id"}
|
|
)
|
|
def issue_status():
|
|
"""GET /api/certs/issue/<request_id> — Poll status of an issuance request."""
|
|
|
|
|
|
@daemon_route(POST_ACME_RENEW, bp, rule="/<domain>/renew")
|
|
def renew_bp():
|
|
"""POST /api/certs/<domain>/renew — Start an (async) certificate renewal."""
|
|
|
|
|
|
@daemon_route(
|
|
GET_ACME_RENEW_STATUS, bp, rule="/renew/<request_id>", params={"id": "request_id"}
|
|
)
|
|
def renew_status():
|
|
"""GET /api/certs/renew/<request_id> — Poll status of a certificate renewal."""
|
|
|
|
|
|
@daemon_route(DELETE_ACME_REMOVE, bp, rule="/<domain>", transform=void_transform)
|
|
def remove_bp():
|
|
"""DELETE /api/certs/<domain> — Remove a certificate from ACME management."""
|
|
|
|
|
|
@daemon_route(POST_ACME_EMAIL, bp, body=_email_body, transform=_email_echo)
|
|
def set_email_bp():
|
|
"""POST /api/certs/email — Set the ACME account email address."""
|
|
|
|
|
|
@daemon_route(GET_ACME_ACCOUNT, bp)
|
|
def account():
|
|
"""GET /api/certs/account — Return ACME account information."""
|
|
|
|
|
|
@daemon_route(POST_ACME_ACCOUNT_REGISTER, bp, body=_register_body)
|
|
def register_account():
|
|
"""POST /api/certs/account/register — Register a new ACME account."""
|
|
|
|
|
|
@daemon_route(DELETE_ACME_ACCOUNT_DEACTIVATE, bp, rule="/account", body=NO_BODY)
|
|
def deactivate_account():
|
|
"""DELETE /api/certs/account — Deactivate the ACME account."""
|