xc-agent-cloud-probe/app.py

"""Credential-free probe for the owner's ModelHub cloud deployment."""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from threading import Thread
from network_probe import run

external_token = os.environ.get("EXTERNAL_SERVICE_TOKEN", "")
metadata = {"name": "xc-cloud-build-probe", "version": "probe-20261026-04",
    "strategyIdPresent": bool(os.environ.get("STRATEGY_ID")),
    "externalCredentialPresent": bool(external_token),
    "externalCredentialLength": len(external_token),
    "externalCredentialShape": "jwt-like" if external_token.count(".") == 2 else "opaque" if external_token else "empty",
    "credentialEnvironmentNames": sorted(name for name in os.environ
        if any(marker in name for marker in ("TOKEN", "AUTH", "CREDENTIAL", "SECRET", "API_KEY")))}
print(json.dumps(metadata), flush=True)
Thread(target=run, daemon=True).start()


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        data = {"status": "ok"} if self.path.split("?", 1)[0] == "/health" else metadata
        payload = json.dumps(data).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)


ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()