Spaces:
Running
Running
| """Adversarial tests for the authoritative runtime dispatch boundary.""" | |
| from __future__ import annotations | |
| import asyncio | |
| from pathlib import Path | |
| import pytest | |
| from trace2mcp.codegen import generate_project, generate_server_py | |
| from trace2mcp.models import PolicyViolationError, Trace, TraceEvent, Workflow, WorkflowNode | |
| from trace2mcp.policy import ToolContract | |
| from trace2mcp.runtime import ( | |
| SchemaValidationError, | |
| ToolNotFoundError, | |
| ToolRegistry, | |
| WorkflowExecutionError, | |
| execute_workflow, | |
| invoke_capability, | |
| ) | |
| def reviewed(name: str, side_effect: str = "read", **kwargs: object) -> ToolContract: | |
| return ToolContract(name, side_effect=side_effect, reviewed=True, **kwargs) | |
| def strict_registry( | |
| handlers: dict[str, object], contracts: dict[str, ToolContract] | |
| ) -> ToolRegistry: | |
| return ToolRegistry( | |
| handlers, # type: ignore[arg-type] | |
| contracts=contracts, | |
| require_contracts=True, | |
| require_reviewed_contracts=True, | |
| ) | |
| def test_direct_destructive_operation_is_denied_before_handler() -> None: | |
| calls = 0 | |
| def destroy() -> None: | |
| nonlocal calls | |
| calls += 1 | |
| workflow = Workflow("deny-delete", {}, [WorkflowNode("delete", "files.delete", {})]) | |
| registry = strict_registry( | |
| {"files.delete": destroy}, | |
| {"files.delete": reviewed("files.delete", "delete")}, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="denied"): | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert calls == 0 | |
| def test_reviewed_contract_defeats_trace_metadata_confusion() -> None: | |
| calls = 0 | |
| def send() -> str: | |
| nonlocal calls | |
| calls += 1 | |
| return "sent" | |
| workflow = Workflow( | |
| "metadata-confusion", | |
| {}, | |
| [WorkflowNode("send", "gmail.send", {}, side_effect="read")], | |
| ) | |
| registry = strict_registry( | |
| {"gmail.send": send}, | |
| {"gmail.send": reviewed("gmail.send", "external_side_effect")}, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="approval"): | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert calls == 0 | |
| result = asyncio.run(execute_workflow(workflow, registry, approvals={"send"})) | |
| assert result.output == "sent" | |
| assert calls == 1 | |
| def test_alias_and_registry_proxy_cannot_bypass_policy() -> None: | |
| dangerous_calls = 0 | |
| def dangerous() -> None: | |
| nonlocal dangerous_calls | |
| dangerous_calls += 1 | |
| alias_workflow = Workflow( | |
| "alias", | |
| {}, | |
| [WorkflowNode("alias", "safe-looking-alias", {}, side_effect="read")], | |
| ) | |
| alias_registry = strict_registry( | |
| {"safe-looking-alias": dangerous}, | |
| {"safe-looking-alias": reviewed("safe-looking-alias", "delete")}, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="denied"): | |
| asyncio.run(execute_workflow(alias_workflow, alias_registry)) | |
| assert dangerous_calls == 0 | |
| registry = ToolRegistry( | |
| {"dangerous.delete": dangerous}, | |
| contracts={"dangerous.delete": reviewed("dangerous.delete", "delete")}, | |
| ) | |
| mediated = registry.get("dangerous.delete") | |
| with pytest.raises(PolicyViolationError, match="active Trace2MCP runtime"): | |
| asyncio.run(mediated()) | |
| assert dangerous_calls == 0 | |
| def test_unknown_and_unreviewed_capabilities_fail_closed() -> None: | |
| unknown = Workflow("unknown", {}, [WorkflowNode("missing", "missing.tool", {})]) | |
| with pytest.raises(ToolNotFoundError): | |
| asyncio.run(execute_workflow(unknown, ToolRegistry())) | |
| calls = 0 | |
| def unreviewed() -> None: | |
| nonlocal calls | |
| calls += 1 | |
| workflow = Workflow("unreviewed", {}, [WorkflowNode("tool", "tool", {})]) | |
| registry = ToolRegistry( | |
| {"tool": unreviewed}, | |
| contracts={"tool": ToolContract("tool", side_effect="read", reviewed=False)}, | |
| require_contracts=True, | |
| require_reviewed_contracts=True, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="reviewed contract"): | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert calls == 0 | |
| def test_approval_is_necessary_but_does_not_override_other_denials() -> None: | |
| calls = 0 | |
| def send() -> str: | |
| nonlocal calls | |
| calls += 1 | |
| return "sent" | |
| workflow = Workflow("approval", {}, [WorkflowNode("send", "send", {})]) | |
| registry = strict_registry( | |
| {"send": send}, | |
| {"send": reviewed("send", "external_side_effect")}, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="approval"): | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert calls == 0 | |
| assert asyncio.run(execute_workflow(workflow, registry, approvals={"send"})).output == "sent" | |
| denied_registry = strict_registry( | |
| {"send": send}, | |
| {"send": reviewed("send", "delete")}, | |
| ) | |
| with pytest.raises(PolicyViolationError, match="denied"): | |
| asyncio.run(execute_workflow(workflow, denied_registry, approvals={"send"})) | |
| assert calls == 1 | |
| def test_nested_capability_escalation_returns_to_same_gate() -> None: | |
| dangerous_calls = 0 | |
| safe_calls = 0 | |
| registry: ToolRegistry | |
| async def parent() -> str: | |
| dynamic_name = ".".join(("dangerous", "delete")) | |
| return await registry.invoke(dynamic_name) | |
| def dangerous() -> str: | |
| nonlocal dangerous_calls | |
| dangerous_calls += 1 | |
| return "destroyed" | |
| def safe_child(value: int) -> int: | |
| nonlocal safe_calls | |
| safe_calls += 1 | |
| return value + 1 | |
| registry = strict_registry( | |
| { | |
| "allowed.read": parent, | |
| "dangerous.delete": dangerous, | |
| "safe.child": safe_child, | |
| }, | |
| { | |
| "allowed.read": reviewed("allowed.read"), | |
| "dangerous.delete": reviewed("dangerous.delete", "delete"), | |
| "safe.child": reviewed("safe.child"), | |
| }, | |
| ) | |
| workflow = Workflow("nested-deny", {}, [WorkflowNode("parent", "allowed.read", {})]) | |
| with pytest.raises(WorkflowExecutionError) as captured: | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert isinstance(captured.value.cause, PolicyViolationError) | |
| assert dangerous_calls == 0 | |
| async def safe_parent() -> int: | |
| return await invoke_capability("safe.child", value=4) | |
| registry.register("allowed.read", safe_parent) | |
| assert asyncio.run(execute_workflow(workflow, registry)).output == 5 | |
| assert safe_calls == 1 | |
| def test_generated_server_has_no_direct_handler_route(tmp_path: Path) -> None: | |
| workflow = Workflow("generated", {}, [WorkflowNode("read", "records.read", {})]) | |
| server = generate_server_py(workflow) | |
| assert "result = await execute_workflow(" in server | |
| assert "WORKFLOW," in server | |
| assert "REGISTRY," in server | |
| assert "runtime_guard=RUNTIME_GUARD" in server | |
| assert "HANDLERS[" not in server | |
| assert "require_contracts=True" in server | |
| assert "require_reviewed_contracts=True" in server | |
| generated = generate_project(workflow, tmp_path) | |
| assert "contracts" in generated | |
| contracts = (tmp_path / "contracts.py").read_text(encoding="utf-8") | |
| assert "reviewed=False" in contracts | |
| assert "Set reviewed=True only after checking" in contracts | |
| def test_registry_swap_between_preflight_and_invoke_fails_closed() -> None: | |
| old_calls = 0 | |
| new_calls = 0 | |
| registry: ToolRegistry | |
| def old_target() -> str: | |
| nonlocal old_calls | |
| old_calls += 1 | |
| return "old" | |
| def new_target() -> str: | |
| nonlocal new_calls | |
| new_calls += 1 | |
| return "new" | |
| def mutate_registry() -> str: | |
| registry.register("target", new_target) | |
| return "mutated" | |
| workflow = Workflow( | |
| "toctou", | |
| {}, | |
| [ | |
| WorkflowNode("mutate", "mutate", {}), | |
| WorkflowNode("target", "target", {}, depends_on=("mutate",)), | |
| ], | |
| ) | |
| registry = strict_registry( | |
| {"mutate": mutate_registry, "target": old_target}, | |
| {"mutate": reviewed("mutate"), "target": reviewed("target")}, | |
| ) | |
| with pytest.raises(WorkflowExecutionError) as captured: | |
| asyncio.run(execute_workflow(workflow, registry)) | |
| assert isinstance(captured.value.cause, PolicyViolationError) | |
| assert "changed after authorization" in str(captured.value.cause) | |
| assert old_calls == 0 | |
| assert new_calls == 0 | |
| def test_contract_schema_is_enforced_before_and_after_handler() -> None: | |
| calls = 0 | |
| def handler(value: object) -> object: | |
| nonlocal calls | |
| calls += 1 | |
| return value | |
| contract = reviewed( | |
| "typed", | |
| input_schema={ | |
| "type": "object", | |
| "properties": {"value": {"type": "integer"}}, | |
| "required": ["value"], | |
| "additionalProperties": False, | |
| }, | |
| output_schema={"type": "string"}, | |
| ) | |
| registry = strict_registry({"typed": handler}, {"typed": contract}) | |
| invalid_input = Workflow( | |
| "invalid-input", {}, [WorkflowNode("typed", "typed", {"value": "not-int"})] | |
| ) | |
| with pytest.raises(WorkflowExecutionError) as captured: | |
| asyncio.run(execute_workflow(invalid_input, registry)) | |
| assert isinstance(captured.value.cause, SchemaValidationError) | |
| assert calls == 0 | |
| invalid_output = Workflow("invalid-output", {}, [WorkflowNode("typed", "typed", {"value": 1})]) | |
| with pytest.raises(WorkflowExecutionError) as captured: | |
| asyncio.run(execute_workflow(invalid_output, registry)) | |
| assert isinstance(captured.value.cause, SchemaValidationError) | |
| assert calls == 1 | |
| def test_handler_exception_stops_dependents_and_approval_does_not_leak() -> None: | |
| send_calls = 0 | |
| def fail() -> None: | |
| raise RuntimeError("boom") | |
| def send() -> None: | |
| nonlocal send_calls | |
| send_calls += 1 | |
| workflow = Workflow( | |
| "exception", | |
| {}, | |
| [ | |
| WorkflowNode("fail", "fail", {}), | |
| WorkflowNode("send", "send", {}, depends_on=("fail",)), | |
| ], | |
| ) | |
| registry = strict_registry( | |
| {"fail": fail, "send": send}, | |
| {"fail": reviewed("fail"), "send": reviewed("send", "external_side_effect")}, | |
| ) | |
| with pytest.raises(WorkflowExecutionError): | |
| asyncio.run(execute_workflow(workflow, registry, approvals={"send"})) | |
| assert send_calls == 0 | |
| send_only = Workflow("send-only", {}, [WorkflowNode("send", "send", {})]) | |
| with pytest.raises(PolicyViolationError, match="approval"): | |
| asyncio.run(execute_workflow(send_only, registry)) | |
| assert send_calls == 0 | |
| def test_trace_compiled_server_still_uses_authoritative_runtime() -> None: | |
| workflow = Workflow( | |
| "compiled", | |
| {}, | |
| [WorkflowNode("read", "read", {}, side_effect="read")], | |
| ) | |
| source = generate_server_py(workflow) | |
| assert source.count("execute_workflow(") == 1 | |
| assert "result = await execute_workflow" in source | |
| trace = Trace( | |
| inputs={}, | |
| events=[TraceEvent("read", "read", {}, "ok", side_effect="read")], | |
| ) | |
| assert trace.events[0].tool == "read" | |