fix(relay): scope interactive prompt resolution to the session owner

gateway/relay/adapter.py

@@ -27,10 +27,15 @@
 from gateway.relay.descriptor import CapabilityDescriptor
 from gateway.relay.media import RelayMediaClient
 from gateway.relay.transport import RelayTransport
-from gateway.session import SessionSource
+from gateway.session import SessionSource, build_session_key
 
 logger = logging.getLogger(__name__)
 
+# Cap on _session_owner_by_key (one small string pair per per-user channel
+# session this gateway has seen). Comfortably above the per-user session count
+# of any single relayed workspace, so trimming only ever drops long-idle rows.
+_SESSION_OWNER_CACHE_MAX = 2048
+
 
 def _utf16_len(text: str) -> int:
     """Count UTF-16 code units (Telegram's length unit)."""
@@ -103,6 +108,19 @@ def __init__(
         # Entries expire lazily (see _pop_prompt) so an unanswered prompt
         # never leaks. Keyed by our own minted 8-hex ids.
         self._pending_prompts: Dict[str, Dict[str, Any]] = {}
+        # session_key -> the participant id that session belongs to, for keys
+        # that isolate ONE member (per-user group/channel sessions). Recorded
+        # from the inbound source that minted the key (_remember_session_owner)
+        # rather than re-derived when an answer arrives, because a prompt answer
+        # can come back on a DIFFERENT lane whose source normalizes differently:
+        # a Discord component press travels the passthrough plane, where
+        # _discord_interaction_to_event stamps Platform.RELAY and carries no
+        # thread_id, so build_session_key over it yields a different STRING for
+        # the very same conversation (agent:main:relay:channel:… vs
+        # agent:main:discord:thread:…). Participant ids are identical on both
+        # lanes, so ownership is compared on them. Shared sessions are never
+        # recorded — an absent entry means "no single owner" and fails open.
+        self._session_owner_by_key: Dict[str, str] = {}
 
     # ── capability surface (from descriptor) ─────────────────────────────
     @property
@@ -314,6 +332,10 @@ def _capture_scope(self, event) -> None:
             egress guard declines the reply as 'target not routed to an
             onboarded tenant'. See gateway-gateway routedEgressGuard.ts /
             discordTenant.ts (makeDiscordTenantOf).
+
+        Also the capture point for per-user session ownership
+        (_remember_session_owner) — same "learn it from the inbound source"
+        shape, read back when an interactive prompt is answered.
         """
         try:
             src = getattr(event, "source", None)
@@ -344,9 +366,50 @@ def _capture_scope(self, event) -> None:
             scope = getattr(src, "scope_id", None)
             if scope:
                 self._scope_by_chat[str(chat)] = str(scope)
+            self._remember_session_owner(src)
         except Exception:  # noqa: BLE001 - scope tracking must never break inbound
             pass
 
+    def _remember_session_owner(self, src) -> None:
+        """Record which participant owns this event's session key (Phase 3 authz).
+
+        Only keys that ISOLATE one member are recorded. ``build_session_key``
+        appends the participant id last and only when isolation applies, so a
+        key ending in this sender's own id is per-user; anything else (a shared
+        group/channel, a shared thread, a DM) is deliberately left unowned so
+        every member can still answer its prompts.
+
+        Captured here — on the lane that CREATED the session — because the lane
+        that answers a prompt may normalize the same conversation to a different
+        key (see ``_session_owner_by_key``). Never raises: this runs inside
+        ``_capture_scope``'s inbound-safe try.
+        """
+        if getattr(src, "chat_type", None) == "dm":
+            return
+        participant = getattr(src, "user_id_alt", None) or getattr(src, "user_id", None)
+        if not participant:
+            return
+        key = build_session_key(
+            src,
+            group_sessions_per_user=self.config.extra.get(
+                "group_sessions_per_user", True
+            ),
+            thread_sessions_per_user=self.config.extra.get(
+                "thread_sessions_per_user", False
+            ),
+        )
+        if not key.endswith(f":{participant}"):
+            return  # shared session — no single owner to protect
+        self._session_owner_by_key[str(key)] = str(participant)
+        # Bounded: an entry is only useful while a prompt for that session can
+        # still be answered (_pop_prompt expires them), so trimming the oldest
+        # insertions can never strand a live prompt for long. Dicts preserve
+        # insertion order, so the tail is the most recently seen.
+        excess = len(self._session_owner_by_key) - _SESSION_OWNER_CACHE_MAX
+        if excess > 0:
+            for stale in list(self._session_owner_by_key)[:excess]:
+                self._session_owner_by_key.pop(stale, None)
+
     def _with_scope(self, chat_id: str, metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
         """Ensure the outbound metadata carries the discriminator(s) the connector's
         egress guard needs to resolve the owning tenant.
