diff --git a/gateway/session.py b/gateway/session.py index e8221bf6bcf83..53116372c1c48 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -1258,6 +1258,14 @@ class SessionStore: try: from hermes_state import SessionDB self._db = SessionDB() + except RuntimeError as e: + if "live-system guard" in str(e): + # Test-isolation guard fired: a pytest-context process + # resolved the developer's production state.db. Never + # swallow this into the JSONL fallback — the whole point + # is a loud, hard failure. + raise + print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}") except Exception as e: print(f"[gateway] Warning: SQLite session store unavailable, falling back to JSONL: {e}") diff --git a/hermes_state.py b/hermes_state.py index 0991f3cd2ff23..47e7e47cf3e0a 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -283,6 +283,109 @@ def _default_db_path() -> Path: return DEFAULT_DB_PATH return get_hermes_home() / "state.db" + +# --------------------------------------------------------------------------- +# Live-DB test-isolation guard +# --------------------------------------------------------------------------- +# Forensic evidence (Aug 2026, live developer machine): the production +# ~/.hermes/state.db accumulated pytest fixture rows — sessions with +# chat_id='chat-1'/'123'/'wx-chat' and gateway_routing scopes literally under +# /tmp/pytest-of-*/ — and a pytest-spawned process flipped the journal mode +# out from under the WAL-mode gateway writer, destroying committed +# transcripts ("Persisted transcript lagged live cached history ... possible +# FTS write corruption"). The hermetic conftest redirects HERMES_HOME per +# test, but any escape (a session-scoped fixture running before the autouse +# fixture, a subprocess child launched without HERMES_HOME, a stale worktree +# without the re-pin, or a developer shell that exports HERMES_HOME to the +# real home so the conftest session sandbox is skipped) silently fell +# through to the real database. +# +# This guard is the single choke point: EVERY ``SessionDB`` construction +# resolves its path here, so under pytest a resolution that lands on a +# production state.db fails hard instead of corrupting live data. It is +# env-based (``PYTEST_CURRENT_TEST`` / ``PYTEST_VERSION`` are set by pytest +# and inherited by subprocess children), so it also protects children that +# never import the test conftest. + +#: Escape hatch for the rare legitimate case (a test that genuinely needs +#: the real DB). The in-tree conftest sets this for tests marked +#: ``@pytest.mark.live_system_guard_bypass``; scripts may set it explicitly. +_STATE_DB_GUARD_BYPASS = False + +#: Additional production roots to refuse (beyond the platform default +#: ``~/.hermes``). The test conftest injects the pre-sandbox production +#: root here so custom-``HERMES_HOME`` deployments are covered too. +_STATE_DB_GUARD_EXTRA_DENY_ROOTS: Tuple[Path, ...] = () + + +def _running_under_pytest() -> bool: + """True when this process (or a parent test process) is a pytest run.""" + return bool( + os.environ.get("PYTEST_CURRENT_TEST") + or os.environ.get("PYTEST_VERSION") + ) + + +def _production_state_roots() -> List[Path]: + roots: List[Path] = [] + try: + from hermes_constants import _get_platform_default_hermes_home + + roots.append(_get_platform_default_hermes_home().resolve()) + except Exception: + pass + for extra in _STATE_DB_GUARD_EXTRA_DENY_ROOTS: + try: + roots.append(Path(extra).expanduser().resolve()) + except Exception: + continue + return roots + + +def _is_production_state_db(resolved: Path, root: Path) -> bool: + """True when *resolved* is a DB file of the real Hermes home *root*. + + Matches files directly in the root (``/state.db``) and profile + homes (``/profiles//state.db``). Deliberately does NOT + match deeper scratch paths (e.g. repo worktrees that happen to live + under ``~/.hermes/hermes-agent/...``) so hermetic tests using unusual + tempdirs cannot false-positive. + """ + if resolved.parent == root: + return True + try: + rel = resolved.relative_to(root) + except ValueError: + return False + parts = rel.parts + return len(parts) == 3 and parts[0] == "profiles" + + +def _ensure_test_isolation(db_path: Path) -> None: + """Fail hard when a pytest-context process resolves a production DB. + + Raises ``RuntimeError`` before any connection, mkdir, journal-mode + pragma, or byte probe can touch the live database. No-op outside + pytest and for hermetic (tmp ``HERMES_HOME``) paths. + """ + if _STATE_DB_GUARD_BYPASS or not _running_under_pytest(): + return + try: + resolved = Path(db_path).expanduser().resolve() + except Exception: + return + for root in _production_state_roots(): + if _is_production_state_db(resolved, root): + raise RuntimeError( + "live-system guard: test attempted to open production " + f"state.db at {resolved} (under real Hermes root {root}). " + "Tests must run against a temporary HERMES_HOME — pass an " + "explicit tmp db_path or let the hermetic conftest redirect " + "HERMES_HOME. If this test genuinely needs the live " + "database, mark it with " + "@pytest.mark.live_system_guard_bypass." + ) + # --------------------------------------------------------------------------- # WAL-compatibility fallback # --------------------------------------------------------------------------- @@ -2080,6 +2183,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) def __init__(self, db_path: Path = None, read_only: bool = False): self.db_path = db_path or _default_db_path() + # Fail hard (before any connection/pragma/mkdir) if a pytest-context + # process resolved the developer's production state.db — see the + # live-DB test-isolation guard block near _default_db_path(). + _ensure_test_isolation(self.db_path) self.read_only = read_only self._lock = threading.Lock() diff --git a/tests/conftest.py b/tests/conftest.py index 26c034edc96f3..21857f5a72559 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -58,7 +58,35 @@ if str(PROJECT_ROOT) not in sys.path: # would silently stop protecting the operator's actual ~/.hermes (#69385). _PRE_SANDBOX_KANBAN_OVERRIDE = os.environ.get("HERMES_KANBAN_HOME", "").strip() _PRE_SANDBOX_HERMES_HOME = os.environ.get("HERMES_HOME", "") -if not os.environ.get("HERMES_HOME"): + + +def _hermes_home_points_at_production(value: str) -> bool: + """True when a pre-set HERMES_HOME resolves to the real production root. + + Gateway-launched shells (and developer shells that ``export + HERMES_HOME=~/.hermes``) hand pytest the PRODUCTION home. Historically + the session sandbox below honored any pre-set value, so collection-time + imports (logging handlers, ``hermes_state.DEFAULT_DB_PATH``) froze paths + inside the real ``~/.hermes`` — the escape vector that landed pytest + fixture rows (chat-1 / wx-chat sessions, /tmp/pytest-of-* routing + scopes) in the live state.db and flipped its journal mode under the + WAL-mode gateway writer. Only a genuinely custom (non-production) + HERMES_HOME is honored now. + """ + if not value: + return True + try: + resolved = Path(value).expanduser().resolve() + real_root = (Path.home() / ".hermes").resolve() + except Exception: + return True + if resolved == real_root: + return True + # Profile home directly under the production root: /profiles/ + return resolved.parent.name == "profiles" and resolved.parent.parent == real_root + + +if _hermes_home_points_at_production(os.environ.get("HERMES_HOME", "")): _SESSION_HERMES_HOME = tempfile.mkdtemp(prefix="hermes-test-home-") os.environ["HERMES_HOME"] = _SESSION_HERMES_HOME atexit.register(shutil.rmtree, _SESSION_HERMES_HOME, True) @@ -578,9 +606,14 @@ def _capture_real_kanban_root() -> Path: """ if _PRE_SANDBOX_KANBAN_OVERRIDE: return Path(_PRE_SANDBOX_KANBAN_OVERRIDE).expanduser().resolve() - if _PRE_SANDBOX_HERMES_HOME: - # HERMES_HOME was genuinely set before the sandbox — honor it via the - # normal resolver (it may be a profile dir whose root matters). + if _PRE_SANDBOX_HERMES_HOME and not _hermes_home_points_at_production( + _PRE_SANDBOX_HERMES_HOME + ): + # HERMES_HOME was genuinely set to a CUSTOM root before the sandbox + # (production-pointing values are sandboxed away above, in which case + # the env still holds the tempdir and the resolver would be wrong) — + # honor it via the normal resolver (it may be a profile dir whose + # root matters). from hermes_constants import get_default_hermes_root return get_default_hermes_root().resolve() # No pre-existing HERMES_HOME: the real root is the platform default, @@ -645,6 +678,45 @@ def _kanban_write_guard(_hermetic_environment, monkeypatch): monkeypatch.setattr(_kdb, "connect", _guarded_connect) +# ── Live state.db write guard ─────────────────────────────────────────────── +# Companion to the kanban guard above, for the MAIN state database. +# ``hermes_state._ensure_test_isolation`` (the single choke point every +# ``SessionDB()`` construction goes through) refuses, under pytest, any DB +# path that resolves inside the REAL Hermes root. This fixture wires the +# test-side knobs: +# • honors ``@pytest.mark.live_system_guard_bypass`` (the established +# escape-hatch marker) by disabling the state-db guard for that test; +# • injects the pre-sandbox CUSTOM production root (Docker/portable +# installs where HERMES_HOME is not ~/.hermes) into the guard's +# deny-list, mirroring the kanban deny-list capture above. +# The guard itself is env-activated (PYTEST_CURRENT_TEST / PYTEST_VERSION), +# so subprocess children that import hermes_state directly are covered even +# without this fixture. + + +@pytest.fixture(autouse=True) +def _state_db_write_guard(request, monkeypatch): + _hs = sys.modules.get("hermes_state") + if _hs is None or not hasattr(_hs, "_STATE_DB_GUARD_BYPASS"): + yield + return + if request.node.get_closest_marker("live_system_guard_bypass") is not None: + monkeypatch.setattr(_hs, "_STATE_DB_GUARD_BYPASS", True) + yield + return + extra_roots = [] + if _PRE_SANDBOX_HERMES_HOME and not _hermes_home_points_at_production( + _PRE_SANDBOX_HERMES_HOME + ): + extra_roots.append( + Path(_PRE_SANDBOX_HERMES_HOME).expanduser().resolve() + ) + monkeypatch.setattr( + _hs, "_STATE_DB_GUARD_EXTRA_DENY_ROOTS", tuple(extra_roots) + ) + yield + + # ── Module-level state reset — replaced by per-file process isolation ────── # # Each test FILE runs in a freshly-spawned ``python -m pytest `` diff --git a/tests/hermes_state/test_live_db_isolation_guard.py b/tests/hermes_state/test_live_db_isolation_guard.py new file mode 100644 index 0000000000000..1825b06510510 --- /dev/null +++ b/tests/hermes_state/test_live_db_isolation_guard.py @@ -0,0 +1,192 @@ +"""Behavioral tests for the live-DB test-isolation guard. + +Forensic background (Aug 2026): pytest fixture rows (chat-1 / wx-chat +sessions, gateway_routing scopes under /tmp/pytest-of-*) were found in the +developer's REAL ~/.hermes/state.db, and a pytest-spawned process flipped +the journal mode under the WAL-mode gateway writer, destroying committed +transcripts. The guard under test makes any pytest-context ``SessionDB`` +construction that resolves to a production state.db fail hard instead of +falling through. + +These tests are behavioral: they construct real ``SessionDB`` objects (or +drive the real guard function) and assert outcomes — no source reading. +""" + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +import hermes_state +from gateway.config import GatewayConfig +from gateway.session import SessionStore +from hermes_state import SessionDB + +REAL_ROOT = (Path.home() / ".hermes").resolve() + + +class TestProductionPathRefused: + def test_explicit_production_db_path_raises(self): + """SessionDB pointed at the real ~/.hermes/state.db must fail hard.""" + with pytest.raises(RuntimeError, match="live-system guard"): + SessionDB(db_path=REAL_ROOT / "state.db") + + def test_production_profile_db_path_raises(self): + """Profile homes under the real root are production too.""" + with pytest.raises(RuntimeError, match="live-system guard"): + SessionDB(db_path=REAL_ROOT / "profiles" / "work" / "state.db") + + def test_read_only_open_of_production_db_raises(self): + """Read-only opens are refused too — tests must not READ live data.""" + with pytest.raises(RuntimeError, match="live-system guard"): + SessionDB(db_path=REAL_ROOT / "state.db", read_only=True) + + def test_unnormalized_production_path_raises(self): + """Symlink-free but unnormalized spellings still resolve and refuse.""" + sneaky = Path.home() / "subdir" / ".." / ".hermes" / "state.db" + with pytest.raises(RuntimeError, match="live-system guard"): + SessionDB(db_path=sneaky) + + def test_default_resolution_to_production_raises(self, monkeypatch): + """The argless-construction path is guarded, not just explicit paths. + + Simulates the escape vector: HERMES_HOME leaked/reset to the real + home (subprocess child, stale worktree, gateway-launched shell) so + ``_default_db_path()`` resolves the production DB. + """ + monkeypatch.setenv("HERMES_HOME", str(REAL_ROOT)) + # Neutralize the conftest's DEFAULT_DB_PATH re-pin so the default + # resolver follows the (production-pointing) env, as it would in a + # process that never imported the hermetic conftest. + monkeypatch.setattr( + hermes_state, "DEFAULT_DB_PATH", hermes_state._IMPORT_DEFAULT_DB_PATH + ) + with pytest.raises(RuntimeError, match="live-system guard"): + SessionDB() + + +class TestHermeticPathsAllowed: + def test_tmp_db_path_works(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session("iso-guard-session", "cli") + assert db.get_session("iso-guard-session") is not None + finally: + db.close() + + def test_tmp_hermes_home_default_resolution_works(self, tmp_path, monkeypatch): + """Argless SessionDB() under a hermetic HERMES_HOME must succeed.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermetic-home")) + monkeypatch.setattr( + hermes_state, "DEFAULT_DB_PATH", hermes_state._IMPORT_DEFAULT_DB_PATH + ) + db = SessionDB() + try: + assert str(tmp_path) in str(db.db_path) + finally: + db.close() + + +class TestBypassMarker: + @pytest.mark.live_system_guard_bypass + def test_bypass_marker_disables_state_db_guard(self): + """The established escape-hatch marker must let production paths pass. + + Drives the guard function directly (never actually opens the live + DB) — with the bypass marker active it must not raise. + """ + hermes_state._ensure_test_isolation(REAL_ROOT / "state.db") + + +class TestSessionStoreLoudFailure: + def test_guard_error_is_not_swallowed_into_jsonl_fallback( + self, tmp_path, monkeypatch + ): + """SessionStore must re-raise the guard error, not degrade to JSONL. + + The historical failure mode: SessionDB() blew up (or silently + opened the live DB) inside SessionStore.__init__'s blanket + ``except Exception`` and the gateway carried on. A guard trip must + be loud. + """ + + def _boom(*args, **kwargs): + raise RuntimeError( + "live-system guard: test attempted to open production state.db" + ) + + monkeypatch.setattr(hermes_state, "SessionDB", _boom) + with pytest.raises(RuntimeError, match="live-system guard"): + SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + + def test_ordinary_db_failure_still_degrades_to_jsonl( + self, tmp_path, monkeypatch + ): + """Non-guard SQLite failures keep the existing graceful fallback.""" + + def _boom(*args, **kwargs): + raise RuntimeError("disk on fire") + + monkeypatch.setattr(hermes_state, "SessionDB", _boom) + store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig()) + assert store._db is None + + +class TestSubprocessChildCovered: + def test_child_without_hermes_home_is_refused(self, tmp_path): + """A subprocess child of a test (no HERMES_HOME) must be blocked. + + This is the real leak vector: tests spawning ``python -m ...`` + children that never import the hermetic conftest. The guard is + env-activated (PYTEST_CURRENT_TEST / PYTEST_VERSION are inherited), + so the child's argless SessionDB() must fail hard instead of + opening the developer's real state.db. + """ + env = { + k: v + for k, v in os.environ.items() + if k not in ("HERMES_HOME", "PYTEST_PLUGINS", "PYTHONPATH") + } + env["PYTEST_CURRENT_TEST"] = "tests/fake.py::test_child (call)" + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[2]) + code = ( + "from hermes_state import SessionDB\n" + "SessionDB()\n" + ) + proc = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + env=env, + timeout=120, + ) + assert proc.returncode != 0 + assert "live-system guard" in proc.stderr + + def test_child_with_tmp_hermes_home_succeeds(self, tmp_path): + """Same child, hermetic HERMES_HOME: must work — no false positive.""" + env = { + k: v + for k, v in os.environ.items() + if k not in ("PYTEST_PLUGINS", "PYTHONPATH") + } + env["PYTEST_CURRENT_TEST"] = "tests/fake.py::test_child (call)" + env["HERMES_HOME"] = str(tmp_path / "child-home") + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[2]) + code = ( + "from hermes_state import SessionDB\n" + "db = SessionDB()\n" + "db.close()\n" + "print('OK', db.db_path)\n" + ) + proc = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + env=env, + timeout=120, + ) + assert proc.returncode == 0, proc.stderr + assert "OK" in proc.stdout