teleo-infrastructure/tests/test_leo_oos_readonly_guard.py

140 lines
4.2 KiB
Python

"""Fail-closed tests for the live OOS Hermes tool boundary."""
from __future__ import annotations
import json
import subprocess
import sys
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_wrapper(tmp_path: Path) -> tuple[Path, Path]:
profile = tmp_path / "profile"
wrapper = profile / "bin" / "teleo-kb"
wrapper.parent.mkdir(parents=True)
wrapper.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
wrapper.chmod(0o700)
return profile, wrapper
@pytest.mark.parametrize(
"command",
[
"teleo-kb status --format json",
"teleo-kb search 'Orchid forecast' --limit 5 --context-limit 4 --format json",
"teleo-kb list-proposals --status approved --limit 10 --format markdown",
"teleo-kb show 11111111-1111-4111-8111-111111111111 --format json",
],
)
def test_guard_accepts_only_exact_read_only_bridge_argv(tmp_path: Path, command: str) -> None:
profile, wrapper = make_wrapper(tmp_path)
argv, reason = guard.classify_terminal_command(
wrapper,
profile,
{"command": command},
allow_kb_reads=True,
)
assert argv is not None
assert reason is None
def test_guard_accepts_unquoted_multiword_read_query_as_one_safe_invocation(tmp_path: Path) -> None:
profile, wrapper = make_wrapper(tmp_path)
argv, reason = guard.classify_terminal_command(
wrapper,
profile,
{"command": "teleo-kb context Atlas status snapshot --limit 1 --context-limit 1 --format markdown"},
allow_kb_reads=True,
)
assert argv == [
"teleo-kb",
"context",
"Atlas",
"status",
"snapshot",
"--limit",
"1",
"--context-limit",
"1",
"--format",
"markdown",
]
assert reason is None
@pytest.mark.parametrize(
"tool_args",
[
{"command": "teleo-kb propose-source /tmp/input"},
{"command": "teleo-kb record-document-evaluation /tmp/input"},
{"command": "teleo-kb status; rm -rf /tmp/x"},
{"command": "bash -lc 'teleo-kb status'"},
{"command": "teleo-kb status", "background": True},
{"command": "teleo-kb status", "pty": True},
{"command": "teleo-kb status", "workdir": "/tmp"},
{"command": "teleo-kb evidence not-a-uuid"},
{"command": "teleo-kb search x --limit 1000"},
],
)
def test_guard_rejects_writes_shell_escape_and_unbounded_arguments(
tmp_path: Path, tool_args: dict[str, object]
) -> None:
profile, wrapper = make_wrapper(tmp_path)
argv, reason = guard.classify_terminal_command(
wrapper,
profile,
tool_args,
allow_kb_reads=True,
)
assert argv is None
assert reason
def test_no_db_ablation_preserves_terminal_schema_but_denies_execution(tmp_path: Path) -> None:
profile, wrapper = make_wrapper(tmp_path)
argv, reason = guard.classify_terminal_command(
wrapper,
profile,
{"command": "teleo-kb status --format json"},
allow_kb_reads=False,
)
assert argv is None
assert reason == "database tools are disabled for the no-DB ablation"
def test_terminal_child_uses_temp_profile_and_no_provider_credentials(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
profile, wrapper = make_wrapper(tmp_path)
captured: dict[str, object] = {}
def fake_run(argv, **kwargs):
captured["argv"] = argv
captured["env"] = kwargs["env"]
return subprocess.CompletedProcess(argv, 0, stdout="{}\n", stderr="")
monkeypatch.setattr(guard.subprocess, "run", fake_run)
output = json.loads(
guard.restricted_bridge_terminal_handler(
wrapper,
profile,
{"command": "teleo-kb status --format json"},
allow_kb_reads=True,
)
)
assert output["exit_code"] == 0
assert captured["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 captured["env"])