npm-acme-authority/functions/common.py

"""
Shared helpers for NPM ACME authority mutations.
"""

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
import os


AUTHORITY_FILE = "/acme_authority/state/acme-authority"

HELPER_MARKER = "const acmeAuthorityFile ="

AUTHORITY_HELPER = f"""const acmeAuthorityFile = "{AUTHORITY_FILE}";

const requireAcmeAuthority = () => {{
    try {{
        fs.accessSync(acmeAuthorityFile, fs.constants.R_OK);
    }} catch {{
        throw new error.ValidationError(
            "ACME operations are disabled on this cluster node",
        );
    }}
}};
"""


@dataclass(frozen=True)
class GateConfig:
    request: bool
    renew: bool
    revoke: bool

    @property
    def any_enabled(self) -> bool:
        return self.request or self.renew or self.revoke


def log(message: str) -> None:
    print(f"[acme-authority] {message}", flush=True)


def env_bool(name: str, default: bool = False) -> bool:
    value = os.environ.get(name)

    if value is None:
        return default

    value = value.strip().lower()

    if value in {"1", "true", "yes", "on"}:
        return True

    if value in {"0", "false", "no", "off"}:
        return False

    raise RuntimeError(
        f"{name} must be true/false, yes/no, on/off, or 1/0; got {value!r}"
    )


def load_config() -> GateConfig:
    return GateConfig(
        request=env_bool("ACME_GATE_REQUEST"),
        renew=env_bool("ACME_GATE_RENEW"),
        revoke=env_bool("ACME_GATE_REVOKE"),
    )


def load_source(path: Path) -> str:
    if not path.is_file():
        raise RuntimeError(f"NPM certificate source not found: {path}")

    return path.read_text()


def write_source(path: Path, source: str) -> None:
    path.write_text(source)


def replace_exactly_once(
    source: str,
    old: str,
    new: str,
    *,
    description: str,
) -> str:
    count = source.count(old)

    if count != 1:
        raise RuntimeError(
            f"{description}: expected anchor exactly once, found {count}"
        )

    return source.replace(old, new, 1)


def require_fragment(
    source: str,
    fragment: str,
    *,
    description: str,
) -> None:
    if fragment not in source:
        raise RuntimeError(
            f"{description}: required fragment not found"
        )


def apply_authority_helper(source: str) -> str:
    """
    Install the shared requireAcmeAuthority() helper.

    The helper is inserted immediately after the top-level fs import so the
    patch remains close to the dependency it requires.
    """

    expected_declaration = (
        f'const acmeAuthorityFile = "{AUTHORITY_FILE}";'
    )

    if expected_declaration in source:
        require_fragment(
            source,
            "const requireAcmeAuthority = () => {",
            description="existing authority helper",
        )
        require_fragment(
            source,
            "fs.accessSync(acmeAuthorityFile, fs.constants.R_OK);",
            description="existing authority helper",
        )

        log("authority_helper=already-present")
        return source

    if HELPER_MARKER in source:
        raise RuntimeError(
            "authority helper already exists but does not match "
            f"configured authority file {AUTHORITY_FILE}"
        )

    # Prefer the end of the known ESM import block used by current NPM.
    # Older CommonJS releases fall back to their fs import.
    supported_anchors = (
        'import internalNginx from "./nginx.js";\n',
        'const fs = require("fs");\n',
        'const fs = require("node:fs");\n',
    )

    matches = [
        anchor
        for anchor in supported_anchors
        if anchor in source
    ]

    if len(matches) != 1:
        raise RuntimeError(
            "authority helper: expected exactly one supported insertion "
            f"anchor, found {len(matches)}"
        )

    anchor = matches[0]
    replacement = anchor + "\n" + AUTHORITY_HELPER

    source = replace_exactly_once(
        source,
        anchor,
        replacement,
        description="authority helper",
    )

    log(f"authority_helper=added anchor={anchor.strip()!r}")
    return source


def verify(source: str, config: GateConfig) -> None:
    """
    Verify shared patch state.

    Individual mutation modules will add their own verification markers as
    they are implemented.
    """

    if config.any_enabled:
        require_fragment(
            source,
            f'const acmeAuthorityFile = "{AUTHORITY_FILE}";',
            description="authority helper verification",
        )

        require_fragment(
            source,
            "const requireAcmeAuthority = () => {",
            description="authority helper verification",
        )

        require_fragment(
            source,
            "fs.accessSync(acmeAuthorityFile, fs.constants.R_OK);",
            description="authority helper verification",
        )

        require_fragment(
            source,
            "ACME operations are disabled on this cluster node",
            description="authority helper verification",
        )

    log("verification=passed")