hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 30 Packages Projects Releases Wiki Activity

test: C7 two-tenant canary through one multiplexed messaging gateway

Browse Source 
Class C7 (multiplex / tenancy isolation). A real 'hermes gateway run' with
gateway.multiplex_profiles serves default + alpha + beta, each with its own
loopback provider (accepting only its own key) and distinct canaries: provider
key under the same env var name, API-server key, .env marker, model, MEMORY.md,
SOUL.md, terminal cwd, cron prompt and an exported shell variable. Interleaved
api_server turns (/v1, /p/<name>/v1) each run 'env | sort; pwd' in the terminal
tool; a cron tick fires one job per profile; cross-profile API keys; profile
create/attach/delete next to the live host (PID must not change); SIGTERM
restart + a second cron tick. After every phase: no request, tool-subprocess
env snapshot, profile-home file (state.db + WAL, logs, sessions, config, .env)
or gateway output carries another tenant's canary.

Red on the parent of a445f346a8b (cron shared the launch profile's terminal
environment); see REPORT.md for the red-proof table.

(cherry picked from commit ba5bdae9aa9900d719be36bf9efa40037860cf0b)
This commit is contained in:
teknium1
2026-09-23 04:10:15 -07:00
committed by Teknium
parent 616e8d4df6
commit 0ad5838c1c
2 changed files with 728 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

527
tests/e2e/core/tenancy/_helpers.py Normal file
View File

