fix workflow collaboration room authorization

api/controllers/console/socketio/workflow.py

@@ -51,7 +51,7 @@ def socket_connect(sid, environ, auth):
                 logging.warning("Socket connect rejected: no edit permission (user_id=%s, sid=%s)", user_id, sid)
                 return False
 
-            collaboration_service.save_session(sid, user)
+            collaboration_service.save_socket_identity(sid, user)
             return True
 
     except Exception:
@@ -68,7 +68,7 @@ def handle_user_connect(sid, data):
     if not workflow_id:
         return {"msg": "workflow_id is required"}, 400
 
-    result = collaboration_service.register_session(workflow_id, sid)
+    result = collaboration_service.authorize_and_join_workflow_room(workflow_id, sid)
     if not result:
         return {"msg": "unauthorized"}, 401
 

api/services/workflow_collaboration_service.py

@@ -4,9 +4,15 @@ import logging
 import time
 from collections.abc import Mapping
 
+from sqlalchemy import select
+
+from extensions.ext_database import db
 from models.account import Account
+from models.model import App
 from repositories.workflow_collaboration_repository import WorkflowCollaborationRepository, WorkflowSessionInfo
 
+logger = logging.getLogger(__name__)
+
 
 class WorkflowCollaborationService:
     def __init__(self, repository: WorkflowCollaborationRepository, socketio) -> None:
@@ -16,20 +22,39 @@ class WorkflowCollaborationService:
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}(repository={self._repository})"
 
-    def save_session(self, sid: str, user: Account) -> None:
+    def save_socket_identity(self, sid: str, user: Account) -> None:
+        """Persist the authenticated console user on the raw socket session."""
         self._socketio.save_session(
             sid,
             {
                 "user_id": user.id,
                 "username": user.name,
                 "avatar": user.avatar,
+                "tenant_id": user.current_tenant_id,
             },
         )
 
-    def register_session(self, workflow_id: str, sid: str) -> tuple[str, bool] | None:
+    def authorize_and_join_workflow_room(self, workflow_id: str, sid: str) -> tuple[str, bool] | None:
+        """
+        Join a collaboration room only after validating the socket session and tenant-scoped app access.
+
+        The Socket.IO payload still calls the room key `workflow_id`, but the identifier is the workflow app's
+        `App.id`. Returning `None` lets the controller reject the join before any Redis or room state is created.
+        """
         session = self._socketio.get_session(sid)
         user_id = session.get("user_id")
-        if not user_id:
+        tenant_id = session.get("tenant_id")
+        if not user_id or not tenant_id:
+            return None
+
+        if not self._can_access_workflow(workflow_id, str(tenant_id)):
+            logger.warning(
+                "Workflow collaboration join rejected: workflow_id=%s tenant_id=%s user_id=%s sid=%s",
+                workflow_id,
+                tenant_id,
+                user_id,
+                sid,
+            )
             return None
 
         session_info: WorkflowSessionInfo = {
@@ -52,6 +77,13 @@ class WorkflowCollaborationService:
 
         return str(user_id), is_leader
 
+    def _can_access_workflow(self, workflow_id: str, tenant_id: str) -> bool:
+        """Check that the collaboration room belongs to an active app in the caller's current tenant."""
+        app_id = db.session.scalar(
+            select(App.id).where(App.id == workflow_id, App.tenant_id == tenant_id).limit(1)
+        )
+        return app_id is not None
+
     def disconnect_session(self, sid: str) -> None:
         mapping = self._repository.get_sid_mapping(sid)
         if not mapping:

api/tests/unit_tests/services/test_workflow_collaboration_service.py

@@ -13,19 +13,25 @@ class TestWorkflowCollaborationService:
         socketio = Mock()
         return WorkflowCollaborationService(repository, socketio), repository, socketio
 
-    def test_register_session_returns_leader_status(
+    def test_authorize_and_join_workflow_room_returns_leader_status(
         self, service: tuple[WorkflowCollaborationService, Mock, Mock]
     ) -> None:
         # Arrange
         collaboration_service, repository, socketio = service
-        socketio.get_session.return_value = {"user_id": "u-1", "username": "Jane", "avatar": None}
+        socketio.get_session.return_value = {
+            "user_id": "u-1",
+            "username": "Jane",
+            "avatar": None,
+            "tenant_id": "t-1",
+        }
 
         with (
+            patch.object(collaboration_service, "_can_access_workflow", return_value=True),
             patch.object(collaboration_service, "get_or_set_leader", return_value="sid-1"),
             patch.object(collaboration_service, "broadcast_online_users"),
         ):
             # Act
-            result = collaboration_service.register_session("wf-1", "sid-1")
+            result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
 
         # Assert
         assert result == ("u-1", True)
@@ -33,7 +39,7 @@ class TestWorkflowCollaborationService:
         socketio.enter_room.assert_called_once_with("sid-1", "wf-1")
         socketio.emit.assert_called_once_with("status", {"isLeader": True}, room="sid-1")
 
-    def test_register_session_returns_none_when_missing_user(
+    def test_authorize_and_join_workflow_room_returns_none_when_missing_user(
         self, service: tuple[WorkflowCollaborationService, Mock, Mock]
     ) -> None:
         # Arrange
@@ -41,25 +47,58 @@ class TestWorkflowCollaborationService:
         socketio.get_session.return_value = {}
 
         # Act
-        result = collaboration_service.register_session("wf-1", "sid-1")
+        result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
 
         # Assert
         assert result is None
 
-    def test_repr_and_save_session(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
+    def test_authorize_and_join_workflow_room_returns_none_when_missing_tenant(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        collaboration_service, repository, socketio = service
+        socketio.get_session.return_value = {"user_id": "u-1", "username": "Jane", "avatar": None}
+
+        result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
+
+        assert result is None
+        repository.set_session_info.assert_not_called()
+        socketio.enter_room.assert_not_called()
+        socketio.emit.assert_not_called()
+
+    def test_authorize_and_join_workflow_room_returns_none_when_workflow_is_not_accessible(
+        self, service: tuple[WorkflowCollaborationService, Mock, Mock]
+    ) -> None:
+        collaboration_service, repository, socketio = service
+        socketio.get_session.return_value = {
+            "user_id": "u-1",
+            "username": "Jane",
+            "avatar": None,
+            "tenant_id": "t-1",
+        }
+
+        with patch.object(collaboration_service, "_can_access_workflow", return_value=False):
+            result = collaboration_service.authorize_and_join_workflow_room("wf-1", "sid-1")
+
+        assert result is None
+        repository.set_session_info.assert_not_called()
+        socketio.enter_room.assert_not_called()
+        socketio.emit.assert_not_called()
+
+    def test_repr_and_save_socket_identity(self, service: tuple[WorkflowCollaborationService, Mock, Mock]) -> None:
         collaboration_service, _repository, socketio = service
         user = Mock()
         user.id = "u-1"
         user.name = "Jane"
         user.avatar = "avatar.png"
+        user.current_tenant_id = "t-1"
 
         assert "WorkflowCollaborationService" in repr(collaboration_service)
 
-        collaboration_service.save_session("sid-1", user)
+        collaboration_service.save_socket_identity("sid-1", user)
 
         socketio.save_session.assert_called_once_with(
             "sid-1",
-            {"user_id": "u-1", "username": "Jane", "avatar": "avatar.png"},
+            {"user_id": "u-1", "username": "Jane", "avatar": "avatar.png", "tenant_id": "t-1"},
         )
 
     def test_relay_collaboration_event_unauthorized(