#!/usr/bin/env python3 """Create or update the Pangolin resource/target for Dispatcharr. Expected env vars: - PANGOLIN_API_URL e.g. https://api.paccoco.com/v1 - PANGOLIN_API_KEY bearer token for Pangolin integration API - PANGOLIN_ORG_ID e.g. paccoco Optional env defaults: - PANGOLIN_SITE_ID default 4 (Plausible Deniability) - PANGOLIN_DOMAIN_ID default domain1 Default desired resource: - host/domain: tv.paccoco.com - site: Plausible Deniability (siteId 4) - target: dispatcharr:9191 - Pangolin auth: disabled (sso=false) """ from __future__ import annotations import argparse import json import os import sys from typing import Any from urllib import error, parse, request DEFAULT_API_URL = "https://api.paccoco.com/v1" DEFAULT_ORG_ID = "paccoco" DEFAULT_DOMAIN_ID = "domain1" DEFAULT_SITE_ID = 4 DEFAULT_NAME = "dispatcharr" DEFAULT_SUBDOMAIN = "tv" DEFAULT_TARGET_IP = "dispatcharr" DEFAULT_TARGET_PORT = 9191 USER_AGENT = "doris-pangolin-dispatcharr/1.0" def env(name: str, default: str | None = None) -> str | None: value = os.environ.get(name) if value is None or value == "": return default return value def api_request(base_url: str, api_key: str, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]: url = base_url.rstrip("/") + path headers = { "Authorization": f"Bearer {api_key}", "Accept": "application/json", "User-Agent": USER_AGENT, } data = None if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" req = request.Request(url, data=data, headers=headers, method=method.upper()) try: with request.urlopen(req, timeout=30) as resp: raw = resp.read().decode("utf-8", "replace") return resp.status, json.loads(raw) if raw else None except error.HTTPError as exc: raw = exc.read().decode("utf-8", "replace") try: payload = json.loads(raw) except Exception: payload = raw return exc.code, payload def fail(msg: str, payload: Any = None) -> int: print(msg, file=sys.stderr) if payload is not None: if isinstance(payload, (dict, list)): print(json.dumps(payload, indent=2, sort_keys=True), file=sys.stderr) else: print(payload, file=sys.stderr) return 1 def find_resource(resources: list[dict[str, Any]], *, name: str, full_domain: str) -> dict[str, Any] | None: for resource in resources: if resource.get("fullDomain") == full_domain: return resource for resource in resources: if resource.get("name") == name: return resource return None def main() -> int: parser = argparse.ArgumentParser(description="Upsert Pangolin Dispatcharr resource") parser.add_argument("--api-url", default=env("PANGOLIN_API_URL", DEFAULT_API_URL)) parser.add_argument("--api-key", default=env("PANGOLIN_API_KEY")) parser.add_argument("--org-id", default=env("PANGOLIN_ORG_ID", DEFAULT_ORG_ID)) site_id_default = env("PANGOLIN_SITE_ID", str(DEFAULT_SITE_ID)) or str(DEFAULT_SITE_ID) parser.add_argument("--site-id", type=int, default=int(site_id_default)) parser.add_argument("--domain-id", default=env("PANGOLIN_DOMAIN_ID", DEFAULT_DOMAIN_ID)) parser.add_argument("--name", default=DEFAULT_NAME) parser.add_argument("--subdomain", default=DEFAULT_SUBDOMAIN) parser.add_argument("--target-ip", default=DEFAULT_TARGET_IP) parser.add_argument("--target-port", type=int, default=DEFAULT_TARGET_PORT) parser.add_argument("--hc-path", default="/") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() if not args.api_key: return fail("Missing PANGOLIN_API_KEY / --api-key") full_domain = f"{args.subdomain}.paccoco.com" status, resources_resp = api_request( args.api_url, args.api_key, "GET", f"/org/{parse.quote(args.org_id)}/resources?pageSize=200", ) if status != 200: return fail("Failed to list Pangolin resources", resources_resp) resources = resources_resp.get("data", {}).get("resources", []) resource = find_resource(resources, name=args.name, full_domain=full_domain) create_body = { "name": args.name, "subdomain": args.subdomain, "http": True, "protocol": "tcp", "domainId": args.domain_id, } if resource is None: if args.dry_run: print(json.dumps({"action": "create_resource", "body": create_body}, indent=2)) resource_id = None else: status, create_resp = api_request( args.api_url, args.api_key, "PUT", f"/org/{parse.quote(args.org_id)}/resource", create_body, ) if status not in (200, 201): return fail("Failed to create Pangolin resource", create_resp) resource_id = create_resp.get("data", {}).get("resourceId") or create_resp.get("data", {}).get("resource", {}).get("resourceId") if not resource_id: return fail("Create resource succeeded but resourceId was missing", create_resp) status, resource_resp = api_request(args.api_url, args.api_key, "GET", f"/resource/{resource_id}") if status != 200: return fail("Failed to read back created resource", resource_resp) resource = resource_resp.get("data", {}).get("resource") or resource_resp.get("data") or {} else: resource_id = resource["resourceId"] resource_update = { "name": args.name, "niceId": resource.get("niceId") if resource else None, "subdomain": args.subdomain, "ssl": True, "sso": False, "blockAccess": False, "emailWhitelistEnabled": False, "applyRules": False, "domainId": args.domain_id, "enabled": True, "stickySession": False, "tlsServerName": None, "setHostHeader": None, "skipToIdpId": None, "headers": None, "maintenanceModeEnabled": False, "maintenanceModeType": "forced", "maintenanceTitle": None, "maintenanceMessage": None, "maintenanceEstimatedTime": None, "postAuthPath": None, } target_body = { "siteId": args.site_id, "ip": args.target_ip, "port": args.target_port, "enabled": True, "hcEnabled": True, "hcPath": args.hc_path, "hcScheme": "http", "hcMode": "http", "hcHostname": args.target_ip, "hcPort": args.target_port, "hcInterval": 30, "hcUnhealthyInterval": 30, "hcTimeout": 10, "hcHeaders": None, "hcFollowRedirects": True, "hcMethod": "GET", "hcStatus": 200, "hcTlsServerName": None, "hcHealthyThreshold": 1, "hcUnhealthyThreshold": 3, "path": None, "pathMatchType": None, "rewritePath": None, "rewritePathType": None, } existing_targets = resource.get("targets", []) if resource else [] matching_target = None for target in existing_targets: if target.get("siteId") == args.site_id: matching_target = target break if args.dry_run: print(json.dumps({ "resource_id": resource_id, "full_domain": full_domain, "resource_update": resource_update, "target_action": "update" if matching_target else "create", "target_id": matching_target.get("targetId") if matching_target else None, "target_body": target_body, }, indent=2)) return 0 if resource_id is not None: status, update_resp = api_request(args.api_url, args.api_key, "POST", f"/resource/{resource_id}", resource_update) if status not in (200, 201): return fail("Failed to update Pangolin resource", update_resp) if matching_target: target_id = matching_target["targetId"] status, target_resp = api_request(args.api_url, args.api_key, "POST", f"/target/{target_id}", target_body) if status not in (200, 201): return fail("Failed to update Pangolin target", target_resp) else: status, target_resp = api_request(args.api_url, args.api_key, "PUT", f"/resource/{resource_id}/target", target_body) if status not in (200, 201): return fail("Failed to create Pangolin target", target_resp) status, final_resp = api_request(args.api_url, args.api_key, "GET", f"/resource/{resource_id}") if status != 200: return fail("Failed to read back final Pangolin resource", final_resp) payload = final_resp.get("data", {}).get("resource") or final_resp.get("data") or final_resp print(json.dumps(payload, indent=2)) return 0 return fail("Unexpected state: resource_id is missing") if __name__ == "__main__": raise SystemExit(main())