From 6b961f267a513354a79d1a7e0c1f5c2fd715fe43 Mon Sep 17 00:00:00 2001 From: Samsul Date: Mon, 24 Aug 2026 05:37:26 +0000 Subject: [PATCH 1/3] Add HOL Guard middleware sample --- .../middleware/hol_guard_middleware.py | 231 ++++++++++++++++++ .../agents/test_hol_guard_middleware.py | 98 ++++++++ 2 files changed, 329 insertions(+) create mode 100644 python/samples/02-agents/middleware/hol_guard_middleware.py create mode 100644 python/tests/samples/agents/test_hol_guard_middleware.py 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..19c19d22bd3 --- /dev/null +++ b/python/samples/02-agents/middleware/hol_guard_middleware.py @@ -0,0 +1,231 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "agent-framework-foundry", +# ] +# /// +# 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. + +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 + +""" +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 on an explicit allow verdict +from HOL Guard (https://github.com/hashgraph-online/hol-guard). + +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 (see its docstring in +agent_framework's middleware module): 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. + +The real HOL Guard engine is invoked locally through its documented, side-effect-free CLI +contract (`hol-guard command test --json`; see "Inspect command protection without +running it" in https://github.com/hashgraph-online/hol-guard). hol-guard is intentionally NOT +listed in this file's PEP 723 dependencies -- it is not imported as a Python package here, +only shelled out to. Install it separately with `pipx install hol-guard` for real protection. +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 that is far weaker than actual HOL Guard. + +OPEN QUESTION (resolve before merging): hol-guard's `command test` contract is documented for +shell-command-shaped strings, not typed (function_name, arguments) tool calls. This sample +renders a call as a single command-like string as a best-effort mapping onto that contract. +Confirm the intended integration surface for arbitrary FunctionMiddleware calls with the HOL +Guard maintainers (see discussion on #7833) before treating this as final. + +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. +""" + +logger = logging.getLogger(__name__) + +_CLI_NAME = "hol-guard" +_OFFLINE_DENY_SUBSTRINGS = ("drop table", "rm -rf", "delete_production", "sudo ", "curl|sh") + + +class GuardDecision(str, Enum): + 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 or an 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" + + try: + proc = await asyncio.create_subprocess_exec( + cli_path, + "command", + "test", + command_string, + "--json", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout_seconds) + payload = json.loads(stdout) + decision = GuardDecision(payload.get("decision", "deny")) + return decision, payload.get("reason", payload.get("summary", "")) + except Exception as exc: # noqa: BLE001 - any failure here must fail closed + 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: + 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: + 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' allowed by HOL Guard.", 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: + 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=True lets this demo run without the real hol-guard engine + # installed. In production, omit it (default False) so an unavailable Guard fails + # closed instead of falling back to the much weaker built-in deny-list. + middleware=[HOLGuardMiddleware(offline_fallback=True)], + ) 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()) 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..92bfb8833fc --- /dev/null +++ b/python/tests/samples/agents/test_hol_guard_middleware.py @@ -0,0 +1,98 @@ +# Copyright (c) Microsoft. All rights reserved. + +"""Tests for the HOL Guard FunctionMiddleware sample (issue #7833).""" + +import importlib.util +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 + + +def _context(name: str, arguments: dict) -> FunctionInvocationContext: + fake_tool = SimpleNamespace(name=name) + return FunctionInvocationContext(function=fake_tool, arguments=arguments) + + +# --- evaluate_with_hol_guard: free function, no middleware needed --- + + +async def test_evaluate_fails_closed_when_cli_unavailable() -> None: + 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: + 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: + decision, _ = await evaluate_with_hol_guard("get_weather", {"location": "Tokyo"}, offline_fallback=True) + assert decision is GuardDecision.ALLOW + + +# --- HOLGuardMiddleware wiring: allow => tool runs once, deny/unavailable => zero times --- + + +async def test_allow_runs_tool_once() -> None: + middleware = HOLGuardMiddleware(offline_fallback=True) + context = _context("get_weather", {"location": "Tokyo"}) + calls = 0 + + async def call_next() -> None: + nonlocal calls + calls += 1 + + await middleware.process(context, call_next) + assert calls == 1 + + +async def test_deny_runs_tool_zero_times() -> None: + middleware = HOLGuardMiddleware(offline_fallback=True) + context = _context("delete_production_database", {"confirm": True}) + calls = 0 + + async def call_next() -> None: + 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: + middleware = HOLGuardMiddleware(offline_fallback=False) + context = _context("get_weather", {"location": "Tokyo"}) + calls = 0 + + async def call_next() -> None: + nonlocal calls + calls += 1 + + with pytest.raises(MiddlewareFailure): + await middleware.process(context, call_next) + assert calls == 0 From 48c3919bc22078e0b3464119b91814f0d3d8a298 Mon Sep 17 00:00:00 2001 From: Samsul Date: Mon, 24 Aug 2026 10:03:35 +0000 Subject: [PATCH 2/3] feat: add HOL Guard middleware sample --- python/samples/02-agents/middleware/README.md | 17 +++ .../middleware/hol_guard_middleware.py | 141 +++++++++++------- .../agents/test_hol_guard_middleware.py | 121 +++++++++++++-- 3 files changed, 218 insertions(+), 61 deletions(-) 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 index 19c19d22bd3..4aafbfa3451 100644 --- a/python/samples/02-agents/middleware/hol_guard_middleware.py +++ b/python/samples/02-agents/middleware/hol_guard_middleware.py @@ -2,6 +2,8 @@ # requires-python = ">=3.10" # dependencies = [ # "agent-framework-foundry", +# "azure-identity", +# "python-dotenv", # ] # /// # Run with any PEP 723 compatible runner, e.g.: @@ -9,6 +11,43 @@ # 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 on an explicit allow verdict +from HOL Guard (https://github.com/hashgraph-online/hol-guard). + +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. + +The real HOL Guard engine is invoked locally through its CLI (`hol-guard command test +--json`). That command is a PREVIEW-ONLY pattern check: its JSON response reports +`status` ("no_match" or "review") and explicitly marks `policy_evaluation: "not_run"`. So an +ALLOW from this sample means "no known attack pattern matched", not "HOL Guard's full org +policy approved this call" -- the two are different guarantees. 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. + +OPEN QUESTION (track on #7833): confirm with the HOL Guard maintainers whether a command exists +that runs full policy evaluation (not just the pattern-preview `command test`), and use that +instead once available. + +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 @@ -27,41 +66,6 @@ ) from pydantic import BaseModel, Field -""" -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 on an explicit allow verdict -from HOL Guard (https://github.com/hashgraph-online/hol-guard). - -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 (see its docstring in -agent_framework's middleware module): 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. - -The real HOL Guard engine is invoked locally through its documented, side-effect-free CLI -contract (`hol-guard command test --json`; see "Inspect command protection without -running it" in https://github.com/hashgraph-online/hol-guard). hol-guard is intentionally NOT -listed in this file's PEP 723 dependencies -- it is not imported as a Python package here, -only shelled out to. Install it separately with `pipx install hol-guard` for real protection. -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 that is far weaker than actual HOL Guard. - -OPEN QUESTION (resolve before merging): hol-guard's `command test` contract is documented for -shell-command-shaped strings, not typed (function_name, arguments) tool calls. This sample -renders a call as a single command-like string as a best-effort mapping onto that contract. -Confirm the intended integration surface for arbitrary FunctionMiddleware calls with the HOL -Guard maintainers (see discussion on #7833) before treating this as final. - -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. -""" - logger = logging.getLogger(__name__) _CLI_NAME = "hol-guard" @@ -69,6 +73,8 @@ class GuardDecision(str, Enum): + """Outcome of evaluating a tool call against HOL Guard.""" + ALLOW = "allow" DENY = "deny" REVIEW = "review" @@ -98,9 +104,10 @@ async def evaluate_with_hol_guard( 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 or an evaluation error returns UNAVAILABLE/ERROR, never ALLOW, unless - offline_fallback=True. + """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) @@ -110,21 +117,41 @@ async def evaluate_with_hol_guard( 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: - proc = await asyncio.create_subprocess_exec( - cli_path, - "command", - "test", - command_string, - "--json", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout_seconds) + 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) - decision = GuardDecision(payload.get("decision", "deny")) - return decision, payload.get("reason", payload.get("summary", "")) - except Exception as exc: # noqa: BLE001 - any failure here must fail closed + # `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 attack pattern matched (preview check only)" + 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}" @@ -134,6 +161,14 @@ 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: @@ -148,6 +183,7 @@ async def process( 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, @@ -187,6 +223,7 @@ def delete_production_database( 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 @@ -228,4 +265,4 @@ async def main() -> None: if __name__ == "__main__": - asyncio.run(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 index 92bfb8833fc..6243610a926 100644 --- a/python/tests/samples/agents/test_hol_guard_middleware.py +++ b/python/tests/samples/agents/test_hol_guard_middleware.py @@ -1,17 +1,16 @@ # 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" -) +_HOL_GUARD_SAMPLE_PATH = Path(__file__).parents[3] / "samples" / "02-agents" / "middleware" / "hol_guard_middleware.py" def _load_hol_guard_module() -> ModuleType: @@ -29,40 +28,140 @@ def _load_hol_guard_module() -> ModuleType: 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) -# --- evaluate_with_hol_guard: free function, no middleware needed --- +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: - decision, _ = await evaluate_with_hol_guard( - "delete_production_database", {"confirm": True}, offline_fallback=True - ) + """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 @@ -71,11 +170,13 @@ async def call_next() -> None: 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 @@ -85,14 +186,16 @@ async def call_next() -> None: 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 + assert calls == 0 \ No newline at end of file From cd51443dee1f51b036ba884a4a56da1f1cd3851d Mon Sep 17 00:00:00 2001 From: Samsul Date: Thu, 27 Aug 2026 09:06:50 +0000 Subject: [PATCH 3/3] Address review feedback: fix HOL Guard response schema, avoid policy-approval framing, default demo to fail-closed --- .../middleware/hol_guard_middleware.py | 71 ++++++++++++------- 1 file changed, 44 insertions(+), 27 deletions(-) diff --git a/python/samples/02-agents/middleware/hol_guard_middleware.py b/python/samples/02-agents/middleware/hol_guard_middleware.py index 4aafbfa3451..805bd8a09b3 100644 --- a/python/samples/02-agents/middleware/hol_guard_middleware.py +++ b/python/samples/02-agents/middleware/hol_guard_middleware.py @@ -10,15 +10,12 @@ # 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 on an explicit allow verdict -from HOL Guard (https://github.com/hashgraph-online/hol-guard). +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. @@ -27,27 +24,38 @@ continue -- matching the issue's "terminate before the wrapped tool executes" requirement for all three non-allow outcomes, not just an outright deny. -The real HOL Guard engine is invoked locally through its CLI (`hol-guard command test ---json`). That command is a PREVIEW-ONLY pattern check: its JSON response reports -`status` ("no_match" or "review") and explicitly marks `policy_evaluation: "not_run"`. So an -ALLOW from this sample means "no known attack pattern matched", not "HOL Guard's full org -policy approved this call" -- the two are different guarantees. 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. - -OPEN QUESTION (track on #7833): confirm with the HOL Guard maintainers whether a command exists -that runs full policy evaluation (not just the pattern-preview `command test`), and use that -instead once available. +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 @@ -147,7 +155,10 @@ async def evaluate_with_hol_guard( # 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 attack pattern matched (preview check only)" + 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}" @@ -191,7 +202,10 @@ async def process( timeout_seconds=self._timeout_seconds, ) if decision is GuardDecision.ALLOW: - logger.info("[HOLGuardMiddleware] Tool '%s' allowed by HOL Guard.", context.function.name) + logger.info( + "[HOLGuardMiddleware] Tool '%s': no known attack pattern found by HOL Guard; proceeding.", + context.function.name, + ) await call_next() return @@ -242,10 +256,13 @@ async def main() -> None: name="OpsAgent", instructions="You are a helpful assistant with access to weather and admin tools.", tools=[get_weather, delete_production_database], - # offline_fallback=True lets this demo run without the real hol-guard engine - # installed. In production, omit it (default False) so an unavailable Guard fails - # closed instead of falling back to the much weaker built-in deny-list. - middleware=[HOLGuardMiddleware(offline_fallback=True)], + # 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 ---")