Проверка доступа к чатам при показе истории и доставке событий

chats/admin.py

@@ -211,6 +211,7 @@ def publish_change(self, chat, event_type, content):
             event = {
                 "type": event_type,
                 "content": {"chat_id": chat.pk, "chat_type": ChatType.DIRECT, **content},
+                "_direct_chat_id": chat.pk,
             }
             for user in chat.users.all():
                 channel = cache.get(get_user_channel_cache_key(user))

chats/consumers/chat.py

@@ -7,12 +7,12 @@
 from channels.generic.websocket import AsyncJsonWebsocketConsumer
 from django.conf import settings
 from django.core.cache import cache
-from django.db.models import Q
 from django.utils import timezone
 
 from chats.exceptions import ChatException
-from chats.models import BaseChat
+from chats.permissions import accessible_chats, direct_chat_is_accessible
 from chats.project_reads import personalize_project_message_event
+from chats.services.messages import get_direct_chat_for_participants
 from chats.utils import get_user_channel_cache_key
 from chats.websockets_settings import (
     Event,
@@ -22,41 +22,26 @@
 )
 from core.constants import ONE_DAY_IN_SECONDS, ONE_WEEK_IN_SECONDS
 from core.utils import get_user_online_cache_key, get_users_online_cache_key
-from projects.models import Collaborator
 from users.models import CustomUser
 from chats.consumers.event_types import DirectEvent, ProjectEvent
 from chats.utils import get_chat_and_user_ids_from_content
-from chats.models import DirectChat, ProjectChat
+from chats.models import ProjectChat
 
 
 @database_sync_to_async
-def get_user_project_ids(user_id: int) -> list[int]:
-    return list(
-        Collaborator.objects.filter(user_id=user_id).values_list("project", flat=True)
-    )
+def get_user_project_ids(user) -> list[int]:
+    return list(accessible_chats(user, ProjectChat).values_list("pk", flat=True))
 
 
 @database_sync_to_async
-def can_access_project_chat(user_id, chat_id):
-    try:
-        chat_id = int(chat_id)
-    except (TypeError, ValueError):
-        return False
-    return (
-        ProjectChat.objects.filter(pk=chat_id)
-        .filter(Q(project__leader_id=user_id) | Q(project__collaborator__user_id=user_id))
-        .exists()
-    )
+def can_access_project_chat(user, chat_id):
+    return accessible_chats(user, ProjectChat).filter(pk=chat_id).exists()
 
 
 class ChatConsumer(AsyncJsonWebsocketConsumer):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.room_name: str = ""
         self.user: Optional[CustomUser] = None
-        self.chat_type = None
-        self.chat: Optional[BaseChat] = None
-        self.event = None
         self.joined_rooms: set[str] = set()
 
     async def connect(self):
@@ -72,17 +57,11 @@ async def connect(self):
         )
 
         if not settings.RUNNING_TESTS:
-            # get all projects that user is a member of
-            project_ids_list = await get_user_project_ids(self.user.id)
+            project_ids_list = await get_user_project_ids(self.user)
             for project_id in project_ids_list:
-                # FIXME: if a user is a leader but not a collaborator, this doesn't work
-                #  upd: it seems not possible to be a leader without being a collaborator
-                # join room for each project -
-                # It's currently not possible to do this in a single call, -
-                #  so we have to do it in a loop (e.g. that's O(N) calls to layer backend, redis cache that would be) -
-                room_name = f"{EventGroupType.CHATS_RELATED}_{project_id}"
-                await self.channel_layer.group_add(room_name, self.channel_name)
-                self.joined_rooms.add(room_name)
+                await self._ensure_room_subscription(
+                    f"{EventGroupType.CHATS_RELATED}_{project_id}"
+                )
 
         # set user online
         user_cache_key = get_user_online_cache_key(self.user)
@@ -120,6 +99,16 @@ async def disconnect(self, close_code):
             # don't need to proccess logic for disconnect
             #  if we are not logged in!
             return
