Initialize modular NPM ACME authority patcher

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.py[cod]
+*.swp
+*.swo
+*~
+.env
+state/
+test-output/

README.md

@@ -0,0 +1,52 @@
+# NPM ACME Authority
+
+Runtime ACME write fencing for Nginx Proxy Manager.
+
+The project patches NPM certificate handling at container startup so that
+selected Let's Encrypt mutation paths require a Pacemaker-controlled
+authority token.
+
+## Gate configuration
+
+- `ACME_GATE_REQUEST=true`
+- `ACME_GATE_RENEW=true`
+- `ACME_GATE_REVOKE=true`
+
+These variables determine which ACME operations require authority.
+
+They do **not** grant authority.
+
+## Runtime authority
+
+Authority is represented by:
+
+    /acme_authority/state/acme-authority
+
+The intended production mapping is:
+
+    /etc/npm-acme-authority  -> /acme_authority
+    /run/npm-acme-authority -> /acme_authority/state
+
+The persistent code mount is read-only.
+
+The runtime state mount is read-only from the container's perspective and
+is controlled by Pacemaker on the host.
+
+## Intended policy
+
+REQUEST:
+- Gate Let's Encrypt creation before database insertion.
+
+RENEW:
+- Gate automatic and manual renewal before Certbot execution.
+
+REVOKE:
+- Gate certificate deletion before the database row is marked deleted.
+- Gate Certbot revocation again as defense in depth.
+
+## Safety
+
+Requested mutations must fail closed.
+
+If an expected NPM source anchor cannot be identified or verification fails,
+the patcher must exit non-zero rather than start an ungated NPM backend.

bootstrap.sh

@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+readonly ACME_AUTHORITY_ROOT="${ACME_AUTHORITY_ROOT:-/acme_authority}"
+readonly PATCHER="${ACME_AUTHORITY_ROOT}/patch_acme.py"
+
+log()
+{
+    printf '[acme-authority] %s\n' "$*"
+}
+
+log "Bootstrap starting"
+
+python3 "$PATCHER"
+
+log "Bootstrap completed"

functions/__init__.py

@@ -0,0 +1 @@
+"""NPM ACME authority mutation modules."""

functions/common.py

@@ -0,0 +1,83 @@
+"""
+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"
+
+
+@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 apply_authority_helper(source: str) -> str:
+    """
+    Install the common requireAcmeAuthority() helper.
+
+    Implemented in the next development step.
+    """
+    raise NotImplementedError("authority helper mutation not implemented yet")
+
+
+def verify(source: str, config: GateConfig) -> None:
+    """
+    Verify every requested mutation before certificate.js is written.
+
+    Implemented after the individual mutation modules.
+    """
+    raise NotImplementedError("patch verification not implemented yet")

functions/renew.py

@@ -0,0 +1,12 @@
+"""
+Let's Encrypt certificate renewal gate.
+"""
+
+from __future__ import annotations
+
+
+def apply(source: str) -> str:
+    """
+    Gate internalCertificate.renew() before renewal dispatch.
+    """
+    raise NotImplementedError("renew gate not implemented yet")

functions/request.py

@@ -0,0 +1,12 @@
+"""
+Let's Encrypt certificate request/create gate.
+"""
+
+from __future__ import annotations
+
+
+def apply(source: str) -> str:
+    """
+    Gate NPM create() for provider == letsencrypt before DB insertion.
+    """
+    raise NotImplementedError("request gate not implemented yet")

functions/revoke.py

@@ -0,0 +1,14 @@
+"""
+Let's Encrypt certificate delete/revoke gates.
+"""
+
+from __future__ import annotations
+
+
+def apply(source: str) -> str:
+    """
+    Install both:
+      - delete() pre-database-mutation authority gate
+      - revokeLetsEncryptSsl() execution authority gate
+    """
+    raise NotImplementedError("revoke gate not implemented yet")

patch_acme.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+
+"""
+Nginx Proxy Manager ACME authority patch orchestrator.
+
+This module:
+  1. Reads ACME gate configuration from environment variables.
+  2. Loads NPM certificate.js.
+  3. Applies only the requested mutations.
+  4. Verifies every requested mutation.
+  5. Writes the resulting source only after all enabled gates succeed.
+
+Runtime ACME authority is NOT controlled here.
+Patched NPM code reads the Pacemaker-controlled authority token.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from functions import common
+from functions import request
+from functions import renew
+from functions import revoke
+
+
+DEFAULT_CERTIFICATE_JS = "/app/internal/certificate.js"
+
+
+def main() -> None:
+    certificate_js = Path(
+        os.environ.get(
+            "NPM_CERTIFICATE_JS",
+            DEFAULT_CERTIFICATE_JS,
+        )
+    )
+
+    config = common.load_config()
+
+    common.log(f"certificate.js={certificate_js}")
+    common.log(f"request_gate={config.request}")
+    common.log(f"renew_gate={config.renew}")
+    common.log(f"revoke_gate={config.revoke}")
+
+    source = common.load_source(certificate_js)
+
+    if config.any_enabled:
+        source = common.apply_authority_helper(source)
+
+    if config.request:
+        source = request.apply(source)
+
+    if config.renew:
+        source = renew.apply(source)
+
+    if config.revoke:
+        source = revoke.apply(source)
+
+    common.verify(source, config)
+    common.write_source(certificate_js, source)
+
+    common.log("All requested ACME authority gates applied successfully")
+
+
+if __name__ == "__main__":
+    main()