security(gateway): fail closed on blank-source rows in /resume scoping

Addresses egilewski (Codex) CR on PR #52355: the persisted-row fallback in
_resume_target_allowed() skipped the platform/source check when sessions.source
was blank (the row_src guard only rejects a *mismatching* non-blank source),
then accepted the row on user_id equality alone. A legacy/malformed row with a
blank source but a matching user_id was therefore resumable — an identified
caller could bind to a transcript whose origin it can't prove.

Now an identity-bearing caller is allowed only when the row proves BOTH the
same owner (non-blank user_id match) AND the same platform/origin (non-blank
source match). A blank/legacy source fails closed, exactly like a missing
user_id. No-identity (single-user) callers are unaffected.

Adds a regression replaying the reviewer's blank-source same-uid probe.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

gateway/slash_commands.py

@@ -780,12 +780,21 @@ class GatewaySlashCommandsMixin:
         row_uid = str(row.get("user_id") or "")
         if caller_uid:
             # Identity-bearing caller: allow only when the row PROVES the same
-            # owner. A row with no/blank user_id cannot be proven to belong to
-            # this caller, so fail closed — an identified user must not bind to
-            # an unowned or other-owned persisted session by id/title. (Legacy
-            # NULL-owner rows are intentionally not resumable this way; use a
-            # live session or an explicit admin override.)
-            return bool(row_uid) and row_uid == caller_uid
+            # owner AND the same platform/origin. A row with no/blank user_id
+            # cannot be proven to belong to this caller; a row with no/blank
+            # source cannot be proven to share the caller's platform (the
+            # row_src check above only rejects a *mismatching* non-blank source,
+            # so a blank/legacy source would otherwise slip through on user_id
+            # equality alone). Either gap fails closed — an identified user must
+            # not bind to an unowned, other-owned, or unproven-origin persisted
+            # session by id/title. (Legacy NULL-owner or blank-source rows are
+            # intentionally not resumable this way; use a live session or an
+            # explicit admin override.)
+            return (
+                bool(row_uid) and row_uid == caller_uid
+                and bool(row_src) and bool(caller_src)
+                and str(row_src) == str(caller_src)
+            )
         # No caller identity (single-user / no-identity context): there is no
         # cross-user boundary to enforce beyond the same-platform check above.
         return True

tests/gateway/test_resume_command.py

@@ -473,6 +473,33 @@ class TestHandleSessionsCommand:
             assert "Resumed" not in result, name
         db.close()
 
+    @pytest.mark.asyncio
+    async def test_resume_blocks_blank_source_same_uid_row(self, tmp_path):
+        """A persisted row whose `source` is blank/legacy cannot prove it shares
+        the caller's platform, so user_id equality alone must NOT authorize a
+        resume — the blank source fails closed exactly like a missing user_id
+        (IDOR regression: an identified caller could otherwise bind to an
+        unproven-origin transcript)."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("blank_source_same_uid", "telegram", user_id="12345")
+        db.set_session_title("blank_source_same_uid", "Blank Source Same UID")
+        # Simulate a malformed/legacy row that does not record its origin.
+        db._conn.execute(
+            "UPDATE sessions SET source = '' WHERE id = ?", ("blank_source_same_uid",)
+        )
+        db._conn.commit()
+        db.create_session("current_session_001", "telegram", user_id="12345")
+
+        for name in ("Blank Source Same UID", "blank_source_same_uid"):
+            event = _make_event(text=f"/resume {name}")
+            runner = _make_runner(session_db=db, current_session_id="current_session_001",
+                                  event=event)
+            result = await runner._handle_resume_command(event)
+            runner.session_store.switch_session.assert_not_called()
+            assert "Resumed" not in result, name
+        db.close()
+
     @pytest.mark.asyncio
     async def test_gateway_dispatches_sessions_command(self, tmp_path):
         from hermes_state import SessionDB