+        for room_name in self.joined_rooms:
+            await self.channel_layer.group_discard(room_name, self.channel_name)
+        self.joined_rooms.clear()
+        await self.channel_layer.group_discard(
+            EventGroupType.GENERAL_EVENTS, self.channel_name
+        )
+        user_channel_key = get_user_channel_cache_key(self.user)
+        if cache.get(user_channel_key) != self.channel_name:
+            return
+        cache.delete(user_channel_key)
         online_users = cache.get(get_users_online_cache_key(), set())
         online_users.discard(self.user.id)
         cache.set(get_users_online_cache_key(), online_users)
@@ -135,7 +124,9 @@ async def disconnect(self, close_code):
     async def receive_json(self, content, **kwargs):
         """Receive message from WebSocket in JSON format"""
 
-        # todo reply_to key is not required
+        if not isinstance(content, dict) or not isinstance(content.get("type"), str):
+            await self.send_json({"error": "Некорректный формат события."})
+            return
         event = Event(type=content["type"], content=content.get("content"))
 
         # two event types - related to group chat and related to leave/connect
@@ -146,26 +137,33 @@ async def receive_json(self, content, **kwargs):
             EventType.DELETE_MESSAGE,
             EventType.EDIT_MESSAGE,
         ]:
-            if event.content["chat_type"] == ChatType.DIRECT:
-                self.event = DirectEvent(self.user, self.channel_layer, self.channel_name)
-            elif event.content["chat_type"] == ChatType.PROJECT:
-                if not await can_access_project_chat(
-                    self.user.pk, event.content.get("chat_id")
-                ):
+            if not isinstance(event.content, dict):
+                await self.send_json({"error": "Некорректные данные события."})
+                return
+            chat_type = event.content.get("chat_type")
+            chat_id = event.content.get("chat_id")
+            room_name = ""
+            if chat_type == ChatType.DIRECT:
+                handler = DirectEvent(self.user, self.channel_layer, self.channel_name)
+            elif chat_type == ChatType.PROJECT:
+                try:
+                    if type(chat_id) not in (int, str):
+                        raise ValueError
+                    project_id = int(chat_id)
+                except ValueError:
+                    await self.send_json({"error": "Некорректный идентификатор чата."})
+                    return
+                if not await can_access_project_chat(self.user, project_id):
                     await self.send_json({"error": "У вас нет доступа к чату проекта."})
                     return
-                self.event = ProjectEvent(
-                    self.user, self.channel_layer, self.channel_name
-                )
-
-            chat_id = event.content.get("chat_id")
-            if event.content["chat_type"] == ChatType.PROJECT:
-                chat_id = int(chat_id)
-            room_name = f"{EventGroupType.CHATS_RELATED}_{chat_id}"
-            if event.content["chat_type"] == ChatType.PROJECT:
+                handler = ProjectEvent(self.user, self.channel_layer, self.channel_name)
+                room_name = f"{EventGroupType.CHATS_RELATED}_{project_id}"
                 await self._ensure_room_subscription(room_name)
+            else:
+                await self.send_json({"error": "Неизвестный тип чата."})
+                return
             try:
-                await self.__process_chat_related_event(event, room_name)
+                await self.__process_chat_related_event(handler, event, room_name)
             except ChatException as e:
                 await self.send_json({"error": str(e.get_error())})
             except KeyError as e:
@@ -175,20 +173,19 @@ async def receive_json(self, content, **kwargs):
             room_name = EventGroupType.GENERAL_EVENTS
             await self.__process_general_event(event, room_name)
         else:
-            return self.disconnect(400)
+            await self.send_json({"error": "Неизвестный тип события."})
 
-    async def __process_chat_related_event(self, event, room_name):
-        self.event: DirectEvent | ProjectEvent
+    async def __process_chat_related_event(self, handler, event, room_name):
         if event.type == EventType.NEW_MESSAGE:
-            await self.event.process_new_message_event(event, room_name)
+            await handler.process_new_message_event(event, room_name)
         elif event.type == EventType.TYPING:
             await self.__process_typing_event(event, room_name)
         elif event.type == EventType.READ_MESSAGE:
