homelab-codex-ws/scripts/npm/npm_api.py

#!/usr/bin/env python3
# Manage Nginx Proxy Manager (PIHA + VPS) via its REST API instead of the web UI.
# Zero-dep (stdlib only) — see README.md for rationale.

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_ENV_FILE = SCRIPT_DIR / ".env"

DEFAULT_TIMEOUT = 15
# NPM issues the Let's Encrypt cert synchronously inside POST /nginx/certificates
# (nginx reload -> certbot -> nginx reload), so that one call needs a long leash.
CERT_ISSUE_TIMEOUT = 120

INSTANCES = {
    "piha": {
        "url_env": "NPM_PIHA_URL",
        "url_default": "http://192.168.31.5:81",
        "user_env": "NPM_PIHA_USER",
        "pass_env": "NPM_PIHA_PASS",
    },
    "vps": {
        "url_env": "NPM_VPS_URL",
        # Tailscale mesh address of the admin panel — the public IP
        # (135.181.153.108:81) must never be used here, see docs/backlog.md.
        "url_default": "http://100.95.58.48:81",
        "user_env": "NPM_VPS_USER",
        "pass_env": "NPM_VPS_PASS",
    },
}


class NpmApiError(Exception):
    pass


class NpmTimeoutError(NpmApiError):
    """The request did not complete in time — the server may still be working on it."""


def load_env_file(path):
    values = {}
    if not path.exists():
        return values
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        values[key.strip()] = value.strip().strip('"').strip("'")
    return values


class NpmClient:
    def __init__(self, base_url, identity, secret):
        self.base_url = base_url.rstrip("/")
        self.identity = identity
        self.secret = secret
        self.token = None
        self.token_expires = None

    def authenticate(self):
        result = self._request(
            "POST", "/tokens", {"identity": self.identity, "secret": self.secret}, auth=False
        )
        if "token" not in result:
            raise NpmApiError("login requires 2FA — not supported by this tool")
        self.token = result["token"]
        self.token_expires = result.get("expires")
        return self.token

    def get(self, path, params=None):
        if params:
            path = f"{path}?{urllib.parse.urlencode(params)}"
        return self._request("GET", path)

    def post(self, path, body, timeout=DEFAULT_TIMEOUT):
        return self._request("POST", path, body, timeout=timeout)

    def put(self, path, body):
        return self._request("PUT", path, body)

    def list_proxy_hosts(self):
        return self.get("/nginx/proxy-hosts", {"expand": "certificate"})

    def get_proxy_host(self, host_id):
        return self.get(f"/nginx/proxy-hosts/{host_id}")

    def list_certificates(self):
        return self.get("/nginx/certificates")

    def get_certificate(self, cert_id):
        return self.get(f"/nginx/certificates/{cert_id}")

    def set_certificate(self, host_id, cert_id, ssl_forced=None):
        body = {"certificate_id": cert_id}
        if ssl_forced is not None:
            body["ssl_forced"] = ssl_forced
        return self.put(f"/nginx/proxy-hosts/{host_id}", body)

    def create_proxy_host(self, body):
        return self.post("/nginx/proxy-hosts", body)

    def create_certificate(self, body, timeout=CERT_ISSUE_TIMEOUT):
        return self.post("/nginx/certificates", body, timeout=timeout)

    def test_http_challenge(self, domains):
        return self.post("/nginx/certificates/test-http", {"domains": domains}, timeout=60)

    def _request(self, method, path, body=None, auth=True, timeout=DEFAULT_TIMEOUT):
        url = f"{self.base_url}/api{path}"
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Content-Type", "application/json")
        req.add_header("User-Agent", "homelab-codex-npm-api/1.0")
        if auth:
            if self.token is None:
                self.authenticate()
            req.add_header("Authorization", f"Bearer {self.token}")
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                raw = resp.read()
                return json.loads(raw) if raw else None
        except urllib.error.HTTPError as e:
            raw = e.read()
            message = None
            try:
                message = json.loads(raw).get("error", {}).get("message")
            except (json.JSONDecodeError, AttributeError):
                pass
            raise NpmApiError(
                f"{method} {path} -> HTTP {e.code}: {message or raw.decode(errors='replace')}"
            ) from e
        except urllib.error.URLError as e:
            if isinstance(e.reason, TimeoutError):
                raise NpmTimeoutError(f"{method} {path} -> timed out after {timeout}s") from e
            raise NpmApiError(f"{method} {path} -> connection failed: {e.reason}") from e
        except TimeoutError as e:
            raise NpmTimeoutError(f"{method} {path} -> timed out after {timeout}s") from e


