vacuum-wall/webui/api/certs.py

"""ACME certificate management API blueprint.

Exposed at /api/certs/* and delegates to vacuum-walld.
"""

import logging

from flask import Blueprint, request

from daemon.client import BadRequest, NotFound, delete, get, post
from daemon.iface import (
    DELETE_ACME_ACCOUNT_DEACTIVATE,
    DELETE_ACME_REMOVE,
    GET_ACME_ACCOUNT,
    GET_ACME_INFO,
    GET_ACME_ISSUE_STATUS,
    GET_ACME_LIST,
    POST_ACME_ACCOUNT_REGISTER,
    POST_ACME_EMAIL,
    POST_ACME_ISSUE,
    POST_ACME_RENEW,
    POST_ACME_VALIDATE,
)
from webui.api.common import _error, _ok

logger = logging.getLogger(__name__)
bp = Blueprint("certs", __name__)


@bp.route("/list", methods=["GET"])
def list_certs_bp():
    """GET /api/certs/list — list all managed ACME certificates.

    Returns:
        Response containing the list of certificates or an error message.
    """
    try:
        return _ok(get(GET_ACME_LIST))
    except RuntimeError as exc:
        logger.error("Failed to list certificates: %s", exc)
        return _error(str(exc), 500)


@bp.route("/<domain>", methods=["GET"])
def cert_details(domain: str):
    """GET /api/certs/<domain> — get details for a specific certificate.

    Args:
        domain: Domain name to look up.

    Returns:
        Response containing certificate info or an error message.
    """
    try:
        return _ok(get(GET_ACME_INFO, {"domain": domain}))
    except NotFound as exc:
        logger.info("Cert for '%s' not found: %s", domain, exc)
        return _error(str(exc), 404)
    except RuntimeError as exc:
        logger.error("Failed to get cert info for '%s': %s", domain, exc)
        return _error(str(exc), 500)


@bp.route("/validate", methods=["POST"])
def validate():
    """POST /api/certs/validate — run pre-flight checks for certificate issuance.

    Expects JSON body with ``{``domain``}``.

    Returns:
        Response containing validation results or an error message.
    """
    body = request.get_json(silent=True) or {}
    domain = (body.get("domain") or "").strip()
    if not domain:
        return _error("'domain' is required", 400)
    try:
        result = post(POST_ACME_VALIDATE, {"domain": domain})
        return _ok(result)
    except BadRequest as exc:
        logger.info("Validation rejected: %s", exc)
        return _error(str(exc), 400)
    except RuntimeError as exc:
        logger.error("Failed to validate cert for '%s': %s", domain, exc)
        return _error(str(exc), 500)


@bp.route("/issue/start", methods=["POST"])
def issue_start():
    """POST /api/certs/issue/start — create a new certificate issuance request.

    Expects JSON body with ``{``domain``}``; optional ``email`` and ``webroot``.

    Returns:
        Response containing an issuance request ID or an error message.
    """
    body = request.get_json(silent=True) or {}
    domain = (body.get("domain") or "").strip()
    if not domain:
        return _error("'domain' is required", 400)
    email = (body.get("email") or "").strip() or None
    webroot = body.get("webroot")
    try:
        logger.info("Certificate issuance requested for '%s' via API", domain)
        result = post(
            POST_ACME_ISSUE, {"domain": domain, "webroot": webroot, "email": email}
        )
        logger.info(
            "Certificate issuance started for '%s' (id=%s)",
            domain,
            result.get("request_id"),
        )
        return _ok(result)
    except BadRequest as exc:
        logger.info("Cert issue for '%s' rejected: %s", domain, exc)
        return _error(str(exc), 400)
    except RuntimeError as exc:
        logger.error("Failed to start cert issue for '%s': %s", domain, exc)
        return _error(str(exc), 500)


@bp.route("/issue/<request_id>", methods=["GET"])
def issue_status(request_id: str):
    """GET /api/certs/issue/<request_id> — poll status of a certificate issuance request.

    Args:
        request_id: Issuance request identifier returned by issue_start.

    Returns:
        Response containing issuance status or an error message.
    """
    try:
        result = get(GET_ACME_ISSUE_STATUS, {"id": request_id})
        return _ok(result)
    except NotFound as exc:
        logger.info("Issuance request '%s' not found: %s", request_id, exc)
        return _error(str(exc), 404)
    except RuntimeError as exc:
        logger.error("Failed to get issuance status for '%s': %s", request_id, exc)
        return _error(str(exc), 500)


