326 lines
16 KiB
Python
326 lines
16 KiB
Python
"""Private harness for test_secret_redaction*.py: scenario table, scripted model, sink collection.
|
|
|
|
Secrets are allocated so that every scenario owns disjoint values: the values a scenario leaves raw
|
|
BY DESIGN (a tool argument, the user's own prompt) never coincide with the values another scenario
|
|
asserts are masked, so one Hermes home can serve every scenario and each sink is scanned once.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import secrets
|
|
import string
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Mapping
|
|
|
|
import pytest
|
|
|
|
from tests.e2e.core.delivery._fake_platform import GatewayProcess, read_jsonl, wait_until
|
|
from tests.e2e.core.security._helpers import REPO_ROOT, BoundaryBreach, db_blob, files_containing, run_hermes
|
|
from tests.fakes.fake_llm_provider import Error, Response, Text, ToolCall
|
|
|
|
|
|
def _alnum(n: int) -> str:
|
|
return "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(n))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Secrets:
|
|
"""Every secret value one Hermes home sees; see the module docstring for the allocation."""
|
|
|
|
provider: str = field(default_factory=lambda: "sk-proj-" + _alnum(40)) # .env OPENAI_API_KEY
|
|
env_opaque: str = field(default_factory=lambda: secrets.token_hex(20)) # .env ACME_SERVICE_TOKEN
|
|
env_github: str = field(default_factory=lambda: "ghp_" + _alnum(36)) # .env GITHUB_TOKEN
|
|
echo_github: str = field(default_factory=lambda: "ghp_" + _alnum(36)) # only in a workspace file
|
|
echo_slack: str = field(default_factory=lambda: f"xoxb-{secrets.randbelow(10**12):012d}-{_alnum(24)}")
|
|
echo_openai: str = field(default_factory=lambda: "sk-" + _alnum(40))
|
|
echo_bearer: str = field(default_factory=lambda: _alnum(40)) # opaque, no vendor prefix
|
|
user_key: str = field(default_factory=lambda: "sk-" + _alnum(44)) # pasted by the user
|
|
nonce: str = field(default_factory=lambda: "nonce" + secrets.token_hex(6))
|
|
|
|
def env(self) -> dict[str, str]:
|
|
return {"ACME_SERVICE_TOKEN": self.env_opaque, "GITHUB_TOKEN": self.env_github}
|
|
|
|
|
|
# The user explicitly runs without approval prompts, so the credential-in-command scan does not block the
|
|
# tool-argument scenario before the command executes (that guard is covered by the approvals lane).
|
|
CONFIG = "approvals:\n mode: \"off\"\n"
|
|
|
|
|
|
def core(value: str) -> str:
|
|
"""The part of a secret no mask may keep: a head/tail mask (``sk-pro...AbCd``) is not a leak."""
|
|
return value[6:-4]
|
|
|
|
|
|
# Sinks a scenario is held to; the table below documents which ones the contract covers.
|
|
LOGS, STORE, EXPORT, EXPORT_REDACTED, WIRE, PLATFORM = (
|
|
"logs", "store", "export", "export --redact", "next provider request", "platform wire")
|
|
CONTENT_SINKS = (LOGS, STORE, EXPORT, EXPORT_REDACTED, WIRE, PLATFORM)
|
|
# The session database keeps tool arguments and the user's own prompt as executed (docs:
|
|
# user-guide/security.md "The session database itself still holds the command as it was executed");
|
|
# the provider replay of the model's own tool call and the default (non --redact) export read it back.
|
|
RAW_BY_DESIGN_SINKS = (LOGS, EXPORT_REDACTED, PLATFORM)
|
|
# Test-id spelling of each sink: one cell per (scenario, sink), so a KNOWN names exactly one sink.
|
|
SINK_IDS = {LOGS: "logs", STORE: "store", EXPORT: "export", EXPORT_REDACTED: "export_redact", WIRE: "next_request",
|
|
PLATFORM: "platform"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Scenario:
|
|
name: str
|
|
secrets: Callable[[Secrets], list[str]]
|
|
script: Callable[["Ctx"], list[Response]]
|
|
sinks: tuple[str, ...]
|
|
prompt: str = "please run the task"
|
|
followup: bool = False # a second user turn, so the NEXT request after the answer exists
|
|
|
|
|
|
@dataclass
|
|
class Ctx:
|
|
"""What a scenario script may reference: the secrets, the workspace, the provider port, the .env."""
|
|
|
|
keys: Secrets
|
|
ws: Path
|
|
env_file: Path
|
|
port: int = 0
|
|
|
|
|
|
def _tee(c: Ctx, name: str, src: "str | Path") -> str:
|
|
return f"cat {src} | tee {c.ws / name}"
|
|
|
|
|
|
SCENARIOS: dict[str, Scenario] = {s.name: s for s in (
|
|
# (a) the terminal tool prints secret-shaped strings that are NOT in the env
|
|
Scenario("tool_echoes_prefixed", lambda k: [k.echo_github, k.echo_slack, k.echo_openai, k.echo_bearer],
|
|
lambda c: [ToolCall("terminal", {"command": _tee(c, "echoed_a.txt", c.ws / "fixture.txt")}),
|
|
Text("printed the fixture")], CONTENT_SINKS),
|
|
# (b) the tool prints the profile's real .env (provider key + an opaque token) and a plain file with the key
|
|
Scenario("tool_reads_env_file", lambda k: [k.provider, k.env_opaque],
|
|
lambda c: [ToolCall("terminal", {"command": _tee(c, "echoed_b.txt", c.env_file)}),
|
|
ToolCall("terminal", {"command": _tee(c, "echoed_b2.txt", c.ws / "notes.txt")}),
|
|
Text("read both files")], CONTENT_SINKS),
|
|
# (c) the model's own answer contains the provider key
|
|
Scenario("assistant_text", lambda k: [k.provider],
|
|
lambda c: [Text(f"Your key is {c.keys.provider} - keep it safe.")], CONTENT_SINKS,
|
|
followup=True),
|
|
# (d) the key is a tool ARGUMENT (curl -H 'Authorization: Bearer KEY') to a local endpoint
|
|
Scenario("tool_arg", lambda k: [k.env_github],
|
|
lambda c: [ToolCall("terminal", {"command": (
|
|
f"curl -s -o {c.ws / 'models.json'} -H 'Authorization: Bearer {c.keys.env_github}' "
|
|
f"http://127.0.0.1:{c.port}/v1/models")}), Text("called the endpoint")], RAW_BY_DESIGN_SINKS),
|
|
# (e) the user pastes a key into the prompt (the turn preview is logged)
|
|
Scenario("user_prompt", lambda k: [k.user_key], lambda c: [Text("noted")], RAW_BY_DESIGN_SINKS),
|
|
# (f) the provider echoes the request's key back in an error body
|
|
Scenario("provider_error_echo", lambda k: [k.provider],
|
|
lambda c: [Error(401, f"Incorrect API key provided: {c.keys.provider} ({c.keys.nonce})")],
|
|
(LOGS, STORE, EXPORT, EXPORT_REDACTED, PLATFORM)),
|
|
)}
|
|
|
|
|
|
def prompt_for(name: str, k: Secrets, *, followup: bool = False) -> str:
|
|
tag = f"[scn:{name}{':followup' if followup else ''}]"
|
|
if name == "user_prompt":
|
|
return f"{tag} {k.nonce} my key is {k.user_key}"
|
|
return f"{tag} {'thanks, anything else?' if followup else SCENARIOS[name].prompt}"
|
|
|
|
|
|
_TAG_RE = re.compile(r"\[scn:([a-z_]+)(:followup)?\]")
|
|
|
|
|
|
def _text(msg: dict) -> str:
|
|
content = msg.get("content")
|
|
if isinstance(content, list):
|
|
return " ".join(p.get("text", "") for p in content if isinstance(p, dict))
|
|
return content or ""
|
|
|
|
|
|
class Director:
|
|
"""Scripted model: the newest user message's ``[scn:<name>]`` tag selects the script and the
|
|
number of assistant messages since it selects the step."""
|
|
|
|
def __init__(self, ctx: Ctx) -> None:
|
|
self.ctx = ctx
|
|
|
|
def __call__(self, record: dict) -> Response:
|
|
msgs = record["body"].get("messages", [])
|
|
last_user = max((i for i, m in enumerate(msgs) if m.get("role") == "user"), default=-1)
|
|
m = _TAG_RE.search(_text(msgs[last_user])) if last_user >= 0 else None
|
|
if m is None:
|
|
return Text("no scenario")
|
|
if m.group(2):
|
|
return Text(f"nothing else for {m.group(1)}")
|
|
step = sum(1 for x in msgs[last_user + 1:] if x.get("role") == "assistant")
|
|
script = SCENARIOS[m.group(1)].script(self.ctx)
|
|
return script[min(step, len(script) - 1)] if isinstance(script[-1], Error) or step < len(script) \
|
|
else Text(f"done {m.group(1)}")
|
|
|
|
|
|
def seed_workspace(ws: Path, k: Secrets) -> None:
|
|
ws.mkdir(parents=True, exist_ok=True)
|
|
(ws / "fixture.txt").write_text(
|
|
f"github {k.echo_github}\nslack {k.echo_slack}\nopenai {k.echo_openai}\n"
|
|
f"Authorization: Bearer {k.echo_bearer}\n", encoding="utf-8")
|
|
(ws / "notes.txt").write_text(f"the deploy key is {k.provider}\n", encoding="utf-8")
|
|
|
|
|
|
def echo_preconditions(ws: Path, k: Secrets, llm_gets: list[dict], logs: str) -> dict[str, list[str]]:
|
|
"""Per scenario: the evidence that the secret really reached the tool/log path (harness facts)."""
|
|
def has(name: str, *values: str) -> list[str]:
|
|
path = ws / name
|
|
text = path.read_text(encoding="utf-8") if path.exists() else ""
|
|
return [f"{name} lacks {v[:8]}…" for v in values if v not in text]
|
|
|
|
auths = {g.get("auth") for g in llm_gets}
|
|
return {
|
|
"tool_echoes_prefixed": has("echoed_a.txt", k.echo_github, k.echo_slack, k.echo_openai, k.echo_bearer),
|
|
"tool_reads_env_file": has("echoed_b.txt", k.provider, k.env_opaque) + has("echoed_b2.txt", k.provider),
|
|
"assistant_text": [],
|
|
"tool_arg": [] if f"Bearer {k.env_github}" in auths else ["curl never reached the endpoint with the key"],
|
|
"user_prompt": [] if k.nonce in logs else ["the turn preview never reached the logs"],
|
|
"provider_error_echo": [] if k.nonce in logs else ["the provider error never reached the logs"],
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Sinks:
|
|
"""Raw text of every persisted/egress sink of one Hermes home, keyed by sink name."""
|
|
|
|
texts: dict[str, dict[str, str]] = field(default_factory=dict) # sink -> {location: text}
|
|
|
|
def hits(self, sink: str, needles: list[str], only: str = "") -> list[str]:
|
|
"""``only`` narrows a per-chat sink (the platform journal) to one scenario's chat."""
|
|
out = []
|
|
for loc, text in self.texts.get(sink, {}).items():
|
|
if only and loc != only and sink == PLATFORM:
|
|
continue
|
|
for n in needles:
|
|
at = text.find(core(n))
|
|
if at >= 0:
|
|
ctx = text[max(0, at - 80):at + len(n) + 20].replace(core(n), "<SECRET>")
|
|
out.append(f"{sink} [{loc}] holds {n[:10]}…: …{ctx!r}…")
|
|
return out
|
|
|
|
|
|
def collect(home: Path, requests: list[dict], platform_journal: Path | None = None) -> Sinks:
|
|
hh = home / ".hermes"
|
|
logs = {str(p.relative_to(hh / "logs")): p.read_text(encoding="utf-8", errors="replace")
|
|
for p in sorted((hh / "logs").rglob("*")) if p.is_file()}
|
|
store = {"state.db (decoded rows)": db_blob(hh / "state.db")}
|
|
for p in sorted(hh.rglob("*")): # raw bytes: WAL, FTS shadow pages, sessions/, caches
|
|
if p.is_file() and p.name != ".env" and "logs" not in p.relative_to(hh).parts:
|
|
store[str(p.relative_to(hh))] = p.read_bytes().decode("utf-8", "replace")
|
|
out = home / "exports"
|
|
out.mkdir(exist_ok=True)
|
|
exports = {}
|
|
for flag, name in (([], "plain.jsonl"), (["--redact"], "redacted.jsonl")):
|
|
r = run_hermes(["sessions", "export", str(out / name), *flag], home, timeout=90)
|
|
assert r.returncode == 0 and (out / name).exists(), f"sessions export failed: {r.stdout}\n{r.stderr}"
|
|
exports[name] = (out / name).read_text(encoding="utf-8")
|
|
wire = {f"request #{i} ({r['kind']})": json.dumps(r["body"]) for i, r in enumerate(requests)
|
|
if r["kind"] in ("main", "aux")}
|
|
platform: dict[str, str] = {} # every post AND every intermediate edit (streamed text), per chat
|
|
chat_of: dict[str, str] = {}
|
|
for rec in read_jsonl(platform_journal) if platform_journal is not None else []:
|
|
chat = chat_of.setdefault(rec["message_id"], str(rec.get("chat_id", "?")))
|
|
platform[f"chat {chat}"] = platform.get(f"chat {chat}", "") + json.dumps(rec) + "\n"
|
|
return Sinks({LOGS: logs, STORE: store, EXPORT: {"plain.jsonl": exports["plain.jsonl"]},
|
|
EXPORT_REDACTED: {"redacted.jsonl": exports["redacted.jsonl"]}, WIRE: wire, PLATFORM: platform})
|
|
|
|
|
|
def chat_for(name: str) -> str:
|
|
return f"c-{name}"
|
|
|
|
|
|
def assert_harness_sane(sinks: Sinks, *, gateway: bool = False) -> None:
|
|
"""A sink that is empty cannot prove absence: every scanned sink must actually exist."""
|
|
for sink in (LOGS, STORE, EXPORT, EXPORT_REDACTED, WIRE) + ((PLATFORM,) if gateway else ()):
|
|
assert any(t.strip() for t in sinks.texts.get(sink, {}).values()), f"sink {sink!r} is empty"
|
|
wanted = {"agent.log", "errors.log"} | ({"gateway.log"} if gateway else set())
|
|
assert wanted <= set(sinks.texts[LOGS]), sorted(sinks.texts[LOGS])
|
|
|
|
|
|
@dataclass
|
|
class World:
|
|
"""One Hermes home after every scenario ran: its sinks and each scenario's travel evidence."""
|
|
|
|
keys: Secrets
|
|
sinks: Sinks
|
|
pre: dict[str, list[str]]
|
|
runs: dict[str, str]
|
|
|
|
|
|
def cell_id(scenario: str, sink: str) -> str:
|
|
"""Test id and KNOWN key of one (scenario, sink) cell."""
|
|
return f"{scenario}-{SINK_IDS[sink]}"
|
|
|
|
|
|
def cells(known: Mapping[str, Any], *, platform: bool) -> list[Any]:
|
|
"""One ``pytest.param(scenario, sink)`` per sink a scenario is held to (``platform``: the surface has a
|
|
platform wire). A KNOWN key is a cell id (:func:`cell_id`) and gates ONLY that sink (``known_gate``), so
|
|
a new leak of the same scenario into any other sink is a plain red, never absorbed by the known gap."""
|
|
out, ids = [], set()
|
|
for name, scenario in SCENARIOS.items():
|
|
for sink in scenario.sinks:
|
|
if sink == PLATFORM and not platform:
|
|
continue
|
|
cid = cell_id(name, sink)
|
|
ids.add(cid)
|
|
out.append(pytest.param(name, sink, id=cid))
|
|
stale = sorted(set(known) - ids)
|
|
assert not stale, f"KNOWN names cells that do not exist: {stale}"
|
|
return out
|
|
|
|
|
|
def check(world: World, scenario: str, sink: str) -> None:
|
|
"""Harness precondition (plain AssertionError), then the boundary for ONE sink (BoundaryBreach)."""
|
|
missing = world.pre[scenario]
|
|
assert not missing, (f"{scenario}: the secret never travelled, so absence proves nothing: {missing}\n"
|
|
f"{world.runs[scenario]}")
|
|
leaks = world.sinks.hits(sink, SCENARIOS[scenario].secrets(world.keys), only=f"chat {chat_for(scenario)}")
|
|
if leaks:
|
|
raise BoundaryBreach(f"{scenario}: secret reached the redacted sink {sink!r}:\n " + "\n ".join(leaks))
|
|
|
|
|
|
# Gateway ------------------------------------------------------------------------------------------
|
|
|
|
|
|
class LoggingGateway(GatewayProcess):
|
|
"""The delivery suite's real GatewayRunner child, with the gateway's file logging installed the way
|
|
``start_gateway`` does (agent.log / errors.log / gateway.log under the child's HERMES_HOME)."""
|
|
|
|
def start(self) -> "LoggingGateway":
|
|
assert self.proc is None
|
|
self.boots += 1
|
|
ready_before = len(read_jsonl(self.spool / "ready.jsonl"))
|
|
with open(self.log, "wb") as log:
|
|
self.proc = subprocess.Popen(
|
|
[sys.executable, "-m", "tests.e2e.core.security._redact", "serve", str(self.spool)],
|
|
cwd=str(REPO_ROOT), env=self.env(), stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
|
|
self.pids.append(self.proc.pid)
|
|
pid = self.proc.pid
|
|
rec = wait_until(lambda: next((r for r in read_jsonl(self.spool / "ready.jsonl")[ready_before:]
|
|
if r["pid"] == pid), None),
|
|
"gateway child ready", timeout=240, proc=self.proc, log=self.log)
|
|
import socket
|
|
self.sock = socket.create_connection(("127.0.0.1", rec["port"]), timeout=120)
|
|
self._rfile = self.sock.makefile("rb")
|
|
return self
|
|
|
|
|
|
def files_with(root: Path, needles: list[str]) -> list[str]:
|
|
return files_containing(root, [core(n) for n in needles])
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - gateway child entry
|
|
if len(sys.argv) >= 3 and sys.argv[1] == "serve":
|
|
from hermes_logging import setup_logging
|
|
|
|
setup_logging(mode="gateway")
|
|
from tests.e2e.core.delivery._fake_platform import _serve
|
|
|
|
_serve(Path(sys.argv[2]))
|