security: enforce approval ownership on every transport, not just Slack

coworker/server/manager.py

@@ -3696,6 +3696,42 @@ def _slack_actor_owns_item(
                 return False
         return bool(actor_id) and actor_id in self.slack_approval_owner_ids(owner_team)
 
+    # Protected item kinds: resolving one runs a consequential action / grants a filesystem
+    # root / commits a plan, so only an authorized owner may resolve it — never a bare
+    # allow-listed chat member. Questions and free-text answers are NOT protected.
+    _PROTECTED_ITEM_KINDS = {"approval", "directory", "plan"}
+
+    def _actor_owns_protected_item(
+        self,
+        item,
+        *,
+        platform: str,
+        actor_id: str,
+        chat_id: str,
+        team_id: Optional[str],
+    ) -> bool:
+        """Authorize a protected-item resolution over ANY transport.
+
+        The owner check used to run only on the Slack lane, so a Telegram reply (or button)
+        resolved protected approval/directory/plan items with no owner or channel-binding
+        validation at all — and items mirrored to Slack were resolvable from Telegram,
+        bypassing Slack's enforcement. This gates every transport:
+
+        - cross-transport is refused: an item bound to one channel can only be resolved from
+          that same channel (an in-app-only item is not remotely resolvable at all);
+        - Slack defers to the existing owner + bound-channel check;
+        - no other transport (e.g. Telegram) has an approval-owner model, so it cannot
+          resolve protected items remotely — they stay pending for in-app resolution.
+        """
+        binding = self.inbox_routing.binding_for(item.inbox)
+        if binding.channel and binding.channel != platform:
+            return False
+        if platform == "slack":
+            return self._slack_actor_owns_item(
+                item, actor_id=actor_id, chat_id=chat_id, team_id=team_id
+            )
+        return False
+
     def set_inbox_binding(
         self, name: str, *, channel: Optional[str], target: str
     ) -> dict[str, Any]:
@@ -4383,15 +4419,11 @@ async def _on_interaction(self, event) -> None:
         item = self.inbox.get(item_id)
         if item is None:
             return
-        protected_kinds = {"approval", "directory", "plan"}
-        if (
-            getattr(event, "platform", "") == "slack"
-            and item.kind in protected_kinds
-        ):
-            actor_id = str(getattr(event, "user_id", "") or "")
-            if not self._slack_actor_owns_item(
+        if item.kind in self._PROTECTED_ITEM_KINDS:
+            if not self._actor_owns_protected_item(
                 item,
-                actor_id=actor_id,
+                platform=str(getattr(event, "platform", "") or ""),
+                actor_id=str(getattr(event, "user_id", "") or ""),
                 chat_id=getattr(event, "chat_id", "") or "",
                 team_id=getattr(event, "team_id", None),
             ):
@@ -4429,14 +4461,11 @@ def _resolve(item_id: str, resolution: str) -> bool:
             item = self.inbox.get(item_id)
             if item is None:
                 return False
-            if (
-                getattr(event.source, "platform", "") == "slack"
-                and item.kind in {"approval", "directory", "plan"}
-            ):
-                actor_id = str(getattr(event.source, "user_id", "") or "")
-                if not self._slack_actor_owns_item(
+            if item.kind in self._PROTECTED_ITEM_KINDS:
+                if not self._actor_owns_protected_item(
                     item,
-                    actor_id=actor_id,
+                    platform=str(getattr(event.source, "platform", "") or ""),
+                    actor_id=str(getattr(event.source, "user_id", "") or ""),
                     chat_id=getattr(event.source, "chat_id", "") or "",
                     team_id=getattr(event.source, "team_id", None),
                 ):

tests/test_telegram_approval_bypass.py

@@ -0,0 +1,100 @@
+"""Approval ownership must hold on EVERY transport, not just the Slack lane.
+
+The owner check for protected items (approval/directory/plan) used to run only when
+platform == "slack". A Telegram reply or button therefore resolved protected items with no
+owner or channel-binding validation, and items mirrored to Slack were resolvable from
+Telegram. These tests pin the cross-transport enforcement. Questions stay answerable by any
+allow-listed member.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from coworker.connectors.base import InteractionEvent, MessageEvent, SessionSource
+from coworker.interactions import encode
+from coworker.providers import ModelCapabilities, ProviderClient
+from coworker.server.manager import SessionManager
+
+
+class NoTurnsProvider(ProviderClient):
+    def complete(self, *, model, messages, tools=None, **settings):
+        raise AssertionError("no model turns expected")
+
+    def capabilities(self, model):
+        return ModelCapabilities()
+
+
+def _manager(tmp_path) -> SessionManager:
+    return SessionManager(data_dir=tmp_path / "data", provider=NoTurnsProvider())
+
+
+def _tg_reply(text: str) -> MessageEvent:
+    return MessageEvent(
+        text=text,
+        source=SessionSource(
+            platform="telegram", chat_id="99", user_id="tg-user", user_name="TG"
+        ),
+    )
+
+
+def test_telegram_reply_cannot_resolve_a_protected_approval(tmp_path):
+    manager = _manager(tmp_path)
+    approval = manager.inbox.add_approval("s1", "Run it?")
+    consumed = manager._resolve_inbox_reply(_tg_reply(f"approve [ow:{approval.id}]"))
+    # The token was recognized (consumed), but the protected item must NOT be resolved.
+    assert consumed is True
+    assert manager.inbox.get(approval.id).state == "pending"
+
+
+def test_telegram_button_cannot_resolve_a_protected_approval(tmp_path):
+    manager = _manager(tmp_path)
+
+    class GatewayStub:
+        def __init__(self):
+            self.rejections = []
+
+        async def reject_interaction(self, event, text=""):
+            self.rejections.append(getattr(event, "user_id", None))
+
+        async def update_message(self, *args):
+            raise AssertionError("a refused interaction must not update the message")
+
+    gateway = GatewayStub()
+    manager.gateway = gateway
+    approval = manager.inbox.add_approval("s1", "Run it?")
+
+    asyncio.run(
+        manager._on_interaction(
+            InteractionEvent(
+                platform="telegram",
+                chat_id="99",
+                message_id="1",
+                value=encode(approval.id, "allow"),
+                user_id="tg-user",
+                user_name="TG",
+            )
+        )
+    )
+    assert manager.inbox.get(approval.id).state == "pending"
+    assert gateway.rejections == ["tg-user"]
+
+
+def test_telegram_reply_can_still_answer_a_question(tmp_path):
+    # Questions are not protected — an allow-listed member may answer from any transport.
+    manager = _manager(tmp_path)
+    question = manager.inbox.add_question("s1", "Which region?", options=["A", "B"])
+    consumed = manager._resolve_inbox_reply(_tg_reply(f"A [ow:{question.id}]"))
+    assert consumed is True
+    assert manager.inbox.get(question.id).resolution == "A"
+
+
+def test_slack_bound_item_is_not_resolvable_from_telegram(tmp_path):
+    # Cross-transport: an item whose inbox is bound to Slack must not be resolved by a
+    # Telegram reply, even though Slack's own owner check would apply on the Slack lane.
+    manager = _manager(tmp_path)
+    manager.inbox_routing.set_binding("ops", channel="slack", target="T1:C1")
+    approval = manager.inbox.add_approval("s1", "Deploy?", inbox="ops")
+    consumed = manager._resolve_inbox_reply(_tg_reply(f"approve [ow:{approval.id}]"))
+    assert consumed is True
+    assert manager.inbox.get(approval.id).state == "pending"