def print_table(headers, rows):
    if not rows:
        print("(no results)")
        return
    str_rows = [[str(cell) for cell in row] for row in rows]
    widths = [len(h) for h in headers]
    for row in str_rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))

    def fmt(row):
        return "  ".join(cell.ljust(widths[i]) for i, cell in enumerate(row))

    print(fmt(headers))
    print(fmt(["-" * w for w in widths]))
    for row in str_rows:
        print(fmt(row))


def cert_label(cert):
    if not cert:
        return "-"
    name = cert.get("nice_name") or ",".join(cert.get("domain_names", []))
    return f"{cert['id']} ({name})"


def cmd_token(client, args):
    client.authenticate()
    print(f"OK — authenticated as {client.identity} @ {client.base_url}")
    print(f"Token expires: {client.token_expires}")


def cmd_list_hosts(client, args):
    hosts = client.list_proxy_hosts()
    rows = []
    for h in hosts:
        label = cert_label(h.get("certificate")) if h.get("certificate") else (
            str(h["certificate_id"]) if h.get("certificate_id") else "-"
        )
        rows.append(
            [
                h["id"],
                ",".join(h["domain_names"]),
                f"{h['forward_scheme']}://{h['forward_host']}:{h['forward_port']}",
                label,
                "yes" if h["ssl_forced"] else "no",
                "yes" if h["enabled"] else "no",
            ]
        )
    print_table(["ID", "DOMAINS", "FORWARD", "CERT", "SSL_FORCED", "ENABLED"], rows)


def cmd_list_certs(client, args):
    certs = client.list_certificates()
    rows = [
        [
            c["id"],
            c.get("provider", "-"),
            c.get("nice_name", "-"),
            ",".join(c.get("domain_names", [])),
            c.get("expires_on", "-"),
        ]
        for c in certs
    ]
    print_table(["ID", "PROVIDER", "NICE_NAME", "DOMAINS", "EXPIRES_ON"], rows)


def resolve_id(positional, flag, name):
    """set-cert accepts both `set-cert 14 42` and `set-cert --host-id 14 --cert-id 42`."""
    if positional is not None and flag is not None and positional != flag:
        raise NpmApiError(f"conflicting values for {name}: {positional} (positional) vs {flag} (--{name})")
    value = flag if flag is not None else positional
    if value is None:
        raise NpmApiError(f"missing {name} — pass it positionally or as --{name}")
    return value


def cmd_set_cert(client, args):
    host_id = resolve_id(args.host_id_pos, args.host_id, "host-id")
    cert_id = resolve_id(args.cert_id_pos, args.cert_id, "cert-id")
    host = client.get_proxy_host(host_id)
    cert = client.get_certificate(cert_id)
    print(f"Host #{host['id']}: {', '.join(host['domain_names'])}")
    print(f"  current certificate_id: {host.get('certificate_id') or 0}")
    print(f"  new certificate_id:     {cert_label(cert)}")
    if args.ssl_forced is not None:
        print(f"  ssl_forced:             {host['ssl_forced']} -> {args.ssl_forced}")
    if not args.apply:
        print("\nDRY RUN — no changes made. Re-run with --apply to perform this change.")
        return
    updated = client.set_certificate(host_id, cert_id, args.ssl_forced)
    print(
        f"\nOK — host #{updated['id']} now uses certificate_id "
        f"{updated['certificate_id']} (ssl_forced={updated['ssl_forced']}); "
        "NPM reloaded nginx automatically."
    )


def build_cert_body(domains):
    # Schema (GET /api/schema, POST /nginx/certificates on NPM 2.14.0): only
    # provider / nice_name / domain_names / meta are accepted, and `meta` itself is
    # additionalProperties:false. There is NO letsencrypt_email / letsencrypt_agree —
    # certbot is always run with --agree-tos and -m <email of the owning NPM user>.
    return {
        "provider": "letsencrypt",
        "domain_names": domains,
        "meta": {"dns_challenge": False},
    }


