diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c2b547c..3dd9948 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -74,6 +74,10 @@ jobs: - name: Code sandbox (in-pod test runner) run: python scripts/test-sandbox.py + - name: Security boundary enforcement (secret/workspace/sandbox) (#19) + env: { REDIS_FAKE: "1" } + run: python scripts/test-security-boundary.py + - name: Quality instrumentation (Group B) env: { REDIS_FAKE: "1" } run: python scripts/test-quality.py diff --git a/docs/integration/security-boundary.md b/docs/integration/security-boundary.md index c49af85..21d5124 100644 --- a/docs/integration/security-boundary.md +++ b/docs/integration/security-boundary.md @@ -1,6 +1,10 @@ # 安全边界（Secret / Workspace / Tool / Approval / Tenant / Sandbox） -> 状态：**部分已实现，部分待接入**（依据 `heicode-mananger/docs/heicode.md §六/七`、`docs/heicode-runtime-auth-newapi-secret-design.md §三/五`、`docs/integration/heicode-am-contract.md §3.1/§4`）。 +> 状态：**已强制边界 FROZEN v1（可验证）+ 部分待接入**（依据 `heicode-mananger/docs/heicode.md §六/七`、`docs/heicode-runtime-auth-newapi-secret-design.md §三/五`、`docs/integration/heicode-am-contract.md §3.1/§4`）。回应 issue #19。 +> +> **本仓强制并可验证的边界**（契约测试 `scripts/test-security-boundary.py`，CI 守护）：① secret 仅 `azkv://` 引用、明文密钥入口拒绝、回调/持久化脱敏；② workspace 写入路径越界（绝对路径 / `..` 逃逸）拒绝；③ 代码沙箱 **fail-closed**（未确认隔离则拒绝执行）。 +> +> **本次不扩展（确认的边界/后续层，非本仓可改）**：tool/MCP 权限引擎（**无 SK/MCP 工具层**可治理）、租户隔离（**有意**按 `user/channelId` 归因、不引入 tenant）、Pod 级强化沙箱（seccomp/只读根/NetworkPolicy，归 Infra/Security）。详见 §9。 > > 配套：[`runtime-contract.md`](./runtime-contract.md)、[`usage-billing-schema.md`](./usage-billing-schema.md)。 @@ -84,17 +88,21 @@ ## 9. 覆盖与缺口 +✅✔ = 已实现**且有契约测试**（`scripts/test-security-boundary.py`）。 + | 边界 | 状态 | |---|---| -| `azkv://` `secret_ref` 强校验 / 明文拒绝 / 脱敏 / `.gitignore` | ✅ 已实现（本仓） | -| Workspace 路径越界校验 | ✅ 已实现 | -| 审批状态机（waiting_approval + approvals 回执） | ✅ 已实现（逐项校验待加强） | +| `azkv://` `secret_ref` 强校验 / 明文拒绝 / 脱敏 / `.gitignore` | ✅✔ 已实现 + 测试（`validate_create_request`/`_reject_plaintext_secrets`/`_redact_sensitive`） | +| Workspace 路径越界校验（绝对路径 / `..` 逃逸拒绝） | ✅✔ 已实现 + 测试（`task_executor._resolve_workspace_path`） | +| 代码沙箱 fail-closed（`assert_isolated`：未确认隔离拒绝执行；启动 `assert_quality_eval_safe`） | ✅✔ 已实现 + 测试（§8.1 双门控 `ENABLE_QUALITY_EVAL`+`HEICODE_SANDBOX_ISOLATED`） | +| 审批状态机（waiting_approval + approvals 回执） | ✅ 已实现（逐项 TTL/范围校验待加强） | | 短期凭证派生注入、Workload Identity | 🟡 AM/K8s 侧 | -| Tool/MCP 权限引擎 | 🔴 未实现 | -| allowed_paths 运行时强制、强隔离 | 🔴 未实现 | -| 代码测试沙箱（Pod 内纵深防御 + 环境清洗 + 超时/限额；**fail-closed 双门控** `ENABLE_QUALITY_EVAL` + `HEICODE_SANDBOX_ISOLATED`，启动/运行时硬失败） | ✅ 已实现（本仓，§8.1）；OS 级隔离仍依赖 Pod | -| 强化执行沙箱（seccomp/只读根/网络策略/能力裁剪，Pod 层） | 🔴 未实现（Infra/Security） | -| 租户隔离 | ⛔ 不在本仓（归因按 user/channelId） | +| Tool/MCP 权限引擎 | ⏸ 本次不做 —— **无 SK/MCP 工具层可治理**；待工具层落地后随其设计权限边界 | +| allowed_paths 运行时强制（按 grant 限定）、强隔离 | ⏸ 本次不做 —— 当前仅 workspace 根越界强制；按 grant 的 allowed_paths 待资源授权链接入 | +| 强化执行沙箱（seccomp/只读根/网络策略/能力裁剪，Pod 层） | ⏸ 本次不做 —— 归 Infra/Security（Pod manifest） | +| 租户隔离 | ⛔ **有意**不在本仓（归因按 `user/channelId`，不引入 tenant 概念） | + +> ⏸ = 确认的后续/外部层，非本仓本次范围（按规则 #9 不伪造「已强制」）。本仓**已强制**的三类边界均可由 `scripts/test-security-boundary.py` 验证。 ## 10. 待对齐对象 diff --git a/scripts/test-security-boundary.py b/scripts/test-security-boundary.py new file mode 100644 index 0000000..e8b2738 --- /dev/null +++ b/scripts/test-security-boundary.py @@ -0,0 +1,141 @@ +"""Security-boundary enforcement tests (issue #19). + +Consolidates the *enforced, in-repo* security boundaries into one verifiable suite, so the +contract in docs/integration/security-boundary.md is testable rather than aspirational: + + 1. Secret: `azkv://` strict on refs; plaintext secret-like fields rejected at intake. + 2. Redaction: plaintext creds -> [redacted] on the callback-safe path; `secret_ref` (azkv ref) + passes through (HM re-redacts for the client). + 3. Workspace: file writes are confined to the task workspace — absolute paths and `..` escapes + are rejected. + 4. Sandbox fail-closed: code execution refuses to run unless isolation is explicitly confirmed. + +Out of scope (documented confirmed-limitations in security-boundary.md, NOT tested here): tool/MCP +permission engine (no tool/MCP layer exists), tenant isolation (by design — user/channelId), and +Pod-level hardening (Infra: seccomp/read-only-root/NetworkPolicy). + +Run from agent_swarm_v6 (install deps first): + pip install -r orchestrator/requirements.txt -r agent/requirements.txt + REDIS_FAKE=1 python scripts/test-security-boundary.py +""" +import os +import sys +import tempfile +from pathlib import Path + +os.environ["REDIS_FAKE"] = "1" +# Ensure the sandbox isolation assertion starts from the *unconfirmed* state for the fail-closed test. +os.environ.pop("HEICODE_SANDBOX_ISOLATED", None) +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from orchestrator.swarm_runtime import swarm_runtime, RuntimeValidationError +from orchestrator import sandbox as sandbox_mod +from agent.task_executor import TaskExecutor + +failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +def raises(fn, exc=RuntimeValidationError): + try: + fn() + return False + except exc: + return True + except Exception: + return False + + +def valid_body(**overrides): + body = { + "mode": "swarm", + "orchestration_plan": {"objective": "do a thing", "agents": []}, + "callback": {"url": "http://manager.local/cb", "subscribed_events": []}, + "metadata": {"manager_deployment_id": "dep_1"}, + } + body.update(overrides) + return body + + +def test_secret_boundary(): + # billing_context.secret_ref must be azkv:// + check("intake rejects non-azkv billing secret_ref", + raises(lambda: swarm_runtime.validate_create_request( + valid_body(billing_context={"secret_ref": "plain-token-123"})))) + check("intake accepts azkv billing secret_ref", + not raises(lambda: swarm_runtime.validate_create_request( + valid_body(billing_context={"secret_ref": "azkv://kv/secrets/billing"})))) + + # plaintext secret-like fields rejected anywhere in the tree; refs must be azkv + check("plaintext password field rejected", + raises(lambda: swarm_runtime._reject_plaintext_secrets({"password": "hunter2"}, "body"))) + check("plaintext access_token field rejected", + raises(lambda: swarm_runtime._reject_plaintext_secrets({"creds": {"access_token": "x"}}, "body"))) + check("non-azkv *_ref rejected", + raises(lambda: swarm_runtime._reject_plaintext_secrets({"secret_ref": "vault://old/x"}, "body"))) + check("azkv secret_ref accepted", + not raises(lambda: swarm_runtime._reject_plaintext_secrets({"secret_ref": "azkv://kv/secrets/x"}, "body"))) + + +def test_redaction_boundary(): + red = swarm_runtime._redact_sensitive({ + "access_token": "PLAINTEXT", + "api_key": "PLAINTEXT", + "nested": {"password": "PLAINTEXT"}, + "secret_ref": "azkv://kv/secrets/x", + "note": "fine", + }) + check("plaintext token redacted", red["access_token"] == "[redacted]") + check("plaintext api_key redacted", red["api_key"] == "[redacted]") + check("nested plaintext password redacted", red["nested"]["password"] == "[redacted]") + check("azkv secret_ref preserved (HM re-redacts for client)", red["secret_ref"] == "azkv://kv/secrets/x") + check("non-sensitive field untouched", red["note"] == "fine") + + +def test_workspace_boundary(): + # TaskExecutor's constructor requires a model key; the path resolver never calls a model. + os.environ.setdefault("OPENAI_API_KEY", "test-dummy-key") + with tempfile.TemporaryDirectory() as ws: + ex = TaskExecutor("agent-x", ws) + # in-bounds path resolves under the workspace root + ok = ex._resolve_workspace_path("src/app.py") + check("in-workspace relative path resolves under root", + str(ok).startswith(str(Path(ws).resolve()))) + check("absolute path rejected", raises(lambda: ex._resolve_workspace_path("/etc/passwd"), ValueError)) + check("parent-escape (..) rejected", raises(lambda: ex._resolve_workspace_path("../../etc/passwd"), ValueError)) + check("empty path rejected", raises(lambda: ex._resolve_workspace_path(""), ValueError)) + + +def test_sandbox_fail_closed(): + os.environ.pop("HEICODE_SANDBOX_ISOLATED", None) + check("isolation not confirmed by default", sandbox_mod.isolation_confirmed() is False) + check("assert_isolated raises without confirmation", + raises(sandbox_mod.assert_isolated, sandbox_mod.SandboxIsolationError)) + os.environ["HEICODE_SANDBOX_ISOLATED"] = "1" + try: + check("isolation confirmed when explicitly set", sandbox_mod.isolation_confirmed() is True) + check("assert_isolated passes once confirmed", + not raises(sandbox_mod.assert_isolated, sandbox_mod.SandboxIsolationError)) + finally: + os.environ.pop("HEICODE_SANDBOX_ISOLATED", None) + + +def main(): + test_secret_boundary() + test_redaction_boundary() + test_workspace_boundary() + test_sandbox_fail_closed() + print() + if failures: + print(f"{len(failures)} security-boundary check(s) FAILED: {failures}") + sys.exit(1) + print("all security-boundary checks passed") + + +if __name__ == "__main__": + main()