From b38d4d10c8989173600bd4456742eb6c6a97e2c4 Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Sun, 23 Aug 2026 15:05:29 -0400 Subject: [PATCH 1/3] docs: add HOL Guard security analyzer example --- .../43_hol_guard_security_analyzer.py | 180 ++++++++++++++++++ tests/examples/test_examples.py | 1 + 2 files changed, 181 insertions(+) create mode 100644 examples/01_standalone_sdk/43_hol_guard_security_analyzer.py diff --git a/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py b/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py new file mode 100644 index 0000000000..b7cd681e5a --- /dev/null +++ b/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py @@ -0,0 +1,180 @@ +"""OpenHands Agent SDK — HOL Guard Security Analyzer Example. + +This example adapts HOL Guard's side-effect-free shell command inspection to +OpenHands' SecurityAnalyzerBase. Install HOL Guard separately before running: + + pipx install --pip-args='--pre' hol-guard + +Project: https://github.com/hashgraph-online/hol-guard + +`hol-guard command test` classifies a shell command without executing it. This +adapter fails closed: only commands HOL Guard marks explicitly benign are LOW +risk; review, block, unknown, unsupported actions, or analyzer errors are HIGH +risk and flow through OpenHands' normal ConfirmRisky confirmation path. +""" + +import json +import os +import signal +import subprocess +from collections.abc import Callable + +from pydantic import SecretStr + +from openhands.sdk import LLM, Agent, BaseConversation, Conversation +from openhands.sdk.conversation.state import ( + ConversationExecutionStatus, + ConversationState, +) +from openhands.sdk.event import ActionEvent +from openhands.sdk.security.analyzer import SecurityAnalyzerBase +from openhands.sdk.security.confirmation_policy import ConfirmRisky +from openhands.sdk.security.risk import SecurityRisk +from openhands.sdk.tool import Tool +from openhands.tools.terminal import TerminalAction, TerminalTool + + +signal.signal(signal.SIGINT, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) + + +class HolGuardSecurityAnalyzer(SecurityAnalyzerBase): + """Classify OpenHands terminal actions with the local HOL Guard CLI.""" + + def __init__( + self, + guard_executable: str = "hol-guard", + *, + workspace: str = ".", + timeout_seconds: float = 10.0, + ) -> None: + self.guard_executable = guard_executable + self.workspace = workspace + self.timeout_seconds = timeout_seconds + + def security_risk(self, action: ActionEvent) -> SecurityRisk: + terminal_action = action.action + if not isinstance(terminal_action, TerminalAction): + return SecurityRisk.HIGH + if terminal_action.is_input or not terminal_action.command.strip(): + return SecurityRisk.HIGH + + try: + result = subprocess.run( + [ + self.guard_executable, + "command", + "test", + terminal_action.command, + "--json", + ], + cwd=self.workspace, + capture_output=True, + check=False, + text=True, + timeout=self.timeout_seconds, + ) + if result.returncode != 0: + return SecurityRisk.HIGH + payload = json.loads(result.stdout) + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): + return SecurityRisk.HIGH + + minimum_action = str(payload.get("minimum_action") or "").lower() + if minimum_action in {"review", "block", "require-reapproval"}: + return SecurityRisk.HIGH + + classification = payload.get("classification") + if ( + isinstance(classification, dict) + and classification.get("explicitly_benign") is True + ): + return SecurityRisk.LOW + + return SecurityRisk.HIGH + + +def _print_blocked_actions(pending_actions) -> None: + print(f"\nHOL Guard flagged {len(pending_actions)} action(s) for confirmation:") + for i, action in enumerate(pending_actions, start=1): + headline = action.summary or "(no summary provided)" + snippet = str(action.action)[:100].replace("\n", " ") + print(f" {i}. [{action.tool_name}] {headline}") + print(f" {snippet}...") + + +def confirm_high_risk_in_console(pending_actions) -> bool: + """Approve or reject actions that HOL Guard did not classify as benign.""" + _print_blocked_actions(pending_actions) + while True: + try: + answer = ( + input("\nExecute these flagged actions anyway? (yes/no): ") + .strip() + .lower() + ) + except (EOFError, KeyboardInterrupt): + print("\nNo input received; rejecting by default.") + return False + + if answer in ("yes", "y"): + return True + if answer in ("no", "n"): + return False + print("Please enter 'yes' or 'no'.") + + +def run_until_finished_with_security( + conversation: BaseConversation, confirmer: Callable[[list], bool] +) -> None: + """Run until completion, rejecting flagged pending actions by default.""" + while conversation.state.execution_status != ConversationExecutionStatus.FINISHED: + if ( + conversation.state.execution_status + == ConversationExecutionStatus.WAITING_FOR_CONFIRMATION + ): + pending = ConversationState.get_unmatched_actions(conversation.state.events) + if not pending: + raise RuntimeError( + "Agent is waiting for confirmation but no pending actions " + "were found." + ) + if not confirmer(pending): + conversation.reject_pending_actions( + "User rejected HOL Guard flagged actions" + ) + continue + conversation.run() + + +def main() -> None: + api_key = os.getenv("LLM_API_KEY") + assert api_key is not None, "LLM_API_KEY environment variable is not set." + model = os.getenv("LLM_MODEL", "gpt-5.5") + base_url = os.getenv("LLM_BASE_URL") + llm = LLM( + usage_id="hol-guard-security-analyzer", + model=model, + base_url=base_url, + api_key=SecretStr(api_key), + ) + + agent = Agent(llm=llm, tools=[Tool(name=TerminalTool.name)]) + conversation = Conversation( + agent=agent, persistence_dir="./.conversations", workspace="." + ) + conversation.set_security_analyzer(HolGuardSecurityAnalyzer(workspace=".")) + conversation.set_confirmation_policy(ConfirmRisky()) + + print("\n1) Safe command: HOL Guard should classify this as benign.") + conversation.send_message("Use the terminal to run exactly: pwd") + run_until_finished_with_security(conversation, confirm_high_risk_in_console) + + print("\n2) Risky command: HOL Guard should require confirmation.") + conversation.send_message( + "Use the terminal to run exactly: rm -rf /tmp/openhands-hol-guard-demo" + ) + run_until_finished_with_security(conversation, confirm_high_risk_in_console) + + +if __name__ == "__main__": + main() diff --git a/tests/examples/test_examples.py b/tests/examples/test_examples.py index 533a999fb3..04a375860a 100644 --- a/tests/examples/test_examples.py +++ b/tests/examples/test_examples.py @@ -54,6 +54,7 @@ "examples/01_standalone_sdk/08_mcp_with_oauth.py", "examples/01_standalone_sdk/15_browser_use.py", "examples/01_standalone_sdk/16_llm_security_analyzer.py", + "examples/01_standalone_sdk/43_hol_guard_security_analyzer.py", "examples/01_standalone_sdk/27_observability_laminar.py", "examples/01_standalone_sdk/35_subscription_login.py", # Requires interactive input() which fails in CI with EOFError From cfeafae3b7413dcdab675defd71bddbb01d0aaba Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Sun, 23 Aug 2026 15:22:50 -0400 Subject: [PATCH 2/3] fix: harden HOL Guard analyzer example --- .../43_hol_guard_security_analyzer.py | 29 +++++++++---------- 1 file changed, 13 insertions(+), 16 deletions(-) diff --git a/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py b/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py index b7cd681e5a..c78aa14dbb 100644 --- a/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py +++ b/examples/01_standalone_sdk/43_hol_guard_security_analyzer.py @@ -15,11 +15,10 @@ import json import os -import signal import subprocess from collections.abc import Callable -from pydantic import SecretStr +from pydantic import Field, SecretStr from openhands.sdk import LLM, Agent, BaseConversation, Conversation from openhands.sdk.conversation.state import ( @@ -34,22 +33,12 @@ from openhands.tools.terminal import TerminalAction, TerminalTool -signal.signal(signal.SIGINT, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())) - - class HolGuardSecurityAnalyzer(SecurityAnalyzerBase): """Classify OpenHands terminal actions with the local HOL Guard CLI.""" - def __init__( - self, - guard_executable: str = "hol-guard", - *, - workspace: str = ".", - timeout_seconds: float = 10.0, - ) -> None: - self.guard_executable = guard_executable - self.workspace = workspace - self.timeout_seconds = timeout_seconds + guard_executable: str = "hol-guard" + workspace: str = "." + timeout_seconds: float = Field(default=10.0, gt=0.0) def security_risk(self, action: ActionEvent) -> SecurityRisk: terminal_action = action.action @@ -79,8 +68,16 @@ def security_risk(self, action: ActionEvent) -> SecurityRisk: except (OSError, subprocess.SubprocessError, json.JSONDecodeError): return SecurityRisk.HIGH + if not isinstance(payload, dict): + return SecurityRisk.HIGH + minimum_action = str(payload.get("minimum_action") or "").lower() - if minimum_action in {"review", "block", "require-reapproval"}: + if minimum_action in { + "review", + "block", + "require-reapproval", + "sandbox-required", + }: return SecurityRisk.HIGH classification = payload.get("classification") From 63e3e5b371b7a655aec2b18343d092e140c49f68 Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Sun, 23 Aug 2026 15:25:23 -0400 Subject: [PATCH 3/3] test: cover HOL Guard security analyzer example --- ...est_hol_guard_security_analyzer_example.py | 187 ++++++++++++++++++ 1 file changed, 187 insertions(+) create mode 100644 tests/sdk/security/test_hol_guard_security_analyzer_example.py diff --git a/tests/sdk/security/test_hol_guard_security_analyzer_example.py b/tests/sdk/security/test_hol_guard_security_analyzer_example.py new file mode 100644 index 0000000000..02f1335ef2 --- /dev/null +++ b/tests/sdk/security/test_hol_guard_security_analyzer_example.py @@ -0,0 +1,187 @@ +"""Regression tests for the HOL Guard SecurityAnalyzerBase example.""" + +import importlib.util +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +from openhands.sdk.event import ActionEvent +from openhands.sdk.llm import MessageToolCall, TextContent +from openhands.sdk.security.risk import SecurityRisk +from openhands.sdk.tool import Action +from openhands.tools.terminal import TerminalAction + + +EXAMPLE_PATH = ( + Path(__file__).parents[3] + / "examples" + / "01_standalone_sdk" + / "43_hol_guard_security_analyzer.py" +) +spec = importlib.util.spec_from_file_location("hol_guard_security_analyzer", EXAMPLE_PATH) +assert spec is not None and spec.loader is not None +example = importlib.util.module_from_spec(spec) +spec.loader.exec_module(example) + + +class DummyAction(Action): + value: str = "not-terminal" + + +def _event(action: Action) -> ActionEvent: + command = getattr(action, "command", "") + return ActionEvent( + thought=[TextContent(text="HOL Guard analyzer test")], + action=action, + tool_name="terminal", + tool_call_id="hol-guard-test", + tool_call=MessageToolCall( + id="hol-guard-test", + name="terminal", + arguments=json.dumps({"command": command}), + origin="completion", + ), + llm_response_id="hol-guard-test", + ) + + +def _completed(payload: Any, *, returncode: int = 0) -> subprocess.CompletedProcess[str]: + stdout = payload if isinstance(payload, str) else json.dumps(payload) + return subprocess.CompletedProcess( + args=["hol-guard"], returncode=returncode, stdout=stdout, stderr="" + ) + + +def test_analyzer_constructs_as_pydantic_model() -> None: + analyzer = example.HolGuardSecurityAnalyzer( + guard_executable="guard-bin", + workspace="/tmp", + timeout_seconds=3.5, + ) + + assert analyzer.guard_executable == "guard-bin" + assert analyzer.workspace == "/tmp" + assert analyzer.timeout_seconds == 3.5 + + with pytest.raises(ValidationError): + example.HolGuardSecurityAnalyzer(timeout_seconds=0) + + +def test_explicitly_benign_terminal_command_is_low_risk( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[list[str], dict[str, Any]]] = [] + + def fake_run(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: + calls.append((argv, kwargs)) + return _completed( + { + "minimum_action": "allow", + "classification": {"explicitly_benign": True}, + } + ) + + monkeypatch.setattr(example.subprocess, "run", fake_run) + analyzer = example.HolGuardSecurityAnalyzer(workspace="/workspace") + + risk = analyzer.security_risk(_event(TerminalAction(command="pwd"))) + + assert risk == SecurityRisk.LOW + assert calls == [ + ( + ["hol-guard", "command", "test", "pwd", "--json"], + { + "cwd": "/workspace", + "capture_output": True, + "check": False, + "text": True, + "timeout": 10.0, + }, + ) + ] + + +@pytest.mark.parametrize( + "minimum_action", + ["review", "block", "require-reapproval", "sandbox-required"], +) +def test_guarded_minimum_actions_are_high_risk( + monkeypatch: pytest.MonkeyPatch, + minimum_action: str, +) -> None: + monkeypatch.setattr( + example.subprocess, + "run", + lambda *_args, **_kwargs: _completed( + { + "minimum_action": minimum_action, + "classification": {"explicitly_benign": True}, + } + ), + ) + + risk = example.HolGuardSecurityAnalyzer().security_risk( + _event(TerminalAction(command="rm -rf /tmp/demo")) + ) + + assert risk == SecurityRisk.HIGH + + +@pytest.mark.parametrize( + ("result", "action"), + [ + (_completed("not-json"), TerminalAction(command="pwd")), + (_completed(["not", "an", "object"]), TerminalAction(command="pwd")), + (_completed({}, returncode=2), TerminalAction(command="pwd")), + ( + _completed( + { + "minimum_action": "allow", + "classification": {"explicitly_benign": False}, + } + ), + TerminalAction(command="pwd"), + ), + ], +) +def test_untrusted_or_failed_guard_results_fail_closed( + monkeypatch: pytest.MonkeyPatch, + result: subprocess.CompletedProcess[str], + action: TerminalAction, +) -> None: + monkeypatch.setattr( + example.subprocess, + "run", + lambda *_args, **_kwargs: result, + ) + + risk = example.HolGuardSecurityAnalyzer().security_risk(_event(action)) + + assert risk == SecurityRisk.HIGH + + +def test_missing_guard_executable_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None: + def missing(*_args: Any, **_kwargs: Any) -> subprocess.CompletedProcess[str]: + raise FileNotFoundError("hol-guard") + + monkeypatch.setattr(example.subprocess, "run", missing) + + risk = example.HolGuardSecurityAnalyzer().security_risk( + _event(TerminalAction(command="pwd")) + ) + + assert risk == SecurityRisk.HIGH + + +def test_non_terminal_and_terminal_input_actions_fail_closed() -> None: + analyzer = example.HolGuardSecurityAnalyzer() + + assert analyzer.security_risk(_event(DummyAction())) == SecurityRisk.HIGH + assert ( + analyzer.security_risk(_event(TerminalAction(command="C-c", is_input=True))) + == SecurityRisk.HIGH + )