feat: add interhub bootstrap helper

src/ops_hub/interhub_gate_probe.py

@@ -0,0 +1,111 @@
+"""Probe whether Inter-Hub production exposes the ops-hub bootstrap API."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from dataclasses import asdict, dataclass
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urljoin
+from urllib.request import Request, urlopen
+
+REQUIRED_OPENAPI_PATHS = (
+    "/hubs",
+    "/hub-capability-manifests",
+    "/api-consumers",
+    "/policy-scopes",
+)
+
+
+@dataclass(frozen=True)
+class HttpObservation:
+    url: str
+    status: int | None
+    ok: bool
+    error: str | None = None
+
+
+@dataclass(frozen=True)
+class GateResult:
+    base_url: str
+    passed: bool
+    hubs_status: int | None
+    required_paths_present: list[str]
+    required_paths_missing: list[str]
+    observations: list[HttpObservation]
+
+
+def normalize_base_url(base_url: str) -> str:
+    return base_url.rstrip("/") + "/"
+
+
+def api_url(base_url: str, path: str) -> str:
+    return urljoin(normalize_base_url(base_url), path.lstrip("/"))
+
+
+def fetch_json(url: str, timeout: float) -> tuple[HttpObservation, dict[str, Any] | None]:
+    request = Request(url, headers={"Accept": "application/json", "User-Agent": "ops-hub-gate-probe/0.1"})
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            payload = json.loads(response.read().decode("utf-8"))
+            return HttpObservation(url=url, status=response.status, ok=True), payload
+    except HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace")
+        return HttpObservation(url=url, status=exc.code, ok=False, error=body[:240]), None
+    except (URLError, TimeoutError, json.JSONDecodeError) as exc:
+        return HttpObservation(url=url, status=None, ok=False, error=str(exc)), None
+
+
+def observe_status(url: str, timeout: float) -> HttpObservation:
+    request = Request(url, headers={"Accept": "application/json", "User-Agent": "ops-hub-gate-probe/0.1"})
+    try:
+        with urlopen(request, timeout=timeout) as response:
+            response.read()
+            return HttpObservation(url=url, status=response.status, ok=True)
+    except HTTPError as exc:
+        exc.read()
+        return HttpObservation(url=url, status=exc.code, ok=False, error=exc.reason)
+    except (URLError, TimeoutError) as exc:
+        return HttpObservation(url=url, status=None, ok=False, error=str(exc))
+
+
+def evaluate_openapi_paths(openapi: dict[str, Any] | None) -> tuple[list[str], list[str]]:
+    paths = set((openapi or {}).get("paths", {}).keys())
+    present = [path for path in REQUIRED_OPENAPI_PATHS if path in paths]
+    missing = [path for path in REQUIRED_OPENAPI_PATHS if path not in paths]
+    return present, missing
+
+
+def probe_interhub_gate(base_url: str, timeout: float = 10.0) -> GateResult:
+    normalized = normalize_base_url(base_url)
+    hubs = observe_status(api_url(normalized, "/api/v2/hubs"), timeout)
+    openapi_observation, openapi = fetch_json(api_url(normalized, "/api/v2/openapi.json"), timeout)
+    present, missing = evaluate_openapi_paths(openapi)
+    passed = hubs.status == 401 and not missing
+    return GateResult(
+        base_url=normalized.rstrip("/"),
+        passed=passed,
+        hubs_status=hubs.status,
+        required_paths_present=present,
+        required_paths_missing=missing,
+        observations=[hubs, openapi_observation],
+    )
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base", default="https://hub.coulomb.social", help="Inter-Hub base URL")
+    parser.add_argument("--timeout", default=10.0, type=float, help="HTTP timeout in seconds")
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    result = probe_interhub_gate(args.base, timeout=args.timeout)
+    print(json.dumps(asdict(result), indent=2, sort_keys=True))
+    return 0 if result.passed else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())