npm-acme-authority/functions/cluster_cert.py

"""
Cluster-managed custom certificate mutations.

This module does NOT implement ACME renewal.

It only:
  1. Adds cluster lifecycle flags to the certificate meta schema.
  2. Preserves logical domain_names when custom certificate material is uploaded.
"""

from __future__ import annotations

import json

from functions.common import log
from functions.common import replace_exactly_once
from functions.common import require_fragment


SCHEMA_PROPERTIES = {
    "cluster_managed": {
        "type": "boolean",
        "default": False,
    },
    "acme_renew": {
        "type": "boolean",
        "default": False,
    },
}


UPLOAD_DOMAIN_ANCHOR = '''\t\tconst certificate = await internalCertificate.update(access, {
\t\t\tid: data.id,
\t\t\texpires_on: moment(validations.certificate.dates.to, "X").format("YYYY-MM-DD HH:mm:ss"),
\t\t\tdomain_names: validations.certificate.cn ? [validations.certificate.cn] : [],
\t\t\tmeta: _.clone(row.meta), // Prevent the update method from changing this value that we'll use later
\t\t});
'''

UPLOAD_DOMAIN_FRAGMENT = '''\t\tconst certificate = await internalCertificate.update(access, {
\t\t\tid: data.id,
\t\t\texpires_on: moment(validations.certificate.dates.to, "X").format("YYYY-MM-DD HH:mm:ss"),
\t\t\tdomain_names:
\t\t\t\tArray.isArray(row.domain_names) && row.domain_names.length
\t\t\t\t\t? row.domain_names
\t\t\t\t\t: validations.certificate.cn
\t\t\t\t\t\t? [validations.certificate.cn]
\t\t\t\t\t\t: [],
\t\t\tmeta: _.clone(row.meta), // Prevent the update method from changing this value that we'll use later
\t\t});
'''


def apply_certificate_js(source: str) -> str:
    if UPLOAD_DOMAIN_FRAGMENT in source:
        log("cluster_cert_upload_domains=already-present")
        return source

    source = replace_exactly_once(
        source,
        UPLOAD_DOMAIN_ANCHOR,
        UPLOAD_DOMAIN_FRAGMENT,
        description="cluster certificate upload domain preservation",
    )

    log("cluster_cert_upload_domains=added")
    return source


def verify_certificate_js(source: str) -> None:
    require_fragment(
        source,
        UPLOAD_DOMAIN_FRAGMENT,
        description="cluster certificate upload domain verification",
    )

    require_fragment(
        source,
        "await internalCertificate.writeCustomCert(certificate);",
        description="cluster certificate custom write verification",
    )

    log("cluster_cert_upload_domains_verification=passed")


def apply_schema(source: str) -> str:
    try:
        document = json.loads(source)
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"cluster certificate schema: invalid JSON: {exc}"
        ) from exc

    try:
        meta = document["properties"]["meta"]
        meta_properties = meta["properties"]
    except (KeyError, TypeError) as exc:
        raise RuntimeError(
            "cluster certificate schema: "
            "properties.meta.properties not found"
        ) from exc

    if meta.get("additionalProperties") is not False:
        raise RuntimeError(
            "cluster certificate schema: "
            "meta.additionalProperties is not false"
        )

    changed = False

    for name, definition in SCHEMA_PROPERTIES.items():
        existing = meta_properties.get(name)

        if existing is None:
            meta_properties[name] = definition
            changed = True
        elif existing != definition:
            raise RuntimeError(
                "cluster certificate schema: "
                f"{name} already exists with unexpected definition"
            )

    if not changed:
        verify_schema(source)
        log("cluster_cert_schema=already-present")
        return source

    result = json.dumps(
        document,
        indent="\t",
        ensure_ascii=False,
    ) + "\n"

    verify_schema(result)
    log("cluster_cert_schema=added")
    return result


def verify_schema(source: str) -> None:
    try:
        document = json.loads(source)
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"cluster certificate schema verification: invalid JSON: {exc}"
        ) from exc

    try:
        meta = document["properties"]["meta"]
        meta_properties = meta["properties"]
    except (KeyError, TypeError) as exc:
        raise RuntimeError(
            "cluster certificate schema verification: "
            "properties.meta.properties not found"
        ) from exc

    if meta.get("additionalProperties") is not False:
        raise RuntimeError(
            "cluster certificate schema verification: "
            "meta.additionalProperties is not false"
        )

    for name, definition in SCHEMA_PROPERTIES.items():
        if meta_properties.get(name) != definition:
            raise RuntimeError(
                "cluster certificate schema verification: "
                f"{name} definition is missing or unexpected"
            )

    log("cluster_cert_schema_verification=passed")