security(gateway): fail closed on persisted /resume when caller keys on user_id_alt

The persisted (DB-fallback) branch of _resume_target_allowed() compared only
sessions.user_id against source.user_id, but build_session_key() keys the
participant on `user_id_alt or user_id` (Signal/Feishu carry the canonical
participant in user_id_alt). The sessions table has no user_id_alt column, so a
per-user row a caller shares the user_id of — but not the user_id_alt — maps to a
DIFFERENT live session key, yet the row's user_id matched both participants:
a co-member could resume/enumerate another member's persisted per-user group or
no-chat_id DM session (IDOR, CWE-639).

The live-origin guard (_same_origin_chat) already compares user_id_alt; the
persisted fallback couldn't. Fail closed on both identity-bearing per-user
branches (non-DM per-user group, no-chat_id DM) whenever the caller carries a
user_id_alt. Shared group/thread sessions (no participant scoping) and DMs keyed
on a present chat_id are unaffected; callers keyed on user_id (e.g. Telegram)
still resume their own rows; admin --all override still applies.

Regression: tests/gateway/test_resume_command.py::
test_resume_persisted_fallback_fails_closed_on_user_id_alt.

gateway/slash_commands.py

@@ -810,6 +810,20 @@ class GatewaySlashCommandsMixin:
         row_thread = str(row.get("thread_id") or "")
         chat_type = (getattr(source, "chat_type", "") or "").lower()
         caller_is_dm = chat_type in {"dm", "direct", "private", ""}
+        # build_session_key keys the participant on ``user_id_alt or user_id``
+        # (Signal/Feishu carry the canonical participant in user_id_alt), but the
+        # sessions table only ever stored user_id — it has no user_id_alt column.
+        # So when the caller carries a user_id_alt, the row CANNOT prove the
+        # canonical participant that the live session key is built from: two
+        # members sharing one user_id but different user_id_alt map to DIFFERENT
+        # session keys, yet the persisted row's user_id would match both. The
+        # live-origin guard (_same_origin_chat) compares user_id_alt correctly;
+        # the persisted fallback cannot, so any per-user comparison that would
+        # otherwise rely on row_uid == caller_uid must fail closed here to stay
+        # in lock-step with the key boundary (CWE-639). Shared group/thread
+        # sessions are unaffected (they don't scope by participant at all), and
+        # an admin --all override still bypasses this above.
+        caller_keys_on_alt = bool(str(getattr(source, "user_id_alt", "") or ""))
         if caller_uid:
             # Identity-bearing caller: allow only when the row PROVES the same
             # owner AND the same platform/origin AND the same chat/thread. A row
@@ -840,6 +854,15 @@ class GatewaySlashCommandsMixin:
                 # legitimately absent on both sides for a no-chat_id DM (scoped
                 # by user_id), but a mismatching chat_id (when present) is still
                 # rejected.
+                #
+                # A no-chat_id DM is keyed PURELY on the participant
+                # (``user_id_alt or user_id``). If the caller keys on user_id_alt
+                # the persisted row (user_id only) cannot prove that participant,
+                # so fail closed. When chat_id is present on both sides it is the
+                # DM key and equal chat_id is sufficient, so the alt gap doesn't
+                # apply there.
+                if caller_keys_on_alt and not (bool(row_chat) and bool(caller_chat)):
+                    return False
                 return (
                     bool(row_uid) and row_uid == caller_uid
                     and row_chat == caller_chat
@@ -865,6 +888,12 @@ class GatewaySlashCommandsMixin:
             )
             if shared:
                 return True
+            # Per-user non-DM: the session key includes the participant
+            # (``user_id_alt or user_id``). If the caller keys on user_id_alt,
+            # the persisted row (user_id only) cannot prove the canonical
+            # participant, so fail closed rather than matching on user_id alone.
+            if caller_keys_on_alt:
+                return False
             return bool(row_uid) and row_uid == caller_uid
         # No caller identity: the persisted row carries only source + user_id
         # (the sessions table has no chat_id), so a same-platform row can belong

tests/gateway/test_resume_command.py

@@ -639,6 +639,60 @@ class TestHandleSessionsCommand:
                                                    allow_override=False) is False
         db.close()
 
+    @pytest.mark.asyncio
+    async def test_resume_persisted_fallback_fails_closed_on_user_id_alt(self, tmp_path):
+        """egilewski/CodeRabbit probe: Signal/Feishu key the session participant
+        on ``user_id_alt or user_id`` (build_session_key), but the sessions table
+        stores only user_id. So a persisted per-user row that a caller shares the
+        user_id of — but NOT the user_id_alt — maps to a DIFFERENT live session
+        key; the persisted fallback must NOT match it on user_id alone (IDOR).
+
+        The live-origin guard already compares user_id_alt correctly; here the
+        target is persisted-only, so the fallback fails closed whenever the
+        caller keys on user_id_alt and the row can't prove that participant."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        # Persisted rows carry only user_id (no user_id_alt column).
+        db.create_session("victim_alt_group", "signal", user_id="+15550001111",
+                          chat_id="signal-group", chat_type="group")
+        db.create_session("victim_alt_dm", "signal", user_id="+15550001111")  # no chat_id
+        runner = _make_runner(session_db=db)
+        runner._gateway_session_origin_for_id = lambda sid: None  # persisted-only
+
+        # Per-user group: attacker shares user_id but has a different user_id_alt
+        # → different session key → must fail closed (was: allowed via user_id).
+        attacker = SessionSource(platform=Platform.SIGNAL, chat_id="signal-group",
+                                 chat_type="group", user_id="+15550001111",
+                                 user_id_alt="attacker-uuid")
+        assert await runner._resume_target_allowed(attacker, "victim_alt_group",
+                                                   allow_override=False) is False
+        # No-chat_id DM keyed purely on the participant: same block.
+        dm_attacker = SessionSource(platform=Platform.SIGNAL, chat_id=None,
+                                    chat_type="dm", user_id="+15550001111",
+                                    user_id_alt="attacker-uuid")
+        assert await runner._resume_target_allowed(dm_attacker, "victim_alt_dm",
+                                                   allow_override=False) is False
+
+        # Regression: a caller WITHOUT user_id_alt (Telegram-style, keyed on
+        # user_id) still resumes its own persisted per-user group row.
+        tg_db = SessionDB(db_path=tmp_path / "state_tg.db")
+        tg_db.create_session("own_group", "telegram", user_id="12345",
+                             chat_id="chat-a", chat_type="group")
+        tg_runner = _make_runner(session_db=tg_db)
+        tg_runner._gateway_session_origin_for_id = lambda sid: None
+        tg_caller = SessionSource(platform=Platform.TELEGRAM, chat_id="chat-a",
+                                  chat_type="group", user_id="12345")
+        assert await tg_runner._resume_target_allowed(tg_caller, "own_group",
+                                                      allow_override=False) is True
+
+        # Regression: an EXPLICITLY-shared group is unaffected — participant
+        # scoping doesn't apply, so an alt-keyed co-member still resumes.
+        runner.config.group_sessions_per_user = False
+        assert await runner._resume_target_allowed(attacker, "victim_alt_group",
+                                                   allow_override=False) is True
+        db.close()
+        tg_db.close()
+
     @pytest.mark.asyncio
     async def test_gateway_dispatches_sessions_command(self, tmp_path):
         from hermes_state import SessionDB