2 Commits
v1.0.1 ... main

Author SHA1 Message Date
f18eeaf334 harden probe agent v1.0.3
安全鉴权、默认只读、显式提交开关、状态接口与测试
2026-08-26 17:16:02 +08:00
5db73110a5 probe v1.0.2 container submit test 2026-08-24 13:13:11 +08:00
4 changed files with 368 additions and 101 deletions

2
.gitignore vendored Normal file
View File

@@ -0,0 +1,2 @@
__pycache__/
*.py[cod]

View File

@@ -1,39 +1,69 @@
# huni-probe-agent # huni-probe-agent
信创模盒 ModelHub XC 适配智能体 · **只读探针骨架 (probe-only)** 信创模盒 ModelHub XC 适配智能体安全探针 `v1.0.3`
对照官方样例 `luopingyi/xc_agent_platform_demo`(最小合规骨架)扩了一个零副作用探针 它用于验证「建仓 → Kaniko 构建 → 部署 → `/health` → 平台 API 鉴权 → 可控提交」链路。默认只读,只有显式开启提交开关后才会创建验证任务
## 用途 ## 安全行为
1. 验证平台「建仓 → Kaniko 构建 → 部署 → `/health` → 运行/停止」整条链路是否通 - 默认只调用一次只读接口 `GET /api/adapt/task/page`
2. 零副作用观测平台注入的 `STRATEGY_ID` / `EXTERNAL_SERVICE_TOKEN` 是否为有效凭证: - 空凭证和 `tmp``placeholder` 等占位值会被拒绝,不会继续请求。
启动时用注入 token 调一次**只读** `GET /api/adapt/task/page`,看返回 `code` - 任何日志和 HTTP 状态响应都不包含 token 值或长度。
`0` = 凭证有效;`40100` = 未登录/凭证无效 - 只有 `ALLOW_SUBMIT=1` 且只读鉴权成功,才会调用 `build-config``task/add`
- 每个进程生命周期最多调用一次 `task/add`,没有自动重试。
- `/health` 始终只代表容器存活;链路结果请查看 `/status`
## 满足的运行时契约 ## 凭证注入
- 根目录 `Dockerfile``EXPOSE 8080` 按以下优先级读取凭证:
- `GET /health` 返回 HTTP 200K8s livenessProbe
- 读取 `STRATEGY_ID``EXTERNAL_SERVICE_TOKEN` 环境变量
- 处理 `SIGTERM`优雅停机30s 窗口内退出)
- 资源占用极小,符合 requests 100m/256Mi、limits 1C/512Mi
## 明确不做 1. `XC_TOKEN_FILE`Kubernetes Secret 挂载文件路径,推荐。
2. `XC_TOKEN`:由 Kubernetes Secret 注入的环境变量。
3. `EXTERNAL_SERVICE_TOKEN`:兼容平台旧变量名。
- **绝不提交任何验证任务**(不调 `task/add`,不调 `build-config` 不要把 token 写进代码、Dockerfile、Git 仓库或智能体名称。平台侧应通过 Kubernetes Secret 注入真实凭证;如果仍注入 `tmp`,探针会进入 `blocked_missing_auth`,不会误提交
- 不打印 token 值、不落盘任何凭证;探针仅调只读 GET仅记录 `http/code/计数`
## 本地运行 ## 运行时变量
| 变量 | 默认值 | 说明 |
| --- | --- | --- |
| `STRATEGY_ID` | 无 | 平台策略 ID必填 |
| `ALLOW_SUBMIT` | `0` | 设置为 `1` 才允许单次提交 |
| `MAIN_HOST` | `https://modelhub.org.cn` | 平台 API 地址 |
| `PORT` | `8080` | HTTP 监听端口 |
| `TEST_MODEL` | TrialSpace-1225-GGUF | 验证模型地址 |
| `GPU` | `MetaX_c-500` | 目标 GPU |
| `FW` | `vllm` | 推理框架 |
| `TT` | `text-generation` | 任务类型 |
## 状态接口
- `GET /health`容器存活检查HTTP 200。
- `GET /status`(或 `/`):返回安全状态,不返回凭证。
关键 `phase`
- `blocked_missing_strategy_id`:平台未注入策略 ID。
- `blocked_missing_auth`:没有可用凭证,或只拿到了占位凭证。
- `auth_failed`:只读鉴权请求未通过。
- `read_only_complete`:鉴权通过,默认只读流程完成。
- `build_config_failed`:配置生成失败,已阻止后续提交。
- `submit_complete` / `submit_failed`:显式开启后,单次提交成功或失败。
## 本地验证
```bash ```bash
python main.py python -m unittest -v
curl http://localhost:8080/health # {"status":"ok"}
curl http://localhost:8080/ # 查看探针结果 PORT=8080 STRATEGY_ID=local-test python main.py
curl http://localhost:8080/health
curl http://localhost:8080/status
``` ```
本地没有凭证时,预期状态是 `blocked_missing_auth`
## 上平台 ## 上平台
推到 `https://dev.modelhub.org.cn/<user>/<repo>`保留一个**真 git 标签**(如 `v1.0.0` 推到 `https://dev.modelhub.org.cn/<user>/<repo>`创建真实 Git 标签(`v1.0.3`)。在「我的适配智能体 → 新增智能体」填写名称、仓库地址、标签。平台 Kaniko 按 `refs/tags/<tag>` 构建,不能用分支名代替标签。
平台 Kaniko 按 `refs/tags/<tag>` 解析,用分支名会失败)。再在「我的适配智能体 → 新增智能体」
填 名称 / 仓库地址 / 标签 提交 建议先保持 `ALLOW_SUBMIT=0`,确认 `/status``read_only_complete`;再由平台管理员把它改为 `1`,仅跑一次验证任务,验证后立即恢复为 `0`

275
main.py
View File

@@ -1,18 +1,10 @@
"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only) v1.0.1 """ModelHub XC 适配智能体安全探针 v1.0.3。
用途: 默认只做一次只读鉴权检查。只有显式设置 ALLOW_SUBMIT=1且鉴权检查成功后
1) 验证平台「建仓 -> Kaniko 构建 -> 部署 -> /health -> 运行/停止」整条链路 才会调用 build-config 和 task/add每个进程生命周期最多提交一次
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 copy
import json import json
import os import os
import signal import signal
@@ -21,107 +13,234 @@ import time
import urllib.error import urllib.error
import urllib.request import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Callable, Mapping, Optional, Tuple
MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn")
TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "") VERSION = "1.0.3"
STRATEGY_ID = os.getenv("STRATEGY_ID", "") MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn").rstrip("/")
STRATEGY_ID = os.getenv("STRATEGY_ID", "").strip()
PORT = int(os.getenv("PORT", "8080")) PORT = int(os.getenv("PORT", "8080"))
ALLOW_SUBMIT = os.getenv("ALLOW_SUBMIT", "").strip().lower() in {"1", "true", "yes", "on"}
TEST_MODEL = os.getenv(
"TEST_MODEL",
"https://www.modelscope.cn/models/mradermacher/TrialSpace-1225-GGUF",
)
GPU = os.getenv("GPU", "MetaX_c-500")
FW = os.getenv("FW", "vllm")
TT = os.getenv("TT", "text-generation")
_TOKEN_PLACEHOLDERS = {"tmp", "placeholder", "changeme", "change-me", "example", "test"}
shutdown = threading.Event() shutdown = threading.Event()
PROBE = {"done": False, "results": []} STATE_LOCK = threading.Lock()
def _initial_state() -> dict:
return {
"version": VERSION,
"phase": "starting",
"done": False,
"auth_ready": False,
"submit_enabled": ALLOW_SUBMIT,
"token_source": "none",
"results": {},
}
STATE = _initial_state()
def log(msg: str) -> None: def log(msg: str) -> None:
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
def _mask(v: str) -> str: def _set_state(**changes) -> None:
"""短值(必是废值)原样看;长值只留头尾,避免泄露可能的真 token。""" with STATE_LOCK:
return v if len(v) <= 8 else (v[:2] + "" + v[-2:]) STATE.update(changes)
def _readonly_get(path: str, headers: dict, timeout: int = 20) -> dict: def _snapshot() -> dict:
req = urllib.request.Request(MAIN + path, method="GET", headers=headers) with STATE_LOCK:
return copy.deepcopy(STATE)
def _reset_state() -> None:
"""Reset process state; kept separate to make the probe deterministically testable."""
with STATE_LOCK:
STATE.clear()
STATE.update(_initial_state())
def _valid_token(value: str) -> bool:
value = value.strip()
return bool(value) and value.lower() not in _TOKEN_PLACEHOLDERS
def _resolve_token(
env: Optional[Mapping[str, str]] = None,
read_text: Optional[Callable[[str], str]] = None,
) -> Tuple[str, str, Optional[str]]:
"""Resolve a credential without exposing its value or length.
Precedence is XC_TOKEN_FILE, XC_TOKEN, then the legacy
EXTERNAL_SERVICE_TOKEN. An explicitly configured but unreadable token file
is treated as an error instead of silently falling back.
"""
values = os.environ if env is None else env
file_reader = read_text or (lambda path: Path(path).read_text(encoding="utf-8"))
token_file = values.get("XC_TOKEN_FILE", "").strip()
if token_file:
try:
value = file_reader(token_file).strip()
except (OSError, UnicodeError):
return "", "XC_TOKEN_FILE", "token_file_unreadable"
if _valid_token(value):
return value, "XC_TOKEN_FILE", None
return "", "XC_TOKEN_FILE", "token_missing_or_placeholder"
for name in ("XC_TOKEN", "EXTERNAL_SERVICE_TOKEN"):
value = values.get(name, "").strip()
if _valid_token(value):
return value, name, None
if value:
return "", name, "token_missing_or_placeholder"
return "", "none", "token_missing_or_placeholder"
def _req(method: str, path: str, token: str, body=None, timeout: int = 30) -> dict:
headers = {"Accept": "application/json", "Xc-Token": token}
data = body.encode("utf-8") if isinstance(body, str) else body
if data is not None:
headers["Content-Type"] = "application/json"
req = urllib.request.Request(MAIN + path, data=data, method=method, headers=headers)
try: try:
with urllib.request.urlopen(req, timeout=timeout) as r: with urllib.request.urlopen(req, timeout=timeout) as response:
body = r.read().decode("utf-8", "ignore") text = response.read().decode("utf-8", "ignore")
status = r.status status = response.status
except urllib.error.HTTPError as e: except urllib.error.HTTPError as error:
body = e.read().decode("utf-8", "ignore") text = error.read().decode("utf-8", "ignore")
status = e.code status = error.code
except Exception as e: # noqa: BLE001 except Exception as error: # noqa: BLE001 - network diagnostics are returned safely
return {"http": -1, "err": str(e)[:120]} return {"http": -1, "error": type(error).__name__}
code = count = None
try: try:
j = json.loads(body) payload = json.loads(text)
code = j.get("code") except (TypeError, ValueError):
d = j.get("data") return {"http": status, "code": None, "message": "non_json_response"}
if isinstance(d, dict): return {
count = d.get("total") "http": status,
if count is None and isinstance(d.get("records"), list): "code": payload.get("code"),
count = len(d["records"]) "message": str(payload.get("message") or "")[:100],
elif isinstance(d, list): "data": payload.get("data"),
count = len(d) }
except Exception: # noqa: BLE001
pass
return {"http": status, "code": code, "count": count} def _api_ok(result: dict) -> bool:
try:
status = int(result.get("http", -1))
except (TypeError, ValueError):
return False
return 200 <= status < 300 and result.get("code") in {0, "0"}
def _result_summary(result: dict) -> dict:
summary = {"http": result.get("http"), "code": result.get("code")}
if result.get("message"):
summary["message"] = result["message"]
if result.get("error"):
summary["error"] = result["error"]
return summary
def _task_id(data) -> Optional[str]:
if not isinstance(data, dict):
return None
for key in ("id", "taskId", "task_id"):
if data.get(key) is not None:
return str(data[key])
return None
def probe() -> None: def probe() -> None:
# 1) dump 所有 env 的 key + 值长度(不打值),找隐藏的真 token """Run one guarded probe; no retry loop and at most one task/add call."""
env_lens = {k: len(str(os.environ.get(k, ""))) for k in sorted(os.environ)} token, source, token_error = _resolve_token()
log("env keys+lens: " + json.dumps(env_lens, ensure_ascii=False)) _set_state(token_source=source, submit_enabled=ALLOW_SUBMIT)
# 2) EXTERNAL_SERVICE_TOKEN 的值(短则全打,长则打码)+ STRATEGY_ID if not STRATEGY_ID:
log(f"strategy_id={STRATEGY_ID or 'MISSING'} | " _set_state(phase="blocked_missing_strategy_id", done=True)
f"EXTERNAL_SERVICE_TOKEN value={_mask(TOKEN)!r} len={len(TOKEN)}") log("probe blocked: STRATEGY_ID is missing")
return
if token_error:
_set_state(phase="blocked_missing_auth", done=True, results={"auth": token_error})
log(f"probe blocked: usable credential is unavailable (source={source})")
return
path = "/api/adapt/task/page?current=1&pageSize=1" _set_state(phase="checking_auth")
page = _req("GET", "/api/adapt/task/page?current=1&pageSize=1", token)
results = {"task_page": _result_summary(page)}
if not _api_ok(page):
_set_state(phase="auth_failed", done=True, results=results)
log("read-only auth check failed: " + json.dumps(results["task_page"], ensure_ascii=False))
return
# 3) 对每个「长得像 token」(len>=16) 的 env 值,试当 Xc-Token,看哪个能通 _set_state(auth_ready=True, results=results)
candidates = [k for k in sorted(os.environ) if len(str(os.environ.get(k, ""))) >= 16] log(f"read-only auth check passed (source={source})")
log("token-like env candidates (len>=16): " + json.dumps(candidates, ensure_ascii=False)) if not ALLOW_SUBMIT:
for k in candidates: _set_state(phase="read_only_complete", done=True)
res = _readonly_get(path, {"Xc-Token": os.environ[k]}) log("probe complete in read-only mode; task/add was not called")
PROBE["results"].append({"try_env": k, "as": "Xc-Token", **res}) return
log(f"try Xc-Token from ${k} -> {json.dumps(res, ensure_ascii=False)}")
# 4) EXTERNAL_SERVICE_TOKEN 三种头对照 _set_state(phase="preparing_submission")
trials = [ build_path = f"/api/adapt/task/build-config?gpuType={GPU}&framework={FW}&taskType={TT}"
("Xc-Token", {"Xc-Token": TOKEN}), build_config = _req("POST", build_path, token)
("Authorization-Bearer", {"Authorization": "Bearer " + TOKEN}), results["build_config"] = _result_summary(build_config)
("Authorization-raw", {"Authorization": TOKEN}), config = build_config.get("data")
] if not _api_ok(build_config) or not isinstance(config, str) or not config:
for name, headers in trials: _set_state(phase="build_config_failed", done=True, results=results)
res = {"skip": "no-token"} if not TOKEN else _readonly_get(path, headers) log("build-config failed; task/add was not called")
PROBE["results"].append({"est_token_hdr": name, **res}) return
log(f"EST-token {name} -> {json.dumps(res, ensure_ascii=False)}")
PROBE["done"] = True body = json.dumps(
log("probe done | 判读: 任一 code=0 => 该来源凭证有效; 全 401/40100 => 无有效凭证") {
"modelAddress": TEST_MODEL,
"taskType": TT,
"targetGpu": GPU,
"framework": FW,
"strategyId": STRATEGY_ID,
"configParams": config,
}
)
task_add = _req("POST", "/api/adapt/task/add", token, body)
results["task_add"] = _result_summary(task_add)
task_id = _task_id(task_add.get("data"))
if task_id:
results["task_add"]["task_id"] = task_id
phase = "submit_complete" if _api_ok(task_add) else "submit_failed"
_set_state(phase=phase, done=True, results=results)
log("single task/add attempt finished: " + json.dumps(results["task_add"], ensure_ascii=False))
class Handler(BaseHTTPRequestHandler): class Handler(BaseHTTPRequestHandler):
def _json(self, obj: dict, status: int = 200) -> None: def _json(self, obj: dict, status: int = 200) -> None:
payload = json.dumps(obj, ensure_ascii=False).encode() payload = json.dumps(obj, ensure_ascii=False).encode("utf-8")
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload))) self.send_header("Content-Length", str(len(payload)))
self.end_headers() self.end_headers()
self.wfile.write(payload) self.wfile.write(payload)
def do_GET(self) -> None: def do_GET(self) -> None:
if self.path == "/health": if self.path == "/health":
self._json({"status": "ok"}) self._json({"status": "ok", "version": VERSION})
return return
if self.path == "/": if self.path in {"/", "/status"}:
self._json({"name": "huni-probe-agent", "mode": "probe-only-v1.0.1", self._json({"name": "huni-probe-agent", "probe": _snapshot()})
"strategy_id_present": bool(STRATEGY_ID),
"token_present": bool(TOKEN), "probe": PROBE})
return return
self._json({"error": "not found"}, 404) self._json({"error": "not found"}, 404)
def log_message(self, *_a) -> None: def log_message(self, *_args) -> None:
pass pass
@@ -137,7 +256,7 @@ def main() -> None:
server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
server.timeout = 1 server.timeout = 1
log(f"probe-only agent v1.0.1 listening on 0.0.0.0:{PORT}") log(f"probe agent v{VERSION} listening on 0.0.0.0:{PORT} (submit_enabled={ALLOW_SUBMIT})")
while not shutdown.is_set(): while not shutdown.is_set():
server.handle_request() server.handle_request()
server.server_close() server.server_close()

