Files
omnigent-ai--omnigent/tests/native_hook_helpers.py
T
ckcuslife-source 1d897ca0fd fix(harnesses): fail closed for TOOL_CALL when native policy hook can't reach the server (#579)
* fix(harnesses): fail closed for TOOL_CALL when native policy hook can't reach the server

Fixes #536.

For native harnesses (claude-native, codex-native) the PreToolUse/
PostToolUse hook subprocess is the entire policy-governance layer: it
gates Bash/Write/Edit, the native Skill tool, and connector-native
mcp__* tools by POSTing to /v1/sessions/{id}/policies/evaluate. Every
error/edge path returned exit 0 with no stdout — "no opinion" — so any
condition that prevented a well-formed verdict (server unreachable,
non-2xx, empty body, malformed JSON) silently disabled all DENY/ASK
enforcement. A transient AP outage turned a blocked tool into an
allowed one, with only a stderr line. P0 bypass.

Make the hooks' default phase-aware, mirroring the runner-side fix in
PR #163. Once a session is known to be governed (active session id +
configured ap_server_url) and the evaluate round-trip cannot yield a
usable verdict, a PreToolUse (PHASE_TOOL_CALL) call fails CLOSED with a
deny — the authoritative, only-enforcement-point gate — while
UserPromptSubmit (advisory request gate) and PostToolUse (the tool
already ran) keep failing OPEN. Pre-evaluation short-circuits that mean
the session simply isn't governed (no session, no ap_server_url,
unparseable payload, relay-gated mcp__omnigent__* tools) still emit "no
opinion" so non-Omnigent sessions are never blocked.

The client timeout is intentionally left unchanged: the long timeout
backs the server-side ASK long-poll, and shortening it would break ASK
and reintroduce a fail-open. A hung server still blocks (the safe
direction) rather than failing open.

Changes:
- native_policy_hook.py: new shared fail_closed_hook_output() helper.
- claude_native_hook.py / codex_native_hook.py: the HTTP-error,
  empty-body, and malformed-response branches now fail closed for the
  tool-call gate instead of returning no opinion.
- Tests: unit coverage for the helper plus integration tests asserting
  PreToolUse denies across connect-error/non-2xx/empty/malformed while
  PostToolUse and UserPromptSubmit stay fail-open.

* test/fix(harnesses): address Polly review — clearer non-2xx log, shared test helper, unknown-event guard

Non-blocking follow-ups from the Polly AI review on PR #579:

- Log non-2xx responses distinctly from connection errors. Both native
  hooks now catch httpx.HTTPStatusError before the broad httpx.HTTPError
  branch and log the status code, so a real AP outage (e.g. 503) is
  distinguishable from an unreachable server in production diagnostics.
  Behavior is unchanged — both still fail closed for the tool-call gate.
- Deduplicate the failing-client test stub into
  tests/native_hook_helpers.make_failing_client, imported by both the
  claude- and codex-native hook test modules, so the four failure modes
  can't drift.
- Add an explicit unknown-event test for fail_closed_hook_output
  ("SomeNewEvent" -> None) documenting the fail-open-for-unknowns contract.
2026-06-17 23:50:36 -07:00

45 lines
1.6 KiB
Python

"""Shared test helpers for the native-harness policy hooks.
Used by both ``tests/test_claude_native_hook.py`` and
``tests/test_codex_native_hook.py`` so the fail-closed failure-mode stub
lives in one place and can't drift if new modes are added.
"""
from __future__ import annotations
import httpx
def make_failing_client(mode: str) -> type:
"""
Build an ``httpx.Client`` stub that fails the policy POST a given way.
:param mode: One of ``"connect_error"`` (POST raises), ``"non_2xx"``
(503 → ``raise_for_status``), ``"empty_body"`` (200, no content),
or ``"malformed_json"`` (200, non-JSON body).
:returns: A class usable as a drop-in for :class:`httpx.Client`.
"""
class _FailingHttpxClient:
def __init__(self, *, headers: dict[str, str], timeout: object) -> None:
del headers, timeout
def __enter__(self) -> _FailingHttpxClient:
return self
def __exit__(self, *args: object) -> None:
del args
def post(self, url: str, *, json: dict[str, object]) -> httpx.Response:
del json
req = httpx.Request("POST", url)
if mode == "connect_error":
raise httpx.ConnectError("AP unreachable", request=req)
if mode == "non_2xx":
return httpx.Response(503, text="upstream down", request=req)
if mode == "empty_body":
return httpx.Response(200, content=b"", request=req)
return httpx.Response(200, text="not json at all", request=req)
return _FailingHttpxClient