-            await self.event.process_read_message_event(event, room_name)
+            await handler.process_read_message_event(event, room_name)
         elif event.type == EventType.DELETE_MESSAGE:
-            await self.event.process_delete_message_event(event, room_name)
+            await handler.process_delete_message_event(event, room_name)
         elif event.type == EventType.EDIT_MESSAGE:
-            await self.event.process_edit_message_event(event, room_name)
+            await handler.process_edit_message_event(event, room_name)
 
     async def __process_typing_event(self, event: Event, room_name: str):
         """Send typing event to room group."""
@@ -204,29 +201,37 @@ async def __process_typing_event(self, event: Event, room_name: str):
 
         if event.content["chat_type"] == ChatType.PROJECT:
             event_data["_project_chat_id"] = event.content["chat_id"]
+            await self.channel_layer.group_send(room_name, event_data)
 
-        if event.content["chat_type"] == ChatType.DIRECT:
+        else:
             _, other_user = await get_chat_and_user_ids_from_content(
                 event.content, self.user
             )
 
-            await database_sync_to_async(DirectChat.get_chat)(self.user, other_user)
+            chat = await database_sync_to_async(get_direct_chat_for_participants)(
+                self.user, other_user
+            )
+            event_data["content"]["chat_id"] = chat.pk
+            event_data["_direct_chat_id"] = chat.pk
 
             # send message to user's channel
             other_user_channel = cache.get(get_user_channel_cache_key(other_user), None)
 
-            if other_user_channel:
+            if other_user_channel and other_user_channel != self.channel_name:
                 await self.channel_layer.send(other_user_channel, event_data)
 
-        await self.channel_layer.group_send(
-            room_name,
-            event_data,
-        )
-
     async def _send_chat_event(self, event):
+        direct_chat_id = event.get("_direct_chat_id")
+        if direct_chat_id is not None and (
+            not self.user
+            or not await database_sync_to_async(direct_chat_is_accessible)(
+                self.user, direct_chat_id
+            )
+        ):
+            return
         chat_id = event.get("_project_chat_id")
         if chat_id is not None:
-            if not self.user or not await can_access_project_chat(self.user.pk, chat_id):
+            if not self.user or not await can_access_project_chat(self.user, chat_id):
                 room_name = f"{EventGroupType.CHATS_RELATED}_{int(chat_id)}"
                 await self.channel_layer.group_discard(room_name, self.channel_name)
                 self.joined_rooms.discard(room_name)
@@ -285,7 +290,7 @@ async def __process_general_event(self, event: Event, room_name: str):
         elif event.type == EventType.SET_OFFLINE:
             cache.delete(cache_key)
             users_online_list = cache.get_or_set(users_online_list_key, set())
-            users_online_list.remove(self.user.pk)
+            users_online_list.discard(self.user.pk)
             cache.set(users_online_list_key, users_online_list, ONE_DAY_IN_SECONDS)
 
             # sent everyone online event that user X is offline
@@ -301,6 +306,5 @@ async def __process_general_event(self, event: Event, room_name: str):
     async def decode_json(self, text_data) -> dict:
         try:
             return json.loads(text_data)
-        except JSONDecodeError as error:
-            await self.disconnect(400)
-            raise error
+        except JSONDecodeError:
+            return {}

chats/consumers/event_types/DirectEvent.py

@@ -1,16 +1,18 @@
 from channels.db import database_sync_to_async
 
-from chats.models import DirectChatMessage, DirectChat
+from chats.models import DirectChat
 from chats.websockets_settings import ChatType, Event, EventType
-from chats.exceptions import WrongChatIdException
 from django.core.cache import cache
 from chats.utils import (
     get_user_channel_cache_key,
     get_chat_and_user_ids_from_content,
-    orm_get,
-    orm_save,
 )
-from chats.services.messages import create_direct_message, delete_message, edit_message
+from chats.services.messages import (
+    create_direct_message,
+    delete_message,
+    edit_message,
+    mark_direct_message_read,
+)
 
 
 class DirectEvent:
