修复运行时凭证与网络出口兼容性

README.md

@@ -17,13 +17,14 @@
 - 仓库根目录包含 `Dockerfile`，监听 `8080`。
 - `GET /health` 始终返回 HTTP 200，并附带扫描器最近状态。
 - 从平台注入的 `STRATEGY_ID` 获取自身策略 ID。
-- 优先读取 `MODELHUB_XC_TOKEN`，同时兼容官方参考策略使用的 `EXTERNAL_SERVICE_TOKEN`。
+- 优先读取平台 Secret 挂载路径 `XC_TOKEN_FILE`，同时兼容 `MODELHUB_XC_TOKEN`、`XC_TOKEN` 与官方参考策略使用的 `EXTERNAL_SERVICE_TOKEN`。
+- 保留 Python 默认的 HTTP(S) 代理处理，兼容平台容器的受控网络出口。
 - 正确处理 `SIGTERM`，在平台 30 秒窗口内退出。
 - 仅使用 Python 标准库，符合 1 CPU / 512 MiB 的平台限制。
 
 ## 安全边界
 
-- 令牌只从运行环境读取，不写入仓库、响应或日志。
+- 令牌只从运行环境或平台挂载的 Secret 文件读取，不写入仓库、响应或日志。
 - stdout 只记录令牌是否存在，不记录令牌值、请求头或完整 API 响应。
 - 提交前必须完成精确去重；API 返回非成功业务码时不会伪报成功。
 - 未获得令牌或策略 ID 时会明确记录 `task_submission_blocked`，不会静默假运行。

main.py

@@ -14,13 +14,14 @@ import signal
 import threading
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from typing import Any, Mapping
+from pathlib import Path
+from typing import Any, Callable, Mapping
 from urllib.parse import quote, urlencode, urlsplit
-from urllib.request import ProxyHandler, Request, build_opener
+from urllib.request import Request, build_opener
 
 
 AGENT_NAME = "xc-model-auto-adaptation-agent"
-AGENT_VERSION = "1.0.0"
+AGENT_VERSION = "1.0.1"
 TARGET_VENDOR = "天数智芯"
 TARGET_CARD = "天垓100"
 DEFAULT_TARGET_GPU = "Iluvatar_bi-100"
@@ -103,7 +104,10 @@ CONFIG_PARAMS = os.getenv("MODELHUB_CONFIG_PARAMS", DEFAULT_CONFIG_PARAMS)
 SECRET_ASSIGNMENT_RE = re.compile(
     r"(?i)(token|secret|password|api[_-]?key)=([^&\s]+)"
 )
-HTTP_OPENER = build_opener(ProxyHandler({}))
+# Keep urllib's standard proxy handling. The hosted runtime may provide its
+# outbound route through HTTP(S)_PROXY, so disabling proxies can isolate the
+# scanner even though the container itself remains healthy.
+HTTP_OPENER = build_opener()
 STATE_LOCK = threading.Lock()
 STOP = threading.Event()
 SCANNER_STATE: dict[str, Any] = {
@@ -125,15 +129,39 @@ class PlatformAPIError(RuntimeError):
     """Raised when ModelHub returns a non-success business response."""
 
 
-def resolve_runtime_token(environ: Mapping[str, str] | None = None) -> str:
+TOKEN_PLACEHOLDERS = {"tmp", "placeholder", "changeme", "change-me", "test"}
-    """Resolve documented/compatible secret names without ever logging values."""
+
+
+def _valid_runtime_token(value: str) -> bool:
+    value = value.strip()
+    return bool(value) and value.lower() not in TOKEN_PLACEHOLDERS
+
+
+def resolve_runtime_token(
+    environ: Mapping[str, str] | None = None,
+    read_text: Callable[[str], str] | None = None,
+) -> str:
+    """Resolve the platform credential without ever logging its value."""
 
     source = environ if environ is not None else os.environ
-    return (
+    token_file = source.get("XC_TOKEN_FILE", "").strip()
-        source.get("MODELHUB_XC_TOKEN", "")
+    if token_file:
-        or source.get("EXTERNAL_SERVICE_TOKEN", "")
+        file_reader = read_text or (
-        or source.get("XC_TOKEN", "")
+            lambda path: Path(path).read_text(encoding="utf-8")
         )
+        try:
+            value = file_reader(token_file).strip()
+        except (OSError, UnicodeError):
+            return ""
+        return value if _valid_runtime_token(value) else ""
+
+    for name in ("MODELHUB_XC_TOKEN", "XC_TOKEN", "EXTERNAL_SERVICE_TOKEN"):
+        value = source.get(name, "").strip()
+        if _valid_runtime_token(value):
+            return value
+        if value:
+            return ""
+    return ""
 
 
 XC_TOKEN = resolve_runtime_token()
@@ -187,7 +215,10 @@ def _http_json(
     """Call one JSON endpoint without logging headers, bodies, or credentials."""
 
     body = None
-    headers = {"Accept": "application/json"}
+    headers = {
+        "Accept": "application/json",
+        "User-Agent": f"{AGENT_NAME}/{AGENT_VERSION}",
+    }
     if payload is not None:
         body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
         headers["Content-Type"] = "application/json"

test_main.py

@@ -42,6 +42,19 @@ class AutoAdaptationTests(unittest.TestCase):
         token = resolve_runtime_token({"EXTERNAL_SERVICE_TOKEN": "private-value"})
         self.assertEqual(token, "private-value")
 
+    def test_runtime_token_file_is_supported(self):
+        token = resolve_runtime_token(
+            {"XC_TOKEN_FILE": "/run/secrets/xc-token"},
+            read_text=lambda path: "mounted-private-value\n"
+            if path == "/run/secrets/xc-token"
+            else "",
+        )
+        self.assertEqual(token, "mounted-private-value")
+
+    def test_placeholder_runtime_token_is_rejected(self):
+        token = resolve_runtime_token({"EXTERNAL_SERVICE_TOKEN": "tmp"})
+        self.assertEqual(token, "")
+
     def test_explicit_modelhub_token_takes_precedence(self):
         token = resolve_runtime_token(
             {