security(gateway): re-resolve hooks directory per call to fix profile isolation

gateway/hooks.py

@@ -39,6 +39,7 @@
 import asyncio
 import importlib.util
 import sys
+from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional
 
 import yaml
@@ -47,6 +48,35 @@
 
 
 HOOKS_DIR = get_hermes_home() / "hooks"
+# Import-time default, used by ``_resolve_hooks_dir`` below to detect a test
+# monkeypatch of the constant (test seam, see tests/gateway/test_hooks.py) vs.
+# an unmodified import-time value (in which case it re-resolves through the
+# active profile override).
+_HOOKS_DIR_IMPORT_DEFAULT = HOOKS_DIR
+
+
+def _resolve_hooks_dir() -> Path:
+    """Resolve the hooks directory, honoring the active profile.
+
+    ``HOOKS_DIR`` is frozen at import time, which pins every later hook
+    discovery to whichever profile's ``HERMES_HOME`` was active when this
+    module was first imported — a cross-profile leak under the multiplexed
+    gateway, where each profile's own ``Gateway``/``HookRegistry`` instance
+    shares one process. Without this, a later-starting profile's
+    ``discover_and_load()`` silently loads and executes the FIRST profile's
+    hook handlers (arbitrary Python) against its own live event context
+    (session_id, message, response text) instead of its own hooks directory.
+    Re-resolve through ``get_hermes_home()`` on every call so the
+    context-local profile override (``set_hermes_home_override``) is
+    honored, mirroring the cache-dir profile-isolation fix.
+
+    A test that monkeypatches the module constant away from its import-time
+    default is respected (test seam preserved).
+    """
+    current = HOOKS_DIR
+    if current != _HOOKS_DIR_IMPORT_DEFAULT:
+        return current
+    return get_hermes_home() / "hooks"
 
 
 class HookRegistry:
@@ -90,10 +120,11 @@ def discover_and_load(self) -> None:
         """
         self._register_builtin_hooks()
 
-        if not HOOKS_DIR.exists():
+        hooks_dir = _resolve_hooks_dir()
+        if not hooks_dir.exists():
             return
 
-        for hook_dir in sorted(HOOKS_DIR.iterdir()):
+        for hook_dir in sorted(hooks_dir.iterdir()):
             if not hook_dir.is_dir():
                 continue
 

tests/test_profile_isolation_runtime.py

@@ -143,6 +143,62 @@ def test_store_path_follows_override(self, two_profiles, monkeypatch):
         assert b_seen.endswith("state/rich_sent_index.json")
 
 
+class TestGatewayHooksDirResolution:
+    """gateway/hooks.py's HookRegistry must discover hooks from the active
+    profile's directory — otherwise one profile's HookRegistry loads and
+    executes a DIFFERENT profile's hook handlers (arbitrary Python) against
+    its own live event context under the multiplexed gateway."""
+
+    def test_hooks_dir_follows_override(self, two_profiles):
+        prof_a, prof_b = two_profiles
+        import gateway.hooks as gh
+
+        a_seen = _under_override(prof_a, lambda: gh._resolve_hooks_dir())
+        b_seen = _under_override(prof_b, lambda: gh._resolve_hooks_dir())
+
+        assert a_seen == prof_a / "hooks"
+        assert b_seen == prof_b / "hooks"
+        assert a_seen != b_seen
+
+    def test_discover_and_load_uses_active_profile_hooks_dir(self, two_profiles):
+        """End-to-end: a hook that only exists under profile B's hooks dir
+        must not be discovered when profile A's override is active, and vice
+        versa — proving the registry doesn't fall back to a frozen profile."""
+        prof_a, prof_b = two_profiles
+        import gateway.hooks as gh
+
+        b_hook_dir = prof_b / "hooks" / "only-in-b"
+        b_hook_dir.mkdir(parents=True)
+        (b_hook_dir / "HOOK.yaml").write_text(
+            "name: only-in-b\nevents: [\"agent:start\"]\n", encoding="utf-8",
+        )
+        (b_hook_dir / "handler.py").write_text(
+            "async def handle(event_type, context):\n    pass\n", encoding="utf-8",
+        )
+
+        def _load_and_names():
+            reg = gh.HookRegistry()
+            reg.discover_and_load()
+            return [h["name"] for h in reg.loaded_hooks]
+
+        a_hooks = _under_override(prof_a, _load_and_names)
+        b_hooks = _under_override(prof_b, _load_and_names)
+
+        assert "only-in-b" not in a_hooks
+        assert "only-in-b" in b_hooks
+
+    def test_monkeypatched_constant_still_wins(self, two_profiles, monkeypatch, tmp_path):
+        """The existing test seam (monkeypatch the module constant, see
+        tests/gateway/test_hooks.py) is preserved."""
+        _prof_a, prof_b = two_profiles
+        import gateway.hooks as gh
+
+        forced = tmp_path / "forced_hooks"
+        monkeypatch.setattr("gateway.hooks.HOOKS_DIR", forced)
+        seen = _under_override(prof_b, lambda: gh._resolve_hooks_dir())
+        assert seen == forced
+
+
 # ---------------------------------------------------------------------------
 # M2 — thread / executor context propagation
 # ---------------------------------------------------------------------------