Add automatic CPA Management API import after mint

Upload xai-*.json to POST /v0/management/auth-files when cpa_auto_import_remote is enabled. Failures are logged only and never block registration.

cpa_export.py

@@ -3,6 +3,10 @@
 OIDC package lives at ./cpa_xai (bundled with this project).
 Optional override: config `api_reverse_tools` / env `API_REVERSE_TOOLS`
 points at a directory that *contains* the `cpa_xai` package.
+
+After mint, optionally:
+  - copy into local CPA auth-dir (`cpa_copy_to_hotload`)
+  - POST into remote CPA Management API (`cpa_auto_import_remote`)
 """
 
 from __future__ import annotations
@@ -13,6 +17,7 @@ import sys
 import time
 from pathlib import Path
 from typing import Any, Callable
+from urllib.parse import quote, urlparse, urlunparse
 
 _REG_DIR = Path(__file__).resolve().parent
 _DEFAULT_OUT = _REG_DIR / "cpa_auths"
@@ -64,6 +69,191 @@ def export_cookies_from_page(page: Any) -> list[dict]:
     return []
 
 
+def _normalize_cpa_remote_base(base: str) -> str:
+    """Normalize user-provided CPA root URL to scheme://host[:port].
+
+    Accepts:
+      http://127.0.0.1:8317
+      http://host:8317/
+      http://host:8317/v0/management
+      http://host:8317/v0/management/auth-files
+    """
+    raw = (base or "").strip()
+    if not raw:
+        return ""
+    if "://" not in raw:
+        raw = "http://" + raw
+    parsed = urlparse(raw)
+    if not parsed.scheme or not parsed.netloc:
+        return raw.rstrip("/")
+    # Keep only scheme + netloc (drop path/query/fragment)
+    return urlunparse((parsed.scheme, parsed.netloc, "", "", "", "")).rstrip("/")
+
+
+def resolve_cpa_remote_settings(cfg: dict | None = None) -> dict[str, Any]:
+    """Resolve remote CPA import settings from config + env."""
+    cfg = cfg or {}
+    enabled = bool(cfg.get("cpa_auto_import_remote", False))
+    base = (
+        (cfg.get("cpa_remote_base") or "").strip()
+        or (os.environ.get("CPA_REMOTE_BASE") or "").strip()
+    )
+    password = (
+        (cfg.get("cpa_remote_password") or "").strip()
+        or (os.environ.get("CPA_REMOTE_PASSWORD") or "").strip()
+        or (os.environ.get("CPA_MANAGEMENT_PASSWORD") or "").strip()
+        or (os.environ.get("MANAGEMENT_PASSWORD") or "").strip()
+    )
+    retries = int(cfg.get("cpa_remote_import_retries", 3) or 3)
+    delay = float(cfg.get("cpa_remote_import_retry_delay", 2) or 2)
+    return {
+        "enabled": enabled,
+        "base": _normalize_cpa_remote_base(base),
+        "password": password,
+        "retries": max(1, retries),
+        "delay": max(0.0, delay),
+    }
+
+
+def import_cpa_auth_to_remote(
+    path: str | Path,
+    *,
+    base: str,
+    password: str,
+    retries: int = 3,
+    delay: float = 2.0,
+    log: Callable[[str], None] | None = None,
+    timeout: float = 12.0,
+) -> dict[str, Any]:
+    """Upload one local xai-*.json to CPA Management API.
+
+    POST {base}/v0/management/auth-files?name=<filename>
+    Authorization: Bearer <password>
+    Body: raw JSON file content
+
+    Failures are returned in the result dict; caller decides whether to ignore.
+    """
+    _log = log or (lambda m: print(m, flush=True))
+    src = Path(path)
+    root = _normalize_cpa_remote_base(base)
+    if not root:
+        return {"ok": False, "error": "empty cpa_remote_base"}
+    if not password:
+        return {"ok": False, "error": "empty cpa_remote_password"}
+    if not src.is_file():
+        return {"ok": False, "error": f"auth file not found: {src}"}
+
+    name = src.name
+    if not name.endswith(".json"):
+        return {"ok": False, "error": f"filename must end with .json: {name}"}
+
+    try:
+        body = src.read_bytes()
+    except Exception as e:  # noqa: BLE001
+        return {"ok": False, "error": f"read file: {e}"}
+
+    url = f"{root}/v0/management/auth-files?name={quote(name)}"
+    headers = {
+        "Authorization": f"Bearer {password}",
+        "X-Management-Key": password,
+        "Content-Type": "application/json",
+    }
+
+    try:
+        from curl_cffi import requests as _req  # type: ignore
+    except Exception as e:  # noqa: BLE001
+        return {"ok": False, "error": f"import curl_cffi failed: {e}"}
+
+    last_err = "unknown"
+    attempts = max(1, int(retries or 1))
+    for i in range(1, attempts + 1):
+        try:
+            # Admin API is local/remote management — do not use outbound proxy.
+            resp = _req.post(
+                url,
+                data=body,
+                headers=headers,
+                timeout=timeout,
+                proxies={},
+                impersonate="chrome",
+            )
+            status = int(getattr(resp, "status_code", 0) or 0)
+            text = ""
+            try:
+                text = (resp.text or "")[:500]
+            except Exception:
+                text = ""
+            if status in (200, 207):
+                # Prefer JSON status field when present
+                ok_status = True
+                try:
+                    j = resp.json()
+                    st = str((j or {}).get("status") or "").lower()
+                    if st and st not in ("ok", "partial", "success"):
+                        ok_status = False
+                        last_err = f"status={st} body={text}"
+                except Exception:
+                    pass
+                if ok_status:
+                    _log(f"[cpa] remote import ok -> {root} name={name}")
+                    return {
+                        "ok": True,
+                        "base": root,
+                        "name": name,
+                        "status_code": status,
+                        "body": text,
+                    }
+            else:
+                last_err = f"HTTP {status}: {text or '(empty)'}"
+        except Exception as e:  # noqa: BLE001
+            last_err = str(e)
+
+        if i < attempts:
+            _log(f"[cpa] remote import retry {i}/{attempts}: {last_err}")
+            if delay > 0:
+                time.sleep(delay)
+
+    _log(f"[cpa] remote import failed: {last_err}")
+    return {
+        "ok": False,
+        "base": root,
+        "name": name,
+        "error": last_err,
+    }
+
+
+def maybe_import_cpa_auth_remote(
+    path: str | Path,
+    *,
+    config: dict | None = None,
+    log_callback: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    """Config-gated remote import helper (safe no-op when disabled)."""
+    cfg = config or {}
+    log = log_callback or (lambda m: print(m, flush=True))
+    settings = resolve_cpa_remote_settings(cfg)
+    if not settings["enabled"]:
+        return {"ok": False, "skipped": True, "reason": "disabled"}
+    if not settings["base"] or not settings["password"]:
+        log(
+            "[cpa] remote import enabled but missing cpa_remote_base / "
+            "cpa_remote_password (or CPA_REMOTE_BASE / CPA_REMOTE_PASSWORD)"
+        )
+        return {
+            "ok": False,
+            "skipped": True,
+            "reason": "missing_base_or_password",
+        }
+    return import_cpa_auth_to_remote(
+        path,
+        base=settings["base"],
+        password=settings["password"],
+        retries=settings["retries"],
+        delay=settings["delay"],
+        log=log,
+    )
+
+
 def export_cpa_xai_for_account(
     email: str,
     password: str,
@@ -220,6 +410,29 @@ def export_cpa_xai_for_account(
             log(f"[cpa] hotload copy failed: {e}")
             result["cpa_copy_error"] = str(e)
 
+    # Optional: push auth file into remote CPA via Management API.
+    # Failures never flip result["ok"] — local mint already succeeded.
+    if result.get("ok") and result.get("path"):
+        try:
+            remote = maybe_import_cpa_auth_remote(
+                result["path"],
+                config=cfg,
+                log_callback=log,
+            )
+            if remote.get("skipped"):
+                result["remote_import_skipped"] = remote.get("reason") or "skipped"
+            elif remote.get("ok"):
+                result["remote_import_ok"] = True
+                result["remote_import_base"] = remote.get("base")
+                result["remote_import_name"] = remote.get("name")
+            else:
+                result["remote_import_ok"] = False
+                result["remote_import_error"] = remote.get("error") or "unknown"
+        except Exception as e:  # noqa: BLE001
+            log(f"[cpa] remote import unexpected error: {e}")
+            result["remote_import_ok"] = False
+            result["remote_import_error"] = str(e)
+
     # failure log under register dir
     if not result.get("ok"):
         fail_path = out_dir / "cpa_auth_failed.txt"