237 lines
13 KiB
Python
237 lines
13 KiB
Python
"""Agent-side HUX hook driven end-to-end against the real service over HTTP.
|
|
|
|
Security obligations exercised: SO-35 (only the human surface decides; the
|
|
hook's worker identity cannot), SO-36 (a ``once`` approval releases the gate
|
|
exactly once), SO-37 (the gate compares the canonical argument hash the hook
|
|
recorded at request time), SO-39 (an external side effect under
|
|
``autonomous`` still waits for a human), SO-40 (an exhausted budget blocks new
|
|
approvals), SO-41 (a stop is done only when the receipt exists), SO-11 and
|
|
SO-07 (raw arguments and tool output never reach the tenant ledger).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
|
|
HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux"
|
|
for entry in (FOUNDATION, HOOK_ROOT):
|
|
if str(entry) not in sys.path:
|
|
sys.path.insert(0, str(entry))
|
|
|
|
from hux import contracts # noqa: E402
|
|
from hux.http import serve # noqa: E402
|
|
from hux.server import build_router # noqa: E402
|
|
from hux_hook import HuxClient, HuxServiceError, after_tool, before_tool, canonical_argument_hash, on_stop, record_spend # noqa: E402
|
|
|
|
SCHEMAS = contracts.load_all()
|
|
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
|
|
SUBJECT = "usr_0123456789abcdef"
|
|
IDENTITY = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "worker", "trust": "worker"}
|
|
HUMAN = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "chat", "trust": "router"}
|
|
CANARY = "CANARY-7f3a-SECRET-VALUE"
|
|
ARGS = {"path": "notes.md", "content": CANARY}
|
|
RUN = "run_hook_1"
|
|
|
|
|
|
def start(tmp_path: Path, flags: str = ALL_ON) -> tuple[str, object]:
|
|
"""Run the real service on an ephemeral loopback port for one test."""
|
|
router = build_router(tmp_path, {"HUX_FLAGS": flags, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"})
|
|
server = serve(router, "127.0.0.1", 0)
|
|
threading.Thread(target=server.serve_forever, daemon=True).start()
|
|
return f"http://127.0.0.1:{server.server_address[1]}", server
|
|
|
|
|
|
@pytest.fixture
|
|
def service(tmp_path):
|
|
base, server = start(tmp_path)
|
|
human = HuxClient(base, HUMAN, key="rk")
|
|
conversation = human.post("/hux/v1/conversations", {"title": "hook test"}).body["id"]
|
|
yield {"base": base, "root": tmp_path, "agent": HuxClient(base, IDENTITY, key="wk"), "human": human, "conv": conversation}
|
|
server.shutdown()
|
|
|
|
|
|
def set_policy(human: HuxClient, autonomy: str, budgets: dict | None = None) -> None:
|
|
body = {"scope": {"level": "global"}, "autonomy": autonomy}
|
|
if budgets:
|
|
body["budgets"] = budgets
|
|
human.put("/hux/v1/policy", body)
|
|
|
|
|
|
def ledger_text(root: Path) -> str:
|
|
return "\n".join(p.read_text(errors="ignore") for p in root.rglob("*") if p.is_file())
|
|
|
|
|
|
def events_of(human: HuxClient, conversation: str) -> list[dict]:
|
|
items = human.get(f"/hux/v1/conversations/{conversation}/events").body["items"]
|
|
for item in items:
|
|
assert contracts.validate_record(item, SCHEMAS) == []
|
|
return items
|
|
|
|
|
|
# --- approval -> pending -> human decision -> gate once ------------------------------
|
|
|
|
def test_once_approval_releases_exactly_once(service):
|
|
"""SO-35, SO-36, SO-37: pending until a human decides, then one release; the second gate is blocked."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
first = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low", turn=1)
|
|
assert (first.proceed, first.reason) == (False, "approval_required")
|
|
assert first.approval_id and first.approval_id.startswith("apr_")
|
|
record = human.get(f"/hux/v1/approvals/{first.approval_id}").body
|
|
assert contracts.validate_record(record, SCHEMAS) == []
|
|
assert record["request"]["evidence"][0]["hash"] == canonical_argument_hash("write_file", ARGS)
|
|
assert CANARY not in json.dumps(record)
|
|
with pytest.raises(HuxServiceError) as denied:
|
|
agent.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"})
|
|
assert denied.value.status == 403
|
|
human.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"})
|
|
second = before_tool(agent, RUN, conv, "write_file", {"content": CANARY, "path": "notes.md"}, "write_files", risk="low")
|
|
assert second == second.__class__(True, first.approval_id, "released")
|
|
third = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low")
|
|
assert third.proceed is False and "consumed" in third.reason
|
|
kinds = [e["kind"] for e in events_of(human, conv)]
|
|
assert kinds.count("side_effect.released") == 1 and "side_effect.blocked" in kinds and "approval.requested" in kinds
|
|
assert CANARY not in ledger_text(service["root"])
|
|
|
|
|
|
def test_different_arguments_after_approval_are_blocked(service):
|
|
"""SO-37: an approval for one argument hash never releases another."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell")
|
|
human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "once"})
|
|
other = before_tool(agent, RUN, conv, "shell", {"cmd": "rm -rf /"}, "shell")
|
|
assert other.proceed is False and other.reason == "approval_required"
|
|
|
|
|
|
def test_denied_and_session_choices(service):
|
|
"""A denial fails closed; a session grant lets a later call in the same conversation auto-approve."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell")
|
|
human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "deny"})
|
|
assert before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") == pending.__class__(False, pending.approval_id, "approval_denied")
|
|
pending2 = before_tool(agent, RUN, conv, "shell", {"cmd": "pwd"}, "shell")
|
|
human.post(f"/hux/v1/approvals/{pending2.approval_id}", {"choice": "session"})
|
|
assert before_tool(agent, "run_hook_2", conv, "shell", {"cmd": "whoami"}, "shell").proceed is True
|
|
|
|
|
|
def test_external_side_effect_under_autonomous_still_asks(service):
|
|
"""SO-39: autonomy never auto-allows an external side effect."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
set_policy(human, "autonomous")
|
|
decision = before_tool(agent, RUN, conv, "send_email", {"to": "x@example.com"}, "send_message", external=True, risk="high")
|
|
assert (decision.proceed, decision.reason) == (False, "approval_required")
|
|
local = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files")
|
|
assert local.proceed is True and local.reason == "released"
|
|
assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").proceed is False
|
|
|
|
|
|
def test_budget_exhaustion_blocks_new_approvals(service):
|
|
"""SO-40: once the run budget is spent the hook reports budget_exhausted and never proceeds."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
set_policy(human, "autonomous", {"tool_calls_per_run": 2})
|
|
state = record_spend(agent, RUN, conv, tool_calls=2, tokens=10, bogus=5)
|
|
assert state is not None and contracts.validate_record(state, SCHEMAS) == []
|
|
assert state["exhausted"] == ["tool_calls_per_run"]
|
|
decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files")
|
|
assert decision == decision.__class__(False, None, "budget_exhausted")
|
|
assert record_spend(agent, RUN, conv, tokens=-3)["spent"]["tokens"] == 10
|
|
assert record_spend(agent, RUN, tool_calls=1)["spent"]["tool_calls"] == 3
|
|
|
|
|
|
def test_stop_receipt_is_written_once(service):
|
|
"""SO-41: the receipt reports what really happened and a repeat returns the same record."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
effects = [{"description": "partial notes.md", "reverted": True}]
|
|
receipt = on_stop(agent, RUN, conv, process_registry_empty=True, side_effects=effects)
|
|
assert contracts.validate_record(receipt, SCHEMAS) == []
|
|
assert receipt["outcome"] == "cancelled" and receipt["side_effects"] == effects
|
|
assert on_stop(agent, RUN, conv, process_registry_empty=False)["id"] == receipt["id"]
|
|
failed = on_stop(agent, "run_hook_3", None, process_registry_empty=False)
|
|
assert failed["outcome"] == "failed_to_cancel" and "conversation_id" not in failed
|
|
assert on_stop(agent, "run_hook_4", conv, True, already_complete=True)["outcome"] == "already_complete"
|
|
assert [e["kind"] for e in events_of(human, conv)].count("run.cancelled") == 2
|
|
|
|
|
|
def test_after_tool_records_status_only(service):
|
|
"""SO-11: tool.result carries sizes and status; the output and arguments never land."""
|
|
agent, conv = service["agent"], service["conv"]
|
|
digest = canonical_argument_hash("shell", {"cmd": CANARY})
|
|
event = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest, duration_ms=40, exit_code=0)
|
|
assert event["kind"] == "tool.result" and event["detail"] == {"tool": "shell", "ok": True, "bytes": 512, "duration_ms": 40, "exit_code": 0}
|
|
assert event["evidence"][0]["hash"] == digest and event["turn"] == 3
|
|
replay = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest)
|
|
assert replay["id"] == event["id"]
|
|
plain = after_tool(agent, RUN, conv, "shell", False, -1)
|
|
assert plain["detail"] == {"tool": "shell", "ok": False, "bytes": 0} and "evidence" not in plain
|
|
assert CANARY not in ledger_text(service["root"])
|
|
|
|
|
|
def test_bad_ids_and_service_errors_fail_closed(service):
|
|
"""Malformed run or conversation ids never reach a tool execution."""
|
|
agent, conv = service["agent"], service["conv"]
|
|
assert before_tool(agent, "run/../x", conv, "shell", {}, "shell").reason == "invalid_run_id"
|
|
assert before_tool(agent, RUN, "not an id", "shell", {}, "shell").reason == "service_error:invalid"
|
|
assert before_tool(agent, RUN, conv, "shell", {}, "teleport").reason == "service_error:invalid"
|
|
|
|
|
|
def test_gate_failures_fail_closed(service, monkeypatch):
|
|
"""A gate that errors, answers nonsense or says no keeps the tool from running."""
|
|
agent, human, conv = service["agent"], service["human"], service["conv"]
|
|
set_policy(human, "autonomous")
|
|
real_post = agent.post
|
|
|
|
def broken_gate(path, body, idempotency_key=None):
|
|
if path.endswith("/gate"):
|
|
raise HuxServiceError(500, "invalid", "boom")
|
|
return real_post(path, body, idempotency_key)
|
|
|
|
monkeypatch.setattr(agent, "post", broken_gate)
|
|
decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files")
|
|
assert decision.proceed is False and decision.reason == "service_error:invalid" and decision.approval_id
|
|
|
|
class Odd:
|
|
body = "not a dict"
|
|
|
|
monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd() if path.endswith("/gate") else real_post(path, body, idempotency_key))
|
|
assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_invalid"
|
|
|
|
class NoProceed:
|
|
body = {"proceed": False}
|
|
|
|
monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: NoProceed() if path.endswith("/gate") else real_post(path, body, idempotency_key))
|
|
assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_blocked"
|
|
monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd())
|
|
assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "approval_invalid"
|
|
|
|
|
|
def test_unreachable_service_fails_closed_for_gate_and_open_for_telemetry(tmp_path):
|
|
"""No service: the gate says no, telemetry returns None, nothing raises into the loop."""
|
|
base, server = start(tmp_path)
|
|
server.shutdown()
|
|
server.server_close()
|
|
agent = HuxClient(base, IDENTITY, key="wk", timeout=1)
|
|
decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell")
|
|
assert decision == decision.__class__(False, None, "hux_unavailable")
|
|
assert after_tool(agent, RUN, "conv_0001abcd", "shell", True, 1) is None
|
|
assert record_spend(agent, RUN, "conv_0001abcd", tool_calls=1) is None
|
|
assert on_stop(agent, RUN, "conv_0001abcd", True) is None
|
|
|
|
|
|
def test_autonomy_flag_off_is_reported_and_fails_closed(tmp_path):
|
|
"""SO-50: with hux.autonomy off, capabilities says so and the hook refuses every side effect."""
|
|
base, server = start(tmp_path, "hux.foundation,hux.activity_timeline,hux.projects")
|
|
try:
|
|
agent = HuxClient(base, IDENTITY, key="wk")
|
|
caps = agent.capabilities()
|
|
assert caps["reachable"] and caps["cards"]["HUX-11"] and caps["cards"]["HUX-01"] and not caps["cards"]["HUX-05"]
|
|
decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell")
|
|
assert decision == decision.__class__(False, None, "autonomy_off")
|
|
finally:
|
|
server.shutdown()
|