probe v1.0.1 dump env + candidate scan

main.py

@@ -1,14 +1,16 @@
-"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only)
+"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only) v1.0.1
 
 用途:
   1) 验证平台「建仓 -> Kaniko 构建 -> 部署 -> /health -> 运行/停止」整条链路。
-  2) 零副作用观测平台注入的 STRATEGY_ID / EXTERNAL_SERVICE_TOKEN 是否为有效凭证
-     (启动时用注入 token 调一次只读 GET /api/adapt/task/page,看返回 code:
-      0 = 凭证有效; 40100 = 未登录/凭证无效)。
+  2) 零副作用诊断平台注入的凭证:
+     - dump 所有环境变量的「名字 + 值长度」(不打值),看真 token 是否藏在别的变量名里。
+     - 打印 EXTERNAL_SERVICE_TOKEN 的值(短值必是废值、直接看;长值打码,避免泄露真 token)。
+     - 对每个「长得像 token」(len>=16) 的环境变量,试当 Xc-Token 调一次只读接口,看哪个能通(code=0)。
+     - 仍用 EXTERNAL_SERVICE_TOKEN 三种头各探一次,作对照。
 
 严格约束:
   - 只服务 GET /health 与 GET /,绝不提交任何验证任务 (不调 task/add, 不调 build-config)。
-  - token 只从环境读,绝不打印其值/落盘;探针只调只读 GET;只记录 http/code/计数,不打印返回体内容(防 PII)。
+  - 只读 GET;只记录 http/code/计数,不打印任何接口返回体内容(防 PII);不打印任何长凭证的完整值。
   - 纯 stdlib 零依赖;正确处理 SIGTERM 优雅停机。
 """
 import json
@@ -21,20 +23,24 @@ import urllib.request
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn")
-TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "")     # 平台注入的服务 token
-STRATEGY_ID = os.getenv("STRATEGY_ID", "")          # 平台注入的自身策略 id
+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": []}              # 探针结果,暴露在 GET / (不含任何 token 值)
+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:
-    """只读 GET,返回 {http, code, count};绝不返回/记录响应体内容。"""
     req = urllib.request.Request(MAIN + path, method="GET", headers=headers)
     try:
         with urllib.request.urlopen(req, timeout=timeout) as r:
@@ -62,10 +68,25 @@ def _readonly_get(path: str, headers: dict, timeout: int = 20) -> dict:
 
 
 def probe() -> None:
-    """一次性只读探针:验证注入凭证有效性 + 记录 STRATEGY_ID。"""
-    log(f"probe start | strategy_id={STRATEGY_ID or 'MISSING'} | "
-        f"token_present={bool(TOKEN)} | token_len={len(TOKEN)}")
+    # 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}),
@@ -73,10 +94,11 @@ def probe() -> None:
     ]
     for name, headers in trials:
         res = {"skip": "no-token"} if not TOKEN else _readonly_get(path, headers)
-        PROBE["results"].append({"auth": name, **res})
-        log(f"probe {name} -> {json.dumps(res, ensure_ascii=False)}")
+        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 => 该头凭证有效; code=40100 => 未登录/凭证无效")
+    log("probe done | 判读: 任一 code=0 => 该来源凭证有效; 全 401/40100 => 无有效凭证")
 
 
 class Handler(BaseHTTPRequestHandler):
@@ -93,13 +115,9 @@ class Handler(BaseHTTPRequestHandler):
             self._json({"status": "ok"})
             return
         if self.path == "/":
-            self._json({
-                "name": "huni-probe-agent",
-                "mode": "probe-only",
-                "strategy_id_present": bool(STRATEGY_ID),
-                "token_present": bool(TOKEN),
-                "probe": PROBE,
-            })
+            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)
 
@@ -119,7 +137,7 @@ def main() -> None:
 
     server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
     server.timeout = 1
-    log(f"probe-only agent listening on 0.0.0.0:{PORT}")
+    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()