316 lines
11 KiB
Python
316 lines
11 KiB
Python
|
#!/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())
|