@@ -41,39 +43,41 @@ async def process_new_message_event(self, event: Event, room_name: str):
         # send message to user's channel
         other_user_channel = cache.get(get_user_channel_cache_key(other_user), None)
 
-        event_data = {"type": EventType.NEW_MESSAGE, "content": content}
+        event_data = {
+            "type": EventType.NEW_MESSAGE,
+            "content": content,
+            "_direct_chat_id": chat_id,
+        }
 
         await self.channel_layer.send(self.channel_name, event_data)
 
         if created and other_user_channel and other_user_channel != self.channel_name:
             await self.channel_layer.send(other_user_channel, event_data)
 
     async def process_read_message_event(self, event: Event, room_name: str):
-        chat_id, other_user = await get_chat_and_user_ids_from_content(
-            event.content, self.user
+        _, other_user = await get_chat_and_user_ids_from_content(event.content, self.user)
+        chat_id = DirectChat.get_chat_id_from_users(self.user, other_user)
+        message_id = await database_sync_to_async(mark_direct_message_read)(
+            chat_id=chat_id,
+            message_id=event.content["message_id"],
+            user=self.user,
+            author=other_user,
         )
-        msg = await orm_get(DirectChatMessage.objects, pk=event.content["message_id"])
-        if msg.chat_id != chat_id or msg.author_id != other_user.id:
-            raise WrongChatIdException(
-                "Some of chat/message ids are wrong, you can't access this message"
-            )
-        msg.is_read = True
-        await orm_save(msg, update_fields=["is_read"])
         # send 2 events to user's channel
         other_user_channel = cache.get(get_user_channel_cache_key(other_user), None)
-        json_thingy = {
+        event_data = {
             "type": EventType.READ_MESSAGE,
             "content": {
-                "chat_id": event.content["chat_id"],
-                "chat_type": event.content["chat_type"],
+                "chat_id": chat_id,
+                "chat_type": ChatType.DIRECT,
                 "user_id": self.user.id,
-                "message_id": event.content["message_id"],
+                "message_id": message_id,
             },
+            "_direct_chat_id": chat_id,
         }
-        await self.channel_layer.send(self.channel_name, json_thingy)
-        if other_user_channel is None:
-            return
-        await self.channel_layer.send(other_user_channel, json_thingy)
+        await self.channel_layer.send(self.channel_name, event_data)
+        if other_user_channel and other_user_channel != self.channel_name:
+            await self.channel_layer.send(other_user_channel, event_data)
 
     async def process_delete_message_event(self, event: Event, room_name: str):
         _, other_user = await get_chat_and_user_ids_from_content(event.content, self.user)
@@ -87,18 +91,19 @@ async def process_delete_message_event(self, event: Event, room_name: str):
 
         # send message to user's channel
         other_user_channel = cache.get(get_user_channel_cache_key(other_user), None)
-        content = {
+        event_data = {
             "type": EventType.DELETE_MESSAGE,
             "content": {
                 "message_id": message_id,
                 "chat_id": chat_id,
                 "chat_type": ChatType.DIRECT,
             },
+            "_direct_chat_id": chat_id,
         }
-        await self.channel_layer.send(self.channel_name, content)
+        await self.channel_layer.send(self.channel_name, event_data)
 
         if other_user_channel and other_user_channel != self.channel_name:
-            await self.channel_layer.send(other_user_channel, content)
+            await self.channel_layer.send(other_user_channel, event_data)
 
     async def process_edit_message_event(self, event, room_name):
         _, other_user = await get_chat_and_user_ids_from_content(event.content, self.user)
@@ -118,7 +123,11 @@ async def process_edit_message_event(self, event, room_name):
         # send message to user's channel
         other_user_channel = cache.get(get_user_channel_cache_key(other_user), None)
 
-        event_data = {"type": EventType.EDIT_MESSAGE, "content": content}
+        event_data = {
+            "type": EventType.EDIT_MESSAGE,
+            "content": content,
+            "_direct_chat_id": chat_id,
+        }
 
         if other_user_channel and other_user_channel != self.channel_name:
             await self.channel_layer.send(other_user_channel, event_data)