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 urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

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

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


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):
        return self._request("POST", path, body)

    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):
        return self.put(f"/nginx/proxy-hosts/{host_id}", {"certificate_id": cert_id})

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

    def _request(self, method, path, body=None, auth=True):
        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=15) 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:
            raise NpmApiError(f"{method} {path} -> connection failed: {e.reason}") 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 cmd_set_cert(client, args):
    host = client.get_proxy_host(args.host_id)
    cert = client.get_certificate(args.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 not args.apply:
        print("\nDRY RUN — no changes made. Re-run with --apply to perform this change.")
        return
    updated = client.set_certificate(args.host_id, args.cert_id)
    print(
        f"\nOK — host #{updated['id']} now uses certificate_id "
        f"{updated['certificate_id']} (NPM reloaded nginx automatically)."
    )


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", type=int)
    p_set_cert.add_argument("cert_id", type=int)
    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 = 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())