1793903f5e
Real traffic showed the first enforcement pass blocking core assistant faculties: skills listing and the sandboxed Python classified as unknown external side effects, browsing denied by default, with no approval surface in the chat flow. Enforcement returns to observe-only fleet-wide while the approvals UX and default grants are reworked, and the capability map now tells the truth about the real toolset: the Python sandbox is internal shell work, skills/todo/clarify/vision are reads, browsing is network (medium), image generation writes an artifact through the trusted broker. Unknown tools remain fail-closed. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf
489 lines
18 KiB
Python
489 lines
18 KiB
Python
"""Adversarial tests for the Hermes HUX execution middleware."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import importlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
PLUGIN = ROOT / "services" / "hermes" / "plugins" / "hux-runtime"
|
|
HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux"
|
|
sys.path.insert(0, str(HOOK_ROOT))
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"hermes_hux_runtime",
|
|
PLUGIN / "__init__.py",
|
|
submodule_search_locations=[str(PLUGIN)],
|
|
)
|
|
assert SPEC and SPEC.loader
|
|
PACKAGE = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = PACKAGE
|
|
SPEC.loader.exec_module(PACKAGE)
|
|
context_ids = importlib.import_module("hermes_hux_runtime.context_ids")
|
|
runtime_module = importlib.import_module("hermes_hux_runtime.runtime")
|
|
tool_policy = importlib.import_module("hermes_hux_runtime.tool_policy")
|
|
|
|
SUBJECT = "usr_" + "a" * 64
|
|
SLOT = "slot-2"
|
|
KEY = bytes(range(32))
|
|
ARGS = {"path": "notes.txt", "content": "CANARY-secret-value"}
|
|
|
|
|
|
class FakeIds:
|
|
"""Stable ids without touching a key file in middleware-only tests."""
|
|
|
|
def conversation(self, raw: str) -> str:
|
|
if not raw:
|
|
raise context_ids.ContextUnavailable("missing")
|
|
return "conv_" + "c" * 32
|
|
|
|
def session(self, raw: str) -> str:
|
|
if not raw:
|
|
raise context_ids.ContextUnavailable("missing")
|
|
return "ses_" + "b" * 32
|
|
|
|
def project(self, raw: str) -> str:
|
|
if not raw:
|
|
raise context_ids.ContextUnavailable("missing")
|
|
return "prj_" + "a" * 32
|
|
|
|
def run(self, raw: str) -> str:
|
|
if not raw:
|
|
raise context_ids.ContextUnavailable("missing")
|
|
return "run_" + "d" * 32
|
|
|
|
|
|
def middleware_metadata() -> dict[str, str]:
|
|
return {"session_id": "session-1", "turn_id": "turn-1", "tool_call_id": "call-1"}
|
|
|
|
|
|
class FakeClient:
|
|
"""Capture only the mandatory context bootstrap request."""
|
|
|
|
def __init__(self):
|
|
self.posts = []
|
|
|
|
def post(self, path, body, idempotency_key=None):
|
|
self.posts.append((path, body, idempotency_key))
|
|
return SimpleNamespace(body={})
|
|
|
|
|
|
def make_runtime(enforce: bool = True):
|
|
return runtime_module.Runtime(FakeClient(), FakeIds(), enforce)
|
|
|
|
|
|
def test_context_id_matches_frozen_webui_contract():
|
|
raw = "session:one+two"
|
|
message = "\0".join(
|
|
("hux.context.id.v1", "conversation", SLOT, SUBJECT, raw)
|
|
).encode()
|
|
expected = "conv_" + hmac.new(KEY, message, hashlib.sha256).hexdigest()[:32]
|
|
assert context_ids.derive_hux_id(KEY, "conv", "conversation", SLOT, SUBJECT, raw) == expected
|
|
run = context_ids.derive_hux_id(KEY, "run", "run", SLOT, SUBJECT, "turn-9")
|
|
assert run.startswith("run_") and len(run) == 36
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("prefix", "purpose", "slot", "subject", "raw", "key"),
|
|
[
|
|
("bad", "conversation", SLOT, SUBJECT, "s", KEY),
|
|
("conv", "conversation", "tenant-2", SUBJECT, "s", KEY),
|
|
("conv", "conversation", SLOT, "usr_short", "s", KEY),
|
|
("conv", "conversation", SLOT, SUBJECT, "slash/not-allowed", KEY),
|
|
("conv", "conversation", SLOT, SUBJECT, "s", b"short"),
|
|
],
|
|
)
|
|
def test_context_id_rejects_noncanonical_inputs(prefix, purpose, slot, subject, raw, key):
|
|
with pytest.raises(context_ids.ContextUnavailable):
|
|
context_ids.derive_hux_id(key, prefix, purpose, slot, subject, raw)
|
|
|
|
|
|
def test_context_key_requires_owner_only_regular_single_link_file(tmp_path: Path):
|
|
path = tmp_path / "context-key"
|
|
path.write_bytes(KEY)
|
|
path.chmod(0o600)
|
|
ids = context_ids.ContextIds(path, SLOT, SUBJECT)
|
|
assert ids.conversation("session-1").startswith("conv_")
|
|
assert ids.session("session-1").startswith("ses_")
|
|
assert ids.project("profile:default").startswith("prj_")
|
|
assert ids.run("turn-1").startswith("run_")
|
|
path.chmod(0o640)
|
|
with pytest.raises(context_ids.ContextUnavailable, match="unsafe"):
|
|
context_ids.ContextIds(path, SLOT, SUBJECT)
|
|
path.chmod(0o600)
|
|
linked = tmp_path / "linked"
|
|
os.link(path, linked)
|
|
with pytest.raises(context_ids.ContextUnavailable, match="unsafe"):
|
|
context_ids.ContextIds(path, SLOT, SUBJECT)
|
|
linked.unlink()
|
|
path.write_bytes(b"short")
|
|
with pytest.raises(context_ids.ContextUnavailable):
|
|
context_ids.ContextIds(path, SLOT, SUBJECT)
|
|
|
|
|
|
def test_context_key_rejects_missing_and_symlink(tmp_path: Path):
|
|
with pytest.raises(context_ids.ContextUnavailable, match="unavailable"):
|
|
context_ids.ContextIds(tmp_path / "missing", SLOT, SUBJECT)
|
|
real = tmp_path / "real"
|
|
real.write_bytes(KEY)
|
|
real.chmod(0o600)
|
|
link = tmp_path / "link"
|
|
link.symlink_to(real)
|
|
with pytest.raises(context_ids.ContextUnavailable):
|
|
context_ids.ContextIds(link, SLOT, SUBJECT)
|
|
with pytest.raises(context_ids.ContextUnavailable, match="identity"):
|
|
context_ids.ContextIds(real, SLOT, "usr_" + "A" * 64)
|
|
|
|
|
|
def test_context_key_handles_short_read(monkeypatch, tmp_path: Path):
|
|
path = tmp_path / "key"
|
|
path.write_bytes(KEY)
|
|
path.chmod(0o600)
|
|
monkeypatch.setattr(context_ids.os, "read", lambda *_args: b"short")
|
|
with pytest.raises(context_ids.ContextUnavailable, match="invalid"):
|
|
context_ids.ContextIds(path, SLOT, SUBJECT)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("name", "capability", "risk", "external"),
|
|
[
|
|
("read_file", "read_files", "low", False),
|
|
("write_file", "write_files", "medium", False),
|
|
("memory", "memory_write", "medium", False),
|
|
("delegate_task", "delegate", "high", False),
|
|
("terminal", "shell", "high", True),
|
|
("web_search", "network", "medium", True),
|
|
("send_email", "send_message", "high", True),
|
|
("kubectl_apply", "deploy", "high", True),
|
|
("image_generate", "artifact_write", "medium", False),
|
|
("artifact_create", "artifact_write", "medium", False),
|
|
("memory_forget", "memory_write", "medium", False),
|
|
("subagent_spawn", "delegate", "high", False),
|
|
("edit_document", "write_files", "medium", False),
|
|
("new_plugin_tool", "external_side_effect", "high", True),
|
|
(None, "external_side_effect", "high", True),
|
|
],
|
|
)
|
|
def test_tool_mapping_is_conservative(name, capability, risk, external):
|
|
assert tool_policy.classify(name) == tool_policy.ToolPolicy(capability, risk, external)
|
|
|
|
|
|
def test_default_off_registers_nothing(monkeypatch):
|
|
hooks, middleware = {}, {}
|
|
ctx = SimpleNamespace(
|
|
register_hook=lambda name, callback: hooks.setdefault(name, callback),
|
|
register_middleware=lambda name, callback: middleware.setdefault(name, callback),
|
|
)
|
|
monkeypatch.delenv("HUX_RUNTIME_ENABLED", raising=False)
|
|
PACKAGE.register(ctx)
|
|
assert not hooks and not middleware
|
|
|
|
|
|
@pytest.mark.parametrize(("enforce", "executes"), [("1", False), ("0", True)])
|
|
def test_misconfigured_opt_in_closes_only_enforcement(monkeypatch, enforce, executes):
|
|
hooks, middleware = {}, {}
|
|
ctx = SimpleNamespace(
|
|
register_hook=lambda name, callback: hooks.setdefault(name, callback),
|
|
register_middleware=lambda name, callback: middleware.setdefault(name, callback),
|
|
)
|
|
monkeypatch.setenv("HUX_RUNTIME_ENABLED", "true")
|
|
monkeypatch.setenv("HUX_TOOL_ENFORCEMENT", enforce)
|
|
monkeypatch.setattr(PACKAGE.Runtime, "from_env", classmethod(lambda cls: (_ for _ in ()).throw(ValueError())))
|
|
PACKAGE.register(ctx)
|
|
called = []
|
|
result = middleware["tool_execution"](args=ARGS, next_call=lambda value: called.append(value) or "ok")
|
|
assert bool(called) is executes
|
|
assert (result == "ok") is executes
|
|
assert set(hooks) == {"on_session_end"}
|
|
|
|
|
|
def test_from_env_reads_subject_only_from_file(tmp_path: Path):
|
|
subject = tmp_path / "subject"
|
|
subject.write_text(SUBJECT + "\n")
|
|
subject.chmod(0o400)
|
|
worker = tmp_path / "worker"
|
|
worker.write_text("worker-key")
|
|
worker.chmod(0o400)
|
|
context = tmp_path / "context"
|
|
context.write_bytes(KEY)
|
|
context.chmod(0o600)
|
|
env = {
|
|
"HUX_TENANT_SLOT": SLOT,
|
|
"HUX_SUBJECT_FILE": str(subject),
|
|
"HUX_WORKER_KEY_FILE": str(worker),
|
|
"HUX_CONTEXT_KEY_FILE": str(context),
|
|
"HUX_SUBJECT": "usr_" + "b" * 64,
|
|
"HUX_TOOL_ENFORCEMENT": "yes",
|
|
"HUX_TIMEOUT_SECONDS": "0.2",
|
|
}
|
|
instance = runtime_module.Runtime.from_env(env)
|
|
assert instance.client.identity == {
|
|
"tenant_slot": SLOT,
|
|
"subject": SUBJECT,
|
|
"surface": "worker",
|
|
"trust": "worker",
|
|
}
|
|
assert instance.enforce is True
|
|
assert instance.project_source == "profile:default"
|
|
|
|
|
|
def test_from_env_rejects_missing_file_contract():
|
|
with pytest.raises(context_ids.ContextUnavailable):
|
|
runtime_module.Runtime.from_env({"HUX_TENANT_SLOT": SLOT})
|
|
|
|
|
|
@pytest.mark.parametrize("project_source", [None, "", "x" * 201])
|
|
def test_runtime_rejects_malformed_project_source(project_source):
|
|
with pytest.raises(context_ids.ContextUnavailable, match="project source"):
|
|
runtime_module.Runtime(FakeClient(), FakeIds(), False, project_source)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"metadata",
|
|
[
|
|
{"session_id": None, "turn_id": "turn-1"},
|
|
{"session_id": "session-1", "turn_id": None},
|
|
],
|
|
)
|
|
def test_scope_requires_host_owned_string_ids(metadata):
|
|
with pytest.raises(context_ids.ContextUnavailable, match="session context"):
|
|
make_runtime(False)._scope(metadata)
|
|
|
|
|
|
def test_denial_never_executes_and_exposes_no_arguments(monkeypatch):
|
|
instance = make_runtime()
|
|
calls, telemetry = [], []
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: telemetry.append((args[1:], kwargs)))
|
|
|
|
def deny(_client, run_id, conversation_id, name, arguments, capability, **policy):
|
|
calls.append((run_id, conversation_id, name, arguments, capability, policy))
|
|
return SimpleNamespace(proceed=False, reason="approval_required", approval_id="apr_safe")
|
|
|
|
monkeypatch.setattr(runtime_module, "before_tool", deny)
|
|
downstream = []
|
|
result = instance.tool_execution(
|
|
tool_name="write_file",
|
|
args=ARGS,
|
|
next_call=lambda args: downstream.append(args),
|
|
**middleware_metadata(),
|
|
)
|
|
assert not downstream and calls[0][3] is ARGS
|
|
assert calls[0][-1] == {"external": False, "risk": "medium"}
|
|
assert json.loads(result) == {
|
|
"approval_id": "apr_safe",
|
|
"error": "HUX policy did not release this tool call",
|
|
"reason": "approval_required",
|
|
"schema": "hux.tool_block.v1",
|
|
"status": "blocked",
|
|
}
|
|
assert ARGS["content"] not in json.dumps(telemetry)
|
|
|
|
|
|
def test_bootstrap_uses_exact_ids_and_replays_in_process(monkeypatch):
|
|
instance = make_runtime(False)
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None)
|
|
for call_id in ("call-1", "call-2"):
|
|
instance.tool_execution(
|
|
tool_name="read_file",
|
|
args={},
|
|
next_call=lambda _args: "ok",
|
|
session_id="session-1",
|
|
turn_id="turn-1",
|
|
tool_call_id=call_id,
|
|
)
|
|
assert instance.client.posts == [
|
|
(
|
|
"/hux/v1/context/bootstrap",
|
|
{
|
|
"raw_session_id": "session-1",
|
|
"project_source": "profile:default",
|
|
"session_id": "ses_" + "b" * 32,
|
|
"conversation_id": "conv_" + "c" * 32,
|
|
"project_id": "prj_" + "a" * 32,
|
|
},
|
|
"context:conv_" + "c" * 32,
|
|
)
|
|
]
|
|
|
|
|
|
def test_bootstrap_failure_blocks_enforcement_and_opens_telemetry(monkeypatch):
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
|
|
for enforce in (True, False):
|
|
instance = make_runtime(enforce)
|
|
instance.client.post = lambda *args, **kwargs: (_ for _ in ()).throw(OSError("offline"))
|
|
called = []
|
|
result = instance.tool_execution(
|
|
tool_name="read_file",
|
|
args={},
|
|
next_call=lambda value: called.append(value) or "ok",
|
|
**middleware_metadata(),
|
|
)
|
|
assert bool(called) is (not enforce)
|
|
assert (result == "ok") is (not enforce)
|
|
|
|
|
|
def test_enforcement_failure_is_closed_but_telemetry_failure_is_open(monkeypatch):
|
|
for enforce in (True, False):
|
|
instance = make_runtime(enforce)
|
|
monkeypatch.setattr(runtime_module, "canonical_argument_hash", lambda *_: (_ for _ in ()).throw(TypeError()))
|
|
called = []
|
|
result = instance.tool_execution(
|
|
tool_name="write_file", args=ARGS, next_call=lambda args: called.append(args) or "done", **middleware_metadata()
|
|
)
|
|
assert bool(called) is (not enforce)
|
|
assert (result == "done") is (not enforce)
|
|
|
|
|
|
def test_released_call_executes_exact_object_once_and_reports_status_only(monkeypatch):
|
|
instance = make_runtime()
|
|
events, after, spend = [], [], []
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args[1:], kwargs)))
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"before_tool",
|
|
lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"),
|
|
)
|
|
monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append((args[1:], kwargs)))
|
|
monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: spend.append((args, kwargs)))
|
|
downstream = []
|
|
|
|
def execute(value):
|
|
downstream.append(value)
|
|
return {"ok": True, "payload": "RAW-RESULT-SECRET"}
|
|
|
|
result = instance.tool_execution(
|
|
tool_name="write_file", args=ARGS, next_call=execute, **middleware_metadata()
|
|
)
|
|
assert downstream == [ARGS] and downstream[0] is ARGS
|
|
assert result["payload"] == "RAW-RESULT-SECRET"
|
|
assert after[0][0][3] is True and after[0][0][4] > 0
|
|
assert spend[0][1]["tool_calls"] == 1
|
|
assert ARGS["content"] not in json.dumps(events)
|
|
assert "RAW-RESULT-SECRET" not in json.dumps(after)
|
|
|
|
|
|
def test_downstream_exception_is_reported_then_re_raised(monkeypatch):
|
|
instance = make_runtime()
|
|
after = []
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(
|
|
runtime_module,
|
|
"before_tool",
|
|
lambda *args, **kwargs: SimpleNamespace(proceed=True, reason="released", approval_id="apr_1"),
|
|
)
|
|
monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: after.append(args))
|
|
monkeypatch.setattr(runtime_module, "record_spend", lambda *args, **kwargs: None)
|
|
|
|
def explode(_args):
|
|
raise RuntimeError("RAW-EXECUTION-SECRET")
|
|
|
|
with pytest.raises(RuntimeError, match="RAW-EXECUTION-SECRET"):
|
|
instance.tool_execution(tool_name="terminal", args=ARGS, next_call=explode, **middleware_metadata())
|
|
assert after[0][4] is False and after[0][5] == 0
|
|
|
|
|
|
def test_after_telemetry_failure_never_changes_tool_result(monkeypatch):
|
|
instance = make_runtime(False)
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: {})
|
|
monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
|
|
assert instance.tool_execution(
|
|
tool_name="read_file", args={}, next_call=lambda _args: "result", **middleware_metadata()
|
|
) == "result"
|
|
|
|
|
|
def test_run_start_is_single_fire_under_concurrency(monkeypatch):
|
|
instance = make_runtime(False)
|
|
events = []
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append(args[2]))
|
|
call = runtime_module.CallContext(
|
|
"conv_" + "c" * 32,
|
|
"run_" + "d" * 32,
|
|
"call",
|
|
"session-1",
|
|
"ses_" + "b" * 32,
|
|
"prj_" + "a" * 32,
|
|
)
|
|
threads = [threading.Thread(target=instance._start_once, args=(call,)) for _ in range(20)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
assert events == ["run.started"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("result", "ok"),
|
|
[
|
|
("plain", True),
|
|
("{not json", True),
|
|
({"status": "ok"}, True),
|
|
({"error": "no"}, False),
|
|
(json.dumps({"status": "cancelled"}), False),
|
|
({"status": "failed"}, False),
|
|
],
|
|
)
|
|
def test_result_status_detection(result, ok):
|
|
assert runtime_module._result_ok(result) is ok
|
|
|
|
|
|
def test_result_size_is_bounded_and_tolerates_bad_values(monkeypatch):
|
|
class Bad:
|
|
def __str__(self):
|
|
raise ValueError
|
|
|
|
assert runtime_module._result_size(b"abc") == 3
|
|
assert runtime_module._result_size("é") == 2
|
|
assert runtime_module._result_size(Bad()) == 0
|
|
monkeypatch.setattr(runtime_module, "MAX_REPORTED_BYTES", 2)
|
|
assert runtime_module._result_size(b"xxx") == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("completed", "interrupted", "kind"),
|
|
[(True, False, "run.completed"), (True, True, "run.failed"), (False, False, "run.failed")],
|
|
)
|
|
def test_session_end_never_claims_unverified_cancellation(monkeypatch, completed, interrupted, kind):
|
|
instance = make_runtime(False)
|
|
events = []
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: events.append((args, kwargs)))
|
|
instance.session_end(completed=completed, interrupted=interrupted, **middleware_metadata())
|
|
assert events[0][0][2] == kind
|
|
assert "cancel" not in events[0][0][3].lower() or "without a verified cancellation" in events[0][0][3]
|
|
|
|
|
|
def test_session_end_and_after_ignore_telemetry_errors(monkeypatch):
|
|
instance = make_runtime(False)
|
|
monkeypatch.setattr(runtime_module, "emit", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
|
|
instance.session_end(session_id="bad session", turn_id="bad turn")
|
|
monkeypatch.setattr(runtime_module, "after_tool", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError()))
|
|
call = runtime_module.CallContext(
|
|
"conv_" + "c" * 32,
|
|
"run_" + "d" * 32,
|
|
"call",
|
|
"session-1",
|
|
"ses_" + "b" * 32,
|
|
"prj_" + "a" * 32,
|
|
)
|
|
instance._after(call, "read_file", "sha256:" + "e" * 64, True, 1, 0.0)
|
|
|
|
|
|
def test_plugin_files_stay_small_and_stop_bridge_is_honest():
|
|
for path in PLUGIN.glob("*.py"):
|
|
assert len(path.read_text().splitlines()) <= 500
|
|
source = (PLUGIN / "runtime.py").read_text()
|
|
notes = (PLUGIN / "NOTES.md").read_text()
|
|
assert "on_stop(" not in source
|
|
assert "process_registry_empty" in notes
|
|
assert "HUX_SUBJECT" not in source.replace("HUX_SUBJECT_FILE", "")
|