def find_cert_by_domains(client, domains, exclude_ids):
    """Locate a cert NPM created for `domains` when the POST response was lost to a timeout."""
    wanted = sorted(domains)
    matches = [
        c
        for c in client.list_certificates()
        if c["id"] not in exclude_ids and sorted(c.get("domain_names", [])) == wanted
    ]
    return max(matches, key=lambda c: c["id"]) if matches else None


def poll_until_issued(client, cert_id, deadline, interval=3):
    """Wait for expires_on to appear on a cert row. Returns the cert, or None on timeout."""
    while True:
        cert = client.get_certificate(cert_id)
        if cert.get("expires_on"):
            return cert
        if time.monotonic() >= deadline:
            return None
        time.sleep(min(interval, max(0, deadline - time.monotonic())))


def cmd_create_cert(client, args):
    body = build_cert_body(args.domains)
    print("Planned certificate request (HTTP-01, no DNS challenge):")
    print(json.dumps(body, indent=2))
    if not args.apply:
        print("\nDRY RUN — no changes made. Re-run with --apply to request the certificate.")
        return

    existing_ids = {c["id"] for c in client.list_certificates()}
    deadline = time.monotonic() + args.timeout
    print(f"\nRequesting certificate from Let's Encrypt (timeout {args.timeout}s) ...")

    cert_id = None
    try:
        created = client.create_certificate(body, timeout=args.timeout)
        cert_id = created["id"]
        print(f"NPM accepted the request — certificate #{cert_id}")
    except NpmTimeoutError:
        # certbot keeps running server-side; find the row NPM inserted before the POST returned.
        print("POST timed out client-side — certbot may still be running, looking for the cert row ...")
        found = find_cert_by_domains(client, args.domains, existing_ids)
        if found is None:
            raise NpmApiError(
                f"certificate request timed out after {args.timeout}s and no certificate for "
                f"{', '.join(args.domains)} was created — check `docker logs npm` on the target node"
            ) from None
        cert_id = found["id"]
        print(f"Found pending certificate #{cert_id}")

    cert = poll_until_issued(client, cert_id, deadline)
    if cert is None:
        raise NpmApiError(
            f"certificate #{cert_id} was created but expires_on is still empty after {args.timeout}s "
            "— the Let's Encrypt challenge has not completed; check `docker logs npm` on the target node"
        )
    print(f"\nOK — certificate #{cert['id']} issued for {', '.join(cert['domain_names'])}")
    print(f"  expires_on: {cert['expires_on']}")
    print(f"\nNext: npm_api.py --npm <inst> set-cert --host-id <id> --cert-id {cert['id']} --ssl-forced --apply")


def build_create_body(args):
    cert_id = args.cert_id or 0
    ssl_forced = args.ssl_forced if args.ssl_forced is not None else cert_id != 0
    body = {
        "domain_names": args.domains,
        "forward_scheme": args.forward_scheme,
        "forward_host": args.forward_host,
        "forward_port": args.forward_port,
        "certificate_id": cert_id,
        "ssl_forced": ssl_forced,
        "http2_support": args.http2_support if args.http2_support is not None else ssl_forced,
        "block_exploits": args.block_exploits if args.block_exploits is not None else True,
        "allow_websocket_upgrade": args.allow_websocket_upgrade
        if args.allow_websocket_upgrade is not None
        else False,
        "caching_enabled": False,
        "enabled": True,
    }
    if args.advanced_config:
        body["advanced_config"] = args.advanced_config
    if args.access_list_id:
        body["access_list_id"] = args.access_list_id
    return body


def cmd_create_host(client, args):
    body = build_create_body(args)
    print("Planned proxy host:")
    print(json.dumps(body, indent=2))
    if not args.apply:
        print("\nDRY RUN — no changes made. Re-run with --apply to perform this change.")
        return
    created = client.create_proxy_host(body)
    print(f"\nOK — created proxy host #{created['id']}: {', '.join(created['domain_names'])}")


