Patched code sample

import asyncio

# The following are mock objects to create a self-contained, representative example.
# In the actual Open WebUI application, these would be part of the server framework.
class MockUser:
    """Represents the currently authenticated user making the request."""
    def __init__(self, user_id: str):
        self.id = user_id

class MockSocketIOManager:
    """Represents the Socket.IO server state."""
    def __init__(self):
        # Maps a client's socket ID (sid) to their session data,
        # which crucially includes the user ID that owns the session.
        self._sessions = {
            "socket_id_for_user_A": {"user_id": "user_A_id"},
            "socket_id_for_user_B": {"user_id": "user_B_id"},
        }
        self._connected_sids = set(self._sessions.keys())

    def is_connected(self, sid: str) -> bool:
        return sid in self._connected_sids

    async def get_session(self, sid: str) -> dict:
        return self._sessions.get(sid)

    async def emit(self, event: str, data: dict, to: str):
        # In a real application, this sends the event to the specified client.
        print(f"-> Emitting '{event}' to session '{to}' (Owner: {self._sessions[to]['user_id']})")

class MockApp:
    """Represents the main application."""
    sio = MockSocketIOManager()

app = MockApp()


async def get_event_call_fixed(session_id: str, event_data: dict, current_user: MockUser):
    """
    Fixed event handler that correctly authorizes execution events.

    This function represents the logic after the security patch was applied.
    """
    event = event_data.get("event")
    data = event_data.get("data")

    if event == "execute:python" or event == "execute:tool":
        # First, retrieve the session data associated with the target socket ID.
        session = await app.sio.get_session(session_id)

        # THE FIX:
        # Before emitting a privileged event, the server now verifies that
        # the user ID of the person making the request (`current_user.id`)
        # matches the user ID stored in the target session (`session.get("user_id")`).
        # The vulnerable version only checked `app.sio.is_connected(session_id)`.
        if app.sio.is_connected(session_id) and session and session.get("user_id") == current_user.id:
            await app.sio.emit(event, data, to=session_id)
            return {"status": True, "detail": "Event sent successfully."}

    # If the check fails, the request is denied.
    return {"status": False, "detail": f"User '{current_user.id}' not authorized for session '{session_id}'."}