Agentswarm/scripts/test-security-boundary.py

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