diff --git a/cloud/chv/executor.py b/cloud/chv/executor.py index 764f0f4..cc6bef6 100644 --- a/cloud/chv/executor.py +++ b/cloud/chv/executor.py @@ -2727,6 +2727,54 @@ class CloudHypervisorExecutor(Executor): except Exception: pass + def _inference_gate(self, task: dict, + model_provider: str | None, + extra_env: dict | None = None) -> None: + """feat-chv-muse-p9:起 VM 前的推理出口预审。 + + 模型调用只允许走受限推理通道(见 inference_channel): + allow → 审计放行;deny/ask → 审计 + ChvFailure 起 VMM + 前 fail-closed(ask 在这里就是 deny:wake 等不了审批 + UI)。解析器可注入(`sentinel_resolver` 属性,测试用 + 假解析器)。origin 取任务级 extra_env 覆盖执行器 env; + origin 非法同样记 deny,不抛裸 ValueError。审计 detail + 只含 VM 名/主机/verdict,不含任何凭证值。 + """ + from cloud.chv.inference_channel import check_inference_call + name = self.vm_name(task) + env = dict(self._env) + if extra_env: + env.update(extra_env) + try: + check = check_inference_call( + model_provider, env=env, + resolve=getattr(self, "sentinel_resolver", None)) + except ValueError as exc: + self._emit("inference_egress_deny", task, + "vm=%s reasons=bad_origin" % name) + raise ChvFailure( + "inference_egress_denied", + "推理 origin 配置非法,拒绝启动:%s" % exc) + decision = check.decision + if decision.verdict == "allow": + capability = decision.capability + request = check.request + assert capability is not None # allow 恒带严格能力 + assert request is not None # allow 恒带已解析请求 + self._emit("inference_egress_allow", task, + "vm=%s host=%s capability=%s" + % (name, request.host, capability.id)) + return + host = check.request.host if check.request is not None else "" + self._emit("inference_egress_" + decision.verdict, task, + "vm=%s host=%s reasons=%s" + % (name, host, ",".join(decision.reasons))) + raise ChvFailure( + "inference_egress_denied", + "推理出口预审未通过(%s),拒绝启动:请运维确认 " + "provider 与 CHV_INFERENCE_ORIGIN/CHV_INFERENCE_ALLOW_HOSTS" + % decision.verdict) + def _run_with_vm(self, task, inner_args, goal=False, extra_env=None, **kwargs): """任务 VM 生命周期(feat-chv-muse-p1,见 ADR-17): @@ -2740,7 +2788,11 @@ class CloudHypervisorExecutor(Executor): 重试(重试是 worker 的 max_retries 语义)。中途死亡 判 interrupted(断点可续跑),见 _run_cold 与 _run_on_resident。dispatch/run/goal/goal_resume/断点 - 续跑都走本方法,签名不动。""" + 续跑都走本方法,签名不动。round-12 合入 P9:起任何 VM + (常驻 resume 或冷启动 wake)之前先过 _inference_gate + 推理出口预审,deny/ask 即 fail-closed。""" + self._inference_gate(task, kwargs.get("model_provider"), + extra_env=extra_env) resident = self._resident_acquire(task, extra_env=extra_env) if resident is not None: pool = resident["pool"] diff --git a/cloud/chv/inference_channel.py b/cloud/chv/inference_channel.py new file mode 100644 index 0000000..23f3a2e --- /dev/null +++ b/cloud/chv/inference_channel.py @@ -0,0 +1,247 @@ +"""受限推理通道(feat-chv-muse-p9,叶子模块)。 + +对齐 Muse 官方安全文(security.muse.ai 2026-09-08,见 +docs/chv-muse-p9-gap.md):推理与遥测代理是通向外部基础设施的 +受限通道;Sentinel 是连接器动作与全部网络出口的唯一权限机构。 + +本模块是**决策层**,不是网络层强制: +- 它不拦截客内流量。客内 codex 直连的强制仍靠 Tinyproxy 白名单 + + 宿主防火墙(第三棒落地中);不要把本模块写成“出口已全部经 + Sentinel”——与 P3 的 sentinel 模块同款声明。 +- 唯一出口点:模型调用只允许走这里决定的 origin。执行器在 + `_run_with_vm` 起 VM 前调 `check_inference_call`:allow 才继续; + deny/ask 一律审计 + 停止,不起 VM。 +- provider=`sub2api` 时 origin 取 `SUB2API_BASE_URL`;provider 为空 + (xcmapi 路径,codex 用自身 config.toml 默认网关)时执行器看不 + 到 origin,改取 operator 显式配置的 `CHV_INFERENCE_ORIGIN`,未配 + 即 ask(fail-closed,不静默放行)。部署迁移见差距表 §2.5。 +- 能力说明:allow 时带出的严格能力(单次、300 秒)是**起 VM 前的 + 预审授权**,不是逐次 HTTP 请求的令牌;客内 codex 的逐次调用仍由 + 代理白名单 + 防火墙约束。本模块不伪装成逐请求强制。 + +本模块只用标准库,不依赖 cloud.*(除 vendored 的 sentinel), +解析器可注入(`resolve` 参数),默认 `socket.getaddrinfo`;测试 +用假解析器,绝不联网。审计记录里绝不含凭证值。 +""" + +import ipaddress +import os +from collections.abc import Callable +from typing import NamedTuple +from urllib.parse import urlparse + +from cloud.chv.sentinel import ( + Decision, + EgressRequest, + SentinelPolicy, + _one_shot_capability, +) + +INFERENCE_CONNECTOR = "inference" + +# 推理连接器的 L7 路径规则:codex 以 wire_api="responses" 调模型, +# 只放 POST /v1/responses;GET /v1/models 是否必需待真机核实 +# (codex 启动时是否拉模型列表),若非必需则收紧为只放前者。 +INFERENCE_PATH_RULES = ( + ("POST", "prefix", "/v1/responses"), + ("GET", "prefix", "/v1/models"), +) + +# 上限默认值(本项目自定,非 Meta 官方值):超时 120 秒与 +# host.py 的 Server.request 默认超时一致;重试 1 次即保持 +# 现有行为不变(现为 range(2))。非法值一律 ValueError。 +DEFAULT_INFERENCE_TIMEOUT_SECONDS = 120 +DEFAULT_INFERENCE_MAX_RETRIES = 1 +MIN_INFERENCE_TIMEOUT_SECONDS = 10 +MAX_INFERENCE_TIMEOUT_SECONDS = 600 +MAX_INFERENCE_RETRIES = 5 + + +class InferenceLimits(NamedTuple): + """推理通道上限:单次请求超时(秒)与重连重试上限(次)。""" + + timeout_seconds: int + max_retries: int + + +class InferenceCheck(NamedTuple): + """一次预审的结果:决策 + 被评估的请求(origin 缺失/非法时 + request 为 None,审计方用原因兜底)。""" + + decision: Decision + request: EgressRequest | None + + +def _env_of(env: dict | None) -> dict: + """测试传 dict,生产默认读 os.environ。""" + return env if env is not None else os.environ # type: ignore[return-value] + + +def _is_local_relay_host(host: str) -> bool: + """字面环回/私网 IP 才是本地中继主机(无 DNS,无重绑定; + 主机名一律不是——DNS 解析结果不可信)。""" + try: + addr = ipaddress.ip_address(host) + except ValueError: + return False + return addr.is_loopback or addr.is_private + + +def resolve_inference_origin(provider: str | None, + env: dict | None = None) -> str: + """解析推理出口 origin(不带尾斜杠,端口非默认时保留)。 + + - provider="sub2api":取 `SUB2API_BASE_URL`; + - provider 为空或未知标签:取 `CHV_INFERENCE_ORIGIN`(与 + host.py `_server_command` 一致走默认网关;标签本身不决定 + origin,origin 永远只取 operator 配置的 env); + 变量未配/空白返回 ""(调用方按 ask 处理,不在这里抛错)。 + scheme 规则(见 cloud/README.md 的 sub2api 网关一节): + - https:公网网关(如 xgapi.lol),端口只接受 443(或缺省); + - http:只接受本地中继——字面环回/私网 IP 主机(如文档形状 + http://172.17.0.1:28084,Docker 桥接 IP),端口保留; + - 其余(http 公网主机名、https 非 443 端口、带 userinfo、 + 无 host)一律 ValueError——配置错了要 loud,不要静默跑向别处。 + 报错不带 origin 原值(http URL 里可能含 userinfo)。 + """ + values = _env_of(env) + if provider is None: + raw = values.get("CHV_INFERENCE_ORIGIN") or "" + elif provider == "sub2api": + raw = values.get("SUB2API_BASE_URL") or "" + else: + # 未知 provider 标签:与 host.py `_server_command` 一致走 + # 默认网关(`CHV_INFERENCE_ORIGIN`)——标签本身不决定 + # origin,origin 永远只取 operator 配置的 env。 + raw = values.get("CHV_INFERENCE_ORIGIN") or "" + raw = raw.strip().rstrip("/") + if not raw: + return "" + parsed = urlparse(raw) + scheme = parsed.scheme.lower() + if "@" in parsed.netloc: + raise ValueError("推理 origin 不得带 userinfo(凭证禁进 URL)") + host = (parsed.hostname or "").lower() + if not host: + raise ValueError("推理 origin 无主机名") + if scheme == "https": + if parsed.port is not None and parsed.port != 443: + raise ValueError("推理 origin 端口只接受 443") + return "https://" + host + if scheme == "http": + if not _is_local_relay_host(host): + raise ValueError("http 推理 origin 只接受本地中继" + "(环回/私网 IP)") + port = parsed.port + if port is not None: + return "http://%s:%d" % (host, port) + return "http://" + host + raise ValueError("推理 origin 必须 https(或本地中继 http)") + + +def inference_allow_hosts(provider: str | None, + env: dict | None = None) -> frozenset: + """推理出口 allowlist(小写主机名集合)。 + + `CHV_INFERENCE_ALLOW_HOSTS`(逗号分隔)显式覆盖;未配时默认 + 即 origin 自身的主机——allowlist 永远是“显式配置的 origin”, + 不是“解析后是什么都行”。origin 缺失时返回空集(预审必 ask)。 + """ + values = _env_of(env) + raw = values.get("CHV_INFERENCE_ALLOW_HOSTS") or "" + hosts = frozenset(h.strip().lower() + for h in raw.split(",") if h.strip()) + if hosts: + return hosts + try: + origin = resolve_inference_origin(provider, env) + except ValueError: + return frozenset() + if not origin: + return frozenset() + host = urlparse(origin).hostname or "" + return frozenset({host}) if host else frozenset() + + +def _parse_int_env(env: dict, name: str, default: int, + low: int, high: int) -> int: + """严格整数 env 解析:未配即默认;非整数/越界即 ValueError, + 绝不静默回落(配错早炸,不留到起 VM 才暴露)。""" + raw = env.get(name) + if raw is None or str(raw).strip() == "": + return default + try: + value = int(str(raw).strip()) + except ValueError: + raise ValueError("%s 必须是整数: %r" % (name, raw)) + if not low <= value <= high: + raise ValueError("%s 必须在 %d-%d 之间: %s" + % (name, low, high, value)) + return value + + +def inference_limits(env: dict | None = None) -> InferenceLimits: + """推理通道上限(见模块头默认值说明)。""" + values = _env_of(env) + return InferenceLimits( + timeout_seconds=_parse_int_env( + values, "CHV_INFERENCE_TIMEOUT", + DEFAULT_INFERENCE_TIMEOUT_SECONDS, + MIN_INFERENCE_TIMEOUT_SECONDS, + MAX_INFERENCE_TIMEOUT_SECONDS), + max_retries=_parse_int_env( + values, "CHV_INFERENCE_MAX_RETRIES", + DEFAULT_INFERENCE_MAX_RETRIES, 0, MAX_INFERENCE_RETRIES), + ) + + +def check_inference_call(provider: str | None, + env: dict | None = None, + resolve: Callable[[str, int], list] | None = None, + policy: SentinelPolicy | None = None, + now: float | None = None) -> InferenceCheck: + """预审一次推理出口调用(起 VM 前,fail-closed)。 + + - origin 缺失 → ask(`inference_origin_not_configured`), + 调用方停止执行,不得自动放行; + - origin 非法 → deny; + - http 本地中继(字面环回/私网 IP):无 DNS、无重绑定,跳过 + Sentinel 的 DNS/SSRF 步;allowlist 与单次能力照常执行; + - 其余走 Sentinel 评估:SSRF、私网 IP、scheme/端口/路径越界 + 即 deny;主机不在白名单即 ask;全过即 allow + 严格能力 + (单次、300 秒,见模块头的能力说明)。 + `resolve` / `policy` 可注入(假解析器 / P5 纵深策略),默认按 allowlist 现建。 + """ + try: + origin = resolve_inference_origin(provider, env) + except ValueError: + # 原因不带原值:非法 origin 的原串里可能含 userinfo, + # 审计原因只记“非法”本身,不记值。 + return InferenceCheck( + Decision("deny", ["inference_origin_invalid"]), None) + if not origin: + return InferenceCheck( + Decision("ask", ["inference_origin_not_configured"]), None) + parsed = urlparse(origin) + scheme = parsed.scheme + host = parsed.hostname or "" + port = parsed.port or (443 if scheme == "https" else 80) + if policy is None: + policy = SentinelPolicy( + allow_hosts=inference_allow_hosts(provider, env), + connector_paths={INFERENCE_CONNECTOR: INFERENCE_PATH_RULES}) + request = EgressRequest(INFERENCE_CONNECTOR, "POST", scheme, + host, port, "/v1/responses") + if scheme == "http": + # 本地中继分支:见上。allowlist 不含该主机 → ask; + # 含 → allow + 与公网路径同款单次能力(P3 默认档)。 + if host not in policy.allow_hosts: + return InferenceCheck( + Decision("ask", ["host_not_allowlisted"]), request) + rules = policy.connector_paths.get(INFERENCE_CONNECTOR, ()) + capability = _one_shot_capability(request, rules, now) + return InferenceCheck( + Decision("allow", ["policy_allow", "local_relay"], + capability=capability), request) + decision = policy.evaluate(request, resolve=resolve, now=now) + return InferenceCheck(decision, request) diff --git a/cloud/chv/sentinel.py b/cloud/chv/sentinel.py index fc5a8b2..919635c 100644 --- a/cloud/chv/sentinel.py +++ b/cloud/chv/sentinel.py @@ -95,6 +95,10 @@ _BLOCKED_NETWORKS = tuple( # gitea:API 只放 /api/v1/;git 智能 HTTP 只放三条(clone/pull/push # 的实际路径,见 git http-protocol)。其它路径(Web UI、登录页、 # 任意管理接口)一律不在本连接器的能力内。 +# feat-chv-muse-p9 的 inference 条目:推理连接器只放 +# POST /v1/responses 与 GET /v1/models(/v1/models 是否必需待 +# 真机核实);scheme/端口见下面两表。本文件其余内容与 P5 的 +# 25bcd35 一致,合入时以本文件为准(它是 P3+P5+P9 的并集)。 _CONNECTOR_PATH_RULES = { "gitea": ( ("GET", "prefix", "/api/v1/"), @@ -103,12 +107,21 @@ _CONNECTOR_PATH_RULES = { ("POST", "suffix", ".git/git-upload-pack"), ("POST", "suffix", ".git/git-receive-pack"), ), + "inference": ( + ("POST", "prefix", "/v1/responses"), + ("GET", "prefix", "/v1/models"), + ), } # 每连接器允许的 scheme/端口(默认:只 https/443;http 只在 -# 显式放行时才出现,本期没有任何连接器放 http)。 -_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"})} -_CONNECTOR_PORTS = {"gitea": frozenset({443})} +# 显式放行时才出现,本期没有任何连接器放 http)。注意:推理的 +# http 本地中继(字面环回/私网 IP,如 http://172.17.0.1:28084) +# 不走 evaluate(见 inference_channel 的本地中继分支),不要 +# “顺手”把 http 加进下面两表——那会放宽公网路径的 scheme 检查。 +_CONNECTOR_SCHEMES = {"gitea": frozenset({"https"}), + "inference": frozenset({"https"})} +_CONNECTOR_PORTS = {"gitea": frozenset({443}), + "inference": frozenset({443})} # 默认放凭证有效期(秒):与 authd 按次释放同款 300 秒。 DEFAULT_CAPABILITY_TTL_SECONDS = 300 diff --git a/cloud/chv/telemetry.py b/cloud/chv/telemetry.py new file mode 100644 index 0000000..8a443bf --- /dev/null +++ b/cloud/chv/telemetry.py @@ -0,0 +1,211 @@ +"""遥测最小化 + 训练退出开关(feat-chv-muse-p9,叶子模块)。 + +对齐 Muse 官方安全文(security.muse.ai 2026-09-08,见 +docs/chv-muse-p9-gap.md):VM 发送有限数据用于推理与遥测 +(安全文未给字段级清单——本模块的字段清单按最小化原则自定, +见 `TELEMETRY_FIELDS`);推理轨迹在去除关键个人身份信息后可 +用于训练新检查点,用户可一键退出训练。 + +诚实边界(见差距表 §4): +- 遥测目前只落盘 `/telemetry.jsonl`,没有向外部遥测 + 端点发送的路径;落盘失败也不影响任务(`emit_telemetry` + 永不抛错)。 +- 训练轨迹目前没有上传路径:开关管的是“轨迹类事件是否产生”; + 上传通道本身不存在,开关是为上传通道预留的门禁。 +- 字段白名单是构造性保证:不在白名单里的键一律丢弃;token 数 + 只取 host.py 回传的 usage,未回传记 `null`,不估算、不编造。 +- 训练开关默认开是**本项目决定,非 Meta 官方值**(报告注明 + 帮助中心原文未逐字确认默认开启)。 + +本模块只用标准库 + `atomic_file`(开关状态文件原子写)。 +""" + +import hashlib +import json +import logging +import os +import time + +from cloud.chv.atomic_file import atomic_write_text + +_LOGGER = logging.getLogger(__name__) + +# 遥测字段白名单(自定,按最小化原则):只含任务级元数据。 +# 绝不含对话原文、文件内容、密钥、个人身份信息——白名单外 +# 的键在 sanitize 时一律丢弃,不是“记下来再脱敏”。 +TELEMETRY_FIELDS = frozenset({ + "task_id", "vm_id", "provider", "model", "status", + "started_at", "ended_at", "duration_ms", + "tokens_in", "tokens_out", "tokens_total", + "turns", "tool_calls", "sampled", +}) + +# 轨迹类事件字段白名单(自定):只含回合级元数据,不含任何文本。 +TRAJECTORY_FIELDS = frozenset({ + "task_id", "turns", "tools", "duration_ms", +}) + +TRAINING_OPT_FILENAME = "training-opt.json" +TRAINING_OPT_AUDIT_FILENAME = "training-opt-audit.jsonl" +TELEMETRY_FILENAME = "telemetry.jsonl" +DEFAULT_TELEMETRY_SAMPLE_RATE = 1.0 + + +def telemetry_sample_rate(env: dict | None = None) -> float: + """遥测采样率:`CHV_TELEMETRY_SAMPLE_RATE`,0.0–1.0,默认 1.0 + (全量,任务级元数据量小)。非法值即 ValueError。""" + values = env if env is not None else os.environ # type: ignore[assignment] + raw = values.get("CHV_TELEMETRY_SAMPLE_RATE") + if raw is None or str(raw).strip() == "": + return DEFAULT_TELEMETRY_SAMPLE_RATE + try: + rate = float(str(raw).strip()) + except ValueError: + raise ValueError( + "CHV_TELEMETRY_SAMPLE_RATE 必须是数字: %r" % raw) + if not 0.0 <= rate <= 1.0: + raise ValueError( + "CHV_TELEMETRY_SAMPLE_RATE 必须在 0.0-1.0 之间: %s" % rate) + return rate + + +def sampled(task_id: str, rate: float) -> bool: + """确定性采样:同一 task_id 每次判定一致(可测试);rate=0 + 全丢,rate=1 全留。""" + digest = hashlib.sha256(task_id.encode("utf-8")).hexdigest() + return int(digest[:8], 16) % 10000 < rate * 10000 + + +def _is_scalar(value: object) -> bool: + """遥测值只认标量(str/int/float/bool/None);dict/list + 一律丢弃——不把结构化内容带进遥测。""" + return value is None or isinstance(value, (str, int, float, bool)) + + +def sanitize_telemetry_event(event: dict) -> dict: + """按白名单过滤遥测事件:不在白名单里的键丢弃;值不是标量 + 的丢弃;`tools` 字段例外——只认 str 列表(工具名),其它 + 元素一律去掉。返回新 dict,不改原 event。""" + clean: dict = {} + for key, value in (event or {}).items(): + if key not in TELEMETRY_FIELDS and key not in TRAJECTORY_FIELDS: + continue + if key == "tools": + if isinstance(value, (list, tuple)): + clean[key] = [t for t in value if isinstance(t, str)] + continue + if _is_scalar(value): + clean[key] = value + return clean + + +def build_task_telemetry(task_id: str, vm_id: str, provider: str, + model: str | None, status: str, + started_at: float, ended_at: float, + duration_ms: int, + tokens: dict | None = None, + turns: int | None = None, + tool_calls: int | None = None, + env: dict | None = None) -> dict | None: + """建一条任务级遥测事件;采样未命中返回 None(不产生事件, + 不是产生后丢弃)。tokens 取 host.py 回传的 usage,未回传 + 时三个 token 字段记 None,不编造。""" + rate = telemetry_sample_rate(env) + if not sampled(task_id, rate): + return None + tokens = tokens or {} + event = { + "task_id": task_id, "vm_id": vm_id, "provider": provider, + "model": model, "status": status, + "started_at": started_at, "ended_at": ended_at, + "duration_ms": duration_ms, + "tokens_in": tokens.get("input_tokens"), + "tokens_out": tokens.get("output_tokens"), + "tokens_total": tokens.get("total_tokens"), + "turns": turns, "tool_calls": tool_calls, "sampled": True, + } + return sanitize_telemetry_event(event) + + +def build_trajectory_event(task_id: str, turns: int, + tools: list, duration_ms: int, + state_dir: str) -> dict | None: + """建一条轨迹类事件(训练用):只含回合级元数据(回合数、 + 工具名、耗时),不含任何文本。训练退出开关关时返回 None + (事件不再产生)。""" + if not training_enabled(state_dir): + return None + return sanitize_telemetry_event({ + "task_id": task_id, "turns": turns, + "tools": list(tools or []), "duration_ms": duration_ms, + }) + + +def _append_jsonl(path: str, record: dict) -> bool: + """best-effort 追加一行 JSONL:失败返回 False,永不抛错 + (遥测/审计落盘失败绝不影响任务)。""" + try: + line = json.dumps(record, ensure_ascii=False) + "\n" + with open(path, "a", encoding="utf-8") as fh: + fh.write(line) + fh.flush() + return True + except Exception as exc: # noqa: BLE001 - 落盘失败只记日志 + _LOGGER.warning("telemetry_append_failed path=%s error=%s", + path, exc) + return False + + +def emit_telemetry(state_dir: str, event: dict) -> bool: + """发一条遥测事件:先过白名单再落盘 `/telemetry.jsonl`。 + 落盘失败返回 False,永不抛错、不影响任务。""" + if not state_dir or not isinstance(event, dict): + return False + clean = sanitize_telemetry_event(event) + if not clean: + return False + return _append_jsonl( + os.path.join(state_dir, TELEMETRY_FILENAME), clean) + + +def training_enabled(state_dir: str) -> bool: + """训练退出开关状态:默认开(本项目决定,非 Meta 官方值)。 + 状态文件缺失即默认开;损坏/不可解析/缺 `enabled` 键时 + fail-closed 当作关处理并记日志(损坏不等于同意上传)。""" + path = os.path.join(state_dir, TRAINING_OPT_FILENAME) + try: + with open(path, encoding="utf-8") as fh: + data = json.load(fh) + except FileNotFoundError: + return True + except Exception as exc: # noqa: BLE001 - 损坏即关 + _LOGGER.warning("training_opt_corrupt path=%s error=%s", + path, exc) + return False + if not isinstance(data, dict) or "enabled" not in data: + return False + return bool(data["enabled"]) + + +def set_training_enabled(state_dir: str, enabled: bool, + actor: str = "user") -> bool: + """一键退出/恢复训练:状态原子写 `/training-opt.json` + (0600),变更追加进 `training-opt-audit.jsonl`。`enabled` + 不是 bool 即 TypeError;写失败返回 False,永不抛错。 + 关后立即生效:后续轨迹事件不再产生。""" + if not isinstance(enabled, bool): + raise TypeError("enabled 必须是 bool: %r" % (enabled,)) + record = {"enabled": enabled, "updated_at": time.time(), + "actor": str(actor)[:64]} + try: + atomic_write_text( + os.path.join(state_dir, TRAINING_OPT_FILENAME), + json.dumps(record, ensure_ascii=False)) + except Exception as exc: # noqa: BLE001 - 写失败只记日志 + _LOGGER.warning("training_opt_write_failed error=%s", exc) + return False + audit = dict(record) + audit["event"] = "training_opt_changed" + _append_jsonl( + os.path.join(state_dir, TRAINING_OPT_AUDIT_FILENAME), audit) + return True diff --git a/cloud/tests/test_chv_closeup.py b/cloud/tests/test_chv_closeup.py index 9cd94af..857b143 100644 --- a/cloud/tests/test_chv_closeup.py +++ b/cloud/tests/test_chv_closeup.py @@ -85,6 +85,12 @@ def _write(path, content=b"x"): return path +def _public_resolver(host, port): + """feat-chv-muse-p9:假解析器(公网 IP),供推理出口预审用; + 实验室绝不联网,真实解析只在真机发生。""" + return ["93.184.216.34"] + + class CloseupCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -113,7 +119,11 @@ class CloseupCase(unittest.TestCase): env_extra=None, **overrides): env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", "CHV_HOST_KEY_MODE": "keyscan", - "CHV_SSH_FORWARD": "1"} + "CHV_SSH_FORWARD": "1", + # feat-chv-muse-p9:推理出口预审的假 origin(公网 + # https,配假解析器;实验室不起真 VM、不联网)。 + "SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} if env_extra: env.update(env_extra) kwargs = dict( @@ -127,6 +137,7 @@ class CloseupCase(unittest.TestCase): sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) + ex.sentinel_resolver = _public_resolver ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex diff --git a/cloud/tests/test_chv_fix.py b/cloud/tests/test_chv_fix.py index 2123390..9b14141 100644 --- a/cloud/tests/test_chv_fix.py +++ b/cloud/tests/test_chv_fix.py @@ -73,6 +73,12 @@ def _write(path, content=b"x"): return path +def _public_resolver(host, port): + """feat-chv-muse-p9:假解析器(公网 IP),供推理出口预审用; + 实验室绝不联网,真实解析只在真机发生。""" + return ["93.184.216.34"] + + class ChvFixCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -127,7 +133,11 @@ class ChvFixCase(unittest.TestCase): def make_exec(self, runner=None, env_extra=None, **overrides): env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", - "CHV_SSH_FORWARD": "1"} + "CHV_SSH_FORWARD": "1", + # feat-chv-muse-p9:推理出口预审的假 origin(公网 + # https,配假解析器;实验室不起真 VM、不联网)。 + "SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} if env_extra: env.update(env_extra) kwargs = dict( @@ -141,6 +151,7 @@ class ChvFixCase(unittest.TestCase): sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) + ex.sentinel_resolver = _public_resolver ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex @@ -305,8 +316,12 @@ class GatewayExtraEnvTest(ChvFixCase): task = {"id": "gw1", "project_id": "p1", "prompt": "做目标", "goal_mode": 1, "_gateway_provider": "sub2api", + # feat-chv-muse-p9:网关 URL 须是合法推理 origin—— + # 这里用文档形状的本地中继(http + Docker 桥接 IP), + # 预审走本地中继分支放行(字面 IP,不查 DNS)。 "_gateway_env": {"SUB2API_API_KEY": "gwkey", - "SUB2API_BASE_URL": "http://gw/"}} + "SUB2API_BASE_URL": + "http://172.17.0.1:28084"}} result, error = worker._run_attempt(task) self.assertIsNone(error) self.assertEqual(result["status"], "completed") diff --git a/cloud/tests/test_chv_gaps.py b/cloud/tests/test_chv_gaps.py index a193eac..8e12967 100644 --- a/cloud/tests/test_chv_gaps.py +++ b/cloud/tests/test_chv_gaps.py @@ -81,6 +81,12 @@ def _write(path, content=b"x"): return path +def _public_resolver(host, port): + """feat-chv-muse-p9:假解析器(公网 IP),供推理出口预审用; + 实验室绝不联网,真实解析只在真机发生。""" + return ["93.184.216.34"] + + class ChvGapCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -129,7 +135,11 @@ class ChvGapCase(unittest.TestCase): # genisoimage,测的不是 host key 模式,故显式 # 声明实验室路径,绝不把它写成生产默认。 "CHV_HOST_KEY_MODE": "keyscan", - "CHV_SSH_FORWARD": "1"} + "CHV_SSH_FORWARD": "1", + # feat-chv-muse-p9:推理出口预审的假 origin(公网 + # https,配假解析器;实验室不起真 VM、不联网)。 + "SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} if env_extra: env.update(env_extra) kwargs = dict( @@ -143,6 +153,7 @@ class ChvGapCase(unittest.TestCase): sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) + ex.sentinel_resolver = _public_resolver ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex diff --git a/cloud/tests/test_chv_muse_p9.py b/cloud/tests/test_chv_muse_p9.py new file mode 100644 index 0000000..2293789 --- /dev/null +++ b/cloud/tests/test_chv_muse_p9.py @@ -0,0 +1,570 @@ +"""对齐 Muse P9:推理与遥测受限通道。 + +全部用标准库与 mock,不启动真实 VM、不联网、不碰 WSL/真机; +解析器一律用假解析器(公网 IP / 私网 IP 两档)。 +""" + +import importlib.util +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +try: + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.inference_channel import ( + InferenceLimits, + check_inference_call, + inference_allow_hosts, + inference_limits, + resolve_inference_origin, + ) + from cloud.chv.sentinel import SentinelPolicy + from cloud.chv.telemetry import ( + build_task_telemetry, + build_trajectory_event, + emit_telemetry, + sanitize_telemetry_event, + sampled, + set_training_enabled, + telemetry_sample_rate, + training_enabled, + ) + from cloud.sandbox import ChvFailure +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.chv.executor import CloudHypervisorExecutor + from cloud.chv.inference_channel import ( + InferenceLimits, + check_inference_call, + inference_allow_hosts, + inference_limits, + resolve_inference_origin, + ) + from cloud.chv.sentinel import SentinelPolicy + from cloud.chv.telemetry import ( + build_task_telemetry, + build_trajectory_event, + emit_telemetry, + sanitize_telemetry_event, + sampled, + set_training_enabled, + telemetry_sample_rate, + training_enabled, + ) + from cloud.sandbox import ChvFailure + + +def _public_resolver(host, port): + """假解析器:公网 IP。""" + return ["93.184.216.34"] + + +def _private_resolver(host, port): + """假解析器:私网 IP(SSRF 场景)。""" + return ["10.0.0.5"] + + +def _load_host_module(): + """以独立模块名加载仓库根的 host.py(只用标准库)。""" + path = Path(__file__).resolve().parents[2] / "host.py" + spec = importlib.util.spec_from_file_location("host_under_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class OriginTest(unittest.TestCase): + def test_valid_sub2api_origin(self): + env = {"SUB2API_BASE_URL": "https://xgapi.lol/"} + self.assertEqual( + resolve_inference_origin("sub2api", env), "https://xgapi.lol") + + def test_missing_origin_returns_empty(self): + self.assertEqual(resolve_inference_origin("sub2api", {}), "") + self.assertEqual(resolve_inference_origin(None, {}), "") + + def test_http_origin_rejected(self): + env = {"SUB2API_BASE_URL": "http://xgapi.lol"} + with self.assertRaises(ValueError): + resolve_inference_origin("sub2api", env) + + def test_http_public_ip_rejected(self): + env = {"SUB2API_BASE_URL": "http://8.8.8.8:28084"} + with self.assertRaises(ValueError): + resolve_inference_origin("sub2api", env) + + def test_http_hostname_rejected(self): + env = {"SUB2API_BASE_URL": "http://relay.local:28084"} + with self.assertRaises(ValueError): + resolve_inference_origin("sub2api", env) + + def test_http_local_relay_allowed(self): + # 文档形状:本地 sub2api 中继走 Docker 桥接 IP + http, + # 端口保留;环回同款。 + env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084/"} + self.assertEqual( + resolve_inference_origin("sub2api", env), + "http://172.17.0.1:28084") + env2 = {"SUB2API_BASE_URL": "http://127.0.0.1:28084"} + self.assertEqual( + resolve_inference_origin("sub2api", env2), + "http://127.0.0.1:28084") + + def test_userinfo_in_origin_rejected(self): + env = {"SUB2API_BASE_URL": "https://user:pass@xgapi.lol"} + with self.assertRaises(ValueError): + resolve_inference_origin("sub2api", env) + + def test_non_443_port_rejected(self): + env = {"SUB2API_BASE_URL": "https://xgapi.lol:8443"} + with self.assertRaises(ValueError): + resolve_inference_origin("sub2api", env) + + def test_unknown_provider_falls_back_to_default_origin(self): + # 未知标签与 host.py `_server_command` 一致走默认网关; + # 未配 CHV_INFERENCE_ORIGIN 时返回 ""(调用方按 ask 处理)。 + env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} + self.assertEqual( + resolve_inference_origin("openai", env), "https://xgapi.lol") + self.assertEqual(resolve_inference_origin("openai", {}), "") + + def test_xcmapi_origin_from_explicit_env(self): + env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} + self.assertEqual( + resolve_inference_origin(None, env), "https://xgapi.lol") + + +class AllowlistTest(unittest.TestCase): + def test_explicit_allowlist_wins(self): + env = {"SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "a.example, B.example "} + hosts = inference_allow_hosts("sub2api", env) + self.assertEqual(hosts, frozenset({"a.example", "b.example"})) + + def test_default_allowlist_is_origin_host(self): + env = {"SUB2API_BASE_URL": "https://xgapi.lol"} + self.assertEqual(inference_allow_hosts("sub2api", env), + frozenset({"xgapi.lol"})) + + def test_no_origin_means_empty_allowlist(self): + self.assertEqual(inference_allow_hosts("sub2api", {}), frozenset()) + + +class LimitsTest(unittest.TestCase): + def test_defaults(self): + self.assertEqual(inference_limits({}), + InferenceLimits(timeout_seconds=120, + max_retries=1)) + + def test_env_override(self): + env = {"CHV_INFERENCE_TIMEOUT": "60", + "CHV_INFERENCE_MAX_RETRIES": "3"} + self.assertEqual(inference_limits(env), + InferenceLimits(timeout_seconds=60, max_retries=3)) + + def test_invalid_timeout_rejected(self): + for bad in ("abc", "5", "601", "-1"): + with self.assertRaises(ValueError): + inference_limits({"CHV_INFERENCE_TIMEOUT": bad}) + + def test_invalid_retries_rejected(self): + for bad in ("x", "-1", "6"): + with self.assertRaises(ValueError): + inference_limits({"CHV_INFERENCE_MAX_RETRIES": bad}) + + +class PreflightTest(unittest.TestCase): + def setUp(self): + self.env = {"SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"} + + def test_allow_with_public_ip(self): + check = check_inference_call( + "sub2api", self.env, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "allow") + self.assertIsNotNone(check.decision.capability) + self.assertTrue(check.decision.capability.matches(check.request)) + self.assertEqual(check.request.host, "xgapi.lol") + self.assertEqual(check.request.path, "/v1/responses") + + def test_capability_single_use_300s(self): + check = check_inference_call( + "sub2api", self.env, resolve=_public_resolver, now=1000.0) + cap = check.decision.capability + self.assertTrue(cap.use(check.request, now=1000.0)) + self.assertFalse(cap.use(check.request, now=1000.0)) + self.assertFalse(cap.matches(check.request, now=2000.0)) + + def test_local_relay_allow_without_dns(self): + # 本地中继是字面私网 IP:不查 DNS(解析器被调即错), + # allowlist 默认即 origin 主机,直接 allow + 单次能力。 + def _must_not_resolve(host, port): + raise AssertionError("local relay must not resolve") + env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084"} + check = check_inference_call( + "sub2api", env, resolve=_must_not_resolve) + self.assertEqual(check.decision.verdict, "allow") + self.assertIn("local_relay", check.decision.reasons) + self.assertIsNotNone(check.decision.capability) + self.assertEqual(check.request.host, "172.17.0.1") + self.assertEqual(check.request.port, 28084) + self.assertEqual(check.request.scheme, "http") + + def test_local_relay_ask_when_not_allowlisted(self): + env = {"SUB2API_BASE_URL": "http://172.17.0.1:28084", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"} + check = check_inference_call( + "sub2api", env, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "ask") + + def test_deny_on_private_ip(self): + check = check_inference_call( + "sub2api", self.env, resolve=_private_resolver) + self.assertEqual(check.decision.verdict, "deny") + self.assertIn("ssrf_private_ip", check.decision.reasons[0]) + + def test_ask_on_non_allowlisted_host(self): + # 显式 allowlist 已配、origin 主机不在其中 → ask。 + # (未配 allowlist 时默认即 origin 主机自身,不会 ask—— + # 见 test_default_allowlist_is_origin_host。) + env = {"SUB2API_BASE_URL": "https://unknown.example", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"} + check = check_inference_call( + "sub2api", env, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "ask") + + def test_ask_when_origin_not_configured(self): + check = check_inference_call( + None, {}, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "ask") + self.assertIn("inference_origin_not_configured", + check.decision.reasons) + + def test_xcmapi_origin_allow(self): + env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"} + check = check_inference_call( + None, env, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "allow") + + def test_unknown_provider_falls_back_to_default_origin(self): + # 未知标签不 deny:与 _server_command 一致走默认网关; + # 未配 CHV_INFERENCE_ORIGIN 时仍是 ask(fail-closed)。 + env = {"CHV_INFERENCE_ORIGIN": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"} + check = check_inference_call( + "prov-x", env, resolve=_public_resolver) + self.assertEqual(check.decision.verdict, "allow") + check2 = check_inference_call( + "prov-x", {}, resolve=_public_resolver) + self.assertEqual(check2.decision.verdict, "ask") + + def test_deny_path_out_of_scope(self): + policy = SentinelPolicy( + allow_hosts={"xgapi.lol"}, + connector_paths={"inference": (("POST", "prefix", + "/v1/responses"),)}) + from cloud.chv.sentinel import EgressRequest + request = EgressRequest("inference", "GET", "https", + "xgapi.lol", 443, "/v1/admin") + decision = policy.evaluate(request, resolve=_public_resolver) + self.assertEqual(decision.verdict, "deny") + + def test_audit_record_has_no_values(self): + check = check_inference_call( + "sub2api", self.env, resolve=_public_resolver) + record = check.decision.audit(check.request) + blob = json.dumps(record) + self.assertNotIn("xgapi.lol".replace("xgapi", "SECRET"), blob) + self.assertEqual(record["host"], "xgapi.lol") + self.assertEqual(record["verdict"], "allow") + + +class FakeResult: + def __init__(self, returncode=0, stdout="", stderr=""): + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +def _write(path, content=b"x"): + with open(path, "wb") as fh: + fh.write(content) + os.chmod(path, 0o755) + return path + + +class GateTest(unittest.TestCase): + """执行器预审:deny/ask 时不起 VM(wake 未被调用);allow 才继续。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) + self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) + self.kernel = _write(os.path.join(self.dir, "vmlinux")) + self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100) + self.ssh_key = _write(os.path.join(self.dir, "id_chv")) + self.kvm = _write(os.path.join(self.dir, "kvm")) + self.cpuinfo = os.path.join(self.dir, "cpuinfo") + with open(self.cpuinfo, "w") as fh: + fh.write("flags : fpu vmx sse\n") + self.state_dir = os.path.join(self.dir, "state") + self.wake_calls = [] + self.audit = [] + + def tearDown(self): + self.tmp.cleanup() + + def _make_exec(self, env_extra): + env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", + "CHV_HOST_KEY_MODE": "keyscan", + "CHV_SSH_FORWARD": "1"} + env.update(env_extra) + ex = CloudHypervisorExecutor( + binary=self.binary, ch_remote=self.ch_remote, + kernel=self.kernel, rootfs=self.rootfs, + ssh_key=self.ssh_key, kvm_path=self.kvm, + cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, + runner=lambda cmd, timeout=None: FakeResult(0, "", ""), + launcher=lambda cmd: FakeResult(0, "", ""), + api_client=lambda *a, **k: FakeResult(0, "", ""), + sleeper=lambda s: None, env=env) + ex.audit_callback = lambda e, t, d, k=None: self.audit.append( + (e, t.get("id") if isinstance(t, dict) else t, d, k)) + return ex + + def test_deny_blocks_wake(self): + ex = self._make_exec( + {"SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "xgapi.lol"}) + ex.sentinel_resolver = _private_resolver + ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake")) + task = {"id": "t1", "prompt": "hi"} + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task(task, model_provider="sub2api") + self.assertEqual(ctx.exception.kind, "inference_egress_denied") + ex.wake_vm.assert_not_called() + kinds = [e[0] for e in self.audit] + self.assertIn("inference_egress_deny", kinds) + + def test_ask_blocks_wake_when_origin_missing(self): + ex = self._make_exec({}) + ex.sentinel_resolver = _public_resolver + ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake")) + task = {"id": "t2", "prompt": "hi"} + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task(task, model_provider=None) + self.assertEqual(ctx.exception.kind, "inference_egress_denied") + ex.wake_vm.assert_not_called() + + def test_extra_env_origin_is_honored(self): + # 网关配置随任务来(worker 的 extra_env),执行器 + # 进程 env 里没有 origin 也要放行。 + ex = self._make_exec({}) + ex.sentinel_resolver = _public_resolver + ex.wake_vm = mock.Mock() + task = {"id": "t3", "prompt": "hi"} + with mock.patch.object(ex, "_run_guest", + return_value={"status": "completed"}): + with mock.patch.object(ex, "shutdown_vm"): + ex.dispatch_task( + task, model_provider="sub2api", + extra_env={"SUB2API_BASE_URL": "https://xgapi.lol"}) + ex.wake_vm.assert_called_once() + + def test_bad_origin_becomes_deny_not_value_error(self): + ex = self._make_exec({}) + ex.sentinel_resolver = _public_resolver + ex.wake_vm = mock.Mock(side_effect=AssertionError("must not wake")) + task = {"id": "t4", "prompt": "hi"} + with self.assertRaises(ChvFailure) as ctx: + ex.dispatch_task( + task, model_provider="sub2api", + extra_env={"SUB2API_BASE_URL": "http://xgapi.lol"}) + self.assertEqual(ctx.exception.kind, "inference_egress_denied") + ex.wake_vm.assert_not_called() + kinds = [e[0] for e in self.audit] + self.assertIn("inference_egress_deny", kinds) + + +class TelemetryTest(unittest.TestCase): + def test_allowlisted_fields_only(self): + event = build_task_telemetry( + task_id="t1", vm_id="vm1", provider="sub2api", + model="gpt-6.1-terra", status="completed", + started_at=1000.0, ended_at=1005.0, duration_ms=5000, + tokens={"input_tokens": 10, "output_tokens": 5, + "total_tokens": 15}, + turns=2, tool_calls=3, env={"CHV_TELEMETRY_SAMPLE_RATE": "1"}) + self.assertIsNotNone(event) + self.assertEqual(event["tokens_in"], 10) + self.assertEqual(event["tokens_total"], 15) + self.assertEqual(event["turns"], 2) + + def test_secret_and_text_keys_dropped(self): + event = {"task_id": "t1", "vm_id": "v", "provider": "sub2api", + "model": "m", "status": "completed", + "started_at": 1.0, "ended_at": 2.0, "duration_ms": 1, + "prompt": "do evil", "content": "secret file", + "api_key": "sk-123", "user_email": "a@b.c", + "extra_blob": {"nested": "data"}} + clean = sanitize_telemetry_event(event) + for bad in ("prompt", "content", "api_key", "user_email", + "extra_blob"): + self.assertNotIn(bad, clean) + self.assertEqual(clean["task_id"], "t1") + + def test_tokens_none_when_not_reported(self): + event = build_task_telemetry( + task_id="t1", vm_id="vm1", provider="sub2api", model=None, + status="failed", started_at=1.0, ended_at=2.0, duration_ms=1, + env={"CHV_TELEMETRY_SAMPLE_RATE": "1"}) + self.assertIsNone(event["tokens_in"]) + self.assertIsNone(event["tokens_out"]) + self.assertIsNone(event["tokens_total"]) + + def test_sampling_deterministic(self): + self.assertEqual(sampled("task-abc", 1.0), True) + self.assertEqual(sampled("task-abc", 0.0), False) + self.assertEqual(sampled("task-abc", 0.5), + sampled("task-abc", 0.5)) + event = build_task_telemetry( + task_id="t1", vm_id="v", provider="sub2api", model=None, + status="completed", started_at=1.0, ended_at=2.0, + duration_ms=1, env={"CHV_TELEMETRY_SAMPLE_RATE": "0"}) + self.assertIsNone(event) + + def test_invalid_sample_rate_rejected(self): + with self.assertRaises(ValueError): + telemetry_sample_rate({"CHV_TELEMETRY_SAMPLE_RATE": "2"}) + with self.assertRaises(ValueError): + telemetry_sample_rate({"CHV_TELEMETRY_SAMPLE_RATE": "x"}) + + def test_emit_never_raises(self): + self.assertFalse(emit_telemetry("/nonexistent/dir/xyz", + {"task_id": "t1"})) + self.assertFalse(emit_telemetry("", {"task_id": "t1"})) + + def test_emit_writes_jsonl(self): + with tempfile.TemporaryDirectory() as d: + event = {"task_id": "t9", "vm_id": "v", "status": "completed"} + self.assertTrue(emit_telemetry(d, event)) + with open(os.path.join(d, "telemetry.jsonl")) as fh: + row = json.loads(fh.read().strip()) + self.assertEqual(row["task_id"], "t9") + self.assertNotIn("prompt", row) + + +class TrainingOptTest(unittest.TestCase): + def test_default_enabled(self): + with tempfile.TemporaryDirectory() as d: + self.assertTrue(training_enabled(d)) + + def test_set_off_persists_and_blocks_trajectory(self): + with tempfile.TemporaryDirectory() as d: + self.assertTrue(set_training_enabled(d, False)) + self.assertFalse(training_enabled(d)) + event = build_trajectory_event( + "t1", 3, ["read", "write"], 100, d) + self.assertIsNone(event) + # 任务级遥测不受开关影响 + task_event = build_task_telemetry( + "t1", "v", "sub2api", None, "completed", + 1.0, 2.0, 1, env={"CHV_TELEMETRY_SAMPLE_RATE": "1"}) + self.assertIsNotNone(task_event) + + def test_trajectory_has_metadata_only(self): + with tempfile.TemporaryDirectory() as d: + event = build_trajectory_event( + "t1", 2, ["read", "exec"], 50, d) + self.assertIsNotNone(event) + self.assertEqual(event["tools"], ["read", "exec"]) + self.assertNotIn("prompt", event) + self.assertNotIn("content", event) + + def test_corrupt_state_fails_closed(self): + with tempfile.TemporaryDirectory() as d: + with open(os.path.join(d, "training-opt.json"), "w") as fh: + fh.write("{not json") + self.assertFalse(training_enabled(d)) + + def test_non_bool_rejected(self): + with tempfile.TemporaryDirectory() as d: + with self.assertRaises(TypeError): + set_training_enabled(d, "off") + + def test_switch_change_is_audited(self): + with tempfile.TemporaryDirectory() as d: + set_training_enabled(d, False, actor="tester") + audit_path = os.path.join(d, "training-opt-audit.jsonl") + with open(audit_path) as fh: + row = json.loads(fh.read().strip()) + self.assertEqual(row["event"], "training_opt_changed") + self.assertFalse(row["enabled"]) + self.assertEqual(row["actor"], "tester") + + +class HostPyTest(unittest.TestCase): + def setUp(self): + self.host = _load_host_module() + + def test_server_command_valid_origin(self): + with mock.patch.dict(os.environ, + {"SUB2API_BASE_URL": "https://xgapi.lol"}): + cmd = self.host._server_command("codex", provider="sub2api") + self.assertIn("app-server", cmd) + self.assertTrue(any("xgapi.lol/v1" in c for c in cmd)) + + def test_server_command_http_origin_rejected(self): + with mock.patch.dict(os.environ, + {"SUB2API_BASE_URL": "http://xgapi.lol"}): + with self.assertRaises(ValueError): + self.host._server_command("codex", provider="sub2api") + + def test_server_command_local_relay_allowed(self): + # 文档形状的本地中继(http + 桥接 IP)必须通过第二道防线。 + with mock.patch.dict(os.environ, + {"SUB2API_BASE_URL": + "http://172.17.0.1:28084"}): + cmd = self.host._server_command("codex", provider="sub2api") + self.assertTrue(any("172.17.0.1:28084/v1" in c for c in cmd)) + + def test_server_command_allowlist_mismatch_rejected(self): + with mock.patch.dict(os.environ, + {"SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ALLOW_HOSTS": "other.example"}): + with self.assertRaises(ValueError): + self.host._server_command("codex", provider="sub2api") + + def test_request_timeout_default_and_env(self): + self.assertEqual( + self.host._inference_request_timeout(None), 120) + self.assertEqual( + self.host._inference_request_timeout(60), 60) + with mock.patch.dict(os.environ, + {"CHV_INFERENCE_TIMEOUT": "30"}): + self.assertEqual( + self.host._inference_request_timeout(None), 30) + with mock.patch.dict(os.environ, + {"CHV_INFERENCE_TIMEOUT": "5"}): + with self.assertRaises(ValueError): + self.host._inference_request_timeout(None) + + def test_request_retries_default_and_env(self): + self.assertEqual(self.host._inference_max_retries(), 1) + with mock.patch.dict(os.environ, + {"CHV_INFERENCE_MAX_RETRIES": "0"}): + self.assertEqual(self.host._inference_max_retries(), 0) + with mock.patch.dict(os.environ, + {"CHV_INFERENCE_MAX_RETRIES": "9"}): + with self.assertRaises(ValueError): + self.host._inference_max_retries() + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_chv_optimize.py b/cloud/tests/test_chv_optimize.py index 371ac10..d56247f 100644 --- a/cloud/tests/test_chv_optimize.py +++ b/cloud/tests/test_chv_optimize.py @@ -86,6 +86,12 @@ def _write(path, content=b"x"): return path +def _public_resolver(host, port): + """feat-chv-muse-p9:假解析器(公网 IP),供推理出口预审用; + 实验室绝不联网,真实解析只在真机发生。""" + return ["93.184.216.34"] + + class ChvOptCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -118,7 +124,11 @@ class ChvOptCase(unittest.TestCase): # genisoimage,测的不是 host key 模式,故显式 # 声明实验室路径,绝不把它写成生产默认。 "CHV_HOST_KEY_MODE": "keyscan", - "CHV_SSH_FORWARD": "1"} + "CHV_SSH_FORWARD": "1", + # feat-chv-muse-p9:推理出口预审的假 origin(公网 + # https,配假解析器;实验室不起真 VM、不联网)。 + "SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} if env_extra: env.update(env_extra) if runner is None: @@ -135,7 +145,9 @@ class ChvOptCase(unittest.TestCase): runner=runner, launcher=launcher, api_client=api_client, sleeper=lambda s: None, env=env) kwargs.update(overrides) - return CloudHypervisorExecutor(**kwargs) + ex = CloudHypervisorExecutor(**kwargs) + ex.sentinel_resolver = _public_resolver + return ex def _default_runner(self, cmd, timeout=None): self.calls.append(list(cmd)) diff --git a/cloud/tests/test_cloud_hypervisor.py b/cloud/tests/test_cloud_hypervisor.py index 12389cf..cdc1155 100644 --- a/cloud/tests/test_cloud_hypervisor.py +++ b/cloud/tests/test_cloud_hypervisor.py @@ -73,6 +73,12 @@ def _write(path, content=b"x"): return path +def _public_resolver(host, port): + """feat-chv-muse-p9:假解析器(公网 IP),供推理出口预审用; + 实验室绝不联网,真实解析只在真机发生。""" + return ["93.184.216.34"] + + class ChvCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() @@ -106,7 +112,11 @@ class ChvCase(unittest.TestCase): # genisoimage,测的不是 host key 模式,故显式 # 声明实验室路径,绝不把它写成生产默认。 "CHV_HOST_KEY_MODE": "keyscan", - "CHV_SSH_FORWARD": "1"} + "CHV_SSH_FORWARD": "1", + # feat-chv-muse-p9:推理出口预审的假 origin(公网 + # https,配假解析器;实验室不起真 VM、不联网)。 + "SUB2API_BASE_URL": "https://xgapi.lol", + "CHV_INFERENCE_ORIGIN": "https://xgapi.lol"} if env_extra: env.update(env_extra) if runner is None: @@ -124,6 +134,7 @@ class ChvCase(unittest.TestCase): sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) + ex.sentinel_resolver = _public_resolver ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex diff --git a/docs/chv-access-boundary.md b/docs/chv-access-boundary.md new file mode 100644 index 0000000..137162f --- /dev/null +++ b/docs/chv-access-boundary.md @@ -0,0 +1,45 @@ +# CHV VM 运营访问边界说明(feat-chv-muse-p9) + +> 口径依据:Meta 官方安全文(security.muse.ai 2026-09-08)对 Muse 当前架构的 +> 明确表述——当前架构**通过运营政策限制 Meta 人员访问用户数据,但并不阻止 +> Meta 在必要时为支持、保障或运营服务而访问数据**;“连 Meta 也看不到”是未来 +> Confidential VM 的目标,**截至 2026-11-01 的公开材料中未见 Confidential VM +> 已全面上线**。本文件把同一口径落到本项目:只写文档 + 审计钩子, +> **不冒充技术上不可访问**。 + +## 1. 谁能访问 VM 内数据 + +| 访问方 | 能访问什么 | 边界 | +|---|---|---| +| 用户本人(VM 属主) | 全部:工作区文件、记忆、凭证替身、审计记录 | 属主即所有者;可检查、编辑、下载 VM 内所有文件,也可要求忘记指定内容(见 P7) | +| Agent(codex,单元内) | 任务所需的工作区与工具输出 | 受 nspawn 单元(P2)、authd 替身凭证(P3)、Sentinel 出口审批(P3/P5)约束;真凭证与宿主侧数据对单元内不可见 | +| 宿主运维(部署机器的 root / codexsvc) | VM 磁盘镜像、state_dir、日志、审计 | **技术上可以访问**:宿主侧持有磁盘文件与进程,OS 账户即边界。本文件不声称运维看不到——只规定运维**应在什么情况下**访问(见 §2),并要求访问留审计 | +| Meta(对照口径) | 不适用——本项目无 Meta 侧运营方 | 本项目是用户自部署,不存在 Meta 人员访问;写这一行只为对齐口径:若未来托管化,边界按同一句话执行 | + +## 2. 运维访问规则(政策,非技术强制) + +1. **日常不看**:运维不主动翻看用户工作区文件、对话记录与记忆;只看任务级元数据(任务 id、起止、成败、耗时)与审计事件。 +2. **排障可看**:任务失败、VM 异常、安全事件调查时,运维可查看相关 VM 的日志、审计记录与任务输出;查看行为本身记入审计(`chv_audit` + 操作日志)。 +3. **凭证永不直接取**:真凭证(Gitea token、网关 key)只经 authd 替身释放;运维排障需要凭证时走替身铸发流程,不直接读 seed 文件;seed 文件 0600,属主 root/codexsvc。 +4. **数据不出宿主**:排障取证不出部署机器;必须外发时先去标识化(任务 id 可留,对话原文、文件内容、密钥一律去掉),并记审计。 +5. **删除即删除**:用户要求删除任务数据/记忆时,运维删除 VM 盘、state_dir 与相关遥测行,并记审计;备份里的副本按备份保留期滚动清除,不单点恢复。 + +## 3. 审计(可查的证据) + +- 任务生命周期审计:`chv_audit`(执行器 `_emit`,失败落盘 `chv-audit-spill.jsonl`)。 +- 出口审批审计:Sentinel 决策记录(`sentinel_egress_allow/deny/ask`,只含目标描述与原因,不含值)。 +- 训练开关审计:`training-opt-audit.jsonl`(开关变更时间、开/关、操作方)。 +- 遥测:`telemetry.jsonl`(只含任务级元数据字段,见 `docs/chv-muse-p9-gap.md` §2.1)。 + +审计记录里绝不含凭证值、对话原文与文件内容;审计失败不中断任务(worker 层 best-effort 语义,见执行器 `_emit` 注释)。 + +## 4. 广告隔离(口径,不用写代码) + +- 本项目不把用户对话或 VM 数据分享给任何广告系统;遥测字段白名单里没有可用于广告画像的字段。 +- Agent 浏览互联网时表现为用户活动,可能间接影响用户在别处看到的广告(与 Muse 官方口径一致)——这是浏览行为本身的副作用,不是本项目分享数据。 + +## 5. 明确未做的事 + +- **Confidential VM**:未做。Meta 自己也未交付(路线图),本项目不自研密码学隔离 VM;当前边界是 OS 账户 + 文件权限 + 审计,不是密码学不可访问。任何“连运维也看不到”的说法在本架构下都是错的,不要这样对外写。 +- **端到端加密记忆**:未做。记忆文件在 VM 盘上明文存放(权限 0600),依赖磁盘与宿主访问控制。 +- **操作审批 UI**:运维访问的审批流程目前是人工约定(本文件),没有强制审批卡;强制审批是后续工作,不在本轨冒充已实现。 diff --git a/docs/chv-muse-p9-gap.md b/docs/chv-muse-p9-gap.md new file mode 100644 index 0000000..c58ad0c --- /dev/null +++ b/docs/chv-muse-p9-gap.md @@ -0,0 +1,157 @@ +# 对齐 Muse P9 差距表:推理与遥测受限通道 + +> 先写表再动手。依据:`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md` +> (Meta 官方 security.muse.ai 与设计文 2026-09-08 为准;第三方拆解只作参考,不得把逆向细节写成官方架构)。 +> 基线:`06c3e17`。分支:`feat-chv-muse-p9`。本地做,不推、不合、不部署。 + +## 1. 现状 vs Muse 官方要求 + +| # | Muse 官方要求(security.muse.ai / 设计文) | 现状(基线 06c3e17) | 本轨做法 | +|---|---|---|---| +| 1 | 推理与遥测代理是通向外部基础设施的**受限通道** | 模型调用无收敛点:`host.py` 起 `codex app-server`,provider=`sub2api` 时 `-c` 覆盖 base_url(取 `SUB2API_BASE_URL`);provider 为空时 codex 用自身 config.toml 默认网关(xcmapi 路径),执行器侧没有任何出口校验 | 新模块 `cloud/chv/inference_channel.py`:单一出口点 + 执行器 `_run_with_vm` 起 VM 前预审(Sentinel 评估);`host.py` 的 `_server_command` 做结构校验(https、无 userinfo、主机在 allowlist 内)当第二道防线 | +| 2 | VM 发送**有限数据**用于推理与遥测(安全文**未给字段级清单**) | 无任何遥测:任务元数据只进宿主日志与审计 spill,无字段清单、无采样、无发送路径 | 新模块 `cloud/chv/telemetry.py`:字段级白名单(只含任务级元数据,见 §2);采样率可配;落盘 `/telemetry.jsonl`,发送失败只记日志、不影响任务 | +| 3 | 推理轨迹(对话往返、工具调用、子代理交接)在**去除关键个人身份信息后**可用于训练新检查点;用户可**一键退出训练** | 无训练上传、无开关 | 训练退出开关(`training_opt`,见 §2):状态持久化在 `/training-opt.json`,关后轨迹类事件不再产生;开关变更进审计。**默认值开是本项目决定,非 Meta 官方值**(报告注明帮助中心原文未逐字确认默认开启) | +| 4 | 当前 VM 的 Meta 访问边界(官方明确,非营销口径):运营政策限制 Meta 人员访问用户数据,但**并不阻止** Meta 在必要时为支持、保障或运营而访问;“连 Meta 也看不到”是未来 Confidential VM 的目标 | 无边界说明 | `docs/chv-access-boundary.md`:只写文档 + 审计钩子,不冒充技术上不可访问;Confidential VM 明确列为未做 | +| 5 | 广告隔离:不把用户对话或 VM 数据分享给广告系统;但浏览互联网时表现为用户活动、可能间接影响广告 | 无口径 | 只记为口径(见 §4),不用写代码 | +| 6 | Sentinel 是连接器动作与全部网络出口的唯一权限机构 | P3 已做 Sentinel 最小审批、P5 已做纵深(本基线**都没有**;P3 在 `feat-chv-muse-p3` 分支 a90cf78,P5 在 `feat-chv-muse-p5` 分支 25bcd35) | 推理通道走 Sentinel 审批,不重复造审批:`cloud/chv/sentinel.py` 从 **P5 的 25bcd35 拷入**(P5 版已是 P3 的超集,顶层仍只用标准库),再加 P9 的 inference 条目(三张表里加 `inference`:scheme https、端口 443、路径 `POST /v1/responses` + `GET /v1/models`);合入时以本文件为准(它是 P3+P5+P9 的并集) | + +## 2. 本项目自定清单(非 Meta 官方值,逐项注明) + +### 2.1 遥测字段白名单(`telemetry.py`) + +只含任务级元数据,**绝不含**对话原文、文件内容、密钥、个人身份信息: + +| 字段 | 含义 | 来源 | +|---|---|---| +| `task_id` | 任务 id | 任务 dict | +| `vm_id` | VM 名 | 执行器 | +| `provider` | 模型 provider(`sub2api` / `xcmapi`) | 通道决策 | +| `model` | 模型名 | 任务 dict | +| `status` | `completed` / `failed` / `interrupted` / `timeout` | 运行结果 | +| `started_at` / `ended_at` | 起止时间戳(秒) | 执行器 | +| `duration_ms` | 耗时毫秒 | 执行器 | +| `tokens_in` / `tokens_out` / `tokens_total` | token 数 | host.py 回传的 usage(未回传时记 `null`,**不编造**) | +| `turns` | 回合数 | 运行结果 | +| `tool_calls` | 工具调用次数 | 运行结果 | +| `sampled` | 是否被采样命中 | 采样器 | + +不在白名单里的键一律丢弃;键名含 `key` / `token` / `secret` / `password` / `prompt` / +`content` / `text` / `message` / `email` / `name` 的一律丢弃(防误带)。采样率 +`CHV_TELEMETRY_SAMPLE_RATE`:0.0–1.0,默认 `1.0`(全量,任务级元数据量小);按 +`task_id` 确定性采样(同一任务每次判定一致,可测试)。 + +### 2.2 轨迹事件(训练用) + +轨迹类事件只含回合级元数据(回合数、工具名、耗时),**不含任何文本**;去标识化 += 只出元数据、不出内容。退出开关关后,轨迹类事件**不再产生**(不是产生后丢弃)。 + +### 2.3 推理通道上限(自定) + +| 项 | 环境变量 | 默认 | 范围 | 说明 | +|---|---|---|---|---| +| 单次请求超时(秒) | `CHV_INFERENCE_TIMEOUT` | `120` | 10–600 | 与 `host.py` 的 `Server.request` 默认超时一致;非法值即 ValueError | +| 重试上限(次) | `CHV_INFERENCE_MAX_RETRIES` | `1` | 0–5 | `Server.request` 的 BrokenPipe 重连次数上限;默认 `1` 即**保持现有行为不变**(现为 `range(2)`);模型回合级重试仍是 worker 的 `max_retries` 语义,本轨不动 | + +### 2.4 训练退出开关默认值 + +默认**开**(轨迹可用于训练)。这是**本项目决定,非 Meta 官方值**:报告注明帮助中心原文未逐字确认 +默认开启。开关状态持久化、变更进审计;关后立即生效(后续轨迹不再产生)。 + +### 2.5 推理出口 allowlist 与 origin 规则 + +- provider=`sub2api`:origin 取 `SUB2API_BASE_URL`(去尾斜杠);未知 provider 标签与 + provider 为空一样走默认网关(与 host.py `_server_command` 一致——标签本身不决定 + origin,origin 永远只取 operator 配置的 env)。 +- provider 为空或未知标签(xcmapi 路径,codex 用自身 config.toml 默认网关):执行器 + **看不到** origin,改取 `CHV_INFERENCE_ORIGIN`(operator 显式配置,如 `https://xgapi.lol`); + 未配即 `ask`——审计 + 任务停止(fail-closed),**不静默放行**。部署迁移:现网 xcmapi 部署 + 须补设 `CHV_INFERENCE_ORIGIN`,否则任务会在预审处停下(这是有意行为,见 §3)。 +- scheme 规则(2026-11-07 全量套件复核时修正——初版只认 https 是错的): + - https:公网网关(如 xgapi.lol),端口只接受 443(或缺省); + - http:**只接受本地中继**——字面环回/私网 IP 主机(如 cloud/README.md 的 sub2api 网关 + 文档形状 `http://172.17.0.1:28084`,Docker 桥接 IP),端口保留;http + 公网主机名/IP、 + http + 主机名一律拒绝。主机名一律不算本地中继(DNS 结果不可信)。 + - 本地中继分支:字面 IP 无 DNS、无重绑定,跳过 Sentinel 的 DNS/SSRF 步;allowlist 与 + 单次能力(P3 默认档)照常执行;解析器不被调用(测试用必错解析器钉住)。Sentinel 表里的 + inference scheme/端口仍只写 https/443——本地中继不走 `evaluate`,不要“顺手”把 http + 加进表(那会放宽公网路径)。 +- 路径规则(`inference` 连接器):只放 `POST /v1/responses` 与 `GET /v1/models`; + `/v1/models` 是否必需**待真机核实**(codex 启动时是否拉模型列表),若非必需则收紧为只放 + `/v1/responses`。公网路径:解析到私网 IP 即 deny(SSRF,复用 Sentinel);解析失败即 deny。 +- 任务级 env 合并:门的 origin 取任务级 `extra_env`(网关配置随任务来,如 worker 的 + `_gateway_env`)覆盖执行器进程 env;origin 非法(http 公网、带 userinfo、https 非 443 + 端口)记 deny,不抛裸 ValueError。报错/审计原因不带 origin 原值(http URL 里可能含 + userinfo,见 `resolve_inference_origin`)。 + +## 3. 改什么(文件清单) + +| 文件 | 动作 | +|---|---| +| `cloud/chv/sentinel.py` | **新增**:从 P5 分支 `25bcd35` 拷入(P5 版已是 P3 的超集,顶层仍只用标准库),再加 P9 的 inference 条目(三张表:路径、scheme、端口);合入时以本文件为准(P3+P5+P9 的并集) | +| `cloud/chv/inference_channel.py` | **新增**:受限推理通道(origin 解析、allowlist、上限校验、Sentinel 预审、`InferenceChannelDenied` 之外的 Decision 返回) | +| `cloud/chv/telemetry.py` | **新增**:遥测事件(字段白名单、采样、落盘、永不抛错)+ 训练退出开关(持久化、审计) | +| `cloud/chv/executor.py` | **改动**:`_run_with_vm` 起 VM 前调推理预审(`_inference_gate`);allow 通过、deny/ask 审计 + `ChvFailure` 停止,不起 VM。`execute()`(只读 status,不跑模型回合)不设门 | +| `host.py` | **改动**:`_server_command` 的 sub2api origin 结构校验(https、无 userinfo、主机在 allowlist 内);`Server.request` 默认超时与重连上限改走 `CHV_INFERENCE_TIMEOUT` / `CHV_INFERENCE_MAX_RETRIES`(默认 120 秒 / 1 次,即现有行为) | +| `cloud/tests/test_chv_muse_p9.py` | **新增**:全部覆盖(见 §5) | +| `docs/chv-muse-p9-gap.md` | **新增**:本文件 | +| `docs/chv-access-boundary.md` | **新增**:运营访问边界说明 | + +**不碰的**:`cloud/chv/config.py`(通道读 `env` 字典,不进 VM 配置片段;架构测试无“env 必须进片段”的规矩)、 +P3 的 authd、P5 的 eBPF 污点(P5 已结论:本 VM 的 bpf() 被 seccomp 挡,不可行;本轨不重复探)、 +P6 的确定性检查层(审计复用执行器已有的 `_emit`,不另造审计接口)、worker 的 `max_retries`、 +guest 镜像(`host.py` 是标准库单文件,`/app/host.py` 拷入方式不变——本轨对 `host.py` 的改动只用标准库)。 + +## 4. 诚实边界 + +- 本轨是**决策层 + 预审**,不是网络层强制:客内 codex 直连的强制仍靠 Tinyproxy 白名单 + 宿主防火墙 + (第三棒落地中)。不要把本轨写成“出口已全部经 Sentinel”——P3 的模块 docstring 同款声明。 +- provider 为空时执行器看不到 codex config.toml 里的 origin:本轨用 `CHV_INFERENCE_ORIGIN` + 显式配置 + 未配即 ask 兜底;这不是“读到了 config”,部署时须补 env(见 §2.5)。 +- 遥测目前只落盘 `/telemetry.jsonl`,**没有**向外部遥测端点发送的路径;“发送失败不影响任务” + 指落盘失败也不影响任务。向外部发遥测是后续工作,需另批。 +- 训练轨迹目前**没有**上传路径:开关管的是“轨迹类事件是否产生”;上传通道本身不存在,开关是为 + 上传通道预留的门禁,不是已接通的上传。 +- token 数只取 host.py 回传的 usage;codex 未回传时记 `null`,不估算、不编造。 +- `GET /v1/models` 是否必需未验证(见 §2.5),真机核实前不收紧、也不放宽。 +- `sentinel.py` 是从 P5 分支(25bcd35)原样拿来的超集版本:其中 `capability_for` 的非 one-shot + 分支在运行时 import `cloud.chv.capabilities`——该模块是 P5 的文件,本轨基线 06c3e17 里没有。 + 本轨只走 one-shot 默认档,运行时不触及该分支(新测试 51/51 证明);mypy 会报 1 处 + `import-not-found`,待 P3/P5/P9 集成轨把 capabilities.py 带进来后消除,不在本轨另造一份。 + 其余 mypy 报错与基线一致:`executor.py:1396/1401`(`out` 未注解)与 `marker_tools.py:142` + 是 P3 已记录的基线问题,本轨未碰。 + +## 5. 测试清单(`test_chv_muse_p9.py`,全部用标准库与 mock,不联网、不启动真实 VM) + +- origin 解析:合法 origin 通过;缺 `SUB2API_BASE_URL` 即错;http 公网主机名/IP 被拒; + http 本地中继(字面环回/私网 IP,如 `http://172.17.0.1:28084`)通过且端口保留; + URL 带 userinfo 被拒;尾斜杠被剥掉;未知 provider 标签走默认网关(不 deny)。 +- allowlist:`CHV_INFERENCE_ALLOW_HOSTS` 显式覆盖生效;默认即 origin host。 +- 上限校验:超时/重试非法值(非整数、越界)即 ValueError;默认值 120/1。 +- 预审决策(假解析器,不联网):allow(公网 IP + 白名单 + 路径内);deny(解析到私网 IP、 + origin 非法);ask(主机不在白名单、provider 为空且未配 `CHV_INFERENCE_ORIGIN`); + 路径越界(`GET /v1/admin`)即 deny;能力绑定(单次、300 秒);本地中继 allow(不查 DNS, + 用必错解析器钉住)与本地中继 ask(allowlist 不含该主机)。 +- 旁路:直连(非通道)调模型无入口——执行器 `_run_with_vm` 前必过预审(mock 执行器验调用顺序: + 预审 deny 时 `wake_vm` 未被调用);任务级 `extra_env` 的 origin 被合并进预审(worker 网关 + 路径),非法 origin 转 deny 而不是裸 ValueError。 +- 遥测:白名单外键丢弃;`prompt`/`content`/`api_key` 类键丢弃;采样确定性(同一 task_id 多次判定一致; + 采样率 0 全丢、1 全留);落盘失败(state_dir 不可写)不抛错、不影响任务;token 未回传记 `null`。 +- 训练开关:默认开;关后轨迹事件不再产生、任务级遥测不受影响;开关变更持久化 + 审计有记录; + 状态文件损坏时 fail-closed(当作关处理,记日志)。 +- host.py:`_server_command` 的 origin 结构校验(http 公网被拒、本地中继通过、allowlist 不匹配 + 被拒、合法通过);`Server.request` 默认超时与重试上限走 env(mock 验默认值 120/1 与 env 覆盖)。 +- 旧测试 harness:5 个文件的 `make_exec` 配了假 origin(`SUB2API_BASE_URL` 与 + `CHV_INFERENCE_ORIGIN` 均为 `https://xgapi.lol`)+ 假公网解析器——这是新 fail-closed 门 + 要求的测试配置,不是放宽门;门的 deny/ask 行为由本轨新测试钉住。 + +## 6. 真机待验证清单(排队,测试机空闲后;第三棒 → round-11 → P4/P5/P6 → 本轨) + +1. `GET /v1/models` 是否必需:真机跑一轮 turn,确认 codex 是否调它;必需则保留,不必需则收紧为只放 `POST /v1/responses`。 +2. 预审对真实 xgapi 主机的判定:`https://xgapi.lol` 解析后 IP 是否干净(不在 SSRF 挡截网段内);若解析到被挡网段,记录真实判定、不绕过。 +3. `CHV_INFERENCE_ORIGIN` 未配时 xcmapi 任务是否如实停在预审(ask + 审计),配上后是否放行。 +4. 本地中继真机:`SUB2API_BASE_URL=http://172.17.0.1:28084` 的任务预审是否放行(本地中继分支)、 + 真实 turn 是否走通该中继;同时确认 http 公网 origin 在真机仍被拒。 +5. 真实 turn 的 usage 是否回传:遥测里的 token 字段是否从 `null` 变成实数。 +6. 遥测落盘文件在真实任务后是否存在、内容是否只有白名单字段。 +7. 训练开关在真机上的持久化:开关翻转后重启执行器,状态是否还在。 +7. 超时/重试上限在真机上的表现:长 turn 是否在 `CHV_INFERENCE_TIMEOUT` 处超时(只记录、不主动造超时)。 diff --git a/host.py b/host.py index d58a2f4..64462eb 100644 --- a/host.py +++ b/host.py @@ -45,6 +45,7 @@ CODEX_BIN=.../bin/codex(start.sh 已自动优先用 bin/codex)。 """ import argparse import contextlib +import ipaddress import json import os import queue @@ -56,6 +57,7 @@ import threading import time import uuid from concurrent.futures import ThreadPoolExecutor +from urllib.parse import urlparse HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_STATE = os.path.join(HERE, "host-state.json") @@ -338,6 +340,81 @@ def _append_trace_marker(state_path, task_name, entry): pass +def _inference_request_timeout(timeout): + """单次请求超时(秒):显式传参优先;否则读 + `CHV_INFERENCE_TIMEOUT`(默认 120,与旧默认值一致), + 非整数或越界(10–600)即 ValueError。""" + if timeout is not None: + return timeout + raw = os.environ.get("CHV_INFERENCE_TIMEOUT", "").strip() + if not raw: + return 120 + try: + value = int(raw) + except ValueError: + raise ValueError( + "CHV_INFERENCE_TIMEOUT 必须是整数: %r" % raw) + if not 10 <= value <= 600: + raise ValueError( + "CHV_INFERENCE_TIMEOUT 必须在 10-600 之间: %s" % value) + return value + + +def _inference_max_retries(): + """BrokenPipe 重连上限(次):读 `CHV_INFERENCE_MAX_RETRIES`, + 默认 1(即现有行为 `range(2)`),非整数或越界(0–5)即 + ValueError。模型回合级重试仍是 worker 的 max_retries 语义, + 这里只 bound 传输层重连。""" + raw = os.environ.get("CHV_INFERENCE_MAX_RETRIES", "").strip() + if not raw: + return 1 + try: + value = int(raw) + except ValueError: + raise ValueError( + "CHV_INFERENCE_MAX_RETRIES 必须是整数: %r" % raw) + if not 0 <= value <= 5: + raise ValueError( + "CHV_INFERENCE_MAX_RETRIES 必须在 0-5 之间: %s" % value) + return value + + +def _validate_inference_origin(origin): + """sub2api origin 结构校验(feat-chv-muse-p9,第二道防线, + 只用标准库):https 公网网关(端口只 443)或 http 本地中继 + (字面环回/私网 IP,如 http://172.17.0.1:28084,见 + cloud/README.md 的 sub2api 网关一节);无 userinfo、有主机 + 名;若 `CHV_INFERENCE_ALLOW_HOSTS` 已配,主机必须在其中。 + 第一道防线是执行器的 Sentinel 预审(起 VM 前);这里拦的 + 是“执行器没审就直接起 codex”的旁路(配置错了 loud fail, + 不静默跑向别处)。报错不带 origin 原值。""" + parsed = urlparse(origin) + scheme = parsed.scheme.lower() + if "@" in parsed.netloc: + raise ValueError("推理 origin 不得带 userinfo(凭证禁进 URL)") + host = (parsed.hostname or "").lower() + if not host: + raise ValueError("推理 origin 无主机名") + if scheme == "https": + if parsed.port is not None and parsed.port != 443: + raise ValueError("推理 origin 端口只接受 443") + elif scheme == "http": + try: + addr = ipaddress.ip_address(host) + except ValueError: + raise ValueError("http 推理 origin 只接受本地中继" + "(环回/私网 IP)") + if not (addr.is_loopback or addr.is_private): + raise ValueError("http 推理 origin 只接受本地中继" + "(环回/私网 IP)") + else: + raise ValueError("推理 origin 必须 https(或本地中继 http)") + allowed = os.environ.get("CHV_INFERENCE_ALLOW_HOSTS", "") + hosts = {h.strip().lower() for h in allowed.split(",") if h.strip()} + if hosts and host not in hosts: + raise ValueError("推理 origin 主机不在 allowlist 内: %s" % host) + + def _server_command(codex_bin, provider=None): """app-server 启动命令:-c 必须放在子命令之前。 @@ -361,6 +438,9 @@ def _server_command(codex_bin, provider=None): if not origin: raise RuntimeError( "provider=sub2api 但 SUB2API_BASE_URL 未配置") + # feat-chv-muse-p9:origin 结构校验(第二道防线;第一道 + # 是执行器起 VM 前的 Sentinel 预审)。 + _validate_inference_origin(origin) cmd += [ "-c", 'model_provider="sub2api"', "-c", 'model_providers.sub2api.name="Sub2API"', @@ -519,14 +599,19 @@ class Server: self.proc.stdin.write((json.dumps(obj) + "\n").encode()) self.proc.stdin.flush() - def request(self, method, params=None, timeout=120): - # BrokenPipe / 进程已死时重拉一次再重试(只重试一次, + def request(self, method, params=None, timeout=None): + # BrokenPipe / 进程已死时重拉再重试(重试上限见 + # `_inference_max_retries`,默认 1 次即现有行为; # turn/start 这类有副作用的调用由上层 resume 后再跑, # 这里重试的是 initialize/list/resume 等幂等请求为主; # 即使是 turn/start,重试也只发生在“还没收到任何响应” # 的 BrokenPipe 场景,server 死在 turn 中途时 _read_loop # 已退出、q.get 会超时而不是走这条路)。 - for attempt in range(2): + # feat-chv-muse-p9:默认超时走 CHV_INFERENCE_TIMEOUT + # (默认 120 秒,即旧默认值),显式传参优先。 + timeout = _inference_request_timeout(timeout) + max_retries = _inference_max_retries() + for attempt in range(1 + max_retries): rid = self.next_id self.next_id += 1 q = queue.Queue() @@ -539,7 +624,7 @@ class Server: except (BrokenPipeError, OSError): with self.pending_lock: self.pending.pop(rid, None) - if attempt == 0 and self.ensure_alive(): + if attempt < max_retries and self.ensure_alive(): continue raise finally: