255 lines
9.0 KiB
Python
Executable File
255 lines
9.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Create or update the Pangolin resource/target for Headscale.
|
|
|
|
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: headscale.paccoco.com
|
|
- site: Plausible Deniability (siteId 4)
|
|
- target: headscale:8080
|
|
- Pangolin auth: disabled (sso=false)
|
|
- healthcheck: GET /health
|
|
"""
|
|
|
|
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 = "headscale"
|
|
DEFAULT_SUBDOMAIN = "headscale"
|
|
DEFAULT_TARGET_IP = "headscale"
|
|
DEFAULT_TARGET_PORT = 8080
|
|
DEFAULT_HC_PATH = "/health"
|
|
USER_AGENT = "doris-pangolin-headscale/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 Headscale 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=DEFAULT_HC_PATH)
|
|
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,
|
|
"method": "http",
|
|
"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())
|