116
test_main.py Normal file
View File

@@ -0,0 +1,116 @@
import unittest
from unittest import mock
import main
class TokenResolutionTests(unittest.TestCase):
def test_placeholder_legacy_token_is_rejected(self):
token, source, error = main._resolve_token({"EXTERNAL_SERVICE_TOKEN": "tmp"})
self.assertEqual(token, "")
self.assertEqual(source, "EXTERNAL_SERVICE_TOKEN")
self.assertEqual(error, "token_missing_or_placeholder")
def test_token_file_has_precedence(self):
token, source, error = main._resolve_token(
{"XC_TOKEN_FILE": "/secret/token", "XC_TOKEN": "env-token"},
read_text=lambda _path: "file-token\n",
)
self.assertEqual(token, "file-token")
self.assertEqual(source, "XC_TOKEN_FILE")
self.assertIsNone(error)
def test_unreadable_explicit_token_file_does_not_fall_back(self):
def fail(_path):
raise OSError("not mounted")
token, source, error = main._resolve_token(
{"XC_TOKEN_FILE": "/secret/token", "XC_TOKEN": "env-token"},
read_text=fail,
)
self.assertEqual(token, "")
self.assertEqual(source, "XC_TOKEN_FILE")
self.assertEqual(error, "token_file_unreadable")
class ProbeTests(unittest.TestCase):
def setUp(self):
main._reset_state()
def test_missing_token_makes_no_request(self):
with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
main, "_resolve_token", return_value=("", "EXTERNAL_SERVICE_TOKEN", "token_missing_or_placeholder")
), mock.patch.object(main, "_req") as request:
main.probe()
request.assert_not_called()
self.assertEqual(main._snapshot()["phase"], "blocked_missing_auth")
def test_read_only_mode_never_calls_post(self):
calls = []
def request(method, path, _token, body=None, timeout=30):
calls.append((method, path, body, timeout))
return {"http": 200, "code": 0, "data": {}}
with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
main, "ALLOW_SUBMIT", False
), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
main, "_req", side_effect=request
):
main.probe()
self.assertEqual([call[0] for call in calls], ["GET"])
state = main._snapshot()
self.assertEqual(state["phase"], "read_only_complete")
self.assertTrue(state["auth_ready"])
def test_failed_auth_never_calls_post(self):
with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
main, "ALLOW_SUBMIT", True
), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
main, "_req", return_value={"http": 401, "code": 40100, "message": "unauthorized"}
) as request:
main.probe()
self.assertEqual(request.call_count, 1)
self.assertEqual(request.call_args.args[0], "GET")
self.assertEqual(main._snapshot()["phase"], "auth_failed")
def test_submit_mode_calls_task_add_once(self):
responses = [
{"http": 200, "code": 0, "data": {}},
{"http": 200, "code": 0, "data": "framework: vllm\n"},
{"http": 200, "code": 0, "data": {"taskId": "task-1"}},
]
with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
main, "ALLOW_SUBMIT", True
), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
main, "_req", side_effect=responses
) as request:
main.probe()
methods_and_paths = [(call.args[0], call.args[1]) for call in request.call_args_list]
self.assertEqual(sum(path == "/api/adapt/task/add" for _, path in methods_and_paths), 1)
state = main._snapshot()
self.assertEqual(state["phase"], "submit_complete")
self.assertEqual(state["results"]["task_add"]["task_id"], "task-1")
def test_failed_build_config_prevents_task_add(self):
responses = [
{"http": 200, "code": 0, "data": {}},
{"http": 500, "code": 50000, "message": "build failed"},
]
with mock.patch.object(main, "STRATEGY_ID", "strategy-1"), mock.patch.object(
main, "ALLOW_SUBMIT", True
), mock.patch.object(main, "_resolve_token", return_value=("valid-token", "XC_TOKEN", None)), mock.patch.object(
main, "_req", side_effect=responses
) as request:
main.probe()
self.assertEqual(request.call_count, 2)
self.assertNotIn("/api/adapt/task/add", [call.args[1] for call in request.call_args_list])
self.assertEqual(main._snapshot()["phase"], "build_config_failed")
if __name__ == "__main__":
unittest.main()