diff --git a/python/samples/02-agents/middleware/README.md b/python/samples/02-agents/middleware/README.md index 205dd01504a..58cb3130c29 100644 --- a/python/samples/02-agents/middleware/README.md +++ b/python/samples/02-agents/middleware/README.md @@ -17,6 +17,7 @@ This folder contains focused middleware samples for `Agent`, chat clients, tools | [`decorator_middleware.py`](./decorator_middleware.py) | Demonstrates middleware registration with decorators. | | [`exception_handling_with_middleware.py`](./exception_handling_with_middleware.py) | Shows how middleware can handle failures and recover cleanly. | | [`function_based_middleware.py`](./function_based_middleware.py) | Shows function-based agent and function middleware. | +| [`hol_guard_middleware.py`](./hol_guard_middleware.py) | Demonstrates a `FunctionMiddleware` that gates tool calls behind a [HOL Guard](https://github.com/hashgraph-online/hol-guard) verdict, running the real engine locally via the `hol-guard` CLI (`pipx install hol-guard`) and raising `MiddlewareFailure` on deny, review, or Guard-unavailable so the tool never executes. | | [`middleware_termination.py`](./middleware_termination.py) | Demonstrates stopping a middleware pipeline early. | | [`message_injection_middleware.py`](./message_injection_middleware.py) | Demonstrates `MessageInjectionMiddleware` with a real Foundry chat client: enqueueing a follow-up message into the active session while a long-running async tool is awaiting. | | [`override_result_with_middleware.py`](./override_result_with_middleware.py) | Shows how middleware can replace regular and streaming results, then post-process the final response. | @@ -50,3 +51,19 @@ agent's latest response to a second, external judge chat client on every iterati or malicious judge endpoint could exfiltrate that data, or return a manipulated verdict/gap analysis that gets fed back into the loop as feedback — a form of indirect prompt injection. Only configure a judge client that points at a service you trust as much as the primary model. + + + + +## Running the HOL Guard sample + +Install the real engine for full protection: + +```bash +pipx install hol-guard +hol-guard init +``` + +Without it installed, the sample still runs using its built-in offline fallback +(`offline_fallback=True`), but that fallback is a small demo deny-list only — +it is **not** a substitute for the real HOL Guard engine. \ No newline at end of file diff --git a/python/samples/02-agents/middleware/hol_guard_middleware.py b/python/samples/02-agents/middleware/hol_guard_middleware.py new file mode 100644 index 00000000000..805bd8a09b3 --- /dev/null +++ b/python/samples/02-agents/middleware/hol_guard_middleware.py @@ -0,0 +1,285 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "agent-framework-foundry", +# "azure-identity", +# "python-dotenv", +# ] +# /// +# Run with any PEP 723 compatible runner, e.g.: +# uv run samples/02-agents/middleware/hol_guard_middleware.py + +# Copyright (c) Microsoft. All rights reserved. +"""Official HOL Guard FunctionMiddleware example for protected tool calls (issue #7833). + +HOLGuardMiddleware evaluates the tool name and validated arguments in +FunctionInvocationContext before call_next() and only proceeds when HOL Guard's pattern check +finds no known attack pattern. It does NOT proceed on a match, on Guard's own review-required +signal, or when Guard is unavailable/errors. + +Unlike ATRValidationMiddleware in this folder, which raises MiddlewareTermination on a match, +this sample raises MiddlewareFailure on deny, review-required, AND Guard-unavailable/error. +MiddlewareFailure is the framework's explicit fail-closed escape: it cancels the in-flight +tool-call batch and propagates to the caller of Agent.run() rather than letting the loop +continue -- matching the issue's "terminate before the wrapped tool executes" requirement for +all three non-allow outcomes, not just an outright deny. + +DESIGN NOTE, per clarification from a HOL Guard maintainer on #7833: `hol-guard command test + --json` is a PREVIEW/INSPECTION-ONLY pattern check. Its JSON response reports `status` +("no_match" or "review") and explicitly marks `policy_evaluation: "not_run"`. A "no_match" from +this command is NOT a HOL Guard policy approval -- it only means no known attack pattern was +found; this sample treats it as "let the call proceed" for demonstration purposes but never +describes it as "approved" or "policy allow". Two stronger alternatives exist and were +deliberately not used here to keep this sample small: + - `hol-guard policy evaluate-command --command "" --json` (release/3.0) + evaluates an explicit policy document you supply -- see + https://github.com/hashgraph-online/hol-guard/blob/release/3.0/src/codex_plugin_scanner/guard/cli/commands_parser_policy.py + This is closer to real policy enforcement, but requires shipping a sample policy.yml as + input and is not itself the active harness/runtime policy either. + - The authoritative enforcement path is HOL Guard's supported runtime/harness integration, + where Guard mediates the actual command/tool execution under the live runtime + policy/approval context. That integration is out of scope for a CLI-wrapping + FunctionMiddleware sample like this one. +A contributor extending this sample toward real enforcement should start from +`policy evaluate-command` with an explicit policy file, not from `command test`. + +hol-guard is intentionally NOT listed as an importable Python dependency here; it is only +shelled out to. Install it with `pipx install hol-guard` for the real engine. Without it on +PATH, or if the call errors, the middleware fails closed by default. Pass offline_fallback=True +only for local demo/dev use without the real engine installed; it applies a small built-in +deny-list far weaker than actual HOL Guard, and is deliberately NOT the default in main() below +-- shipping a fail-closed example that silently downgrades to a weaker check by default would +misrepresent what it demonstrates. + +Provider imports (FoundryChatClient, AzureCliCredential) are deferred to main() rather than +imported at module level, so HOLGuardMiddleware and evaluate_with_hol_guard stay importable +and unit-testable without the agent-framework-foundry/azure-identity extras present. +""" + +import asyncio +import json +import logging +import shutil +from collections.abc import Awaitable, Callable, Mapping +from enum import Enum +from random import randint +from typing import Annotated, Any + +from agent_framework import ( + Agent, + FunctionInvocationContext, + FunctionMiddleware, + MiddlewareFailure, + tool, +) +from pydantic import BaseModel, Field + +logger = logging.getLogger(__name__) + +_CLI_NAME = "hol-guard" +_OFFLINE_DENY_SUBSTRINGS = ("drop table", "rm -rf", "delete_production", "sudo ", "curl|sh") + + +class GuardDecision(str, Enum): + """Outcome of evaluating a tool call against HOL Guard.""" + + ALLOW = "allow" + DENY = "deny" + REVIEW = "review" + UNAVAILABLE = "unavailable" + ERROR = "error" + + +def _call_to_command_string(function_name: str, arguments: BaseModel | Mapping[str, Any]) -> str: + """Render a tool call as a single command-shaped string for `hol-guard command test`.""" + values = arguments.model_dump() if isinstance(arguments, BaseModel) else dict(arguments) + args_text = " ".join(f"{key}={value!r}" for key, value in values.items()) + return f"{function_name} {args_text}".strip() + + +def _offline_check(command_string: str) -> tuple[GuardDecision, str]: + lowered = command_string.lower() + for pattern in _OFFLINE_DENY_SUBSTRINGS: + if pattern in lowered: + return GuardDecision.DENY, f"offline fallback matched '{pattern}'" + return GuardDecision.ALLOW, "offline fallback: no match (reduced protection, real engine not installed)" + + +async def evaluate_with_hol_guard( + function_name: str, + arguments: BaseModel | Mapping[str, Any], + *, + offline_fallback: bool = False, + timeout_seconds: float = 5.0, +) -> tuple[GuardDecision, str]: + """Classify a tool call with the real, local hol-guard engine. + + Fails closed by default: a missing CLI, a timeout, or any evaluation error returns + UNAVAILABLE/ERROR, never ALLOW, unless offline_fallback=True. + """ + cli_path = shutil.which(_CLI_NAME) + command_string = _call_to_command_string(function_name, arguments) + + if cli_path is None: + if offline_fallback: + return _offline_check(command_string) + return GuardDecision.UNAVAILABLE, f"{_CLI_NAME} CLI not found on PATH" + + proc = await asyncio.create_subprocess_exec( + cli_path, + "command", + "test", + command_string, + "--json", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout_seconds) + except TimeoutError: + try: + proc.kill() + except ProcessLookupError: + pass + await proc.communicate() + raise + + if not stdout.strip(): + raise RuntimeError(stderr.decode(errors="replace").strip() or f"exit code {proc.returncode}, no output") + + payload = json.loads(stdout) + # `command test` is a preview-only pattern check (policy_evaluation is always "not_run" + # in its response): status="no_match" means no known attack pattern matched, and + # status="review" means it did. There is no third "definitely safe" state to trust + # here, so anything other than a clean no_match fails closed. + status = payload.get("status") + if status == "no_match": + return ( + GuardDecision.ALLOW, + "hol-guard: no known attack pattern found (pattern-preview check; not a policy decision)", + ) + if status == "review": + return GuardDecision.REVIEW, payload.get("summary", "hol-guard flagged this call for review") + return GuardDecision.ERROR, f"hol-guard returned an unrecognized response: {payload!r}" + except Exception as exc: + if offline_fallback: + return _offline_check(command_string) + return GuardDecision.ERROR, f"{_CLI_NAME} evaluation failed: {exc}" + + +class HOLGuardMiddleware(FunctionMiddleware): + """Gates tool calls behind a HOL Guard verdict; fails closed on anything but allow.""" + + def __init__(self, *, offline_fallback: bool = False, timeout_seconds: float = 5.0) -> None: + """Create the middleware. + + Args: + offline_fallback: When True, fall back to a small built-in deny-list if the + hol-guard CLI is unavailable or errors, instead of failing closed. Demo/dev + use only -- much weaker than the real engine. + timeout_seconds: Timeout for the local hol-guard subprocess call. + """ + self._offline_fallback = offline_fallback + self._timeout_seconds = timeout_seconds + if shutil.which(_CLI_NAME) is None: + logger.warning( + "%s CLI not found on PATH. Install with `pipx install hol-guard` for real " + "protection, or pass offline_fallback=True for local demo/dev use.", + _CLI_NAME, + ) + + async def process( + self, + context: FunctionInvocationContext, + call_next: Callable[[], Awaitable[None]], + ) -> None: + """Evaluate the tool call with HOL Guard and only proceed on an allow verdict.""" + decision, reason = await evaluate_with_hol_guard( + context.function.name, + context.arguments, + offline_fallback=self._offline_fallback, + timeout_seconds=self._timeout_seconds, + ) + if decision is GuardDecision.ALLOW: + logger.info( + "[HOLGuardMiddleware] Tool '%s': no known attack pattern found by HOL Guard; proceeding.", + context.function.name, + ) + await call_next() + return + + logger.warning( + "[HOLGuardMiddleware] Blocked tool '%s': verdict=%s reason=%s", + context.function.name, + decision.value, + reason, + ) + # Fail closed: MiddlewareFailure, not MiddlewareTermination -- see module docstring. + raise MiddlewareFailure(f"HOL Guard {decision.value} for tool '{context.function.name}': {reason}") + + +@tool(approval_mode="never_require") +def get_weather( + location: Annotated[str, Field(description="The location to get the weather for.")], +) -> str: + """Get the weather for a given location.""" + conditions = ["sunny", "cloudy", "rainy", "stormy"] + return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C." + + +@tool(approval_mode="never_require") +def delete_production_database( + confirm: Annotated[bool, Field(description="Confirm the deletion.")], +) -> str: + """Deletes the production database. Obviously dangerous -- used to demo a HOL Guard deny.""" + return "Database deleted." # pragma: no cover - should never actually run + + +async def main() -> None: + """Run the benign and dangerous demo requests against an agent guarded by HOL Guard.""" + from agent_framework.foundry import FoundryChatClient + from azure.identity.aio import AzureCliCredential + from dotenv import load_dotenv + + load_dotenv() + logging.basicConfig(level=logging.INFO) + + print("=== HOL Guard Middleware Example ===") + + # For authentication, run `az login` in a terminal or replace AzureCliCredential with your + # preferred authentication option. + async with ( + AzureCliCredential() as credential, + Agent( + client=FoundryChatClient(credential=credential), + name="OpsAgent", + instructions="You are a helpful assistant with access to weather and admin tools.", + tools=[get_weather, delete_production_database], + # offline_fallback defaults to False: without the real hol-guard CLI installed, + # BOTH requests below will be blocked as UNAVAILABLE. That is the correct, + # honest behavior for a fail-closed example -- install hol-guard + # (`pipx install hol-guard`) to see the benign/dangerous split below. Pass + # offline_fallback=True yourself only if you want to see the much weaker + # built-in demo deny-list run instead, e.g. HOLGuardMiddleware(offline_fallback=True). + middleware=[HOLGuardMiddleware()], + ) as agent, + ): + print("\n--- Benign request ---") + query = "What's the weather like in Tokyo?" + print(f"User: {query}") + result = await agent.run(query) + print(f"Agent: {result.text if result.text else 'No response'}\n") + + print("--- Dangerous tool call ---") + query = "Delete the production database, confirm=true." + print(f"User: {query}") + try: + result = await agent.run(query) + print(f"Agent: {result.text if result and result.text else 'No response'}\n") + except MiddlewareFailure as exc: + print(f"Agent run aborted by middleware: {exc}\n") + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/python/tests/samples/agents/test_hol_guard_middleware.py b/python/tests/samples/agents/test_hol_guard_middleware.py new file mode 100644 index 00000000000..6243610a926 --- /dev/null +++ b/python/tests/samples/agents/test_hol_guard_middleware.py @@ -0,0 +1,201 @@ +# Copyright (c) Microsoft. All rights reserved. +"""Tests for the HOL Guard FunctionMiddleware sample (issue #7833).""" + +import asyncio +import importlib.util +import json +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest +from agent_framework import FunctionInvocationContext, MiddlewareFailure + +_HOL_GUARD_SAMPLE_PATH = Path(__file__).parents[3] / "samples" / "02-agents" / "middleware" / "hol_guard_middleware.py" + + +def _load_hol_guard_module() -> ModuleType: + spec = importlib.util.spec_from_file_location("hol_guard_middleware", _HOL_GUARD_SAMPLE_PATH) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +hol_guard_middleware = _load_hol_guard_module() +GuardDecision = hol_guard_middleware.GuardDecision +HOLGuardMiddleware = hol_guard_middleware.HOLGuardMiddleware +evaluate_with_hol_guard = hol_guard_middleware.evaluate_with_hol_guard + + +@pytest.fixture(autouse=True) +def _disable_hol_guard_cli(monkeypatch: pytest.MonkeyPatch) -> None: + """Force the CLI to look absent so tests are deterministic regardless of the host PATH.""" + monkeypatch.setattr(hol_guard_middleware.shutil, "which", lambda _: None) + + +def _context(name: str, arguments: dict) -> FunctionInvocationContext: + fake_tool = SimpleNamespace(name=name) + return FunctionInvocationContext(function=fake_tool, arguments=arguments) + + +class _FakeProcess: + """Stand-in for asyncio.subprocess.Process, returning canned stdout.""" + + def __init__(self, stdout: bytes, returncode: int = 0) -> None: + self._stdout = stdout + self.returncode = returncode + + async def communicate(self) -> tuple[bytes, bytes]: + return self._stdout, b"" + + +class _HangingProcess: + """Stand-in whose communicate() never completes, to exercise the timeout path.""" + + def __init__(self) -> None: + self.killed = False + + async def communicate(self) -> tuple[bytes, bytes]: + if self.killed: + return b"", b"" + await asyncio.sleep(10) + return b"", b"" # pragma: no cover - unreachable within the test's short timeout + + def kill(self) -> None: + self.killed = True + + +# --- CLI unavailable / offline fallback (free function, no middleware needed) --- + + +async def test_evaluate_fails_closed_when_cli_unavailable() -> None: + """An absent CLI with no offline fallback must return UNAVAILABLE, not allow the call.""" + decision, _ = await evaluate_with_hol_guard("delete_production_database", {"confirm": True}) + assert decision is GuardDecision.UNAVAILABLE + + +async def test_offline_fallback_denies_known_dangerous_call() -> None: + """The offline deny-list still blocks an obviously dangerous call name.""" + decision, _ = await evaluate_with_hol_guard("delete_production_database", {"confirm": True}, offline_fallback=True) + assert decision is GuardDecision.DENY + + +async def test_offline_fallback_allows_benign_call() -> None: + """The offline deny-list allows a call that matches no known-dangerous pattern.""" + decision, _ = await evaluate_with_hol_guard("get_weather", {"location": "Tokyo"}, offline_fallback=True) + assert decision is GuardDecision.ALLOW + + +# --- Real CLI JSON schema: status + policy_evaluation, not decision/reason --- + + +async def test_real_cli_no_match_maps_to_allow(monkeypatch: pytest.MonkeyPatch) -> None: + """A hol-guard status of 'no_match' is an allow.""" + monkeypatch.setattr(hol_guard_middleware.shutil, "which", lambda _: "/usr/local/bin/hol-guard") + + async def fake_exec(*args: object, **kwargs: object) -> _FakeProcess: + await asyncio.sleep(0) + payload = json.dumps({"status": "no_match", "policy_evaluation": "not_run"}).encode() + return _FakeProcess(payload) + + monkeypatch.setattr(hol_guard_middleware.asyncio, "create_subprocess_exec", fake_exec) + + decision, _ = await evaluate_with_hol_guard("get_weather", {"location": "Tokyo"}) + assert decision is GuardDecision.ALLOW + + +async def test_real_cli_review_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """A hol-guard status of 'review' fails closed rather than allowing the call through.""" + monkeypatch.setattr(hol_guard_middleware.shutil, "which", lambda _: "/usr/local/bin/hol-guard") + + async def fake_exec(*args: object, **kwargs: object) -> _FakeProcess: + await asyncio.sleep(0) + payload = json.dumps({"status": "review", "summary": "looks suspicious"}).encode() + return _FakeProcess(payload) + + monkeypatch.setattr(hol_guard_middleware.asyncio, "create_subprocess_exec", fake_exec) + + decision, reason = await evaluate_with_hol_guard("delete_production_database", {"confirm": True}) + assert decision is GuardDecision.REVIEW + assert "suspicious" in reason + + +async def test_real_cli_unrecognized_response_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """An unrecognized response shape fails closed instead of defaulting to allow.""" + monkeypatch.setattr(hol_guard_middleware.shutil, "which", lambda _: "/usr/local/bin/hol-guard") + + async def fake_exec(*args: object, **kwargs: object) -> _FakeProcess: + await asyncio.sleep(0) + return _FakeProcess(json.dumps({"unexpected": "shape"}).encode()) + + monkeypatch.setattr(hol_guard_middleware.asyncio, "create_subprocess_exec", fake_exec) + + decision, _ = await evaluate_with_hol_guard("get_weather", {"location": "Tokyo"}) + assert decision is GuardDecision.ERROR + + +async def test_timeout_kills_process_and_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + """A hung hol-guard process is killed and reaped, and the call fails closed.""" + monkeypatch.setattr(hol_guard_middleware.shutil, "which", lambda _: "/usr/local/bin/hol-guard") + fake_process = _HangingProcess() + + async def fake_exec(*args: object, **kwargs: object) -> _HangingProcess: + await asyncio.sleep(0) + return fake_process + + monkeypatch.setattr(hol_guard_middleware.asyncio, "create_subprocess_exec", fake_exec) + + decision, _ = await evaluate_with_hol_guard("get_weather", {"location": "Tokyo"}, timeout_seconds=0.05) + assert decision is GuardDecision.ERROR + assert fake_process.killed is True + + +# --- HOLGuardMiddleware wiring: allow => tool runs once, deny/unavailable => zero times --- + + +async def test_allow_runs_tool_once() -> None: + """An allow verdict lets the wrapped tool execute exactly once.""" + middleware = HOLGuardMiddleware(offline_fallback=True) + context = _context("get_weather", {"location": "Tokyo"}) + calls = 0 + + async def call_next() -> None: + await asyncio.sleep(0) + nonlocal calls + calls += 1 + + await middleware.process(context, call_next) + assert calls == 1 + + +async def test_deny_runs_tool_zero_times() -> None: + """A deny verdict raises MiddlewareFailure and never runs the wrapped tool.""" + middleware = HOLGuardMiddleware(offline_fallback=True) + context = _context("delete_production_database", {"confirm": True}) + calls = 0 + + async def call_next() -> None: + await asyncio.sleep(0) + nonlocal calls + calls += 1 + + with pytest.raises(MiddlewareFailure): + await middleware.process(context, call_next) + assert calls == 0 + + +async def test_unavailable_fails_closed_by_default() -> None: + """An unavailable Guard with no offline fallback also fails closed.""" + middleware = HOLGuardMiddleware(offline_fallback=False) + context = _context("get_weather", {"location": "Tokyo"}) + calls = 0 + + async def call_next() -> None: + await asyncio.sleep(0) + nonlocal calls + calls += 1 + + with pytest.raises(MiddlewareFailure): + await middleware.process(context, call_next) + assert calls == 0 \ No newline at end of file