@@ -1352,6 +1415,50 @@ async def send_clarify(
             chat_id, question, choices, clarify_id, session_key, metadata=metadata
         )
 
+    def _caller_owns_prompt_session(self, event, session_key: str) -> bool:
+        """True when the prompt answer's author may resolve ``session_key``.
+
+        ``_consume_prompt_response`` resolves a click *before* ``handle_message``
+        (see ``_on_passthrough``), so the normal ``_is_user_authorized`` gate
+        never runs for a consumed prompt. Compare the clicker against the owner
+        ``_remember_session_owner`` recorded for that session — the same
+        ``operator == session_user`` shape the native qqbot adapter uses, but
+        without its session-key parser (``build_session_key``'s trailing segment
+        is a participant id or a thread id depending on the shape above it, which
+        is exactly why ``gateway/run.py``'s ``_parse_session_key`` refuses to
+        interpret it).
+
+        Ownership is compared on the participant id rather than on a re-derived
+        session key because the two relay lanes normalize the same conversation
+        differently — see ``_session_owner_by_key``.
+
+        Fails open where there is no single owner to protect: an empty
+        ``session_key``, a shared group/thread session, a DM (1:1 by
+        construction), or a session this adapter never saw inbound (a cron- or
+        API-started turn, which is unchanged from before this gate existed).
+        A prompt answer carrying no author at all fails closed.
+        """
+        if not session_key:
+            return True
+        owner = self._session_owner_by_key.get(str(session_key))
+        if not owner:
+            return True
+        source = getattr(event, "source", None)
+        if source is None:
+            return False
+        # Either identifier the same authenticated author can present: the
+        # connector may send a platform-stable alt id inbound (Signal UUID,
+        # Feishu union_id) where the passthrough lane only has the raw user id.
+        caller_ids = {
+            str(candidate)
+            for candidate in (
+                getattr(source, "user_id_alt", None),
+                getattr(source, "user_id", None),
+            )
+            if candidate
+        }
+        return owner in caller_ids
+
     async def _consume_prompt_response(self, event) -> bool:
         """Route an inbound prompt_response to its waiting primitive.
 
@@ -1369,6 +1476,25 @@ async def _consume_prompt_response(self, event) -> bool:
         option_id = str(pr.get("option_id") or "")
         if not prompt_id or not option_id:
             return False
+        # Authorization (CWE-639): a prompt answer must come from the
+        # participant who owns the session. Resolution here precedes
+        # handle_message (see _on_passthrough), so the usual _is_user_authorized
+        # gate never runs for a consumed prompt — without this, any channel
+        # co-member who can see the buttons could resolve, and via "always"
+        # permanently allowlist, another user's dangerous-command approval. Peek
+        # (do not consume) so a stray click can't evict a still-pending prompt
+        # from under its real owner.
+        pending = self._pending_prompts.get(prompt_id)
+        if pending is not None and not self._caller_owns_prompt_session(
+            event, str(pending.get("session_key") or "")
+        ):
+            logger.warning(
+                "relay prompt_response for session %s rejected: caller %s is not "
+                "the session owner",
+                pending.get("session_key"),
+                getattr(getattr(event, "source", None), "user_id", None),
+            )
+            return True
         state = self._pop_prompt(prompt_id)
         if state is None:
             logger.info(

tests/gateway/relay/test_relay_interactive.py

@@ -27,7 +27,8 @@
 from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome
 from gateway.relay.adapter import RelayAdapter
 from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
-from gateway.session import SessionSource
+from gateway.relay.ws_transport import _event_from_wire
+from gateway.session import SessionSource, build_session_key
 
 from tests.gateway.relay.stub_connector import StubConnector
 
@@ -414,3 +415,211 @@ async def test_cancelled_outcome_removes_eyes_without_verdict():
     await adapter.on_processing_complete(event, ProcessingOutcome.CANCELLED)
     reacts = [(a["emoji"], a.get("remove", False)) for a in stub.sent if a["op"] == "react"]
     assert reacts == [("👀", True)]  # eyes removed, no ✅/❌
+
+
+# ── owner-scoped prompt authorization (CWE-639) ──────────────────────────
+
+
+def _inbound_channel_event(user_id, *, platform="discord", chat_id="chan42", **src_kw):
+    """A normal inbound channel message, built through the real wire decoder.
+
+    ``_event_from_wire`` keeps the UNDERLYING platform (``discord``), which is
+    what makes the session key differ from the one a Discord button press
+    normalizes to — the split this authorization has to survive.
+    """
+    src = {
+        "platform": platform,
+        "chat_id": chat_id,
+        "chat_type": "channel",
+        "scope_id": "guild9",
+        "user_id": user_id,
+    }
+    src.update(src_kw)
+    return _event_from_wire({"text": "hi", "message_type": "text", "source": src})
+
+
+def _prompt_response_event(base_event, prompt_id, option_id):
+    """The same conversation's source answering a prompt on the inbound lane."""
+    return MessageEvent(
+        text=f"/{option_id}",
+        message_type=MessageType.COMMAND,
+        source=base_event.source,
+        prompt_response={"prompt_id": prompt_id, "option_id": option_id},
+    )
+
+
+def _discord_button_forward(prompt_id, option_id, user_id, channel_id="chan42"):
+    """A real Discord component press as the passthrough plane delivers it."""
+
+    class Forward:
+        platform = "discord"
+        method = "POST"
+        path = "/interactions/bot1"
+        body = (
+            '{"type": 3, "id": "i1", "channel_id": "%s", "guild_id": "guild9",'
+            ' "message": {"id": "pm55"},'
+            ' "member": {"user": {"id": "%s", "username": "%s"}},'
+            ' "data": {"custom_id": "hp1:%s:%s"}}'
+            % (channel_id, user_id, user_id, prompt_id, option_id)
+        ).encode()
+
+    return Forward()
+
+
+def _own_a_session(adapter, event) -> str:
+    """Run ``event`` through the inbound capture and return its session key."""
+    adapter._capture_scope(event)
+    return build_session_key(
+        event.source,
+        group_sessions_per_user=adapter.config.extra.get("group_sessions_per_user", True),
+        thread_sessions_per_user=adapter.config.extra.get("thread_sessions_per_user", False),
+    )
+
+
+@pytest.fixture
+def approval_calls(monkeypatch):
+    calls: list = []
+    monkeypatch.setattr(
+        "tools.approval.resolve_gateway_approval",
+        lambda sk, choice, **kw: calls.append((sk, choice)) or 1,
+    )
+    return calls
+
+
+@pytest.mark.asyncio
+async def test_prompt_response_rejects_non_owner_in_per_user_channel(approval_calls):
+    """A co-member must not resolve another user's approval in a per-user
+    channel session — the relay analog of the native adapters' interaction
+    owner check. Resolution runs before handle_message, so without this the
+    normal auth gate never fires for the click."""
+    adapter, _stub = _adapter()
+    adapter.config.extra["group_sessions_per_user"] = True
+    owner_key = _own_a_session(adapter, _inbound_channel_event("owner"))
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": owner_key, "chat_id": "chan42"}
+    )
+
+    attacker = _inbound_channel_event("attacker")
+    consumed = await adapter._consume_prompt_response(
+        _prompt_response_event(attacker, pid, "always")
+    )
+    assert consumed is True  # dropped, not re-dispatched as the attacker's chat
+    assert approval_calls == []  # the victim's approval was NOT resolved
+    assert pid in adapter._pending_prompts  # left intact for its real owner
+
+
+@pytest.mark.asyncio
+async def test_prompt_response_allows_owner_in_per_user_channel(approval_calls):
+    adapter, _stub = _adapter()
+    adapter.config.extra["group_sessions_per_user"] = True
+    owner = _inbound_channel_event("owner")
+    owner_key = _own_a_session(adapter, owner)
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": owner_key, "chat_id": "chan42"}
+    )
+
+    consumed = await adapter._consume_prompt_response(
+        _prompt_response_event(owner, pid, "always")
+    )
+    assert consumed is True
+    assert approval_calls == [(owner_key, "always")]
+    assert pid not in adapter._pending_prompts
+
+
+@pytest.mark.asyncio
+async def test_prompt_response_shared_channel_allows_any_member(approval_calls):
+    """A shared session (group_sessions_per_user=False) has no participant id in
+    its key, so every co-member is a legitimate responder — the fix must not
+    regress that."""
+    adapter, _stub = _adapter()
+    adapter.config.extra["group_sessions_per_user"] = False
+    shared_key = _own_a_session(adapter, _inbound_channel_event("owner"))
+    assert adapter._session_owner_by_key == {}  # a shared key is never owned
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": shared_key, "chat_id": "chan42"}
+    )
+
+    other = _inbound_channel_event("someone_else")
+    consumed = await adapter._consume_prompt_response(
+        _prompt_response_event(other, pid, "always")
+    )
+    assert consumed is True
+    assert approval_calls == [(shared_key, "always")]
+
+
+@pytest.mark.asyncio
+async def test_prompt_response_unowned_session_still_resolves(approval_calls):
+    """A session this adapter never saw inbound (cron-/API-started turn) has no
+    recorded owner and must keep resolving — the gate adds a check, it must not
+    add a dead end."""
+    adapter, _stub = _adapter()
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": "agent:main:discord:channel:elsewhere", "chat_id": "chan42"}
+    )
+    assert await adapter._consume_prompt_response(
+        _prompt_response_event(_inbound_channel_event("whoever"), pid, "once")
+    ) is True
+    assert approval_calls == [("agent:main:discord:channel:elsewhere", "once")]
+
+
+# ── the same guard across the two relay lanes ────────────────────────────
+#
+# A Discord conversation arrives over the connector inbound lane, which KEEPS
+# the underlying platform (agent:main:discord:…), but its buttons come back on
+# the passthrough plane, where _discord_interaction_to_event normalizes to
+# Platform.RELAY and drops thread_id (agent:main:relay:channel:…). Ownership
+# must therefore be decided on the participant id, not on a re-derived key.
+
+
+@pytest.mark.asyncio
+async def test_discord_button_owner_resolves_across_relay_lanes(approval_calls):
+    adapter, _stub = _adapter(platform="discord")
+    adapter.config.extra["group_sessions_per_user"] = True
+    owner = _inbound_channel_event("owner")
+    owner_key = _own_a_session(adapter, owner)
+    assert owner_key.startswith("agent:main:discord:")  # the inbound lane's key
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": owner_key, "chat_id": "chan42"}
+    )
+
+    await adapter._on_passthrough(_discord_button_forward(pid, "always", "owner"))
+    assert approval_calls == [(owner_key, "always")]
+    assert pid not in adapter._pending_prompts
+
+
+@pytest.mark.asyncio
+async def test_discord_button_non_owner_rejected_across_relay_lanes(approval_calls):
+    adapter, _stub = _adapter(platform="discord")
+    adapter.config.extra["group_sessions_per_user"] = True
+    owner_key = _own_a_session(adapter, _inbound_channel_event("owner"))
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": owner_key, "chat_id": "chan42"}
+    )
+
+    await adapter._on_passthrough(_discord_button_forward(pid, "always", "attacker"))
+    assert approval_calls == []
+    assert pid in adapter._pending_prompts
+
+
+@pytest.mark.asyncio
+async def test_discord_thread_button_owner_resolves_across_relay_lanes(approval_calls):
+    """The lanes disagree about threads too: the connector sends a Discord
+    thread as chat_type=thread + thread_id, while the interaction body only has
+    the thread's channel_id. Per-user threads (thread_sessions_per_user) are the
+    case where that key would diverge on two segments, not just the platform."""
+    adapter, _stub = _adapter(platform="discord")
+    adapter.config.extra["group_sessions_per_user"] = True
+    adapter.config.extra["thread_sessions_per_user"] = True
+    owner = _inbound_channel_event(
+        "owner", chat_id="th9", chat_type="thread", thread_id="th9"
+    )
+    owner_key = _own_a_session(adapter, owner)
+    assert owner_key == "agent:main:discord:thread:th9:th9:owner"
+    pid = adapter._mint_prompt(
+        "exec_approval", {"session_key": owner_key, "chat_id": "th9"}
+    )
+
+    await adapter._on_passthrough(
+        _discord_button_forward(pid, "once", "owner", channel_id="th9")
+    )
+    assert approval_calls == [(owner_key, "once")]