226 lines
8 KiB
Python
226 lines
8 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail-closed Hermes tool surface for live no-post Leo OOS trials."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
READ_ONLY_BRIDGE_COMMANDS = frozenset(
|
|
{
|
|
"search",
|
|
"context",
|
|
"show",
|
|
"evidence",
|
|
"edges",
|
|
"list-proposals",
|
|
"search-proposals",
|
|
"show-proposal",
|
|
"status",
|
|
"decision-matrix-status",
|
|
}
|
|
)
|
|
HARMLESS_TRAILING_REDIRECT_TOKENS = frozenset({"2>/dev/null", "2> /dev/null"})
|
|
|
|
|
|
class ReadOnlyGuardError(RuntimeError):
|
|
"""The requested command cannot be proven read-only."""
|
|
|
|
|
|
class GuardArgumentParser(argparse.ArgumentParser):
|
|
def error(self, message: str) -> None:
|
|
raise ReadOnlyGuardError(message)
|
|
|
|
def exit(self, status: int = 0, message: str | None = None) -> None:
|
|
raise ReadOnlyGuardError(message or f"argument parser exited with status {status}")
|
|
|
|
|
|
def _uuid_argument(value: str) -> str:
|
|
try:
|
|
return str(uuid.UUID(value))
|
|
except ValueError as exc:
|
|
raise argparse.ArgumentTypeError("expected a UUID") from exc
|
|
|
|
|
|
def validate_read_only_bridge_argv(argv: list[str]) -> None:
|
|
"""Accept only the bridge's bounded read commands and exact arguments."""
|
|
|
|
parser = GuardArgumentParser(add_help=False)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
def command(name: str) -> argparse.ArgumentParser:
|
|
item = subparsers.add_parser(name, add_help=False)
|
|
item.add_argument("--help", action="store_true")
|
|
item.add_argument("--format", choices=("markdown", "json"), default="markdown")
|
|
return item
|
|
|
|
for name in ("search", "context"):
|
|
item = command(name)
|
|
item.add_argument("query")
|
|
item.add_argument("--limit", type=int, default=5)
|
|
item.add_argument("--context-limit", type=int, default=5 if name == "search" else 6)
|
|
show = command("show")
|
|
show.add_argument("claim_id", type=_uuid_argument)
|
|
for name in ("evidence", "edges"):
|
|
item = command(name)
|
|
item.add_argument("claim_id", type=_uuid_argument)
|
|
item.add_argument("--limit", type=int, default=12)
|
|
list_proposals = command("list-proposals")
|
|
list_proposals.add_argument(
|
|
"--status",
|
|
choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
|
|
default="pending_review",
|
|
)
|
|
list_proposals.add_argument("--limit", type=int, default=10)
|
|
search_proposals = command("search-proposals")
|
|
search_proposals.add_argument("query")
|
|
search_proposals.add_argument(
|
|
"--status",
|
|
choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
|
|
default="all",
|
|
)
|
|
search_proposals.add_argument("--limit", type=int, default=10)
|
|
show_proposal = command("show-proposal")
|
|
show_proposal.add_argument("proposal_id", type=_uuid_argument)
|
|
command("status")
|
|
command("decision-matrix-status")
|
|
parsed = parser.parse_args(argv)
|
|
for name, value in vars(parsed).items():
|
|
if name.endswith("limit") and not 1 <= value <= 100:
|
|
raise ReadOnlyGuardError(f"{name.replace('_', '-')} must be between 1 and 100")
|
|
|
|
|
|
def classify_terminal_command(
|
|
wrapper: Path,
|
|
temp_profile: Path,
|
|
tool_args: dict[str, Any],
|
|
*,
|
|
allow_kb_reads: bool,
|
|
) -> tuple[list[str] | None, str | None]:
|
|
"""Return verified argv or a fail-closed reason without executing anything."""
|
|
|
|
command = str(tool_args.get("command") or "")
|
|
if len(command) > 8192:
|
|
return None, "command is too long"
|
|
if tool_args.get("background") or tool_args.get("pty") or tool_args.get("workdir"):
|
|
return None, "background, PTY, and workdir overrides are disabled"
|
|
if not allow_kb_reads:
|
|
return None, "database tools are disabled for the no-DB ablation"
|
|
try:
|
|
argv = shlex.split(command)
|
|
except ValueError as exc:
|
|
return None, f"invalid command quoting: {exc}"
|
|
while argv and argv[-1] in HARMLESS_TRAILING_REDIRECT_TOKENS:
|
|
argv.pop()
|
|
command_path = f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}"
|
|
try:
|
|
exact_wrapper = wrapper.resolve(strict=True)
|
|
executable = shutil.which(argv[0], path=command_path) if argv else None
|
|
invoked = Path(executable).resolve(strict=True) if executable else None
|
|
except OSError:
|
|
invoked = None
|
|
exact_wrapper = wrapper.resolve(strict=False)
|
|
if invoked != exact_wrapper:
|
|
return None, "only the exact temporary-profile teleo-kb wrapper is available"
|
|
if len(argv) < 2 or argv[1] not in READ_ONLY_BRIDGE_COMMANDS:
|
|
return None, "only read-only Teleo KB bridge commands are available"
|
|
try:
|
|
validate_read_only_bridge_argv(argv[1:])
|
|
except ReadOnlyGuardError as exc:
|
|
return None, f"invalid read-only KB arguments: {exc}"
|
|
return argv, None
|
|
|
|
|
|
def restricted_bridge_terminal_handler(
|
|
wrapper: Path,
|
|
temp_profile: Path,
|
|
tool_args: dict[str, Any],
|
|
*,
|
|
allow_kb_reads: bool,
|
|
**_kwargs: Any,
|
|
) -> str:
|
|
argv, reason = classify_terminal_command(
|
|
wrapper,
|
|
temp_profile,
|
|
tool_args,
|
|
allow_kb_reads=allow_kb_reads,
|
|
)
|
|
if argv is None:
|
|
return json.dumps({"output": "", "exit_code": 126, "error": reason})
|
|
timeout = min(max(int(tool_args.get("timeout") or 60), 1), 120)
|
|
child_environment = {
|
|
"HOME": str(temp_profile),
|
|
"HERMES_HOME": str(temp_profile),
|
|
"PATH": f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}",
|
|
"TELEO_KB_MODE": "local",
|
|
}
|
|
try:
|
|
proc = subprocess.run(
|
|
argv,
|
|
cwd=temp_profile,
|
|
env=child_environment,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=timeout,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return json.dumps({"output": "", "exit_code": 124, "error": f"bridge command exceeded {timeout}s"})
|
|
return json.dumps(
|
|
{
|
|
"output": proc.stdout,
|
|
"exit_code": proc.returncode,
|
|
"error": proc.stderr or None,
|
|
}
|
|
)
|
|
|
|
|
|
def install_read_only_tool_surface(temp_profile: Path, *, allow_kb_reads: bool) -> dict[str, Any]:
|
|
"""Replace Hermes' registry with skills plus one guarded terminal tool."""
|
|
|
|
import tools.skills_tool
|
|
import tools.terminal_tool # noqa: F401
|
|
import toolsets
|
|
from hermes_cli import tools_config
|
|
from tools.registry import registry
|
|
|
|
toolset_name = "leo-oos-kb-readonly" if allow_kb_reads else "leo-oos-no-db-ablation"
|
|
allowed_tools = ["skills_list", "skill_view", "terminal"]
|
|
toolsets.TOOLSETS[toolset_name] = {
|
|
"description": "Fail-closed Leo OOS tool surface",
|
|
"tools": allowed_tools,
|
|
"includes": [],
|
|
}
|
|
tools_config._get_platform_tools = lambda *_args, **_kwargs: {toolset_name}
|
|
missing = sorted(name for name in allowed_tools if name not in registry._tools)
|
|
if missing:
|
|
raise ReadOnlyGuardError(f"required tools are not registered: {missing}")
|
|
allowed_entries = {name: registry._tools[name] for name in allowed_tools}
|
|
registry._tools.clear()
|
|
registry._tools.update(allowed_entries)
|
|
wrapper = temp_profile / "bin" / "teleo-kb"
|
|
terminal_entry = registry._tools["terminal"]
|
|
terminal_entry.handler = lambda tool_args, **kwargs: restricted_bridge_terminal_handler(
|
|
wrapper,
|
|
temp_profile,
|
|
tool_args,
|
|
allow_kb_reads=allow_kb_reads,
|
|
**kwargs,
|
|
)
|
|
return {
|
|
"mode": "read_only_kb" if allow_kb_reads else "no_db_ablation",
|
|
"allowed_tools": allowed_tools,
|
|
"actual_registry_tools": sorted(registry._tools),
|
|
"read_only_bridge_commands": sorted(READ_ONLY_BRIDGE_COMMANDS) if allow_kb_reads else [],
|
|
"send_message_tool_enabled": "send_message" in registry._tools,
|
|
"terminal_restricted_to_exact_wrapper": True,
|
|
"mutating_bridge_commands_exposed": False,
|
|
"provider_credentials_forwarded_to_terminal": False,
|
|
}
|