def parse_args():
    parser = argparse.ArgumentParser(
        prog="npm_api.py", description="Manage Nginx Proxy Manager (PIHA/VPS) via its REST API"
    )
    parser.add_argument("--npm", choices=sorted(INSTANCES), required=True, help="which NPM instance to target")
    parser.add_argument(
        "--env-file", type=Path, default=DEFAULT_ENV_FILE, help=f"path to credentials file (default: {DEFAULT_ENV_FILE})"
    )
    sub = parser.add_subparsers(dest="command", required=True)

    p_token = sub.add_parser("token", help="authenticate and print token info")
    p_token.set_defaults(func=cmd_token)

    p_hosts = sub.add_parser("list-hosts", help="list proxy hosts")
    p_hosts.set_defaults(func=cmd_list_hosts)

    p_certs = sub.add_parser("list-certs", help="list certificates")
    p_certs.set_defaults(func=cmd_list_certs)

    p_set_cert = sub.add_parser("set-cert", help="attach a certificate to a proxy host")
    p_set_cert.add_argument("host_id_pos", type=int, nargs="?", metavar="host_id")
    p_set_cert.add_argument("cert_id_pos", type=int, nargs="?", metavar="cert_id")
    p_set_cert.add_argument("--host-id", type=int, default=None, help="proxy host ID (alternative to positional)")
    p_set_cert.add_argument("--cert-id", type=int, default=None, help="certificate ID (alternative to positional)")
    p_set_cert.add_argument(
        "--ssl-forced", action=argparse.BooleanOptionalAction, default=None, help="also redirect HTTP to HTTPS"
    )
    p_set_cert.add_argument("--apply", action="store_true", help="perform the change (default: dry-run)")
    p_set_cert.set_defaults(func=cmd_set_cert)

    p_create_cert = sub.add_parser("create-cert", help="request a Let's Encrypt certificate (HTTP-01)")
    p_create_cert.add_argument("--domain", action="append", required=True, dest="domains", help="domain name (repeatable)")
    p_create_cert.add_argument(
        "--timeout",
        type=int,
        default=CERT_ISSUE_TIMEOUT,
        help=f"seconds to wait for issuance (default: {CERT_ISSUE_TIMEOUT})",
    )
    p_create_cert.add_argument("--apply", action="store_true", help="perform the change (default: dry-run)")
    p_create_cert.set_defaults(func=cmd_create_cert)

    p_create = sub.add_parser("create-host", help="create a new proxy host")
    p_create.add_argument("--domain", action="append", required=True, dest="domains", help="domain name (repeatable)")
    p_create.add_argument("--forward-host", required=True)
    p_create.add_argument("--forward-port", type=int, required=True)
    p_create.add_argument("--forward-scheme", choices=["http", "https"], default="http")
    p_create.add_argument("--cert-id", type=int, default=0, help="certificate ID (0 = none, default)")
    p_create.add_argument("--ssl-forced", action=argparse.BooleanOptionalAction, default=None)
    p_create.add_argument("--http2-support", action=argparse.BooleanOptionalAction, default=None)
    p_create.add_argument("--block-exploits", action=argparse.BooleanOptionalAction, default=None)
    p_create.add_argument("--websocket", action=argparse.BooleanOptionalAction, default=None, dest="allow_websocket_upgrade")
    p_create.add_argument("--access-list-id", type=int, default=0)
    p_create.add_argument("--advanced-config", default="")
    p_create.add_argument("--apply", action="store_true", help="perform the change (default: dry-run)")
    p_create.set_defaults(func=cmd_create_host)

    return parser.parse_args()


def main():
    args = parse_args()
    env = {**load_env_file(args.env_file), **os.environ}
    instance = INSTANCES[args.npm]
    base_url = env.get(instance["url_env"], instance["url_default"])

    try:
        identity = env[instance["user_env"]]
        secret = env[instance["pass_env"]]
    except KeyError as e:
        print(
            f"Error: missing {e.args[0]} — set it in {args.env_file} (see env.example)",
            file=sys.stderr,
        )
        return 1

    client = NpmClient(base_url, identity, secret)
    try:
        args.func(client, args)
    except NpmApiError as e:
        print(f"Error: {e}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())