diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..43ae0e2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.py[cod] diff --git a/README.md b/README.md index a7bdcbc..535df20 100644 --- a/README.md +++ b/README.md @@ -1,39 +1,69 @@ # huni-probe-agent -信创模盒 ModelHub XC 适配智能体 · **只读探针骨架 (probe-only)**。 +信创模盒 ModelHub XC 适配智能体安全探针 `v1.0.3`。 -对照官方样例 `luopingyi/xc_agent_platform_demo`(最小合规骨架)扩了一个零副作用探针。 +它用于验证「建仓 → Kaniko 构建 → 部署 → `/health` → 平台 API 鉴权 → 可控提交」链路。默认只读,只有显式开启提交开关后才会创建验证任务。 -## 用途 +## 安全行为 -1. 验证平台「建仓 → Kaniko 构建 → 部署 → `/health` → 运行/停止」整条链路是否通。 -2. 零副作用观测平台注入的 `STRATEGY_ID` / `EXTERNAL_SERVICE_TOKEN` 是否为有效凭证: - 启动时用注入 token 调一次**只读** `GET /api/adapt/task/page`,看返回 `code`: - `0` = 凭证有效;`40100` = 未登录/凭证无效。 +- 默认只调用一次只读接口 `GET /api/adapt/task/page`。 +- 空凭证和 `tmp`、`placeholder` 等占位值会被拒绝,不会继续请求。 +- 任何日志和 HTTP 状态响应都不包含 token 值或长度。 +- 只有 `ALLOW_SUBMIT=1` 且只读鉴权成功,才会调用 `build-config` 和 `task/add`。 +- 每个进程生命周期最多调用一次 `task/add`,没有自动重试。 +- `/health` 始终只代表容器存活;链路结果请查看 `/status`。 -## 满足的运行时契约 +## 凭证注入 -- 根目录 `Dockerfile`,`EXPOSE 8080` -- `GET /health` 返回 HTTP 200(K8s 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 值、不落盘任何凭证;探针仅调只读 GET,仅记录 `http/code/计数`。 +不要把 token 写进代码、Dockerfile、Git 仓库或智能体名称。平台侧应通过 Kubernetes Secret 注入真实凭证;如果仍注入 `tmp`,探针会进入 `blocked_missing_auth`,不会误提交。 -## 本地运行 +## 运行时变量 + +| 变量 | 默认值 | 说明 | +| --- | --- | --- | +| `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 -python main.py -curl http://localhost:8080/health # {"status":"ok"} -curl http://localhost:8080/ # 查看探针结果 +python -m unittest -v + +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//` 并保留一个**真 git 标签**(如 `v1.0.0`; -平台 Kaniko 按 `refs/tags/` 解析,用分支名会失败)。再在「我的适配智能体 → 新增智能体」 -填 名称 / 仓库地址 / 标签 提交。 +推到 `https://dev.modelhub.org.cn//` 并创建真实 Git 标签(例如 `v1.0.3`)。在「我的适配智能体 → 新增智能体」填写名称、仓库地址、标签。平台 Kaniko 按 `refs/tags/` 构建,不能用分支名代替标签。 + +建议先保持 `ALLOW_SUBMIT=0`,确认 `/status` 为 `read_only_complete`;再由平台管理员把它改为 `1`,仅跑一次验证任务,验证后立即恢复为 `0`。 diff --git a/main.py b/main.py index bfd4e9d..70dbb31 100644 --- a/main.py +++ b/main.py @@ -1,15 +1,10 @@ -"""ModelHub XC 适配智能体 · 只读探针 + 容器内提交测试 v1.0.2 +"""ModelHub XC 适配智能体安全探针 v1.0.3。 -关键实验:在容器里用注入的 EXTERNAL_SERVICE_TOKEN(=tmp) + STRATEGY_ID 真调一次 task/add, -验证平台是否靠「请求来自容器 pod」放行智能体提交(浏览器/脚本提交会被 60014 挡)。 - -诊断输出: - - dump 所有 env 的名字+长度(不打值)。 - - EXTERNAL_SERVICE_TOKEN 值(短则原样、长则打码)。 - - 容器内 build-config + task/add(真实模型) 的返回 code/msg。 - -约束:token 值不打(短占位除外);只提交这一次测试(真实模型可能进队列);SIGTERM 优雅停机。 +默认只做一次只读鉴权检查。只有显式设置 ALLOW_SUBMIT=1,且鉴权检查成功后, +才会调用 build-config 和 task/add;每个进程生命周期最多提交一次。 """ + +import copy import json import os import signal @@ -18,100 +13,234 @@ import time import urllib.error import urllib.request 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", "") -STRATEGY_ID = os.getenv("STRATEGY_ID", "") + +VERSION = "1.0.3" +MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn").rstrip("/") +STRATEGY_ID = os.getenv("STRATEGY_ID", "").strip() 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, FW, TT = os.getenv("GPU", "MetaX_c-500"), os.getenv("FW", "vllm"), os.getenv("TT", "text-generation") - -HARDCODED_CFG = ( - "framework: vllm\napi: completion\nlang: en\nmax_model_len: 4096\nmax_tokens: 1024\n" - "temperature: 0.7\nrepetition_penalty: 1.1\ntop_p: 0.9\nsut_config:\n gpu_num: 1\n values:\n" - " command:\n - /opt/conda/bin/vllm\n - serve\n - /model\n - --port\n - '20644'\n" - " - --served-model-name\n - llm\n - --max-model-len\n - '4096'\n" - " - --gpu-memory-utilization\n - '0.9'\n - -tp\n - '1'\n - --enforce-eager\n" - " - --trust-remote-code\nref_config:\n gpu_num: 1\n values:\n command:\n - vllm\n" - " - serve\n - /model\n - --port\n - '80'\n - --served-model-name\n - llm\n" - " - --max-model-len\n - '4096'\n - -tp\n - '1'\n - --enforce-eager\n - --trust-remote-code\n" +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() -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: print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) -def _mask(v: str) -> str: - return v if len(v) <= 8 else (v[:2] + "…" + v[-2:]) +def _set_state(**changes) -> None: + with STATE_LOCK: + STATE.update(changes) -def _req(method: str, path: str, headers: dict, body=None, timeout: int = 30) -> dict: - data = body.encode() if isinstance(body, str) else body +def _snapshot() -> dict: + 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: - with urllib.request.urlopen(req, timeout=timeout) as r: - txt = r.read().decode("utf-8", "ignore") - status = r.status - except urllib.error.HTTPError as e: - txt = e.read().decode("utf-8", "ignore") - status = e.code - except Exception as e: # noqa: BLE001 - return {"http": -1, "err": str(e)[:100]} + with urllib.request.urlopen(req, timeout=timeout) as response: + text = response.read().decode("utf-8", "ignore") + status = response.status + except urllib.error.HTTPError as error: + text = error.read().decode("utf-8", "ignore") + status = error.code + except Exception as error: # noqa: BLE001 - network diagnostics are returned safely + return {"http": -1, "error": type(error).__name__} + try: - j = json.loads(txt) - return {"http": status, "code": j.get("code"), "msg": (j.get("message") or "")[:100], - "data": j.get("data")} - except Exception: # noqa: BLE001 - return {"http": status, "raw": txt[:100]} + payload = json.loads(text) + except (TypeError, ValueError): + return {"http": status, "code": None, "message": "non_json_response"} + return { + "http": status, + "code": payload.get("code"), + "message": str(payload.get("message") or "")[:100], + "data": payload.get("data"), + } + + +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: - 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)) - log(f"strategy_id={STRATEGY_ID or 'MISSING'} | EXTERNAL_SERVICE_TOKEN value={_mask(TOKEN)!r} len={len(TOKEN)}") + """Run one guarded probe; no retry loop and at most one task/add call.""" + token, source, token_error = _resolve_token() + _set_state(token_source=source, submit_enabled=ALLOW_SUBMIT) - # 只读:token 有效性 - page = _req("GET", "/api/adapt/task/page?current=1&pageSize=1", {"Xc-Token": TOKEN}) - log("EST-token task/page -> " + json.dumps({k: page.get(k) for k in ("http", "code")}, ensure_ascii=False)) + if not STRATEGY_ID: + _set_state(phase="blocked_missing_strategy_id", done=True) + 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 - # 关键:容器内 build-config + task/add(真实模型) - bc = _req("POST", f"/api/adapt/task/build-config?gpuType={GPU}&framework={FW}&taskType={TT}", {"Xc-Token": TOKEN}) - log("build-config(EST-token) -> " + json.dumps({k: bc.get(k) for k in ("http", "code", "msg")}, ensure_ascii=False)) - cfg = bc.get("data") if isinstance(bc.get("data"), str) and bc.get("data") else HARDCODED_CFG - body = json.dumps({"modelAddress": TEST_MODEL, "taskType": TT, "targetGpu": GPU, - "framework": FW, "strategyId": STRATEGY_ID, "configParams": cfg}) - ta = _req("POST", "/api/adapt/task/add", {"Xc-Token": TOKEN, "Content-Type": "application/json"}, body) - log("SUBMIT-TEST task/add(EST-token+strategyId, real model) -> " + json.dumps(ta, ensure_ascii=False)) - PROBE["results"] = [{"task_page": page.get("code")}, {"build_config": bc.get("code")}, - {"task_add": ta.get("code"), "msg": ta.get("msg")}] - PROBE["done"] = True - log("probe done | task/add code=0 => 容器提交通; 60014 => 容器来源也被判脚本; 401 => token 无效") + _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 + + _set_state(auth_ready=True, results=results) + log(f"read-only auth check passed (source={source})") + if not ALLOW_SUBMIT: + _set_state(phase="read_only_complete", done=True) + log("probe complete in read-only mode; task/add was not called") + return + + _set_state(phase="preparing_submission") + build_path = f"/api/adapt/task/build-config?gpuType={GPU}&framework={FW}&taskType={TT}" + build_config = _req("POST", build_path, token) + results["build_config"] = _result_summary(build_config) + config = build_config.get("data") + if not _api_ok(build_config) or not isinstance(config, str) or not config: + _set_state(phase="build_config_failed", done=True, results=results) + log("build-config failed; task/add was not called") + return + + body = json.dumps( + { + "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): 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_header("Content-Type", "application/json") + self.send_header("Content-Type", "application/json; charset=utf-8") 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"}) + self._json({"status": "ok", "version": VERSION}) return - if self.path == "/": - self._json({"name": "huni-probe-agent", "mode": "submit-test-v1.0.2", "probe": PROBE}) + if self.path in {"/", "/status"}: + self._json({"name": "huni-probe-agent", "probe": _snapshot()}) return self._json({"error": "not found"}, 404) - def log_message(self, *_a) -> None: + def log_message(self, *_args) -> None: pass @@ -127,7 +256,7 @@ def main() -> None: server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) server.timeout = 1 - log(f"probe agent v1.0.2 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(): server.handle_request() server.server_close() diff --git a/test_main.py b/test_main.py new file mode 100644 index 0000000..a836f46 --- /dev/null +++ b/test_main.py @@ -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()