@bp.route("/<domain>/renew", methods=["POST"])
def renew_bp(domain: str):
    """POST /api/certs/<domain>/renew — renew an existing certificate.

    Args:
        domain: Domain name whose certificate should be renewed.

    Returns:
        Response confirming renewal or an error message.
    """
    try:
        logger.info("Certificate renewal requested for '%s' via API", domain)
        post(POST_ACME_RENEW, {"domain": domain})
        logger.info("Certificate renewed for '%s'", domain)
        return _ok(None)
    except BadRequest as exc:
        logger.info("Cert renew for '%s' rejected: %s", domain, exc)
        return _error(str(exc), 400)
    except RuntimeError as exc:
        logger.error("Failed to renew cert for '%s': %s", domain, exc)
        return _error(str(exc), 500)


@bp.route("/<domain>", methods=["DELETE"])
def remove_bp(domain: str):
    """DELETE /api/certs/<domain> — remove a certificate from ACME management.

    Args:
        domain: Domain name whose certificate should be removed.

    Returns:
        Response confirming removal or an error message.
    """
    try:
        delete(DELETE_ACME_REMOVE, {"domain": domain})
        logger.info("Certificate removed for '%s' via API", domain)
        return _ok(None)
    except NotFound as exc:
        logger.info("Cert '%s' not found: %s", domain, exc)
        return _error(str(exc), 404)
    except RuntimeError as exc:
        logger.error("Failed to remove cert '%s': %s", domain, exc)
        return _error(str(exc), 500)


@bp.route("/email", methods=["POST"])
def set_email_bp():
    """POST /api/certs/email — set the ACME account email address.

    Expects JSON body with ``{``email``}``.

    Returns:
        Response confirming the email was set or an error message.
    """
    body = request.get_json(silent=True) or {}
    email = (body.get("email") or "").strip()
    if not email:
        return _error("'email' is required", 400)
    try:
        post(POST_ACME_EMAIL, {"email": email})
        logger.info("ACME email set via API: %s", email)
        return _ok({"email": email})
    except BadRequest as exc:
        logger.info("ACME email set rejected: %s", exc)
        return _error(str(exc), 400)
    except RuntimeError as exc:
        logger.error("Failed to set ACME email: %s", exc)
        return _error(str(exc), 500)


@bp.route("/account", methods=["GET"])
def account():
    """GET /api/certs/account — return ACME account information.

    Returns:
        Response containing account status or an error message.
    """
    try:
        result = get(GET_ACME_ACCOUNT)
        return _ok(result)
    except RuntimeError as exc:
        logger.error("Failed to get ACME account: %s", exc)
        return _error(str(exc), 500)


@bp.route("/account/register", methods=["POST"])
def register_account():
    """POST /api/certs/account/register — register a new ACME account.

    Expects JSON body with ``{``email``, ``server``?}``.

    Returns:
        Response confirming registration or an error message.
    """
    body = request.get_json(silent=True) or {}
    email = (body.get("email") or "").strip()
    if not email:
        return _error("'email' is required", 400)
    server = (body.get("server") or "").strip()
    try:
        result = post(POST_ACME_ACCOUNT_REGISTER, {"email": email, "server": server})
        return _ok(result)
    except BadRequest as exc:
        return _error(str(exc), 400)
    except RuntimeError as exc:
        logger.error("Failed to register ACME account: %s", exc)
        return _error(str(exc), 500)


@bp.route("/account", methods=["DELETE"])
def deactivate_account():
    """DELETE /api/certs/account — deactivate the ACME account.

    Returns:
        Response confirming deactivation or an error message.
    """
    try:
        result = delete(DELETE_ACME_ACCOUNT_DEACTIVATE)
        return _ok(result)
    except RuntimeError as exc:
        logger.error("Failed to deactivate ACME account: %s", exc)
        return _error(str(exc), 500)