@@ -0,0 +1,527 @@
"""Lane-private harness for the C7 two-tenant canary suites.
Every tenant (profile) owns a set of random canaries: its provider key (same env var NAME in every
profile, different value), its API-server key, a non-secret ``.env`` marker, its model id, memory
text, SOUL text, terminal cwd and a cron prompt. Each tenant also owns its own loopback provider
(``FakeLLMServer``) that accepts only its own key, so every request the provider records is proof of
WHO sent it. The invariant shared by every scenario is ``leaks(...) == []``: no tenant's canary may
appear in another tenant's provider request, tool-subprocess env snapshot, or on-disk file.
Hermes runs for real in child processes with HOME=<tmp>/home and HERMES_HOME=<tmp>/home/.hermes
(profiles resolve under $HOME, never the real install), every credential env var stripped.
"""
from __future__ import annotations
import json
import os
import secrets
import signal
import socket
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable
import yaml
from tests.fakes.fake_llm_provider import FakeLLMServer, Response, Text, ToolCall
REPO_ROOT = Path(__file__).resolve().parents[4]
ENV_PROBE = "echo SNAPSHOT-BEGIN; env | sort; echo CWD=$(pwd); echo SNAPSHOT-END"
PROVIDER_KEY_ENV = "TENANT_PROVIDER_KEY" # same NAME in every profile's .env, distinct VALUE
_STRIP_SUFFIXES = ("_API_KEY", "_TOKEN", "_BASE_URL", "_SECRET", "_ACCESS_KEY", "_KEY_ID", "_KEY")
_STRIP_PREFIXES = ("HERMES_", "OPENAI", "ANTHROPIC", "OPENROUTER", "AWS_", "AZURE_", "GOOGLE_", "GEMINI",
"PYTEST_", "NOUS_", "XAI_", "LLM_", "CUSTOM_", "TERMINAL_", "TENANT_", "API_SERVER_")
def real_user_home() -> Path:
import pwd
return Path(pwd.getpwuid(os.getuid()).pw_dir).resolve()
def hermetic_env(home: Path, extra: dict[str, str] | None = None) -> dict[str, str]:
"""Child env: fake HOME (profile root anchor) + HERMES_HOME under it, nothing credential-shaped."""
home = home.resolve()
assert home != real_user_home() and home / ".hermes" != real_user_home() / ".hermes", home
env = {k: v for k, v in os.environ.items()
if not (k.endswith(_STRIP_SUFFIXES) or k.startswith(_STRIP_PREFIXES))}
for var in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy", "XDG_STATE_HOME",
"XDG_CONFIG_HOME", "XDG_DATA_HOME", "XDG_CACHE_HOME",
# no route to the developer's systemd --user bus: a child can never see or touch the
# live hermes-gateway unit, whatever it decides about service management.
"DBUS_SESSION_BUS_ADDRESS", "XDG_RUNTIME_DIR"):
env.pop(var, None)
env.update(
HOME=str(home),
HERMES_HOME=str(home / ".hermes"),
XDG_STATE_HOME=str(home / ".local" / "state"),
PYTHONPATH=str(REPO_ROOT),
NO_COLOR="1",
TERM="dumb",
NO_PROXY="127.0.0.1,localhost",
no_proxy="127.0.0.1,localhost",
# The live-DB guard treats $HOME/.hermes/state.db of a pytest descendant as production;
# this HOME is the test's own tmp dir (asserted above).
HERMES_STATE_DB_GUARD_BYPASS="1",
HERMES_ACCEPT_HOOKS="1",
)
env.update(extra or {})
return env
def free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def poll(pred: Callable[[], Any], timeout: float, what: str, interval: float = 0.1) -> Any:
deadline = time.monotonic() + timeout
while True:
got = pred()
if got:
return got
if time.monotonic() > deadline:
raise AssertionError(f"timed out after {timeout}s waiting for {what}")
time.sleep(interval)
# Tenants -------------------------------------------------------------------------------------------
@dataclass
class Tenant:
name: str
home: Path # the profile's HERMES_HOME
workdir: Path
tag: str = field(default_factory=lambda: secrets.token_hex(5))
srv: FakeLLMServer | None = None
extra: dict[str, str] = field(default_factory=dict) # canaries minted mid-scenario (RPC writes)
# canaries (each embeds the random tag, so a hit can never be a coincidence)
@property
def provider_key(self) -> str:
return f"sk-prov-{self.name}-{self.tag}"
@property
def api_server_key(self) -> str:
return f"apisrv-{self.name}-{self.tag}"
@property
def env_marker(self) -> str:
return f"envmark-{self.name}-{self.tag}"
@property
def model(self) -> str:
return f"model-{self.name}-{self.tag}"
@property
def memory(self) -> str:
return f"memory-canary-{self.name}-{self.tag}"
@property
def soul(self) -> str:
return f"soul-canary-{self.name}-{self.tag}"
@property
def shell_state(self) -> str:
"""Exported into the tool shell AFTER each env dump: a shell reused by another tenant shows it."""
return f"shellstate-{self.name}-{self.tag}"
@property
def cron_prompt(self) -> str:
return f"cron-canary-{self.name}-{self.tag}: run the environment probe"
def canaries(self) -> dict[str, str]:
return {
"provider_key": self.provider_key, "api_server_key": self.api_server_key,
"env_marker": self.env_marker, "model": self.model, "memory": self.memory, "soul": self.soul,
"workdir": str(self.workdir), "cron_prompt": self.cron_prompt.split(":")[0],
"shell_state": self.shell_state, **self.extra,
}
def secrets(self) -> dict[str, str]:
return {"provider_key": self.provider_key, "api_server_key": self.api_server_key}
def _responder(t: Tenant) -> Callable[[dict[str, Any]], Response]:
"""Every user turn runs the env probe in the terminal tool; the follow-up answers with text."""
def respond(record: dict[str, Any]) -> Response:
msgs = record["body"].get("messages") or []
last = msgs[-1] if msgs else {}
if last.get("role") == "tool":
return Text(f"probe done for {t.name}")
return ToolCall("terminal", {"command": f"{ENV_PROBE}; export TENANT_SHELL_STATE={t.shell_state}"})
return respond
def make_tenants(root: Path, names: Iterable[str], launch: str = "default") -> dict[str, Tenant]:
"""Build tenant homes on disk: launch profile at HOME/.hermes, the rest under profiles/<name>."""
hermes_home = root / "home" / ".hermes"
tenants: dict[str, Tenant] = {}
for name in names:
home = hermes_home if name == launch else hermes_home / "profiles" / name
t = Tenant(name=name, home=home, workdir=root / "work")
t.workdir = root / f"work-{name}-{t.tag}"
t.workdir.mkdir(parents=True)
t.srv = FakeLLMServer(_responder(t), api_key=t.provider_key)
tenants[name] = t
return tenants
def write_tenant_home(t: Tenant, extra_config: dict[str, Any] | None = None,
extra_env: dict[str, str] | None = None) -> None:
assert t.srv is not None
t.home.mkdir(parents=True, exist_ok=True)
cfg: dict[str, Any] = {
"model": {"provider": "custom", "base_url": t.srv.base_url, "default": t.model,
"key_env": PROVIDER_KEY_ENV, "context_length": 128000},
"agent": {"api_max_retries": 1},
"terminal": {"backend": "local", "cwd": str(t.workdir)},
"memory": {"memory_enabled": True},
"compression": {"enabled": False},
}
for k, v in (extra_config or {}).items():
cfg[k] = {**cfg.get(k, {}), **v} if isinstance(v, dict) and isinstance(cfg.get(k), dict) else v
(t.home / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8")
env = {PROVIDER_KEY_ENV: t.provider_key, "API_SERVER_KEY": t.api_server_key, "TENANT_MARKER": t.env_marker,
**(extra_env or {})}
(t.home / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items()), encoding="utf-8")
(t.home / "memories").mkdir(exist_ok=True)
(t.home / "memories" / "MEMORY.md").write_text(f"{t.memory}\n", encoding="utf-8")
(t.home / "SOUL.md").write_text(f"You are {t.soul}.\n", encoding="utf-8")
def assert_profiles_root_under(root: Path, home: Path) -> None:
"""The profile root is HOME-anchored: prove it resolves inside ``root`` before any write."""
probe = subprocess.run(
[sys.executable, "-c", "from hermes_cli.profiles import _get_profiles_root as r; print(r())"],
env=hermetic_env(home), cwd=str(home), capture_output=True, text=True, timeout=120,
stdin=subprocess.DEVNULL,
)
assert probe.returncode == 0, probe.stderr[-2000:]
got = Path(probe.stdout.strip().splitlines()[-1]).resolve()
assert str(got).startswith(str(root.resolve())), f"profiles root escaped the sandbox: {got}"
# Invariants ----------------------------------------------------------------------------------------
def _bearer(record: dict[str, Any]) -> str:
auth = record.get("auth") or ""
return auth[7:] if auth.lower().startswith("bearer ") else auth
def request_leaks(tenants: dict[str, Tenant]) -> list[str]:
"""Every provider request carries its own tenant's key/model and no other tenant's canary."""
problems: list[str] = []
for t in tenants.values():
assert t.srv is not None
for i, r in enumerate(list(t.srv.requests)):
where = f"provider[{t.name}] request #{i} ({r['kind']} {r['path']})"
if _bearer(r) != t.provider_key:
owner = next((u.name for u in tenants.values() if u.provider_key == _bearer(r)), "<none>")
problems.append(f"{where}: Authorization is {owner!r}'s key, not {t.name!r}'s")
if r["kind"] == "main" and r["body"].get("model") != t.model:
problems.append(f"{where}: model {r['body'].get('model')!r} != {t.model!r}")
blob = json.dumps(r["body"]) + json.dumps(r.get("headers") or {})
problems += [f"{where}: carries {u.name}'s {kind} ({value})"
for u in tenants.values() if u is not t
for kind, value in u.canaries().items() if value in blob]
return problems
def env_snapshots(t: Tenant) -> list[tuple[str, str]]:
"""``(origin, output)`` of every ENV_PROBE terminal call as sent back to the tenant's provider;
origin is ``cron job`` when the conversation is the tenant's cron prompt, else ``turn``."""
assert t.srv is not None
seen: dict[str, tuple[str, str]] = {} # tool_call_id -> (origin, output); results are re-sent later
cron_mark = t.cron_prompt.split(":")[0]
for r in list(t.srv.requests):
origin = "cron job" if cron_mark in json.dumps(r["body"].get("messages") or []) else "turn"
for m in r["body"].get("messages") or []:
content = m.get("content")
text = content if isinstance(content, str) else json.dumps(content)
if m.get("role") == "tool" and "SNAPSHOT-BEGIN" in text:
seen.setdefault(str(m.get("tool_call_id") or text), (origin, text))
return list(seen.values())
def snapshot_problems(tenants: dict[str, Tenant], min_per_tenant: int = 1) -> list[str]:
"""Inside a tool subprocess: cwd is the tenant's own, and no other tenant's canary is visible."""
problems: list[str] = []
for t in tenants.values():
snaps = env_snapshots(t)
if len(snaps) < min_per_tenant:
problems.append(f"{t.name}: {len(snaps)} env snapshot(s) reached its provider, expected >= {min_per_tenant}")
for origin, s in snaps:
text = s.encode().decode("unicode_escape", errors="ignore") if "\\n" in s else s
own_dirs = [str(t.workdir), *(v for k, v in t.extra.items() if k.startswith("workdir"))]
if not any(f"CWD={d}\n" in text + "\n" for d in own_dirs):
cwd = next((ln for ln in text.splitlines() if ln.startswith("CWD=")), "<no CWD line>")
problems.append(f"{t.name}: terminal ({origin}) ran in {cwd.strip()!r}, not its own {own_dirs}")
problems += [f"{t.name}: tool subprocess ({origin}) carries {u.name}'s {kind}: "
f"{next((ln.strip() for ln in text.splitlines() if value in ln), value)[:300]!r}"
for u in tenants.values() if u is not t
for kind, value in u.canaries().items() if value in text]
return problems
def file_leaks(tenants: dict[str, Tenant], launch: str = "default") -> list[str]:
"""Byte-scan every file of every tenant home (state.db + WAL, logs, sessions, config, .env,
memories, cron output) for any other tenant's canary. The launch home's ``profiles/`` subtree
belongs to the other tenants and is scanned as theirs."""
problems: list[str] = []
for t in tenants.values():
foreign = [(u.name, kind, value.encode()) for u in tenants.values() if u is not t
for kind, value in u.canaries().items()]
for path in t.home.rglob("*"):
if not path.is_file() or (t.name == launch and "profiles" in path.relative_to(t.home).parts[:1]):
continue
try:
data = path.read_bytes()
except OSError:
continue
problems += [f"{t.name}: {path.relative_to(t.home)} contains {owner}'s {kind}"
for owner, kind, value in foreign if value in data]
return problems
def check_isolation(tenants: dict[str, Tenant], *, launch: str = "default", min_snapshots: int = 1,
extra: Iterable[str] = ()) -> None:
problems = [*request_leaks(tenants), *snapshot_problems(tenants, min_snapshots), *file_leaks(tenants, launch),
*extra]
assert not problems, "cross-tenant leak(s):\n " + "\n ".join(dict.fromkeys(problems))
def text_leaks(label: str, text: str, tenants: dict[str, Tenant], owner: str | None = None) -> list[str]:
"""Canaries of any tenant other than ``owner`` in a log/stream (``owner=None``: secrets only)."""
out: list[str] = []
for u in tenants.values():
if u.name == owner:
continue
values = u.canaries() if owner is not None else u.secrets()
out += [f"{label}: carries {u.name}'s {kind}" for kind, value in values.items() if value in text]
return out
_SEED_CRON = ("import sys; from cron.jobs import create_job, trigger_job, list_jobs; "
"jobs = [j for j in list_jobs(include_disabled=True) if j.get('name') == 'tenancy-canary']; "
"j = jobs[0] if jobs else create_job(prompt=sys.argv[1], schedule='every 1h', name='tenancy-canary', "
"deliver='local'); trigger_job(j['id']); print(j['id'])")
def seed_due_cron_job(t: Tenant, home: Path) -> str:
"""Create (once) an hourly job in the tenant's own cron store and mark it due now (the state the
"run on next tick" action produces), so the multiplexed ticker's first tick fires it."""
r = subprocess.run([sys.executable, "-c", _SEED_CRON, t.cron_prompt], cwd=str(home),
env=hermetic_env(home, {"HERMES_HOME": str(t.home)}), capture_output=True, text=True,
timeout=120, stdin=subprocess.DEVNULL)
assert r.returncode == 0, r.stderr[-2000:]
return r.stdout.strip().splitlines()[-1]
def cron_requests(t: Tenant) -> int:
"""Provider requests driven by the tenant's cron job (its prompt canary is in the user turn)."""
assert t.srv is not None
return sum(t.cron_prompt.split(":")[0] in json.dumps(r["body"]) for r in list(t.srv.requests))
# Processes -----------------------------------------------------------------------------------------
def kill_group(proc: subprocess.Popen, sig: int = signal.SIGKILL) -> None:
try:
os.killpg(proc.pid, sig)
except (ProcessLookupError, PermissionError):
pass
def run_hermes(argv: list[str], home: Path, timeout: float = 120.0,
extra_env: dict[str, str] | None = None) -> subprocess.CompletedProcess:
proc = subprocess.Popen(
[sys.executable, "-m", "hermes_cli.main", *argv], cwd=str(home), env=hermetic_env(home, extra_env),
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
start_new_session=True,
)
try:
out, err = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
kill_group(proc)
out, err = proc.communicate()
err += f"\n[harness] killed after {timeout}s"
kill_group(proc)
return subprocess.CompletedProcess(argv, proc.returncode, out, err)
class TuiBackend:
"""The real stdio JSON-RPC backend the TUI/Desktop drive (``python -m tui_gateway.entry``)."""
def __init__(self, home: Path, log_path: Path, extra_env: dict[str, str] | None = None) -> None:
import queue
import threading
self._log = open(log_path, "a", encoding="utf-8") # noqa: SIM115 - closed in close()
self.proc = subprocess.Popen(
[sys.executable, "-m", "tui_gateway.entry"], cwd=str(home), env=hermetic_env(home, extra_env),
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self._log, text=True, bufsize=1,
start_new_session=True,
)
self._q: queue.Queue[dict[str, Any]] = queue.Queue()
self._rid = 0
self.seen: list[dict[str, Any]] = []
threading.Thread(target=self._read, daemon=True, name="tenancy-tui-reader").start()
def _read(self) -> None:
assert self.proc.stdout is not None
for line in self.proc.stdout:
try:
self._q.put(json.loads(line))
except json.JSONDecodeError:
continue
def wait(self, pred: Callable[[dict[str, Any]], bool], timeout: float = 90.0) -> dict[str, Any]:
import queue
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if self.proc.poll() is not None and self._q.empty():
break
try:
msg = self._q.get(timeout=0.25)
except queue.Empty:
continue
self.seen.append(msg)
if pred(msg):
return msg
tail = [(m.get("method"), (m.get("params") or {}).get("type"), m.get("id")) for m in self.seen[-20:]]
raise AssertionError(f"tui_gateway: condition not met within {timeout}s (rc={self.proc.poll()}); last {tail}")
def call(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
self._rid += 1
rid = self._rid
assert self.proc.stdin is not None
self.proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}) + "\n")
self.proc.stdin.flush()
return self.wait(lambda m: m.get("id") == rid and "method" not in m, timeout)
def ok(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
reply = self.call(method, params, timeout)
assert "result" in reply, f"{method} failed: {reply.get('error')}"
return reply["result"]
def turn(self, sid: str, text: str, timeout: float = 90.0) -> dict[str, Any]:
"""Submit one prompt and wait for that session's terminal turn event."""
self.ok("prompt.submit", {"session_id": sid, "text": text})
def done(m: dict[str, Any]) -> bool:
p = m.get("params") or {}
return (m.get("method") == "event" and p.get("session_id") == sid
and p.get("type") in {"message.complete", "error"})
ev = self.wait(done, timeout)
assert ev["params"]["type"] == "message.complete", f"turn failed: {ev}"
return ev
def close(self) -> None:
try:
if self.proc.stdin:
self.proc.stdin.close()
self.proc.wait(timeout=60)
except (subprocess.TimeoutExpired, OSError, ValueError):
pass
kill_group(self.proc)
try:
self.proc.wait(timeout=10)
except subprocess.TimeoutExpired:
pass
self._log.close()
class ServeBackend(TuiBackend):
"""The real Desktop backend: ``hermes serve --port 0`` with the JSON-RPC surface on ``/api/ws``.
``HERMES_DESKTOP=1`` marks it app-spawned, which is also what runs the cron ticker in-process.
"""
def __init__(self, home: Path, log_path: Path, extra_env: dict[str, str] | None = None) -> None:
import queue
import re
import threading
from websockets.sync.client import connect # ``websockets`` is a core dependency
self.token = secrets.token_urlsafe(24)
self._log = open(log_path, "a", encoding="utf-8") # noqa: SIM115 - closed in close()
env = {"HERMES_DASHBOARD_SESSION_TOKEN": self.token, "HERMES_DESKTOP": "1", **(extra_env or {})}
self.proc = subprocess.Popen(
[sys.executable, "-m", "hermes_cli.main", "serve", "--host", "127.0.0.1", "--port", "0"],
cwd=str(home), env=hermetic_env(home, env), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True, bufsize=1, start_new_session=True,
)
port_box: list[int] = []
def pump() -> None:
assert self.proc.stdout is not None
for line in self.proc.stdout:
self._log.write(line)
self._log.flush()
m = re.search(r"backend listening on [\d.]+:(\d+)", line)
if m and not port_box:
port_box.append(int(m.group(1)))
threading.Thread(target=pump, daemon=True, name="tenancy-serve-stdout").start()
self.port = poll(lambda: port_box[0] if port_box else (self.proc.poll() is not None and -1), 120,
"hermes serve to report its port")
assert self.port > 0, f"hermes serve exited rc={self.proc.returncode}"
self.ws = connect(f"ws://127.0.0.1:{self.port}/api/ws?token={self.token}", open_timeout=90, max_size=None)
self._q: queue.Queue[dict[str, Any]] = queue.Queue()
self._rid = 0
self.seen: list[dict[str, Any]] = []
threading.Thread(target=self._read, daemon=True, name="tenancy-ws-reader").start()
def _read(self) -> None:
while True:
try:
raw = self.ws.recv()
except Exception:
return
if not raw:
return
try:
self._q.put(json.loads(raw))
except json.JSONDecodeError:
continue
def call(self, method: str, params: dict[str, Any], timeout: float = 90.0) -> dict[str, Any]:
self._rid += 1
rid = self._rid
self.ws.send(json.dumps({"jsonrpc": "2.0", "id": rid, "method": method, "params": params}))
return self.wait(lambda m: m.get("id") == rid and "method" not in m, timeout)
def close(self) -> None:
try:
self.ws.close()
except Exception:
pass
kill_group(self.proc, signal.SIGTERM)
try:
self.proc.wait(timeout=60)
except subprocess.TimeoutExpired:
kill_group(self.proc)
self.proc.wait(timeout=10)
kill_group(self.proc)
self._log.close()
def pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
stat = Path(f"/proc/{pid}/stat")
return not (stat.exists() and stat.read_text().split(") ", 1)[-1].startswith("Z"))

201
tests/e2e/core/tenancy/test_two_tenant_gateway.py Normal file
View File

@@ -0,0 +1,201 @@
"""C7 two-tenant canary: ONE multiplexed messaging gateway serving three profiles.
Class: multiplex / tenancy isolation (issue_classes.md C7). Users hit it as "profile B's cron job ran
with profile A's terminal settings", "my secondary profile answered with the launch profile's key",
"deleting a profile killed the shared gateway", "a restart re-pinned one profile's env for all"
(#89315, #101719, #105396, #111151, #102769, #107692).
Harness: a real ``hermes gateway run`` child process with ``gateway.multiplex_profiles: true`` serving
the launch profile ``default`` plus ``alpha`` and ``beta``. Each profile owns its own loopback provider
(accepting only its own key), and a distinct value for EVERY per-profile knob: provider key (same env
var NAME, different value), API-server key, a non-secret .env marker, model id, MEMORY.md, SOUL.md,
terminal cwd, cron prompt, and a shell variable its tool call exports. Turns are driven through the
real api_server platform (``/v1`` for the launch profile, ``/p/<name>/v1`` for the others); every
turn runs ``env | sort; pwd`` in the terminal tool, so the provider receives an env snapshot taken
INSIDE the tool subprocess.
Phases (same invariant after each): interleaved turns + a cron tick firing one job per profile ->
cross-profile API keys -> create/attach/delete a fourth profile (host PID must not change) ->
SIGTERM restart + a second cron tick. Invariant (``check_isolation``): every request a provider
recorded carries only its own tenant's key/model and no other tenant's canary; every env snapshot ran
in its own cwd and shows no other tenant's canary; no file in any profile home (state.db + WAL, logs,
sessions, cron output, config, .env) contains another tenant's canary; the gateway's own output
never contains any tenant's secret.
"""
from __future__ import annotations
import json
import signal
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
import pytest
from . import _helpers as H
# The gateway child is spawned and reaped by this module (fake HOME, no systemd bus in its env).
pytestmark = pytest.mark.spawns_gateway_lookalike
NAMES = ("default", "alpha", "beta")
class MultiplexGateway:
def __init__(self, root: Path, port: int) -> None:
self.root, self.home, self.port = root, root / "home", port
self.log_path = root / "gateway.log"
self.proc: subprocess.Popen | None = None
self.pids: list[int] = []
def start(self) -> None:
log = open(self.log_path, "a", encoding="utf-8") # noqa: SIM115 - handed to the child
self.proc = subprocess.Popen(
[sys.executable, "-m", "hermes_cli.main", "gateway", "run"], cwd=str(self.home),
env=H.hermetic_env(self.home), stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT,
start_new_session=True,
)
log.close()
self.pids.append(self.proc.pid)
H.poll(lambda: self.healthy() or self._died(), 120, "multiplexed gateway /health")
assert self.proc.poll() is None, f"gateway exited rc={self.proc.returncode}: {self.tail()}"
def _died(self) -> bool:
return self.proc is not None and self.proc.poll() is not None
def healthy(self) -> bool:
try:
with urllib.request.urlopen(f"http://127.0.0.1:{self.port}/health", timeout=3) as r:
return r.status == 200
except (OSError, urllib.error.URLError):
return False
def chat(self, prefix: str, api_key: str, text: str) -> tuple[int, Any]:
req = urllib.request.Request(
f"http://127.0.0.1:{self.port}{prefix}/v1/chat/completions",
data=json.dumps({"model": "hermes", "messages": [{"role": "user", "content": text}]}).encode(),
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=90) as r:
return r.status, json.loads(r.read())
except urllib.error.HTTPError as e:
return e.code, e.read()[:500]
def stop(self) -> None:
if self.proc is None:
return
H.kill_group(self.proc, signal.SIGTERM)
try:
self.proc.wait(timeout=90)
except subprocess.TimeoutExpired:
H.kill_group(self.proc)
self.proc.wait(timeout=10)
H.kill_group(self.proc)
def tail(self, n: int = 3000) -> str:
return self.log_path.read_text(encoding="utf-8", errors="replace")[-n:] if self.log_path.exists() else ""
def prefix(name: str) -> str:
return "" if name == "default" else f"/p/{name}"
@pytest.fixture(scope="module")
def fleet(tmp_path_factory: pytest.TempPathFactory):
root = tmp_path_factory.mktemp("c7-gateway")
home = root / "home"
tenants = H.make_tenants(root, NAMES)
port = H.free_port()
for t in tenants.values():
assert t.srv is not None
t.srv.start()
# The profile root is HOME-anchored: prove it is inside the sandbox before writing any profile.
home.mkdir(parents=True)
H.assert_profiles_root_under(root, home)
for t in tenants.values():
extra = {} if t.name != "default" else {
"gateway": {"multiplex_profiles": True},
"platforms": {"api_server": {"enabled": True, "extra": {"host": "127.0.0.1", "port": port}}},
}
H.write_tenant_home(t, extra)
gw = MultiplexGateway(root, port)
try:
yield root, tenants, gw
finally:
gw.stop()
for pid in gw.pids:
assert not H.pid_alive(pid), f"gateway pid {pid} survived teardown"
for t in tenants.values():
assert t.srv is not None
t.srv.stop()
def _turns(gw: MultiplexGateway, tenants: dict[str, H.Tenant], order: list[str]) -> None:
for name in order:
t = tenants[name]
before = len(t.srv.requests) # type: ignore[union-attr]
status, body = gw.chat(prefix(name), t.api_server_key, f"turn for {name}")
assert status == 200, f"{name} turn via api_server failed: {status} {body!r}\n{gw.tail()}"
answer = body["choices"][0]["message"]["content"]
assert f"probe done for {name}" in answer, f"{name} was answered by someone else: {answer!r}"
assert len(t.srv.requests) >= before + 2, f"{name}'s turn never reached its own provider" # type: ignore[union-attr]
def _await_cron(tenants: dict[str, H.Tenant], at_least: int) -> None:
# tool call + follow-up per fire; the ticker's first tick runs right after boot.
H.poll(lambda: all(H.cron_requests(t) >= 2 * at_least for t in tenants.values()), 150,
f"cron fire #{at_least} in every profile")
def test_multiplexed_gateway_never_crosses_tenants(fleet) -> None:
root, tenants, gw = fleet
home = root / "home"
# Phase 1: boot with one due cron job per profile; interleave turns across all three.
for t in tenants.values():
H.seed_due_cron_job(t, home)
gw.start()
host_pid = gw.proc.pid # type: ignore[union-attr]
_turns(gw, tenants, ["alpha", "default", "beta", "alpha", "beta", "default"])
_await_cron(tenants, 1)
H.check_isolation(tenants, min_snapshots=3)
# Phase 2: each profile's API-server key is honoured only on its own route.
for owner in tenants.values():
for route in tenants.values():
if route is owner:
continue
seen = len(route.srv.requests) # type: ignore[union-attr]
status, _ = gw.chat(prefix(route.name), owner.api_server_key, "cross-tenant key")
assert status in (401, 403), f"{owner.name}'s API key opened {route.name}'s route (HTTP {status})"
assert len(route.srv.requests) == seen, f"{owner.name}'s key drove {route.name}'s provider" # type: ignore[union-attr]
# Phase 3: profile churn next to the live host never restarts or kills it.
created = H.run_hermes(["profile", "create", "gamma"], home)
assert created.returncode == 0, created.stderr[-2000:]
attach = H.run_hermes(["-p", "gamma", "gateway", "run"], home, timeout=150)
assert "[harness] killed" not in attach.stderr, "`-p gamma gateway run` started a second gateway"
assert gw.proc.pid == host_pid and gw.proc.poll() is None, "profile create/attach replaced the host" # type: ignore[union-attr]
deleted = H.run_hermes(["profile", "delete", "gamma", "--yes"], home)
assert deleted.returncode == 0, deleted.stderr[-2000:]
assert gw.proc.pid == host_pid and gw.proc.poll() is None, f"profile delete killed the host\n{gw.tail()}" # type: ignore[union-attr]
_turns(gw, tenants, ["alpha", "beta"])
H.check_isolation(tenants, min_snapshots=3)
# Phase 4: SIGTERM restart; the new host re-serves all three and ticks each profile's cron again.
gw.stop()
assert not H.pid_alive(host_pid), "old gateway still alive after SIGTERM"
for t in tenants.values():
H.seed_due_cron_job(t, home)
gw.start()
_turns(gw, tenants, ["beta", "default", "alpha"])
_await_cron(tenants, 2)
H.check_isolation(tenants, min_snapshots=5)
gw.stop()
leaks = H.text_leaks("gateway output", gw.tail(10**9), tenants)
assert not leaks, "\n".join(leaks)