149 lines
5.7 KiB
Python
149 lines
5.7 KiB
Python
"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only) v1.0.1
|
|
|
|
用途:
|
|
1) 验证平台「建仓 -> Kaniko 构建 -> 部署 -> /health -> 运行/停止」整条链路。
|
|
2) 零副作用诊断平台注入的凭证:
|
|
- dump 所有环境变量的「名字 + 值长度」(不打值),看真 token 是否藏在别的变量名里。
|
|
- 打印 EXTERNAL_SERVICE_TOKEN 的值(短值必是废值、直接看;长值打码,避免泄露真 token)。
|
|
- 对每个「长得像 token」(len>=16) 的环境变量,试当 Xc-Token 调一次只读接口,看哪个能通(code=0)。
|
|
- 仍用 EXTERNAL_SERVICE_TOKEN 三种头各探一次,作对照。
|
|
|
|
严格约束:
|
|
- 只服务 GET /health 与 GET /,绝不提交任何验证任务 (不调 task/add, 不调 build-config)。
|
|
- 只读 GET;只记录 http/code/计数,不打印任何接口返回体内容(防 PII);不打印任何长凭证的完整值。
|
|
- 纯 stdlib 零依赖;正确处理 SIGTERM 优雅停机。
|
|
"""
|
|
import json
|
|
import os
|
|
import signal
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn")
|
|
TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "")
|
|
STRATEGY_ID = os.getenv("STRATEGY_ID", "")
|
|
PORT = int(os.getenv("PORT", "8080"))
|
|
|
|
shutdown = threading.Event()
|
|
PROBE = {"done": False, "results": []}
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
|
|
|
|
|
|
def _mask(v: str) -> str:
|
|
"""短值(必是废值)原样看;长值只留头尾,避免泄露可能的真 token。"""
|
|
return v if len(v) <= 8 else (v[:2] + "…" + v[-2:])
|
|
|
|
|
|
def _readonly_get(path: str, headers: dict, timeout: int = 20) -> dict:
|
|
req = urllib.request.Request(MAIN + path, method="GET", headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
body = r.read().decode("utf-8", "ignore")
|
|
status = r.status
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode("utf-8", "ignore")
|
|
status = e.code
|
|
except Exception as e: # noqa: BLE001
|
|
return {"http": -1, "err": str(e)[:120]}
|
|
code = count = None
|
|
try:
|
|
j = json.loads(body)
|
|
code = j.get("code")
|
|
d = j.get("data")
|
|
if isinstance(d, dict):
|
|
count = d.get("total")
|
|
if count is None and isinstance(d.get("records"), list):
|
|
count = len(d["records"])
|
|
elif isinstance(d, list):
|
|
count = len(d)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
return {"http": status, "code": code, "count": count}
|
|
|
|
|
|
def probe() -> None:
|
|
# 1) dump 所有 env 的 key + 值长度(不打值),找隐藏的真 token
|
|
env_lens = {k: len(str(os.environ.get(k, ""))) for k in sorted(os.environ)}
|
|
log("env keys+lens: " + json.dumps(env_lens, ensure_ascii=False))
|
|
|
|
# 2) EXTERNAL_SERVICE_TOKEN 的值(短则全打,长则打码)+ STRATEGY_ID
|
|
log(f"strategy_id={STRATEGY_ID or 'MISSING'} | "
|
|
f"EXTERNAL_SERVICE_TOKEN value={_mask(TOKEN)!r} len={len(TOKEN)}")
|
|
|
|
path = "/api/adapt/task/page?current=1&pageSize=1"
|
|
|
|
# 3) 对每个「长得像 token」(len>=16) 的 env 值,试当 Xc-Token,看哪个能通
|
|
candidates = [k for k in sorted(os.environ) if len(str(os.environ.get(k, ""))) >= 16]
|
|
log("token-like env candidates (len>=16): " + json.dumps(candidates, ensure_ascii=False))
|
|
for k in candidates:
|
|
res = _readonly_get(path, {"Xc-Token": os.environ[k]})
|
|
PROBE["results"].append({"try_env": k, "as": "Xc-Token", **res})
|
|
log(f"try Xc-Token from ${k} -> {json.dumps(res, ensure_ascii=False)}")
|
|
|
|
# 4) EXTERNAL_SERVICE_TOKEN 三种头对照
|
|
trials = [
|
|
("Xc-Token", {"Xc-Token": TOKEN}),
|
|
("Authorization-Bearer", {"Authorization": "Bearer " + TOKEN}),
|
|
("Authorization-raw", {"Authorization": TOKEN}),
|
|
]
|
|
for name, headers in trials:
|
|
res = {"skip": "no-token"} if not TOKEN else _readonly_get(path, headers)
|
|
PROBE["results"].append({"est_token_hdr": name, **res})
|
|
log(f"EST-token {name} -> {json.dumps(res, ensure_ascii=False)}")
|
|
|
|
PROBE["done"] = True
|
|
log("probe done | 判读: 任一 code=0 => 该来源凭证有效; 全 401/40100 => 无有效凭证")
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def _json(self, obj: dict, status: int = 200) -> None:
|
|
payload = json.dumps(obj, ensure_ascii=False).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path == "/health":
|
|
self._json({"status": "ok"})
|
|
return
|
|
if self.path == "/":
|
|
self._json({"name": "huni-probe-agent", "mode": "probe-only-v1.0.1",
|
|
"strategy_id_present": bool(STRATEGY_ID),
|
|
"token_present": bool(TOKEN), "probe": PROBE})
|
|
return
|
|
self._json({"error": "not found"}, 404)
|
|
|
|
def log_message(self, *_a) -> None:
|
|
pass
|
|
|
|
|
|
def _handle_signal(signum: int, _frame) -> None:
|
|
shutdown.set()
|
|
log(f"received signal {signum}, shutting down")
|
|
|
|
|
|
def main() -> None:
|
|
signal.signal(signal.SIGTERM, _handle_signal)
|
|
signal.signal(signal.SIGINT, _handle_signal)
|
|
threading.Thread(target=probe, daemon=True).start()
|
|
|
|
server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
|
|
server.timeout = 1
|
|
log(f"probe-only agent v1.0.1 listening on 0.0.0.0:{PORT}")
|
|
while not shutdown.is_set():
|
|
server.handle_request()
|
|
server.server_close()
|
|
log("stopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|