elmprodvpn/scripts/transport_runbook.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import os
import sys
from typing import Dict, List, Optional, Tuple
import urllib.error
import urllib.parse
import urllib.request


def request_json(api_url: str, method: str, path: str, payload: Optional[Dict] = None) -> Tuple[int, Dict]:
    data = None
    headers = {"Accept": "application/json"}
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"

    req = urllib.request.Request(
        f"{api_url.rstrip('/')}{path}",
        data=data,
        method=method.upper(),
        headers=headers,
    )
    try:
        with urllib.request.urlopen(req, timeout=30.0) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            status = int(resp.getcode() or 200)
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        status = int(e.code or 500)
    except Exception as e:
        return 0, {"ok": False, "message": str(e), "code": "HTTP_CLIENT_ERROR"}

    try:
        parsed = json.loads(raw) if raw else {}
    except Exception:
        parsed = {"raw": raw}
    if not isinstance(parsed, dict):
        parsed = {"raw": parsed}
    return status, parsed


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Transport runbook helper for /api/v1/transport/*")
    parser.add_argument("--api-url", default=os.environ.get("API_URL", "http://127.0.0.1:8080"))
    parser.add_argument("--client-id", default="")
    parser.add_argument("--kind", default="singbox", choices=["singbox", "dnstt", "phoenix"])
    parser.add_argument("--name", default="")
    parser.add_argument("--enabled", action="store_true")
    parser.add_argument("--config-json", default='{"runner":"mock","runtime_mode":"exec"}')
    parser.add_argument("--actions", default="capabilities")
    parser.add_argument("--force-delete", action="store_true")
    parser.add_argument("--allow-fail", default="")
    return parser.parse_args()


def require_client_id(action: str, client_id: str) -> None:
    if action == "capabilities":
        return
    if not client_id.strip():
        raise ValueError(f"--client-id is required for action '{action}'")


def format_summary(action: str, status: int, payload: Dict) -> str:
    ok = payload.get("ok")
    code = payload.get("code")
    msg = payload.get("message")
    extras: List[str] = []
    if "status" in payload:
        extras.append(f"status={payload.get('status')}")
    if "status_before" in payload:
        extras.append(f"before={payload.get('status_before')}")
    if "status_after" in payload:
        extras.append(f"after={payload.get('status_after')}")
    extra_text = f" {' '.join(extras)}" if extras else ""
    return f"[transport_runbook] {action}: http={status} ok={ok} code={code} message={msg}{extra_text}"


def main() -> int:
    args = parse_args()
    api_url = args.api_url.strip()
    if not api_url:
        print("[transport_runbook] ERROR: empty --api-url")
        return 1

    actions = [a.strip().lower() for a in args.actions.split(",") if a.strip()]
    if not actions:
        print("[transport_runbook] ERROR: --actions must not be empty")
        return 1
    allow_fail = {a.strip().lower() for a in args.allow_fail.split(",") if a.strip()}

    try:
        cfg = json.loads(args.config_json)
    except Exception as e:
        print(f"[transport_runbook] ERROR: invalid --config-json: {e}")
        return 1
    if not isinstance(cfg, dict):
        print("[transport_runbook] ERROR: --config-json must be a JSON object")
        return 1

    for action in actions:
        try:
            require_client_id(action, args.client_id)
        except ValueError as e:
            print(f"[transport_runbook] ERROR: {e}")
            return 1

        path = ""
        method = "GET"
        payload: Optional[Dict] = None

        if action == "capabilities":
            path = "/api/v1/transport/capabilities"
            method = "GET"
        elif action == "create":
            path = "/api/v1/transport/clients"
            method = "POST"
            payload = {
                "id": args.client_id,
                "name": args.name.strip() or f"Runbook {args.client_id}",
                "kind": args.kind,
                "enabled": bool(args.enabled),
                "config": cfg,
            }
        elif action == "provision":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/provision"
            method = "POST"
        elif action == "start":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/start"
            method = "POST"
        elif action == "health":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/health"
            method = "GET"
        elif action == "metrics":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/metrics"
            method = "GET"
        elif action == "restart":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/restart"
            method = "POST"
        elif action == "stop":
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}/stop"
            method = "POST"
        elif action == "delete":
            q = "?force=true" if args.force_delete else ""
            path = f"/api/v1/transport/clients/{urllib.parse.quote(args.client_id)}{q}"
            method = "DELETE"
        else:
            print(f"[transport_runbook] ERROR: unsupported action '{action}'")
            return 1

        status, res = request_json(api_url, method, path, payload)
        print(format_summary(action, status, res))
        if status != 200:
            if action in allow_fail:
                continue
            return 1
        if not bool(res.get("ok", False)) and action not in allow_fail:
            return 1

    print("[transport_runbook] done")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())