367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""Fail-closed tests for the live OOS Hermes tool boundary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
|
|
import leo_oos_readonly_guard as guard # noqa: E402
|
|
|
|
|
|
def make_kb_tool(tmp_path: Path) -> tuple[Path, Path, str]:
|
|
profile = tmp_path / "profile"
|
|
kb_tool = profile / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("#!/usr/bin/env python3\n", encoding="utf-8")
|
|
kb_tool.chmod(0o700)
|
|
return profile, kb_tool.resolve(), hashlib.sha256(kb_tool.read_bytes()).hexdigest()
|
|
|
|
|
|
def test_structured_context_compiles_to_one_exact_read_only_argv() -> None:
|
|
argv = guard.structured_read_argv(
|
|
{
|
|
"action": "context",
|
|
"query": "Northstar status snapshot",
|
|
"limit": 1,
|
|
"context_limit": 1,
|
|
"format": "markdown",
|
|
}
|
|
)
|
|
|
|
assert argv == [
|
|
"context",
|
|
"Northstar status snapshot",
|
|
"--limit",
|
|
"1",
|
|
"--context-limit",
|
|
"1",
|
|
"--format",
|
|
"markdown",
|
|
]
|
|
|
|
|
|
def test_structured_read_omits_unsupplied_flags_so_bridge_defaults_remain_canonical() -> None:
|
|
assert guard.structured_read_argv({"action": "evidence", "claim_id": "11111111-1111-4111-8111-111111111111"}) == [
|
|
"evidence",
|
|
"11111111-1111-4111-8111-111111111111",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tool_args",
|
|
[
|
|
{"action": "prepare-source", "query": "x"},
|
|
{"action": "context", "query": "x", "rationale": "write it"},
|
|
{"action": "context"},
|
|
{"action": "context", "query": "x", "limit": 0},
|
|
{"action": "context", "query": "x", "context_limit": True},
|
|
{"action": "context", "query": "x", "format": []},
|
|
{"action": "show", "claim_id": "not-a-uuid"},
|
|
{"action": "list-proposals", "status": []},
|
|
{"action": "status", "query": "irrelevant"},
|
|
],
|
|
)
|
|
def test_structured_read_rejects_writes_bad_types_irrelevant_fields_and_unbounded_values(
|
|
tool_args: dict[str, object],
|
|
) -> None:
|
|
with pytest.raises(guard.ReadOnlyGuardError):
|
|
guard.structured_read_argv(tool_args)
|
|
|
|
|
|
def test_schema_is_provider_compatible_while_handler_enforces_action_contracts() -> None:
|
|
parameters = guard.STRUCTURED_READ_TOOL_SCHEMA["parameters"]
|
|
|
|
assert parameters["required"] == ["action"]
|
|
assert parameters["additionalProperties"] is False
|
|
assert not ({"oneOf", "anyOf", "allOf"} & set(parameters))
|
|
assert set(parameters["properties"]["action"]["enum"]) == set(guard.READ_ONLY_BRIDGE_COMMANDS)
|
|
assert "query is required" in parameters["properties"]["action"]["description"]
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self, argv, **kwargs):
|
|
self.argv = argv
|
|
self.kwargs = kwargs
|
|
self.pid = 8123
|
|
self.returncode = 0
|
|
|
|
def communicate(self, timeout=None):
|
|
return "{}\n", ""
|
|
|
|
|
|
def test_structured_read_child_uses_frozen_absolute_tool_and_no_provider_credentials(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_popen(argv, **kwargs):
|
|
proc = FakeProcess(argv, **kwargs)
|
|
captured["proc"] = proc
|
|
return proc
|
|
|
|
monkeypatch.setattr(guard.subprocess, "Popen", fake_popen)
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{
|
|
"action": "context",
|
|
"query": "Northstar status snapshot",
|
|
"limit": 1,
|
|
"context_limit": 1,
|
|
"format": "markdown",
|
|
},
|
|
allow_kb_reads=True,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output == {"output": "{}\n", "exit_code": 0, "error": None, "error_category": None}
|
|
proc = captured["proc"]
|
|
assert proc.argv == [
|
|
str(Path(sys.executable).resolve()),
|
|
str(kb_tool),
|
|
"--local",
|
|
"context",
|
|
"Northstar status snapshot",
|
|
"--limit",
|
|
"1",
|
|
"--context-limit",
|
|
"1",
|
|
"--format",
|
|
"markdown",
|
|
]
|
|
assert proc.kwargs["start_new_session"] is True
|
|
assert proc.kwargs["env"] == {
|
|
"HOME": str(profile),
|
|
"HERMES_HOME": str(profile),
|
|
"PATH": f"{profile / 'bin'}:{guard.os.environ.get('PATH', '')}",
|
|
"TELEO_KB_MODE": "local",
|
|
}
|
|
assert not any("KEY" in key or "TOKEN" in key or "SECRET" in key for key in proc.kwargs["env"])
|
|
|
|
|
|
def test_structured_read_rejects_hash_drift_before_launch(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
kb_tool.write_text("changed\n", encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
guard.subprocess,
|
|
"Popen",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("must not launch")),
|
|
)
|
|
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{"action": "status"},
|
|
allow_kb_reads=True,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output["exit_code"] == 126
|
|
assert output["error_category"] == "invalid_read_arguments"
|
|
|
|
|
|
def test_structured_read_timeout_terminates_the_process_group(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
calls = 0
|
|
signals: list[tuple[int, signal.Signals]] = []
|
|
|
|
class TimeoutProcess(FakeProcess):
|
|
def communicate(self, timeout=None):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
raise subprocess.TimeoutExpired(self.argv, timeout)
|
|
return "", ""
|
|
|
|
monkeypatch.setattr(guard.subprocess, "Popen", TimeoutProcess)
|
|
monkeypatch.setattr(guard.os, "killpg", lambda pid, sent: signals.append((pid, sent)))
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{"action": "status"},
|
|
allow_kb_reads=True,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output["exit_code"] == 124
|
|
assert output["error_category"] == "bridge_timeout"
|
|
assert signals == [(8123, signal.SIGTERM)]
|
|
|
|
|
|
def test_structured_read_timeout_escalates_to_process_group_kill(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
signals: list[tuple[int, signal.Signals]] = []
|
|
|
|
class StubbornProcess(FakeProcess):
|
|
def communicate(self, timeout=None):
|
|
raise subprocess.TimeoutExpired(self.argv, timeout)
|
|
|
|
monkeypatch.setattr(guard.subprocess, "Popen", StubbornProcess)
|
|
monkeypatch.setattr(guard.os, "killpg", lambda pid, sent: signals.append((pid, sent)))
|
|
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{"action": "status"},
|
|
allow_kb_reads=True,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output["exit_code"] == 124
|
|
assert output["error_category"] == "bridge_timeout"
|
|
assert signals == [(8123, signal.SIGTERM), (8123, signal.SIGKILL)]
|
|
|
|
|
|
def test_structured_read_launch_error_is_categorized(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
monkeypatch.setattr(guard.subprocess, "Popen", lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("nope")))
|
|
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{"action": "status"},
|
|
allow_kb_reads=True,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output["exit_code"] == 126
|
|
assert output["error_category"] == "bridge_launch"
|
|
assert "nope" not in output["error"]
|
|
|
|
|
|
def test_structured_read_ablation_preserves_schema_but_never_executes(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
profile, kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
monkeypatch.setattr(
|
|
guard.subprocess,
|
|
"Popen",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("ablation must not execute")),
|
|
)
|
|
|
|
output = json.loads(
|
|
guard.restricted_structured_read_handler(
|
|
kb_tool,
|
|
profile,
|
|
{"action": "status", "format": "json"},
|
|
allow_kb_reads=False,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
)
|
|
|
|
assert output["exit_code"] == 126
|
|
assert output["error_category"] == "kb_reads_disabled"
|
|
|
|
|
|
class FakeEntry:
|
|
def __init__(self, handler=None, schema=None):
|
|
self.handler = handler
|
|
self.schema = schema or {}
|
|
|
|
|
|
class FakeRegistry:
|
|
def __init__(self):
|
|
self._tools = {"skills_list": FakeEntry(), "skill_view": FakeEntry(), "terminal": FakeEntry()}
|
|
|
|
def register(self, *, name, schema, handler, **_kwargs):
|
|
self._tools[name] = FakeEntry(handler=handler, schema=schema)
|
|
|
|
def get_definitions(self, names, quiet=False):
|
|
del quiet
|
|
return [
|
|
{"type": "function", "function": {**self._tools[name].schema, "name": name}}
|
|
for name in sorted(names)
|
|
if name in self._tools
|
|
]
|
|
|
|
def dispatch(self, name, args):
|
|
return self._tools[name].handler(args)
|
|
|
|
|
|
def test_registry_install_exposes_exact_schema_and_denies_ablation(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
profile, _kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
registry = FakeRegistry()
|
|
toolsets = types.ModuleType("toolsets")
|
|
toolsets.TOOLSETS = {}
|
|
tools = types.ModuleType("tools")
|
|
tools.__path__ = []
|
|
skills_tool = types.ModuleType("tools.skills_tool")
|
|
registry_module = types.ModuleType("tools.registry")
|
|
registry_module.registry = registry
|
|
hermes_cli = types.ModuleType("hermes_cli")
|
|
tools_config = types.SimpleNamespace(_get_platform_tools=lambda: set())
|
|
hermes_cli.tools_config = tools_config
|
|
for name, module in {
|
|
"tools": tools,
|
|
"tools.skills_tool": skills_tool,
|
|
"tools.registry": registry_module,
|
|
"toolsets": toolsets,
|
|
"hermes_cli": hermes_cli,
|
|
}.items():
|
|
monkeypatch.setitem(sys.modules, name, module)
|
|
|
|
surface = guard.install_read_only_tool_surface(
|
|
profile,
|
|
allow_kb_reads=False,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
)
|
|
denied = json.loads(registry.dispatch("teleo-kb", {"action": "status"}))
|
|
|
|
assert surface["actual_registry_tools"] == ["skill_view", "skills_list", "teleo-kb"]
|
|
assert surface["shell_tool_enabled"] is False
|
|
assert len(surface["structured_read_tool_definition_sha256"]) == 64
|
|
assert denied["error_category"] == "kb_reads_disabled"
|
|
|
|
registry._tools["process"] = FakeEntry()
|
|
post_runner = guard.reassert_read_only_tool_surface(
|
|
profile,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
expected_definition_sha256=surface["structured_read_tool_definition_sha256"],
|
|
)
|
|
|
|
assert post_runner["initial_registry_tools"] == ["process", "skill_view", "skills_list", "teleo-kb"]
|
|
assert post_runner["removed_known_runner_tools"] == ["process"]
|
|
assert post_runner["actual_registry_tools"] == ["skill_view", "skills_list", "teleo-kb"]
|
|
assert post_runner["final_surface_matches_preexecution"] is True
|
|
|
|
|
|
def test_post_runner_registry_rejects_unknown_additions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
profile, _kb_tool, source_sha256 = make_kb_tool(tmp_path)
|
|
registry = FakeRegistry()
|
|
registry._tools["teleo-kb"] = FakeEntry(schema=guard.STRUCTURED_READ_TOOL_SCHEMA)
|
|
registry._tools["unknown-network-tool"] = FakeEntry()
|
|
registry_module = types.ModuleType("tools.registry")
|
|
registry_module.registry = registry
|
|
monkeypatch.setitem(sys.modules, "tools.registry", registry_module)
|
|
|
|
with pytest.raises(guard.ReadOnlyGuardError, match="unsupported additions"):
|
|
guard.reassert_read_only_tool_surface(
|
|
profile,
|
|
expected_kb_tool_sha256=source_sha256,
|
|
expected_definition_sha256="0" * 64,
|
|
)
|