Merge pull request #31 from xmindlab-heicode/feat/security-boundary-freeze

安全边界强制点：可验证边界 FROZEN v1 + 契约测试（Refs #19）

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()