From 3c5a8c87e2459809258675b60579c52972f24e06 Mon Sep 17 00:00:00 2001 From: lizerk Date: Mon, 17 Aug 2026 13:54:35 +0300 Subject: [PATCH] Update application functionality and supporting components --- .env.example | 6 + backend/api/auth/endpoints/__init__.py | 2 + backend/api/auth/endpoints/websocket.py | 31 ++ backend/api/auth/models.py | 9 +- backend/api/auth/realtime.py | 38 +++ backend/api/auth/urls.py | 2 + backend/api/auth/views.py | 2 + backend/api/auth/websocket_tickets.py | 85 +++++ backend/communications/consumers.py | 187 ++++++----- backend/communications/realtime.py | 60 ++++ backend/communications/rules.py | 34 +- backend/communications/signals.py | 28 +- backend/communications/utils.py | 142 ++++---- .../docs/architecture/websocket-asyncapi.yaml | 50 ++- backend/docs/guides/COMMENTS_USAGE.md | 11 +- backend/employees/utils.py | 12 +- backend/eusrr_backend/asgi.py | 15 +- backend/eusrr_backend/channels_jwt.py | 71 +++- backend/eusrr_backend/settings.py | 18 + backend/eusrr_backend/settings_test.py | 1 + backend/realtime/consumers.py | 243 +++++++++++--- backend/realtime/protocol.py | 105 ++++++ backend/static/js/components/userWebSocket.js | 7 +- .../js/notifications/notification-manager.js | 84 +---- backend/templates/api/websocket_docs.html | 6 +- .../realtime/test_jwt_session_middleware.py | 33 +- backend/tests/realtime/test_security.py | 275 ++++++++++++++++ docs/guides/WEBSOCKET.md | 30 +- frontend/src/app/documents/page.tsx | 28 +- frontend/src/app/messages/[chatId]/page.tsx | 10 +- frontend/src/app/tasks/page.tsx | 14 +- .../DocumentAcknowledgementAudience.tsx | 180 ++++++++++ .../documents/DocumentDetailModal.tsx | 131 +------- .../components/feed/FeedRegulationCard.tsx | 12 +- frontend/src/contexts/CalendarContext.tsx | 15 +- frontend/src/hooks/useMessagesChatList.ts | 4 + frontend/src/hooks/useTaskBoardRealtime.ts | 14 +- .../document-acknowledgement-audience.test.js | 62 ++++ .../lib/document-acknowledgement-audience.ts | 66 ++++ frontend/src/lib/websocketManager.test.js | 120 +++++++ frontend/src/lib/websocketManager.ts | 307 +++++++++--------- frontend/src/services/calendarService.test.js | 8 + 42 files changed, 1879 insertions(+), 679 deletions(-) create mode 100644 backend/api/auth/endpoints/websocket.py create mode 100644 backend/api/auth/realtime.py create mode 100644 backend/api/auth/websocket_tickets.py create mode 100644 backend/communications/realtime.py create mode 100644 backend/realtime/protocol.py create mode 100644 backend/tests/realtime/test_security.py create mode 100644 frontend/src/components/documents/DocumentAcknowledgementAudience.tsx create mode 100644 frontend/src/lib/document-acknowledgement-audience.test.js create mode 100644 frontend/src/lib/document-acknowledgement-audience.ts diff --git a/.env.example b/.env.example index dcb05390..fb322f10 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,9 @@ DEBUG=False SECRET_KEY=change-me-django-secret-key ALLOWED_HOSTS=localhost,127.0.0.1,web,nginx +WEBSOCKET_ALLOW_QUERY_JWT=false +# Set explicit frontend origins when WebSocket and frontend use different hosts. +# WEBSOCKET_ALLOWED_ORIGINS=https://app.example.com # Database Settings USE_SQLITE=False @@ -14,6 +17,9 @@ DB_PORT=5432 # Redis Settings REDIS_HOST=redis REDIS_PORT=6379 +CHANNEL_LAYER_CAPACITY=1000 +CHANNEL_LAYER_EXPIRY=60 +CHANNEL_LAYER_GROUP_EXPIRY=86400 # Celery Settings CELERY_BROKER_URL=redis://redis:6379/0 diff --git a/backend/api/auth/endpoints/__init__.py b/backend/api/auth/endpoints/__init__.py index da7f8cc8..6d79e29b 100644 --- a/backend/api/auth/endpoints/__init__.py +++ b/backend/api/auth/endpoints/__init__.py @@ -21,6 +21,7 @@ from .sessions import ( SessionListAPIView, ) from .tokens import JWTTokenRefreshView, PhoneOrEmailTokenObtainPairView +from .websocket import WebSocketTicketAPIView __all__ = [ "ChangePasswordAPIView", @@ -42,4 +43,5 @@ __all__ = [ "SessionDetailAPIView", "SessionListAPIView", "VerifyEmailAPIView", + "WebSocketTicketAPIView", ] diff --git a/backend/api/auth/endpoints/websocket.py b/backend/api/auth/endpoints/websocket.py new file mode 100644 index 00000000..dc3a324b --- /dev/null +++ b/backend/api/auth/endpoints/websocket.py @@ -0,0 +1,31 @@ +from drf_spectacular.utils import extend_schema +from rest_framework import serializers, status +from rest_framework.response import Response + +from ..base import SessionBaseAPIView +from ..websocket_tickets import TICKET_MAX_AGE_SECONDS, issue_websocket_ticket + + +class WebSocketTicketResponseSerializer(serializers.Serializer): + ticket = serializers.CharField(read_only=True) + expires_in = serializers.IntegerField(read_only=True) + + +class WebSocketTicketAPIView(SessionBaseAPIView): + throttle_scope = "websocket_ticket" + + @extend_schema( + tags=["Auth"], + summary="Получить одноразовый билет для WebSocket-подключения", + request=None, + responses={200: WebSocketTicketResponseSerializer}, + ) + def post(self, request): + ticket = issue_websocket_ticket( + user=request.user, + session_id=self.get_current_session_id(request), + ) + return Response( + {"ticket": ticket, "expires_in": TICKET_MAX_AGE_SECONDS}, + status=status.HTTP_200_OK, + ) diff --git a/backend/api/auth/models.py b/backend/api/auth/models.py index 47c508ee..1349a86b 100644 --- a/backend/api/auth/models.py +++ b/backend/api/auth/models.py @@ -80,6 +80,12 @@ class UserAuthSession(models.Model): self.revoked_reason = reason if commit: self.save(update_fields=["revoked_at", "revoked_reason"]) + from .realtime import schedule_auth_session_revoked + + schedule_auth_session_revoked( + user_id=self.user_id, + session_id=str(self.session_id), + ) class QrLoginToken(models.Model): @@ -251,7 +257,8 @@ class QrLoginRequest(models.Model): verbose_name_plural = "QR-запросы входа" def __str__(self) -> str: - return f"{self.requester_device_name or 'unknown'}:{self.created_at:%Y-%m-%d %H:%M:%S}" + device = self.requester_device_name or "unknown" + return f"{device}:{self.created_at:%Y-%m-%d %H:%M:%S}" @property def is_expired(self) -> bool: diff --git a/backend/api/auth/realtime.py b/backend/api/auth/realtime.py new file mode 100644 index 00000000..36db981c --- /dev/null +++ b/backend/api/auth/realtime.py @@ -0,0 +1,38 @@ +"""Realtime propagation of authentication-session revocation.""" + +from __future__ import annotations + +import logging + +from asgiref.sync import async_to_sync +from channels.layers import get_channel_layer +from django.db import transaction + +logger = logging.getLogger(__name__) + + +def schedule_auth_session_revoked(*, user_id: int, session_id: str) -> None: + def _send() -> None: + channel_layer = get_channel_layer() + if channel_layer is None: + return + try: + async_to_sync(channel_layer.group_send)( + f"user_{user_id}", + { + "type": "auth_session_revoked", + "session_id": str(session_id), + }, + ) + except Exception as error: + logger.exception( + "Auth session revocation realtime update failed", + extra={ + "event": "authentication.session.realtime_revocation_failed", + "user_id": user_id, + "auth_session_id": str(session_id), + "error_type": type(error).__name__, + }, + ) + + transaction.on_commit(_send) diff --git a/backend/api/auth/urls.py b/backend/api/auth/urls.py index 1c117536..5e70219e 100644 --- a/backend/api/auth/urls.py +++ b/backend/api/auth/urls.py @@ -20,11 +20,13 @@ from .views import ( SessionDetailAPIView, SessionListAPIView, VerifyEmailAPIView, + WebSocketTicketAPIView, ) urlpatterns = [ path("token/", PhoneOrEmailTokenObtainPairView.as_view(), name="token_obtain_pair"), path("token/refresh/", JWTTokenRefreshView.as_view(), name="token_refresh"), + path("ws-ticket/", WebSocketTicketAPIView.as_view(), name="ws-ticket"), path("qr-login/", QrLoginCreateAPIView.as_view(), name="qr-login"), path( "qr-login/exchange/", diff --git a/backend/api/auth/views.py b/backend/api/auth/views.py index e3f62c41..f2bee7b9 100644 --- a/backend/api/auth/views.py +++ b/backend/api/auth/views.py @@ -23,6 +23,7 @@ from .endpoints import ( SessionDetailAPIView, SessionListAPIView, VerifyEmailAPIView, + WebSocketTicketAPIView, ) __all__ = [ @@ -47,5 +48,6 @@ __all__ = [ "SessionDetailAPIView", "SessionListAPIView", "VerifyEmailAPIView", + "WebSocketTicketAPIView", "authenticate", ] diff --git a/backend/api/auth/websocket_tickets.py b/backend/api/auth/websocket_tickets.py new file mode 100644 index 00000000..d564667c --- /dev/null +++ b/backend/api/auth/websocket_tickets.py @@ -0,0 +1,85 @@ +"""Short-lived, single-use credentials for browser WebSocket handshakes.""" + +from __future__ import annotations + +import hashlib +import secrets +from dataclasses import dataclass + +from django.contrib.auth import get_user_model +from django.core import signing +from django.core.cache import cache + +from .models import UserAuthSession + +User = get_user_model() + +TICKET_MAX_AGE_SECONDS = 30 +TICKET_SALT = "eusrr.websocket.ticket.v1" + + +class WebSocketTicketError(ValueError): + pass + + +@dataclass(frozen=True) +class WebSocketTicketIdentity: + user: object + session_id: str | None + + +def issue_websocket_ticket(*, user, session_id: str | None) -> str: + return signing.dumps( + { + "uid": user.pk, + "sid": str(session_id) if session_id else None, + "nonce": secrets.token_urlsafe(24), + }, + salt=TICKET_SALT, + compress=True, + ) + + +def consume_websocket_ticket(raw_ticket: str) -> WebSocketTicketIdentity: + try: + payload = signing.loads( + raw_ticket, + salt=TICKET_SALT, + max_age=TICKET_MAX_AGE_SECONDS, + ) + user_id = int(payload["uid"]) + session_id = payload.get("sid") + nonce = str(payload["nonce"]) + except (signing.BadSignature, KeyError, TypeError, ValueError) as error: + raise WebSocketTicketError("invalid_or_expired_ticket") from error + + replay_digest = hashlib.sha256( + f"{user_id}:{session_id}:{nonce}".encode() + ).hexdigest() + replay_key = f"websocket-ticket-used:{replay_digest}" + try: + is_first_use = cache.add( + replay_key, + True, + timeout=TICKET_MAX_AGE_SECONDS * 2, + ) + except Exception as error: # noqa: BLE001 - backend-neutral cache boundary + raise WebSocketTicketError("ticket_store_unavailable") from error + if not is_first_use: + raise WebSocketTicketError("ticket_already_used") + + # The normal HTTP authentication already decided whether this account can + # issue a ticket. Session validity below remains authoritative for JWT + # sessions; Django-session tickets keep Django's authenticated identity. + user = User.objects.filter(pk=user_id).first() + if user is None: + raise WebSocketTicketError("user_unavailable") + + if session_id and not UserAuthSession.objects.filter( + user=user, + session_id=session_id, + revoked_at__isnull=True, + ).exists(): + raise WebSocketTicketError("session_revoked") + + return WebSocketTicketIdentity(user=user, session_id=session_id) diff --git a/backend/communications/consumers.py b/backend/communications/consumers.py index b4d3bf54..8028c9fc 100644 --- a/backend/communications/consumers.py +++ b/backend/communications/consumers.py @@ -12,13 +12,16 @@ WebSocket Consumer Mixin для чатов. """ import logging + +import rules from channels.db import database_sync_to_async from django.db.models import Prefetch, Q from django.utils import timezone +from realtime.protocol import parse_positive_int, parse_text + from .models import ( Chat, - ChatMembership, ChatReadState, Message, MessageReaction, @@ -80,11 +83,7 @@ class ChatConsumerMixin: Устанавливает активный чат и отправляет начальную историю. """ - chat_id = content.get("chat_id") - if not chat_id: - return - - chat_id = int(chat_id) + chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") # Проверяем доступ chat = await self._get_chat(chat_id) @@ -135,9 +134,9 @@ class ChatConsumerMixin: async def _handle_close_chat(self, content): """Закрытие чата - сброс активного чата""" - chat_id = content.get("chat_id") + chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") - if chat_id and int(chat_id) == self.active_chat_id: + if chat_id == self.active_chat_id: # Сбрасываем индикатор печати if self.active_chat_id: chat = await self._get_chat(self.active_chat_id) @@ -376,12 +375,19 @@ class ChatConsumerMixin: await self.send_json({"type": "error", "error": "No active chat"}) return - text = str(content.get("content") or content.get("text") or "").strip() - if not text: - return + text = parse_text( + content.get("content") or content.get("text"), + field="content", + max_length=20_000, + ) chat = await self._get_chat(self.active_chat_id) - if not chat: + if not chat or not await self._user_can_access(chat, self.user): + await self._deny_chat_action("send_message", self.active_chat_id) + return + + if not await self._user_can_send(chat, self.user): + await self._deny_chat_action("send_message", self.active_chat_id) return # Проверка прав на отправку @@ -415,14 +421,21 @@ class ChatConsumerMixin: if not self.active_chat_id: return - message_id = content.get("message_id") - new_content = content.get("content", "").strip() - - if not message_id or not new_content: - return + message_id = parse_positive_int(content.get("message_id"), field="message_id") + new_content = parse_text( + content.get("content"), + field="content", + max_length=20_000, + ) msg = await self._get_message(message_id) - if not msg or msg.author_id != self.user.id: + if ( + not msg + or msg.chat_id != self.active_chat_id + or not await self._user_can_access(msg.chat, self.user) + or not await self._user_can_change_message(msg, self.user) + ): + await self._deny_chat_action("edit_message", self.active_chat_id) return # Обновляем сообщение @@ -445,12 +458,16 @@ class ChatConsumerMixin: if not self.active_chat_id: return - message_id = content.get("message_id") - if not message_id: - return + message_id = parse_positive_int(content.get("message_id"), field="message_id") msg = await self._get_message(message_id) - if not msg or msg.author_id != self.user.id: + if ( + not msg + or msg.chat_id != self.active_chat_id + or not await self._user_can_access(msg.chat, self.user) + or not await self._user_can_delete_message(msg, self.user) + ): + await self._deny_chat_action("delete_message", self.active_chat_id) return chat_id = msg.chat_id @@ -470,14 +487,16 @@ class ChatConsumerMixin: async def _handle_add_reaction(self, content): """Добавление реакции""" - message_id = content.get("message_id") - emoji = content.get("emoji") - - if not message_id or not emoji: - return + message_id = parse_positive_int(content.get("message_id"), field="message_id") + emoji = parse_text(content.get("emoji"), field="emoji", max_length=32) msg = await self._get_message(message_id) - if not msg: + if ( + not msg + or msg.chat_id != self.active_chat_id + or not await self._user_can_access(msg.chat, self.user) + ): + await self._deny_chat_action("add_reaction", getattr(msg, "chat_id", None)) return # Создаем реакцию @@ -502,14 +521,19 @@ class ChatConsumerMixin: async def _handle_remove_reaction(self, content): """Удаление реакции""" - message_id = content.get("message_id") - emoji = content.get("emoji") - - if not message_id or not emoji: - return + message_id = parse_positive_int(content.get("message_id"), field="message_id") + emoji = parse_text(content.get("emoji"), field="emoji", max_length=32) msg = await self._get_message(message_id) - if not msg: + if ( + not msg + or msg.chat_id != self.active_chat_id + or not await self._user_can_access(msg.chat, self.user) + ): + await self._deny_chat_action( + "remove_reaction", + getattr(msg, "chat_id", None), + ) return # Удаляем реакцию @@ -533,7 +557,11 @@ class ChatConsumerMixin: return chat = await self._get_chat(self.active_chat_id) - if not chat: + if ( + not chat + or not await self._user_can_access(chat, self.user) + or not await self._user_can_send(chat, self.user) + ): return await self._set_typing_status(chat, self.user, True) @@ -552,13 +580,13 @@ class ChatConsumerMixin: }, ) - async def _handle_stop_typing(self): + async def _handle_stop_typing(self, content=None): """Остановка индикатора печати""" if not self.active_chat_id: return chat = await self._get_chat(self.active_chat_id) - if not chat: + if not chat or not await self._user_can_access(chat, self.user): return await self._set_typing_status(chat, self.user, False) @@ -580,16 +608,7 @@ class ChatConsumerMixin: вместо этого. Автоотметка происходит автоматически при загрузке сообщений. """ - chat_id = content.get("chat_id") - if not chat_id: - logger.warning( - "Deprecated WebSocket read marker call rejected", - extra={ - "event": "websocket.chat.deprecated_read_marker_rejected", - "reason": "chat_id_missing", - }, - ) - return + chat_id = parse_positive_int(content.get("chat_id"), field="chat_id") logger.warning( "Deprecated WebSocket read marker call ignored", @@ -606,9 +625,13 @@ class ChatConsumerMixin: async def _handle_vote_poll(self, content): """Голосование в опросе""" - poll_id = content.get("poll_id") + poll_id = parse_positive_int(content.get("poll_id"), field="poll_id") option_ids = content.get("option_ids") - if not self.active_chat_id or not poll_id or not isinstance(option_ids, list): + if ( + not self.active_chat_id + or not isinstance(option_ids, list) + or len(option_ids) > 50 + ): await self.send_json( { "type": "error", @@ -618,6 +641,15 @@ class ChatConsumerMixin: ) return + option_ids = [ + parse_positive_int(option_id, field="option_ids") + for option_id in option_ids + ] + chat = await self._get_chat(self.active_chat_id) + if not chat or not await self._user_can_access(chat, self.user): + await self._deny_chat_action("vote_poll", self.active_chat_id) + return + result = await self._cast_poll_vote( poll_id, option_ids, @@ -678,24 +710,19 @@ class ChatConsumerMixin: @database_sync_to_async def _get_available_chat_ids(self, user): """Получить ID всех доступных чатов пользователя""" - - # Получаем ID через membership - membership_chat_ids = list( - ChatMembership.objects.filter(user=user).values_list("chat_id", flat=True) - ) - - chat_ids = list( + candidates = ( Chat.objects.filter( Q(type="global") - | Q(id__in=membership_chat_ids) + | Q(memberships__user=user, memberships__is_active=True) + | Q(participants=user) + | Q(created_by=user) | Q(include_all_users=True) | Q(type="comments", flags__show_in_messages=True) ) - .values_list("id", flat=True) + .select_related("created_by", "context_content_type") .distinct() ) - - return chat_ids + return [chat.id for chat in candidates if user_can_access_chat(chat, user)] @database_sync_to_async def _get_chat(self, chat_id): @@ -716,30 +743,30 @@ class ChatConsumerMixin: @database_sync_to_async def _user_can_access(self, chat, user): """Проверка доступа пользователя к чату""" - if chat.type == "comments": - return user_can_access_chat(chat, user) + return user_can_access_chat(chat, user) - if chat.type == "global": - return True + @database_sync_to_async + def _user_can_send(self, chat, user): + return rules.test_rule("communications.send_message", user, chat) - if chat.type == "private": - return ChatMembership.objects.filter( - chat=chat, user=user, is_active=True - ).exists() + @database_sync_to_async + def _user_can_change_message(self, message, user): + return rules.test_rule("communications.change_message", user, message) - if chat.type == "group": - return ChatMembership.objects.filter( - chat=chat, user=user, is_active=True - ).exists() + @database_sync_to_async + def _user_can_delete_message(self, message, user): + return rules.test_rule("communications.delete_message", user, message) - if chat.type in ("channel", "announcement", "comments"): - if chat.include_all_users: - return user.is_active - return ChatMembership.objects.filter( - chat=chat, user=user, is_active=True - ).exists() - - return False + async def _deny_chat_action(self, action, chat_id): + await self.send_json( + { + "type": "error", + "code": "access_denied", + "action": action, + "chat_id": chat_id, + "error": "Чат не найден или доступ запрещён.", + } + ) @database_sync_to_async def _create_message(self, chat, user, content): diff --git a/backend/communications/realtime.py b/backend/communications/realtime.py new file mode 100644 index 00000000..aa437ba4 --- /dev/null +++ b/backend/communications/realtime.py @@ -0,0 +1,60 @@ +"""Realtime synchronization for chat access changes. + +The channel layer is an integration boundary: model signals publish only a +small access invalidation event, while the consumer owns group membership. +""" + +from __future__ import annotations + +import logging + +from asgiref.sync import async_to_sync +from channels.layers import get_channel_layer +from django.db import transaction + +logger = logging.getLogger(__name__) + + +def schedule_chat_access_changed( + *, + chat_id: int, + user_id: int, + membership_active: bool, +) -> None: + """Notify every live session after the membership transaction commits.""" + + def _send() -> None: + channel_layer = get_channel_layer() + if channel_layer is None: + logger.warning( + "Chat access update skipped because channel layer is unavailable", + extra={ + "event": "chat.realtime_access.channel_layer_unavailable", + "chat_id": chat_id, + "participant_id": user_id, + }, + ) + return + + try: + async_to_sync(channel_layer.group_send)( + f"user_{user_id}", + { + "type": "chat_access_changed", + "chat_id": chat_id, + "membership_active": membership_active, + }, + ) + except Exception as error: + logger.exception( + "Chat access realtime update failed", + extra={ + "event": "chat.realtime_access.send_failed", + "chat_id": chat_id, + "participant_id": user_id, + "membership_active": membership_active, + "error_type": type(error).__name__, + }, + ) + + transaction.on_commit(_send) diff --git a/backend/communications/rules.py b/backend/communications/rules.py index c249c4e9..2e40ef7b 100644 --- a/backend/communications/rules.py +++ b/backend/communications/rules.py @@ -5,9 +5,10 @@ django-rules: декларативные правила доступа для co https://github.com/dfunckt/django-rules """ -import rules import logging +import rules + logger = logging.getLogger(__name__) @@ -25,36 +26,9 @@ def is_superuser(user): @rules.predicate def is_chat_member(user, chat): """Пользователь является участником чата""" - if chat is None: - return False + from .utils import user_can_access_chat - # Используем прямые запросы к БД в обход prefetch cache - from .models import ChatMembership - - # Глобальный чат доступен всем - if chat.type == "global": - return True - - if chat.type == "comments" and chat.context_object_id: - try: - return chat.get_participants().filter(pk=user.pk).exists() - except Exception as error: - logger.exception( - "Comments chat membership participants could not be resolved", - extra={ - "event": "chat.permission.participant_resolution_failed", - "chat_id": chat.id, - "user_id": user.id, - "permission": "membership", - "error_type": type(error).__name__, - }, - ) - return False - - return ( - ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists() - or chat.participants.filter(pk=user.pk).exists() - ) + return user_can_access_chat(chat, user) @rules.predicate diff --git a/backend/communications/signals.py b/backend/communications/signals.py index cc60bd1d..46959b0e 100644 --- a/backend/communications/signals.py +++ b/backend/communications/signals.py @@ -1,15 +1,39 @@ # backend\communications\signals.py +import logging + from django.conf import settings from django.core.files.base import ContentFile from django.db.models import F -from django.db.models.signals import post_save, pre_save +from django.db.models.signals import post_delete, post_save, pre_save from django.dispatch import receiver from django.utils.module_loading import import_string -import logging logger = logging.getLogger(__name__) +@receiver(post_save, sender="communications.ChatMembership") +def synchronize_chat_membership_realtime(sender, instance, **kwargs): + """Keep live consumer group membership aligned with database access.""" + from communications.realtime import schedule_chat_access_changed + + schedule_chat_access_changed( + chat_id=instance.chat_id, + user_id=instance.user_id, + membership_active=instance.is_active, + ) + + +@receiver(post_delete, sender="communications.ChatMembership") +def revoke_deleted_chat_membership_realtime(sender, instance, **kwargs): + from communications.realtime import schedule_chat_access_changed + + schedule_chat_access_changed( + chat_id=instance.chat_id, + user_id=instance.user_id, + membership_active=False, + ) + + # NOTE: Project-specific chat creation signals should be in your project's app # Example: employees/signals.py - create_main_department_chat() diff --git a/backend/communications/utils.py b/backend/communications/utils.py index b7627b1a..4c882268 100644 --- a/backend/communications/utils.py +++ b/backend/communications/utils.py @@ -6,14 +6,15 @@ from __future__ import annotations import datetime -from datetime import datetime as dt -from datetime import timezone as dt_tz +import logging from django.utils import timezone from django.utils.dateparse import parse_datetime from .models import Chat, ChatMembership +logger = logging.getLogger(__name__) + def _coerce_ts(val: str | None) -> datetime.datetime: """ @@ -44,8 +45,8 @@ def _coerce_ts(val: str | None) -> datetime.datetime: iv = int(val) if iv > 10**12: # пришли миллисекунды iv //= 1000 - return dt.fromtimestamp(iv, tz=dt_tz.utc) - except Exception: + return datetime.datetime.fromtimestamp(iv, tz=datetime.UTC) + except (OSError, OverflowError, TypeError, ValueError): pass # Потом пробуем ISO-строку @@ -57,6 +58,60 @@ def _coerce_ts(val: str | None) -> datetime.datetime: return d +def _membership_or_legacy_participant(chat: Chat, user) -> bool: + membership_state = ( + ChatMembership.objects.filter(chat=chat, user=user) + .values_list("is_active", flat=True) + .first() + ) + if membership_state is not None: + return membership_state + return chat.participants.filter(pk=user.pk).exists() + + +def _context_participant_access(chat: Chat, user) -> bool | None: + try: + return chat.get_participants().filter(pk=user.pk).exists() + except Exception as error: # noqa: BLE001 - configurable resolver boundary + logger.warning( + "Chat access participant resolution failed", + extra={ + "event": "chat.access.participant_resolution_failed", + "chat_id": chat.id, + "user_id": user.id, + "chat_type": chat.type, + "error_type": type(error).__name__, + }, + exc_info=True, + ) + return None + + +def _group_chat_access(chat: Chat, user) -> bool: + if chat.context_object_id: + resolved = _context_participant_access(chat, user) + if resolved is not None: + return resolved + return _membership_or_legacy_participant(chat, user) + + +def _comments_chat_access(chat: Chat, user) -> bool: + if (chat.flags or {}).get("show_in_messages") is True: + return True + if chat.context_object_id: + return _context_participant_access(chat, user) is True + return _membership_or_legacy_participant(chat, user) + + +_CHAT_ACCESS_CHECKERS = { + "private": _membership_or_legacy_participant, + "group": _group_chat_access, + "channel": _membership_or_legacy_participant, + "announcement": _membership_or_legacy_participant, + "comments": _comments_chat_access, +} + + def user_can_access_chat(chat: Chat, user) -> bool: """ Проверяет, имеет ли пользователь доступ к чату. @@ -81,78 +136,13 @@ def user_can_access_chat(chat: Chat, user) -> bool: if not chat or not user: return False - if chat.type == "global": + # Создатель является владельцем ресурса и не должен терять доступ из-за + # рассинхронизации legacy participants/ChatMembership. + if chat.created_by_id == user.pk: return True - def _has_direct_participation() -> bool: - return chat.participants.filter(pk=user.pk).exists() + if chat.type == "global" or chat.include_all_users: + return True - if chat.type == "private": - return ( - ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists() - or _has_direct_participation() - ) - - if chat.type == "group": - # Групповые чаты с контекстным объектом используют resolver - if chat.context_object_id: - try: - participants = chat.get_participants() - return participants.filter(pk=user.pk).exists() - except Exception as e: - import logging - - logger = logging.getLogger(__name__) - logger.warning( - "Chat access participant resolution failed", - extra={ - "event": "chat.access.participant_resolution_failed", - "chat_id": chat.id, - "user_id": user.id, - "chat_type": chat.type, - "error_type": type(e).__name__, - }, - exc_info=True, - ) - return ( - ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists() - or _has_direct_participation() - ) - - if chat.type in ("channel", "announcement"): - if chat.include_all_users: - return user.is_active - return ( - ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists() - or _has_direct_participation() - ) - - if chat.type == "comments": - # Чаты-комментарии используют context-based доступ через - # get_participants() - if chat.context_object_id: - try: - participants = chat.get_participants() - return participants.filter(pk=user.pk).exists() - except Exception as e: - import logging - - logger = logging.getLogger(__name__) - logger.warning( - "Comments chat access participant resolution failed", - extra={ - "event": "chat.access.participant_resolution_failed", - "chat_id": chat.id, - "user_id": user.id, - "chat_type": chat.type, - "error_type": type(e).__name__, - }, - exc_info=True, - ) - return False - return ( - ChatMembership.objects.filter(chat=chat, user=user, is_active=True).exists() - or _has_direct_participation() - ) - - return False + checker = _CHAT_ACCESS_CHECKERS.get(chat.type) + return checker(chat, user) if checker else False diff --git a/backend/docs/architecture/websocket-asyncapi.yaml b/backend/docs/architecture/websocket-asyncapi.yaml index 3cc6e0d6..df28ca7f 100644 --- a/backend/docs/architecture/websocket-asyncapi.yaml +++ b/backend/docs/architecture/websocket-asyncapi.yaml @@ -1,7 +1,7 @@ asyncapi: 2.6.0 info: title: EUSRR Realtime WebSocket API - version: 1.2.0 + version: 1.3.0 description: | Отдельная AsyncAPI-спецификация для WebSocket-контура EUSRR. @@ -9,8 +9,8 @@ info: Этот документ описывает только двусторонний realtime-канал /ws/. Аутентификация: - - query-параметр ?token= - - или заголовок Authorization: Bearer + - browser: одноразовый ?ticket= из POST /api/auth/ws-ticket/ + - non-browser: заголовок Authorization: Bearer defaultContentType: application/json servers: @@ -18,10 +18,16 @@ servers: url: localhost:8000/ws/ protocol: ws description: Локальная разработка + security: + - websocketTicket: [] + - bearerAuth: [] production: url: api.example.com/ws/ protocol: wss description: Продакшен через TLS + security: + - websocketTicket: [] + - bearerAuth: [] channels: userRealtime: address: /ws/ @@ -51,6 +57,7 @@ channels: - $ref: '#/components/messages/MarkReadAction' - $ref: '#/components/messages/CalendarSubscribeAction' - $ref: '#/components/messages/CalendarUnsubscribeAction' + - $ref: '#/components/messages/PongAction' subscribe: operationId: serverEvent summary: События от сервера клиенту @@ -60,6 +67,7 @@ channels: - $ref: '#/components/messages/ErrorEvent' - $ref: '#/components/messages/ChatOpenedEvent' - $ref: '#/components/messages/ChatClosedEvent' + - $ref: '#/components/messages/ChatAccessChangedEvent' - $ref: '#/components/messages/InitialMessagesEvent' - $ref: '#/components/messages/NewMessageEvent' - $ref: '#/components/messages/ListUpdateEvent' @@ -81,6 +89,16 @@ channels: - $ref: '#/components/messages/CalendarSnapshotEvent' - $ref: '#/components/messages/CalendarErrorEvent' components: + securitySchemes: + websocketTicket: + type: apiKey + in: query + name: ticket + description: Одноразовый билет со сроком жизни 30 секунд. + bearerAuth: + type: http + scheme: bearer + bearerFormat: JWT schemas: ChatId: type: integer @@ -367,6 +385,19 @@ components: type: string minLength: 1 maxLength: 80 + PongAction: + name: PongAction + title: Ответ на heartbeat + payload: + type: object + required: [action] + properties: + action: + type: string + enum: [pong] + timestamp: + type: string + format: date-time PingEvent: name: PingEvent payload: @@ -414,6 +445,19 @@ components: enum: [chat_closed] chat_id: $ref: '#/components/schemas/ChatId' + ChatAccessChangedEvent: + name: ChatAccessChangedEvent + payload: + type: object + required: [type, chat_id, can_access] + properties: + type: + type: string + enum: [chat_access_changed] + chat_id: + $ref: '#/components/schemas/ChatId' + can_access: + type: boolean InitialMessagesEvent: name: InitialMessagesEvent payload: diff --git a/backend/docs/guides/COMMENTS_USAGE.md b/backend/docs/guides/COMMENTS_USAGE.md index b1778d49..3dabad5a 100644 --- a/backend/docs/guides/COMMENTS_USAGE.md +++ b/backend/docs/guides/COMMENTS_USAGE.md @@ -131,7 +131,12 @@ comment = Message.objects.create( ```javascript // Frontend обычно использует общий websocketManager. -const ws = new WebSocket(`ws://localhost:8000/ws/?token=${accessToken}`); +const { ticket } = await apiClient.request('/api/auth/ws-ticket/', { + method: 'POST' +}); +const ws = new WebSocket( + `ws://localhost:8000/ws/?ticket=${encodeURIComponent(ticket)}` +); // Сделать comments-chat активным в этой сессии. ws.send(JSON.stringify({ @@ -142,6 +147,10 @@ ws.send(JSON.stringify({ // Получать обновления в реальном времени ws.onmessage = (event) => { const data = JSON.parse(event.data); + if (data.type === 'ping') { + ws.send(JSON.stringify({ action: 'pong', timestamp: data.timestamp })); + return; + } if (data.type === 'new_message' && data.chat_id === commentsChat.id) { // Новый комментарий → обновить UI addCommentToUI(data.message); diff --git a/backend/employees/utils.py b/backend/employees/utils.py index 9045446f..cdddaa6c 100644 --- a/backend/employees/utils.py +++ b/backend/employees/utils.py @@ -3,6 +3,7 @@ from __future__ import annotations import phonenumbers from django.conf import settings from django.contrib.auth import get_user_model +from phonenumbers import PhoneNumberFormat from employees.models import ( Department, @@ -11,8 +12,6 @@ from employees.models import ( EmployeeDepartment, RoleAssignment, ) -from phonenumbers import PhoneNumberFormat - Employee = get_user_model() @@ -370,6 +369,7 @@ def resolve_chat_participants_for_procurement_request( отдел-исполнитель и администраторы. """ from django.db.models import Q + from procurement.models import ProcurementRequest if not isinstance(context_object, ProcurementRequest): @@ -432,6 +432,7 @@ def resolve_chat_participants_for_task(context_object, **kwargs): Доступ соответствует видимости доски задачи. """ from django.db.models import Q + from tasks.models import Task if not isinstance(context_object, Task): @@ -539,9 +540,10 @@ def resolve_chat_participants(chat): # 6. Fallback: ChatMembership или participants from communications.models import ChatMembership - membership_ids = ChatMembership.objects.filter(chat=chat).values_list( - "user_id", flat=True - ) + membership_ids = ChatMembership.objects.filter( + chat=chat, + is_active=True, + ).values_list("user_id", flat=True) participant_ids = chat.participants.values_list("id", flat=True) diff --git a/backend/eusrr_backend/asgi.py b/backend/eusrr_backend/asgi.py index 9d512275..b17a79ff 100644 --- a/backend/eusrr_backend/asgi.py +++ b/backend/eusrr_backend/asgi.py @@ -3,6 +3,8 @@ from importlib import import_module from channels.auth import AuthMiddlewareStack from channels.routing import ProtocolTypeRouter, URLRouter +from channels.security.websocket import AllowedHostsOriginValidator, OriginValidator +from django.conf import settings from django.core.asgi import get_asgi_application from eusrr_backend.logging_utils import WebSocketLoggingMiddleware @@ -14,12 +16,23 @@ django_asgi_app = get_asgi_application() JWTAuthMiddleware = import_module("eusrr_backend.channels_jwt").JWTAuthMiddleware websocket_urlpatterns = import_module("realtime.routing").websocket_urlpatterns +websocket_application = AuthMiddlewareStack( + JWTAuthMiddleware(URLRouter(websocket_urlpatterns)) +) +if settings.WEBSOCKET_ALLOWED_ORIGINS: + websocket_application = OriginValidator( + websocket_application, + settings.WEBSOCKET_ALLOWED_ORIGINS, + ) +else: + websocket_application = AllowedHostsOriginValidator(websocket_application) + application = ProtocolTypeRouter( { "http": django_asgi_app, "websocket": WebSocketLoggingMiddleware( - AuthMiddlewareStack(JWTAuthMiddleware(URLRouter(websocket_urlpatterns))) + websocket_application ), } ) diff --git a/backend/eusrr_backend/channels_jwt.py b/backend/eusrr_backend/channels_jwt.py index 7e63b317..48da3d63 100644 --- a/backend/eusrr_backend/channels_jwt.py +++ b/backend/eusrr_backend/channels_jwt.py @@ -2,12 +2,19 @@ from __future__ import annotations import logging -from typing import Optional from urllib.parse import parse_qs from channels.db import database_sync_to_async from channels.middleware import BaseMiddleware +from django.conf import settings from django.contrib.auth.models import AnonymousUser +from rest_framework.exceptions import AuthenticationFailed +from rest_framework_simplejwt.exceptions import TokenError + +from api.auth.websocket_tickets import ( + WebSocketTicketError, + consume_websocket_ticket, +) from .logging_utils import update_log_context @@ -17,20 +24,32 @@ logger = logging.getLogger(__name__) class JWTAuthMiddleware(BaseMiddleware): """Auth для Channels по JWT. - Поддерживает query ?token=... и header Authorization: Bearer ... + Браузер использует одноразовый query ?ticket=..., программные клиенты могут + передать Authorization: Bearer. Долгоживущий JWT в URL выключен по умолчанию. """ - def _get_raw_token(self, scope) -> Optional[str]: - # 1) query string - qs = parse_qs(scope.get("query_string", b"").decode()) - if "token" in qs and qs["token"]: - return qs["token"][0] - # 2) header + def _get_query_params(self, scope) -> dict[str, list[str]]: + return parse_qs( + scope.get("query_string", b"").decode("utf-8"), + max_num_fields=10, + ) + + def _get_raw_token(self, scope, query_params) -> str | None: + if getattr(settings, "WEBSOCKET_ALLOW_QUERY_JWT", False) and query_params.get( + "token" + ): + return query_params["token"][0] for name, value in scope.get("headers", []): if name == b"authorization" and value.lower().startswith(b"bearer "): - return value.split()[1].decode() + parts = value.split(maxsplit=1) + if len(parts) == 2 and parts[1]: + return parts[1].decode() return None + @database_sync_to_async + def _authenticate_ticket(self, raw_ticket: str): + return consume_websocket_ticket(raw_ticket) + @database_sync_to_async def _authenticate_raw_token(self, raw_token: str): from api.auth.authentication import SessionAwareJWTAuthentication @@ -43,25 +62,41 @@ class JWTAuthMiddleware(BaseMiddleware): async def __call__(self, scope, receive, send): scope.setdefault("user", AnonymousUser()) + credential_type = None try: - raw = self._get_raw_token(scope) - if raw: - user, token = await self._authenticate_raw_token(raw) + query_params = self._get_query_params(scope) + raw_ticket = (query_params.get("ticket") or [None])[0] + raw_token = self._get_raw_token(scope, query_params) + if raw_ticket: + if len(raw_ticket) > 2048: + raise WebSocketTicketError("ticket_too_long") + credential_type = "ticket" + identity = await self._authenticate_ticket(raw_ticket) + scope["user"] = identity.user + scope["auth_session_id"] = identity.session_id + scope["auth"] = {"type": "websocket_ticket"} + elif raw_token: + credential_type = "bearer" + user, token = await self._authenticate_raw_token(raw_token) scope["user"] = user scope["auth"] = token - except Exception as error: - session_user = scope.get("user") - if getattr(session_user, "is_authenticated", False): - update_log_context(user_id=getattr(session_user, "pk", None)) + scope["auth_session_id"] = token.get("session_id") + except ( + AuthenticationFailed, + TokenError, + ValueError, + ) as error: + scope["user"] = AnonymousUser() + scope.pop("auth", None) + scope.pop("auth_session_id", None) logger.info( "WebSocket authentication rejected", extra={ "event": "websocket.authentication.rejected", "error_type": type(error).__name__, + "credential_type": credential_type, }, ) - # Сохраняем уже определённого session/anonymous-пользователя. - pass user = scope.get("user") if getattr(user, "is_authenticated", False): update_log_context(user_id=getattr(user, "pk", None)) diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index f6228b46..3b4669af 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -23,6 +23,16 @@ def _split_env_list(value: str) -> list[str]: ALLOWED_HOSTS = _split_env_list(os.getenv("ALLOWED_HOSTS", "localhost,127.0.0.1")) +# Browser clients authenticate WebSocket handshakes with a short-lived ticket. +# Long-lived JWTs in query strings are accepted only during an explicit legacy +# migration window because URLs can be recorded by reverse proxies. +WEBSOCKET_ALLOW_QUERY_JWT = ( + os.getenv("WEBSOCKET_ALLOW_QUERY_JWT", "false").lower() == "true" +) +WEBSOCKET_ALLOWED_ORIGINS = _split_env_list( + os.getenv("WEBSOCKET_ALLOWED_ORIGINS", "") +) + INSTALLED_APPS = [ "daphne", "channels", @@ -435,6 +445,13 @@ CHANNEL_LAYERS = { int(os.getenv("REDIS_PORT", "6379")), ) ], + # Absorb short notification/task bursts without silently dropping + # events; expiry bounds memory when clients disappear abruptly. + "capacity": int(os.getenv("CHANNEL_LAYER_CAPACITY", "1000")), + "expiry": int(os.getenv("CHANNEL_LAYER_EXPIRY", "60")), + "group_expiry": int( + os.getenv("CHANNEL_LAYER_GROUP_EXPIRY", "86400") + ), }, } } @@ -464,6 +481,7 @@ REST_FRAMEWORK = { "auth_refresh": "60/min", "auth_sensitive": "10/min", "auth_email": "5/hour", + "websocket_ticket": "30/min", "registration": "5/hour", }, "DEFAULT_PAGINATION_CLASS": "api.pagination.StandardPagination", diff --git a/backend/eusrr_backend/settings_test.py b/backend/eusrr_backend/settings_test.py index c9d85871..3ff7ffac 100644 --- a/backend/eusrr_backend/settings_test.py +++ b/backend/eusrr_backend/settings_test.py @@ -35,6 +35,7 @@ REST_FRAMEWORK = { "auth_refresh": "10000/min", "auth_sensitive": "10000/min", "auth_email": "10000/min", + "websocket_ticket": "10000/min", "registration": "10000/min", }, } diff --git a/backend/realtime/consumers.py b/backend/realtime/consumers.py index 646599bf..b52ca736 100644 --- a/backend/realtime/consumers.py +++ b/backend/realtime/consumers.py @@ -13,7 +13,10 @@ WebSocket Consumer для пользователя. """ import asyncio +import json import logging +from time import monotonic + from channels.db import database_sync_to_async from channels.generic.websocket import AsyncJsonWebsocketConsumer from django.contrib.auth.models import AnonymousUser @@ -28,6 +31,8 @@ from scheduling.realtime import ( parse_calendar_subscription, ) +from .protocol import ConnectionRateLimiter, RealtimeProtocolError, parse_action + logger = logging.getLogger(__name__) @@ -43,6 +48,23 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): - Онлайн-статус """ + ACTION_HANDLERS = { + "open_chat": "_handle_open_chat", + "close_chat": "_handle_close_chat", + "send_message": "_handle_send_message", + "edit_message": "_handle_edit_message", + "delete_message": "_handle_delete_message", + "add_reaction": "_handle_add_reaction", + "remove_reaction": "_handle_remove_reaction", + "typing": "_handle_typing", + "stop_typing": "_handle_stop_typing", + "mark_read": "_handle_mark_read", + "vote_poll": "_handle_vote_poll", + "calendar_subscribe": "_handle_calendar_subscribe", + "calendar_unsubscribe": "_handle_calendar_unsubscribe", + "pong": "_handle_pong", + } + def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.ping_task = None @@ -54,6 +76,43 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): self.calendar_subscriptions = {} self.calendar_refresh_task = None self.pending_calendar_refresh_ids = set() + self.last_pong_at = monotonic() + self.rate_limiter = ConnectionRateLimiter() + + async def receive(self, text_data=None, bytes_data=None, **kwargs): + """Reject malformed, binary and oversized commands without crashing.""" + if bytes_data is not None: + await self.send_json( + { + "type": "error", + "code": "binary_not_supported", + "error": "Поддерживаются только JSON-команды.", + } + ) + return + if text_data is None: + return + if len(text_data.encode("utf-8")) > 64 * 1024: + await self.send_json( + { + "type": "error", + "code": "payload_too_large", + "error": "WebSocket-команда превышает лимит 64 КиБ.", + } + ) + return + try: + content = await self.decode_json(text_data) + except (json.JSONDecodeError, TypeError, ValueError): + await self.send_json( + { + "type": "error", + "code": "invalid_json", + "error": "Некорректный JSON.", + } + ) + return + await self.receive_json(content, **kwargs) @database_sync_to_async def _mark_chat_notifications_as_read(self, chat_id): @@ -94,6 +153,7 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): await self.accept() # Запускаем ping цикл для keepalive + self.last_pong_at = monotonic() self.ping_task = asyncio.create_task(self._ping_loop()) logger.info( @@ -159,6 +219,16 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): try: while True: await asyncio.sleep(self.ping_interval) + if monotonic() - self.last_pong_at > self.ping_interval * 3: + logger.warning( + "WebSocket client heartbeat timed out", + extra={ + "event": "websocket.heartbeat.timeout", + "user_id": self.user.id, + }, + ) + await self.close(code=4408) + return await self.send_json( {"type": "ping", "timestamp": timezone.now().isoformat()} ) @@ -176,10 +246,22 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): async def receive_json(self, content, **kwargs): """Обработка входящих сообщений от клиента""" - if not isinstance(content, dict): + try: + action = parse_action(content) + except RealtimeProtocolError as error: + await self._send_protocol_error(error) return - action = content.get("action", "") + if not self.rate_limiter.allow(action): + await self.send_json( + { + "type": "error", + "code": "rate_limited", + "action": action, + "error": "Слишком много WebSocket-команд. Повторите позже.", + } + ) + return logger.debug( "WebSocket action received", @@ -190,45 +272,72 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): }, ) - # Управление активным чатом - if action == "open_chat": - await self._handle_open_chat(content) - elif action == "close_chat": - await self._handle_close_chat(content) + handler_name = self.ACTION_HANDLERS.get(action) + if handler_name is None: + await self.send_json( + { + "type": "error", + "code": "unknown_action", + "action": action, + "error": "Неизвестная WebSocket-команда.", + } + ) + return + handler = getattr(self, handler_name) - # Работа с сообщениями (только для активного чата) - elif action == "send_message": - await self._handle_send_message(content) - elif action == "edit_message": - await self._handle_edit_message(content) - elif action == "delete_message": - await self._handle_delete_message(content) + try: + await handler(content) + except RealtimeProtocolError as error: + await self._send_protocol_error(error, action=action) + except (TypeError, ValueError) as error: + logger.info( + "Invalid WebSocket command payload rejected", + extra={ + "event": "websocket.command.invalid", + "action": action, + "user_id": self.user.id, + "error_type": type(error).__name__, + }, + ) + await self.send_json( + { + "type": "error", + "code": "invalid_payload", + "action": action, + "error": "Некорректные данные WebSocket-команды.", + } + ) + except Exception as error: # noqa: BLE001 - protocol boundary must survive + logger.exception( + "WebSocket command failed", + extra={ + "event": "websocket.command.failed", + "action": action, + "user_id": self.user.id, + "error_type": type(error).__name__, + }, + ) + await self.send_json( + { + "type": "error", + "code": "internal_error", + "action": action, + "error": "Не удалось выполнить WebSocket-команду.", + } + ) - # Реакции - elif action == "add_reaction": - await self._handle_add_reaction(content) - elif action == "remove_reaction": - await self._handle_remove_reaction(content) + async def _send_protocol_error(self, error, *, action=None): + await self.send_json( + { + "type": "error", + "code": error.code, + "action": action, + "error": error.detail, + } + ) - # Индикатор печати - elif action == "typing": - await self._handle_typing(content) - elif action == "stop_typing": - await self._handle_stop_typing() - - # Отметка прочитанного - elif action == "mark_read": - await self._handle_mark_read(content) - - # Голосования - elif action == "vote_poll": - await self._handle_vote_poll(content) - - # Подписки календаря на диапазоны событий - elif action == "calendar_subscribe": - await self._handle_calendar_subscribe(content) - elif action == "calendar_unsubscribe": - await self._handle_calendar_unsubscribe(content) + async def _handle_pong(self, content): + self.last_pong_at = monotonic() @database_sync_to_async def _build_calendar_snapshot(self, subscription): @@ -489,3 +598,61 @@ class UserConsumer(ChatConsumerMixin, AsyncJsonWebsocketConsumer): self.calendar_refresh_task = asyncio.create_task( self._refresh_calendar_subscriptions() ) + + async def chat_access_changed(self, event): + """Apply membership changes to a live connection without reconnecting.""" + try: + chat_id = int(event.get("chat_id")) + except (TypeError, ValueError): + return + + chat = await self._get_chat(chat_id) + can_access = bool(chat and await self._user_can_access(chat, self.user)) + group_name = f"chat_{chat_id}" + + if can_access: + if chat_id not in self.subscribed_chats: + await self.channel_layer.group_add(group_name, self.channel_name) + self.subscribed_chats.add(chat_id) + elif chat_id in self.subscribed_chats: + await self.channel_layer.group_discard(group_name, self.channel_name) + self.subscribed_chats.discard(chat_id) + if self.active_chat_id == chat_id: + self.active_chat_id = None + + await self.send_json( + { + "type": "chat_access_changed", + "chat_id": chat_id, + "can_access": can_access, + } + ) + + logger.info( + "Live chat access synchronized", + extra={ + "event": "websocket.chat.access_synchronized", + "chat_id": chat_id, + "user_id": self.user.id, + "allowed": can_access, + }, + ) + + async def auth_session_revoked(self, event): + """Immediately terminate the socket that belongs to a revoked session.""" + connection_session_id = self.scope.get("auth_session_id") + revoked_session_id = event.get("session_id") + if not connection_session_id or str(connection_session_id) != str( + revoked_session_id + ): + return + + logger.info( + "WebSocket closed after authentication session revocation", + extra={ + "event": "websocket.authentication.session_revoked", + "user_id": self.user.id, + "auth_session_id": str(connection_session_id), + }, + ) + await self.close(code=4401) diff --git a/backend/realtime/protocol.py b/backend/realtime/protocol.py new file mode 100644 index 00000000..261e6cdc --- /dev/null +++ b/backend/realtime/protocol.py @@ -0,0 +1,105 @@ +"""Validation and abuse protection for the realtime command protocol.""" + +from __future__ import annotations + +from collections import defaultdict, deque +from dataclasses import dataclass +from time import monotonic +from typing import Any + + +class RealtimeProtocolError(ValueError): + def __init__(self, code: str, detail: str): + super().__init__(detail) + self.code = code + self.detail = detail + + +def parse_action(content: Any) -> str: + if not isinstance(content, dict): + raise RealtimeProtocolError("invalid_payload", "Ожидался JSON-объект.") + action = content.get("action") + if not isinstance(action, str) or not action.strip(): + raise RealtimeProtocolError("action_required", "Не указано действие.") + action = action.strip() + if len(action) > 64: + raise RealtimeProtocolError("invalid_action", "Некорректное действие.") + return action + + +def parse_positive_int(value: Any, *, field: str) -> int: + if isinstance(value, bool): + raise RealtimeProtocolError("invalid_field", f"Поле {field} некорректно.") + try: + parsed = int(value) + except (TypeError, ValueError) as error: + raise RealtimeProtocolError( + "invalid_field", + f"Поле {field} должно быть положительным целым числом.", + ) from error + if parsed <= 0: + raise RealtimeProtocolError( + "invalid_field", + f"Поле {field} должно быть положительным целым числом.", + ) + return parsed + + +def parse_text(value: Any, *, field: str, max_length: int) -> str: + if not isinstance(value, str): + raise RealtimeProtocolError( + "invalid_field", + f"Поле {field} должно быть строкой.", + ) + parsed = value.strip() + if not parsed: + raise RealtimeProtocolError( + "invalid_field", + f"Поле {field} не может быть пустым.", + ) + if len(parsed) > max_length: + raise RealtimeProtocolError( + "field_too_long", + f"Поле {field} превышает лимит {max_length} символов.", + ) + return parsed + + +@dataclass(frozen=True) +class RateLimit: + requests: int + window_seconds: float + + +class ConnectionRateLimiter: + """Small per-connection limiter; infrastructure limits remain a second layer.""" + + DEFAULT = RateLimit(requests=120, window_seconds=60) + GLOBAL = RateLimit(requests=180, window_seconds=60) + POLICIES = { + "typing": RateLimit(requests=30, window_seconds=10), + "stop_typing": RateLimit(requests=30, window_seconds=10), + "send_message": RateLimit(requests=30, window_seconds=60), + "calendar_subscribe": RateLimit(requests=30, window_seconds=60), + } + + def __init__(self) -> None: + self._events: dict[str, deque[float]] = defaultdict(deque) + self._all_events: deque[float] = deque() + + @staticmethod + def _consume(events: deque[float], policy: RateLimit, now: float) -> bool: + threshold = now - policy.window_seconds + while events and events[0] <= threshold: + events.popleft() + if len(events) >= policy.requests: + return False + events.append(now) + return True + + def allow(self, action: str) -> bool: + now = monotonic() + if not self._consume(self._all_events, self.GLOBAL, now): + return False + policy = self.POLICIES.get(action, self.DEFAULT) + return self._consume(self._events[action], policy, now) diff --git a/backend/static/js/components/userWebSocket.js b/backend/static/js/components/userWebSocket.js index 16cbb48f..c9f2f1d4 100644 --- a/backend/static/js/components/userWebSocket.js +++ b/backend/static/js/components/userWebSocket.js @@ -120,12 +120,15 @@ export function initUserWebSocket(options = {}) { function handleMessage(data) { const { type } = data; - + + window.dispatchEvent(new CustomEvent('user:ws-message', { detail: data })); console.log('[UserWebSocket] Received message:', type, data); switch (type) { case 'ping': - // Keepalive - просто логируем + if (ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify({ action: 'pong', timestamp: data.timestamp })); + } break; case 'chat_opened': diff --git a/backend/static/js/notifications/notification-manager.js b/backend/static/js/notifications/notification-manager.js index a9b87422..3e0776a9 100644 --- a/backend/static/js/notifications/notification-manager.js +++ b/backend/static/js/notifications/notification-manager.js @@ -9,11 +9,6 @@ import { class NotificationManager { constructor() { - this.ws = null; - this.reconnectAttempts = 0; - this.maxReconnectAttempts = 5; - this.reconnectDelay = 1000; - this.isConnected = false; this.notificationPermission = "default"; this.soundEnabled = this.loadSoundPreference(); @@ -107,7 +102,9 @@ class NotificationManager { init() { this.requestNotificationPermission(); - this.connectWebSocket(); + window.addEventListener("user:ws-message", (event) => { + this.handleMessage(event.detail); + }); this.loadInitialNotifications(); this.setupDropdown(); } @@ -138,52 +135,8 @@ class NotificationManager { } } - connectWebSocket() { - const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"; - // Используем универсальный WebSocket endpoint (realtime architecture) - const wsUrl = `${protocol}//${window.location.host}/ws/`; - - console.log("[Notifications] Connecting to:", wsUrl); - + handleMessage(data) { try { - this.ws = new WebSocket(wsUrl); - - this.ws.onopen = () => { - console.log("[Notifications] WebSocket connected"); - this.isConnected = true; - this.reconnectAttempts = 0; - this.updateConnectionStatus(true); - }; - - this.ws.onmessage = (event) => { - this.handleMessage(event); - }; - - this.ws.onclose = () => { - console.log("[Notifications] WebSocket disconnected"); - this.isConnected = false; - this.updateConnectionStatus(false); - this.attemptReconnect(); - }; - - this.ws.onerror = (error) => { - console.error("[Notifications] WebSocket error:", error); - }; - } catch (error) { - console.error("[Notifications] Failed to create WebSocket:", error); - this.attemptReconnect(); - } - } - - handleMessage(event) { - try { - const data = JSON.parse(event.data); - - // Игнорируем ping сообщения для keepalive (не логируем) - if (data.type === "ping") { - return; - } - // Игнорируем list_update - это для списка чатов, не для уведомлений if (data.type === "list_update") { return; @@ -238,35 +191,6 @@ class NotificationManager { } } - attemptReconnect() { - if (this.reconnectAttempts < this.maxReconnectAttempts) { - this.reconnectAttempts++; - const delay = this.reconnectDelay * this.reconnectAttempts; - - console.log( - `[Notifications] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})` - ); - - setTimeout(() => { - this.connectWebSocket(); - }, delay); - } else { - console.error("[Notifications] Max reconnect attempts reached"); - } - } - - updateConnectionStatus(connected) { - // Визуальная индикация статуса подключения (опционально) - const badge = document.getElementById("notificationBadge"); - if (badge) { - if (!connected) { - badge.style.backgroundColor = "#dc3545"; // красный при отключении - } else { - badge.style.backgroundColor = ""; // вернуть стандартный цвет - } - } - } - updateBadge(count) { const badge = document.getElementById("notificationBadge"); if (badge) { diff --git a/backend/templates/api/websocket_docs.html b/backend/templates/api/websocket_docs.html index 14b2b6c3..0787f851 100644 --- a/backend/templates/api/websocket_docs.html +++ b/backend/templates/api/websocket_docs.html @@ -105,8 +105,10 @@

Точка подключения

{{ ws_endpoint_url }}

- Аутентификация поддерживает либо query-параметр - ?token=<jwt_access_token>, либо заголовок + Browser получает одноразовый билет через + POST /api/auth/ws-ticket/ и подключается с + ?ticket=<one_time_ticket>. Non-browser + client может использовать заголовок Authorization: Bearer <jwt_access_token>.

diff --git a/backend/tests/realtime/test_jwt_session_middleware.py b/backend/tests/realtime/test_jwt_session_middleware.py index 60bfbbf8..13c9ad22 100644 --- a/backend/tests/realtime/test_jwt_session_middleware.py +++ b/backend/tests/realtime/test_jwt_session_middleware.py @@ -1,5 +1,4 @@ import pytest - from asgiref.sync import sync_to_async from channels.testing import WebsocketCommunicator from django.contrib.auth import get_user_model @@ -40,7 +39,8 @@ async def test_revoked_session_becomes_anonymous_for_websocket(): communicator = WebsocketCommunicator( JWTAuthMiddleware(app), - f"/ws/?token={str(token)}", + "/ws/", + headers=[(b"authorization", f"Bearer {token}".encode())], ) connected, _ = await communicator.connect() assert connected is True @@ -52,7 +52,34 @@ async def test_revoked_session_becomes_anonymous_for_websocket(): captured.clear() communicator = WebsocketCommunicator( JWTAuthMiddleware(app), - f"/ws/?token={str(token)}", + "/ws/", + headers=[(b"authorization", f"Bearer {token}".encode())], + ) + connected, _ = await communicator.connect() + assert connected is True + assert captured["user"].is_authenticated is False + await communicator.disconnect() + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.asyncio +async def test_query_string_jwt_is_rejected_by_default(): + user = await sync_to_async(User.objects.create_user)( + email="ws-query-auth@example.com", + password="testpass123", + phone_number="+79997770012", + ) + token = AccessToken.for_user(user) + captured = {} + + async def app(scope, receive, send): + captured["user"] = scope["user"] + await send({"type": "websocket.accept"}) + await send({"type": "websocket.close"}) + + communicator = WebsocketCommunicator( + JWTAuthMiddleware(app), + f"/ws/?token={token}", ) connected, _ = await communicator.connect() assert connected is True diff --git a/backend/tests/realtime/test_security.py b/backend/tests/realtime/test_security.py new file mode 100644 index 00000000..a16bb439 --- /dev/null +++ b/backend/tests/realtime/test_security.py @@ -0,0 +1,275 @@ +import time +from unittest.mock import patch + +import pytest +from asgiref.sync import sync_to_async +from channels.layers import get_channel_layer +from channels.testing import WebsocketCommunicator + +from api.auth.models import UserAuthSession +from api.auth.websocket_tickets import ( + WebSocketTicketError, + consume_websocket_ticket, + issue_websocket_ticket, +) +from communications.models import Chat, ChatMembership, Message, MessageReaction +from eusrr_backend.asgi import application +from realtime.protocol import ConnectionRateLimiter + +pytestmark = pytest.mark.django_db(transaction=True) + + +def test_connection_rate_limiter_has_action_specific_limits(): + limiter = ConnectionRateLimiter() + assert all(limiter.allow("typing") for _ in range(30)) + assert limiter.allow("typing") is False + + limiter = ConnectionRateLimiter() + assert all(limiter.allow("open_chat") for _ in range(120)) + assert limiter.allow("open_chat") is False + + limiter = ConnectionRateLimiter() + assert all(limiter.allow(f"unknown-{index}") for index in range(180)) + assert limiter.allow("one-more-unique-action") is False + + +@pytest.mark.asyncio +async def test_inactive_membership_is_not_subscribed(ws_communicator, user): + chat = await sync_to_async(Chat.objects.create)(name="Revoked", type="group") + await sync_to_async(ChatMembership.objects.create)( + chat=chat, + user=user, + role="member", + is_active=False, + ) + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + + await get_channel_layer().group_send( + f"chat_{chat.id}", + { + "type": "chat_message", + "chat_id": chat.id, + "payload": {"id": 1, "content": "private"}, + }, + ) + assert await communicator.receive_nothing(timeout=0.1) + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_reaction_requires_chat_access(ws_communicator, user, user2): + chat = await sync_to_async(Chat.objects.create)(name="Private", type="group") + await sync_to_async(ChatMembership.objects.create)(chat=chat, user=user2) + message = await sync_to_async(Message.objects.create)( + chat=chat, + author=user2, + content="secret", + ) + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + + await communicator.send_json_to( + {"action": "add_reaction", "message_id": message.id, "emoji": "👍"} + ) + response = await communicator.receive_json_from() + assert response["code"] == "access_denied" + reaction_exists = await sync_to_async( + MessageReaction.objects.filter(message=message).exists + )() + assert not reaction_exists + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_membership_revocation_removes_live_group(ws_communicator, user): + chat = await sync_to_async(Chat.objects.create)(name="Live", type="group") + membership = await sync_to_async(ChatMembership.objects.create)( + chat=chat, + user=user, + ) + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + + membership.is_active = False + await sync_to_async(membership.save)(update_fields=["is_active"]) + access_event = await communicator.receive_json_from() + assert access_event == { + "type": "chat_access_changed", + "chat_id": chat.id, + "can_access": False, + } + + await get_channel_layer().group_send( + f"chat_{chat.id}", + { + "type": "chat_message", + "chat_id": chat.id, + "payload": {"id": 2, "content": "after revoke"}, + }, + ) + assert await communicator.receive_nothing(timeout=0.1) + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_membership_addition_joins_live_group(ws_communicator, user): + chat = await sync_to_async(Chat.objects.create)(name="Added live", type="group") + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + + await sync_to_async(ChatMembership.objects.create)(chat=chat, user=user) + access_event = await communicator.receive_json_from() + assert access_event == { + "type": "chat_access_changed", + "chat_id": chat.id, + "can_access": True, + } + + await get_channel_layer().group_send( + f"chat_{chat.id}", + { + "type": "chat_message", + "chat_id": chat.id, + "payload": {"id": 3, "content": "after grant"}, + }, + ) + update = await communicator.receive_json_from() + assert update["type"] == "list_update" + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_send_message_honors_membership_permission(ws_communicator, user): + chat = await sync_to_async(Chat.objects.create)(name="Read only", type="group") + await sync_to_async(ChatMembership.objects.create)( + chat=chat, + user=user, + role="guest", + ) + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + await communicator.send_json_to({"action": "open_chat", "chat_id": chat.id}) + assert (await communicator.receive_json_from())["type"] == "chat_opened" + + await communicator.send_json_to({"action": "send_message", "content": "denied"}) + response = await communicator.receive_json_from() + assert response["code"] == "access_denied" + message_exists = await sync_to_async( + Message.objects.filter(chat=chat, content="denied").exists + )() + assert not message_exists + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_invalid_command_does_not_terminate_connection(ws_communicator, user): + communicator = await ws_communicator(user=user) + connected, _ = await communicator.connect() + assert connected + + await communicator.send_json_to({"action": "open_chat", "chat_id": "invalid"}) + first_error = await communicator.receive_json_from() + assert first_error["code"] == "invalid_field" + + await communicator.send_json_to({"action": "not_supported"}) + second_error = await communicator.receive_json_from() + assert second_error["code"] == "unknown_action" + + await communicator.send_to(text_data="{not-json") + malformed_error = await communicator.receive_json_from() + assert malformed_error["code"] == "invalid_json" + + await communicator.send_to(bytes_data=b"binary") + binary_error = await communicator.receive_json_from() + assert binary_error["code"] == "binary_not_supported" + await communicator.disconnect() + + +def test_websocket_ticket_is_single_use(user): + ticket = issue_websocket_ticket(user=user, session_id=None) + identity = consume_websocket_ticket(ticket) + assert identity.user.pk == user.pk + + with pytest.raises(WebSocketTicketError, match="ticket_already_used"): + consume_websocket_ticket(ticket) + + +def test_websocket_ticket_expires(user): + ticket = issue_websocket_ticket(user=user, session_id=None) + future = time.time() + 31 + with ( + patch("django.core.signing.time.time", return_value=future), + pytest.raises(WebSocketTicketError, match="invalid_or_expired_ticket"), + ): + consume_websocket_ticket(ticket) + + +def test_websocket_ticket_rejects_revoked_session(user): + session = UserAuthSession.objects.create(user=user, refresh_token_hash="stub") + ticket = issue_websocket_ticket(user=user, session_id=str(session.session_id)) + session.revoke(reason="test") + with pytest.raises(WebSocketTicketError, match="session_revoked"): + consume_websocket_ticket(ticket) + + +def test_websocket_ticket_endpoint_requires_authentication( + api_client, + auth_client_factory, + user, +): + anonymous_response = api_client.post("/api/auth/ws-ticket/") + assert anonymous_response.status_code == 401 + + response = auth_client_factory(user).post("/api/auth/ws-ticket/") + assert response.status_code == 200 + assert response.data["expires_in"] == 30 + assert response.data["ticket"] + + +@pytest.mark.asyncio +async def test_untrusted_websocket_origin_is_rejected(user): + ticket = await sync_to_async(issue_websocket_ticket)(user=user, session_id=None) + communicator = WebsocketCommunicator( + application, + f"/ws/?ticket={ticket}", + headers=[(b"origin", b"https://attacker.example")], + ) + connected, _ = await communicator.connect() + assert connected is False + + +@pytest.mark.asyncio +async def test_ticket_connects_from_trusted_origin(user): + ticket = await sync_to_async(issue_websocket_ticket)(user=user, session_id=None) + communicator = WebsocketCommunicator( + application, + f"/ws/?ticket={ticket}", + headers=[(b"origin", b"http://localhost:3000")], + ) + connected, _ = await communicator.connect() + assert connected is True + await communicator.disconnect() + + +@pytest.mark.asyncio +async def test_matching_session_revocation_closes_live_socket(ws_communicator, user): + communicator = await ws_communicator(user=user) + communicator.scope["auth_session_id"] = "session-to-revoke" + connected, _ = await communicator.connect() + assert connected + + await get_channel_layer().group_send( + f"user_{user.id}", + { + "type": "auth_session_revoked", + "session_id": "session-to-revoke", + }, + ) + close_event = await communicator.receive_output() + assert close_event == {"type": "websocket.close", "code": 4401} diff --git a/docs/guides/WEBSOCKET.md b/docs/guides/WEBSOCKET.md index e28efcb2..726ea53f 100644 --- a/docs/guides/WEBSOCKET.md +++ b/docs/guides/WEBSOCKET.md @@ -3,7 +3,7 @@ EUSRR использует одно авторизованное WebSocket-соединение на вкладку приложения: ```text -ws://localhost:8000/ws/?token=ACCESS_TOKEN +POST /api/auth/ws-ticket/ → ws://localhost:8000/ws/?ticket=ONE_TIME_TICKET ``` Отдельных `/ws/chat/`, `/ws/chats/`, `/ws/notifications/` и `/ws/realtime/` в текущем routing нет. @@ -43,21 +43,23 @@ NEXT_PUBLIC_WS_URL=ws://localhost:8000/ws/ ## Аутентификация -Browser передаёт access token как `?token=...`. Middleware также понимает `Authorization: Bearer ...`, что удобно для non-browser clients. +Browser сначала выполняет авторизованный `POST /api/auth/ws-ticket/`, затем передаёт полученный одноразовый билет как `?ticket=...`. Билет действует 30 секунд, подписан сервером и может быть использован только один раз. Поэтому долгоживущий JWT не попадает в URL WebSocket handshake. -JWT проходит `SessionAwareJWTAuthentication`: проверяется подпись, пользователь и связанная `UserAuthSession`. При ошибке scope остаётся anonymous, после чего consumer закрывает соединение кодом `4401`. +Non-browser clients могут передать `Authorization: Bearer `. Legacy `?token=` по умолчанию отключён; временно включить его можно только явной настройкой `WEBSOCKET_ALLOW_QUERY_JWT=true`. -Токен в query string может попасть в access logs. В production маскируйте query parameters в reverse proxy и application monitoring. +ASGI проверяет `Origin` через `AllowedHostsOriginValidator`. Если frontend и WebSocket работают на разных host, разрешённые frontend origins задаются явно через `WEBSOCKET_ALLOWED_ORIGINS=https://app.example.com`. JWT и ticket также проверяют связанную `UserAuthSession`; отзыв сессии немедленно закрывает принадлежащее ей соединение кодом `4401`. ## Подписки consumer-а После connect `UserConsumer`: -- подписывается на все доступные `chat_{id}`; +- подписывается только на доступные активные `chat_{id}`; - подписывается на `user_{user_id}`; - подписывается на `notifications_{user_id}`; - запускает server ping каждые 20 секунд. +Изменение `ChatMembership` синхронизируется с уже открытыми соединениями: при добавлении consumer входит в группу, при отзыве доступа немедленно выходит из неё и закрывает активный чат. + `active_chat_id` определяет, получает соединение полное chat event или только обновление списка. При открытии чата связанные message notifications отмечаются прочитанными; уведомление об уже открытом чате подавляется. ## Команды клиента @@ -77,6 +79,7 @@ JWT проходит `SessionAwareJWTAuthentication`: проверяется п | `mark_read` | `chat_id` | Deprecated; используйте HTTP mark-read action | | `calendar_subscribe` | `subscription_id`, `start`, `end` | Подписаться на snapshot событий диапазона; optional `calendar_id`, `scope` | | `calendar_unsubscribe` | `subscription_id` | Закрыть подписку диапазона календаря | +| `pong` | optional `timestamp` | Ответ на server heartbeat | Пример: @@ -89,6 +92,7 @@ JWT проходит `SessionAwareJWTAuthentication`: проверяется п Основные публичные `type`: - connection: `ping`, `error`; +- access lifecycle: `chat_access_changed`; - chat lifecycle: `chat_opened`, `chat_closed`, `initial_messages`; - messages: `new_message`, `list_update`, `message_edited`, `message_deleted`; - reactions: `reaction_added`, `reaction_removed`; @@ -108,16 +112,18 @@ Channel-layer handler names вроде `chat_message`, `notification_message` и `websocketManager` поддерживает один shared connection, пока есть message/status subscribers. Он: -- подключается только при наличии access token; -- повторно подключается после обновления JWT; +- подключается только при наличии auth-сессии и получает новый одноразовый ticket на каждую попытку; - отключается после очистки auth-сессии; - синхронизирует active chat после reconnect; -- использует exponential backoff до пяти попыток; +- отвечает `pong`, контролирует heartbeat и закрывает зависшее соединение; +- использует exponential backoff с jitter, после пяти быстрых попыток продолжает восстановление раз в минуту; +- немедленно повторяет подключение после browser `online` и возвращения вкладки; - закрывает socket, когда исчезает последний subscriber. Не создавайте WebSocket напрямую в page/component. Chat-specific hook подписывается на shared manager и не владеет отдельным соединением. После reconnect frontend повторяет активные календарные подписки. Если snapshot не приходит за три секунды или исчерпаны попытки reconnect, выполняется HTTP fallback для текущего диапазона. +Чаты синхронизируют сообщения после локального watermark, список чатов перечитывается, а задачи и список календарей выполняют контрольную REST-синхронизацию. Поэтому WebSocket используется как сигнал ускорения, а не как единственный источник истины. ## Изменение протокола @@ -139,10 +145,13 @@ pytest tests/realtime tests/api/test_documentation.py -q Ручная проверка через `wscat`: ```bash -wscat -c 'ws://localhost:8000/ws/?token=ACCESS_TOKEN' +TICKET=$(curl -s -X POST http://localhost:8000/api/auth/ws-ticket/ \ + -H 'Authorization: Bearer ACCESS_TOKEN' | jq -r .ticket) +wscat -c "ws://localhost:8000/ws/?ticket=${TICKET}" \ + -H 'Origin: http://localhost:3000' ``` -После подключения server отправляет `ping`; client-side `pong` action не требуется. +После подключения server отправляет `ping`; client должен ответить `{"action":"pong"}`. При отсутствии ответа в течение трёх интервалов соединение закрывается кодом `4408`. ## Production proxy @@ -154,5 +163,6 @@ Reverse proxy должен: - иметь idle timeout больше 20 секунд; - маршрутизировать все worker-ы к общему Redis channel layer; - передавать корректные `Host`/forwarded headers. +- не подменять и не отбрасывать допустимый `Origin`. `1006` — общий abnormal closure, а не самостоятельная причина. Сначала проверяйте close code/log consumer-а, JWT/session, proxy Upgrade, доступность Redis и совпадение endpoint-а. diff --git a/frontend/src/app/documents/page.tsx b/frontend/src/app/documents/page.tsx index 0189c785..e9c3fff6 100644 --- a/frontend/src/app/documents/page.tsx +++ b/frontend/src/app/documents/page.tsx @@ -34,6 +34,7 @@ import { Link2, } from "lucide-react"; import { DocumentAcknowledgementsReport } from "@/components/documents/DocumentAcknowledgementsReport"; +import { DocumentAcknowledgementAudienceBadges } from "@/components/documents/DocumentAcknowledgementAudience"; import { DocumentMetadataEditor } from "@/components/documents/DocumentMetadataEditor"; import { DocumentDetailModal } from "@/components/documents/DocumentDetailModal"; import { DocumentTaskLinks } from "@/components/documents/DocumentTaskLinks"; @@ -1601,8 +1602,6 @@ function DocumentsPageContent() { const isSelected = selectionMode && isDocumentSelected; const isDocumentMenuOpen = documentMenuOpenId === doc.id; const fileSize = formatFileSize(doc.file_size); - const recipientsCount = doc.recipients?.length || 0; - const departmentsCount = doc.departments?.length || 0; const hasPreview = Boolean(doc.file_url && canPreviewDocument(doc.file_name || doc.title)); const isAcknowledging = acknowledgingDocumentId === doc.id; const isChangingMaximumHiding = maximallyHidingDocumentId === doc.id; @@ -1976,27 +1975,10 @@ function DocumentsPageContent() { - {(doc.sent_to_all || recipientsCount > 0 || departmentsCount > 0) && ( -
- {doc.sent_to_all && ( - - - Для всей компании - - )} - {recipientsCount > 0 && ( - - - {recipientsCount} получ. - - )} - {departmentsCount > 0 && ( - - {departmentsCount} отдел. - - )} -
- )} + {doc.description && (

diff --git a/frontend/src/app/messages/[chatId]/page.tsx b/frontend/src/app/messages/[chatId]/page.tsx index 9c3b1c02..c56354d8 100644 --- a/frontend/src/app/messages/[chatId]/page.tsx +++ b/frontend/src/app/messages/[chatId]/page.tsx @@ -413,7 +413,13 @@ function MessageDialogPageContent() { // WS event handler useEffect(() => { handlers.current.onMessage = (data) => { - if (data.type === "new_message" && data.message) { + if ( + data.type === "chat_access_changed" + && data.chat_id === chatId + && data.can_access === false + ) { + router.replace("/messages"); + } else if (data.type === "new_message" && data.message) { const newMsg = data.message; const isMyMessage = newMsg.author_id === user?.id; const wasNearBottom = scroll.isNearBottom(); @@ -512,7 +518,7 @@ function MessageDialogPageContent() { handlers.current.onConnect = () => { void cm.syncLatestMessages({ limit: 30, scrollOnSync: scroll.isNearBottom() }); }; - }, [handlers, user?.id, chatId, scroll, cm, markRead, updateMessageReactionsSummary]); + }, [handlers, user?.id, chatId, scroll, cm, markRead, router, updateMessageReactionsSummary]); /* ── cleanup timers ── */ useEffect(() => () => { diff --git a/frontend/src/app/tasks/page.tsx b/frontend/src/app/tasks/page.tsx index 7dd68ee9..406fd0ef 100644 --- a/frontend/src/app/tasks/page.tsx +++ b/frontend/src/app/tasks/page.tsx @@ -715,11 +715,14 @@ const TaskCardContent = memo(function TaskCardContent({ /> ) : null}

-
+
#{task.id} -

+

{task.title}

-
+
#{task.id} -

+

{task.title}

diff --git a/frontend/src/components/documents/DocumentAcknowledgementAudience.tsx b/frontend/src/components/documents/DocumentAcknowledgementAudience.tsx new file mode 100644 index 00000000..1f8f39ad --- /dev/null +++ b/frontend/src/components/documents/DocumentAcknowledgementAudience.tsx @@ -0,0 +1,180 @@ +import { BookOpenCheck, Building2, User, Users } from "lucide-react"; + +import { + documentAudiencePersonName, + getDocumentAcknowledgementAudience, +} from "@/lib/document-acknowledgement-audience"; +import type { Document } from "@/types/api"; + +type DocumentAcknowledgementAudienceBadgesProps = { + document: Document; + className?: string; + maxTargets?: number; +}; + +export function DocumentAcknowledgementAudienceBadges({ + document, + className = "", + maxTargets = 3, +}: DocumentAcknowledgementAudienceBadgesProps) { + const audience = getDocumentAcknowledgementAudience(document); + + if (!audience.required) { + return ( +
+ + + Не требуется + +
+ ); + } + + if (audience.companyWide) { + return ( +
+ + + Вся компания + +
+ ); + } + + const targets = [ + ...audience.departments.map((department) => ({ + key: `department-${department.id}`, + label: department.name, + icon: , + })), + ...audience.recipients.map((recipient) => ({ + key: `recipient-${recipient.id}`, + label: documentAudiencePersonName(recipient), + icon: , + })), + ]; + const visibleTargets = targets.slice(0, Math.max(1, maxTargets)); + const hiddenTargets = Math.max(0, targets.length - visibleTargets.length); + + return ( +
+ {visibleTargets.map((target) => ( + + {target.icon} + {target.label} + + ))} + {hiddenTargets > 0 ? ( + + +{hiddenTargets} + + ) : null} + {targets.length === 0 ? ( + + + {audience.total > 0 ? `${audience.total} сотрудн.` : "Выбранные сотрудники"} + + ) : null} +
+ ); +} + +export function DocumentAcknowledgementAudienceDetails({ + document, +}: { + document: Document; +}) { + const audience = getDocumentAcknowledgementAudience(document); + + return ( +
+

+ Для ознакомления +

+ + {!audience.required ? ( +
+ + Подтверждение ознакомления не требуется +
+ ) : ( +
+ {audience.companyWide ? ( +
+ + Вся компания +
+ ) : null} + + {audience.departments.length > 0 ? ( +
+
+ + Отделы для ознакомления ({audience.departments.length}) +
+
+ {audience.departments.map((department) => ( +
+ + {department.name} +
+ ))} +
+
+ ) : null} + + {audience.recipients.length > 0 ? ( +
+
+ + Сотрудники для ознакомления ({audience.recipients.length}) +
+
+ {audience.recipients.map((recipient) => ( +
+ + {documentAudiencePersonName(recipient)} +
+ ))} +
+
+ ) : null} + + {!audience.companyWide + && audience.departments.length === 0 + && audience.recipients.length === 0 ? ( +
+ + {audience.total > 0 + ? `Выбрано сотрудников: ${audience.total}` + : "Выбранная аудитория ознакомления"} +
+ ) : null} + + {audience.total > 0 ? ( +

Всего должны ознакомиться: {audience.total}

+ ) : null} +
+ )} +
+ ); +} diff --git a/frontend/src/components/documents/DocumentDetailModal.tsx b/frontend/src/components/documents/DocumentDetailModal.tsx index cc141582..118d934e 100644 --- a/frontend/src/components/documents/DocumentDetailModal.tsx +++ b/frontend/src/components/documents/DocumentDetailModal.tsx @@ -10,8 +10,6 @@ import { Calendar, HardDrive, FolderOpen, - Users, - Building2, Tag, Clock, CheckCircle, @@ -29,6 +27,7 @@ import { X, } from "lucide-react"; import { DocumentAcknowledgeButton, DocumentAcknowledgement } from "./DocumentAcknowledgement"; +import { DocumentAcknowledgementAudienceDetails } from "./DocumentAcknowledgementAudience"; import { DocumentComments } from "./DocumentComments"; import { DocumentRelated } from "./DocumentRelated"; import { DocumentTaskLinks } from "./DocumentTaskLinks"; @@ -411,69 +410,7 @@ export function DocumentDetailModal({
)} - {/* Recipients */} - {(document.sent_to_all || - (document.departments && document.departments.length > 0) || - (document.recipients && document.recipients.length > 0)) && ( -
-

- Получатели -

-
- {document.sent_to_all && ( -
- - Все сотрудники -
- )} - - {document.departments && document.departments.length > 0 && ( -
-
- - Отделы ({document.departments.length}) -
-
- {document.departments.map((dept) => ( -
- - {dept.name} -
- ))} -
-
- )} - - {document.recipients && document.recipients.length > 0 && ( -
-
- - Получатели ({document.recipients.length}) -
-
- {document.recipients.slice(0, 5).map((recipient) => ( -
- - {recipient.last_name} {recipient.first_name} -
- ))} - {document.recipients.length > 5 && ( -
- ... и ещё {document.recipients.length - 5} -
- )} -
-
- )} -
-
- )} +
)} @@ -687,69 +624,7 @@ export function DocumentDetailModal({
)} - {/* Recipients */} - {(document.sent_to_all || - (document.departments && document.departments.length > 0) || - (document.recipients && document.recipients.length > 0)) && ( -
-

- Получатели -

-
- {document.sent_to_all && ( -
- - Все сотрудники -
- )} - - {document.departments && document.departments.length > 0 && ( -
-
- - Отделы ({document.departments.length}) -
-
- {document.departments.map((dept) => ( -
- - {dept.name} -
- ))} -
-
- )} - - {document.recipients && document.recipients.length > 0 && ( -
-
- - Получатели ({document.recipients.length}) -
-
- {document.recipients.slice(0, 5).map((recipient) => ( -
- - {recipient.last_name} {recipient.first_name} -
- ))} - {document.recipients.length > 5 && ( -
- ... и ещё {document.recipients.length - 5} -
- )} -
-
- )} -
-
- )} +
)} diff --git a/frontend/src/components/feed/FeedRegulationCard.tsx b/frontend/src/components/feed/FeedRegulationCard.tsx index 65a28f47..3d4735f3 100644 --- a/frontend/src/components/feed/FeedRegulationCard.tsx +++ b/frontend/src/components/feed/FeedRegulationCard.tsx @@ -19,6 +19,7 @@ import { } from "lucide-react"; import { ExpandableFeedText } from "@/components/feed/ExpandableFeedText"; +import { DocumentAcknowledgementAudienceBadges } from "@/components/documents/DocumentAcknowledgementAudience"; import { RequestAvatar } from "@/components/requests/RequestAvatar"; import TaskLinkPill from "@/components/tasks/TaskLinkPill"; import { canPreviewDocument } from "@/lib/document-preview"; @@ -299,16 +300,7 @@ export function FeedRegulationCard({ ) : null}
- {document.sent_to_all ? ( - - Для всей компании - - ) : null} - {(document.departments || []).slice(0, 3).map((department) => ( - - {department.name} - - ))} + {acknowledgementRequiredForUser ? ( diff --git a/frontend/src/contexts/CalendarContext.tsx b/frontend/src/contexts/CalendarContext.tsx index ae52af0e..02083448 100644 --- a/frontend/src/contexts/CalendarContext.tsx +++ b/frontend/src/contexts/CalendarContext.tsx @@ -78,7 +78,9 @@ export function CalendarProvider({ children }: { children: ReactNode }) { useEffect(() => { if (!user) return undefined; - const unsubscribe = wsManager.subscribe((event) => { + let hasConnected = false; + let wasConnected = false; + const unsubscribeMessages = wsManager.subscribe((event) => { if (event.type !== "calendar_update") return; const data = event.data as { @@ -100,9 +102,18 @@ export function CalendarProvider({ children }: { children: ReactNode }) { } }, 100); }); + const unsubscribeStatus = wsManager.subscribeStatus((status) => { + if (status.isConnected && hasConnected && !wasConnected) { + calendarService.clearCache(); + void loadCalendars(); + } + if (status.isConnected) hasConnected = true; + wasConnected = status.isConnected; + }); return () => { - unsubscribe(); + unsubscribeMessages(); + unsubscribeStatus(); if (realtimeSyncTimerRef.current) { clearTimeout(realtimeSyncTimerRef.current); realtimeSyncTimerRef.current = null; diff --git a/frontend/src/hooks/useMessagesChatList.ts b/frontend/src/hooks/useMessagesChatList.ts index caf80f1f..b3e13557 100644 --- a/frontend/src/hooks/useMessagesChatList.ts +++ b/frontend/src/hooks/useMessagesChatList.ts @@ -135,6 +135,10 @@ function startRealtime() { const unsubscribeMessages = wsManager.subscribe((event) => { const data = event as RealtimeChatEvent; + if (data.type === "chat_access_changed") { + void loadAllChats(false); + return; + } if ( (data.type === "list_update" || data.type === "message_deleted") && data.event === "message_deleted" diff --git a/frontend/src/hooks/useTaskBoardRealtime.ts b/frontend/src/hooks/useTaskBoardRealtime.ts index d043d310..a21fb1ca 100644 --- a/frontend/src/hooks/useTaskBoardRealtime.ts +++ b/frontend/src/hooks/useTaskBoardRealtime.ts @@ -36,7 +36,9 @@ export function useTaskBoardRealtime(options: TaskBoardRealtimeOptions): void { }, [options]); useEffect(() => { - const unsubscribe = wsManager.subscribe((event) => { + let hasConnected = false; + let wasConnected = false; + const unsubscribeMessages = wsManager.subscribe((event) => { const update = parseTaskBoardSocketUpdate(event); if (!update) return; @@ -74,9 +76,17 @@ export function useTaskBoardRealtime(options: TaskBoardRealtimeOptions): void { optionsRef.current.onBoardChange(update.boardId); }, current.debounceMs ?? 250); }); + const unsubscribeStatus = wsManager.subscribeStatus((status) => { + if (status.isConnected && hasConnected && !wasConnected) { + optionsRef.current.onBoardChange(0); + } + if (status.isConnected) hasConnected = true; + wasConnected = status.isConnected; + }); return () => { - unsubscribe(); + unsubscribeMessages(); + unsubscribeStatus(); if (syncTimerRef.current) { clearTimeout(syncTimerRef.current); syncTimerRef.current = null; diff --git a/frontend/src/lib/document-acknowledgement-audience.test.js b/frontend/src/lib/document-acknowledgement-audience.test.js new file mode 100644 index 00000000..8c249d54 --- /dev/null +++ b/frontend/src/lib/document-acknowledgement-audience.test.js @@ -0,0 +1,62 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { getDocumentAcknowledgementAudience } from "./document-acknowledgement-audience.ts"; + +const accounting = { id: 10, name: "Бухгалтерия" }; +const procurement = { id: 20, name: "Закупки" }; +const employee = { id: 7, first_name: "Иван", last_name: "Иванов" }; + +test("company-wide badge is based on acknowledgement audience, not access alone", () => { + assert.equal( + getDocumentAcknowledgementAudience({ + sent_to_all: true, + acknowledgement_required: true, + acknowledgement_for_all: false, + acknowledgement_departments: [accounting], + }).companyWide, + false, + ); +}); + +test("acknowledgement for everyone inherits a selectively shared audience", () => { + const audience = getDocumentAcknowledgementAudience({ + sent_to_all: false, + departments: [accounting], + recipients: [employee], + acknowledgement_required: true, + acknowledgement_for_all: true, + acknowledgement_departments: [procurement], + }); + + assert.equal(audience.companyWide, false); + assert.deepEqual(audience.departments, [accounting]); + assert.deepEqual(audience.recipients, [employee]); +}); + +test("selective acknowledgement ignores the broader access audience", () => { + const audience = getDocumentAcknowledgementAudience({ + sent_to_all: true, + departments: [accounting], + acknowledgement_required: true, + acknowledgement_for_all: false, + acknowledgement_departments: [procurement], + acknowledgement_recipients: [employee], + }); + + assert.deepEqual(audience.departments, [procurement]); + assert.deepEqual(audience.recipients, [employee]); +}); + +test("documents without acknowledgement have no presentation audience", () => { + const audience = getDocumentAcknowledgementAudience({ + sent_to_all: true, + acknowledgement_required: false, + acknowledgement_for_all: true, + }); + + assert.equal(audience.required, false); + assert.equal(audience.companyWide, false); + assert.deepEqual(audience.departments, []); + assert.deepEqual(audience.recipients, []); +}); diff --git a/frontend/src/lib/document-acknowledgement-audience.ts b/frontend/src/lib/document-acknowledgement-audience.ts new file mode 100644 index 00000000..b1edcb75 --- /dev/null +++ b/frontend/src/lib/document-acknowledgement-audience.ts @@ -0,0 +1,66 @@ +import type { Document, User } from "@/types/api"; + +type AudienceDepartment = NonNullable[number]; + +export type DocumentAcknowledgementAudience = { + required: boolean; + companyWide: boolean; + departments: AudienceDepartment[]; + recipients: User[]; + total: number; +}; + +function uniqueById(items: T[]): T[] { + return [...new Map(items.map((item) => [item.id, item])).values()]; +} + +/** + * Resolves the business audience that must acknowledge a document. + * Access fields are used only when acknowledgement is configured for every + * employee who can open a selectively shared document. + */ +export function getDocumentAcknowledgementAudience( + document: Document, +): DocumentAcknowledgementAudience { + if (!document.acknowledgement_required) { + return { + required: false, + companyWide: false, + departments: [], + recipients: [], + total: 0, + }; + } + + const companyWide = Boolean( + document.acknowledgement_for_all && document.sent_to_all, + ); + const inheritsAccessAudience = Boolean(document.acknowledgement_for_all); + + return { + required: true, + companyWide, + departments: companyWide + ? [] + : uniqueById( + inheritsAccessAudience + ? document.departments || [] + : document.acknowledgement_departments || [], + ), + recipients: companyWide + ? [] + : uniqueById( + inheritsAccessAudience + ? document.recipients || [] + : document.acknowledgement_recipients || [], + ), + total: Math.max(0, Number(document.acknowledgement_total) || 0), + }; +} + +export function documentAudiencePersonName(person: User): string { + return [person.last_name, person.first_name, person.patronymic] + .filter(Boolean) + .join(" ") + .trim() || person.email || `Сотрудник #${person.id}`; +} diff --git a/frontend/src/lib/websocketManager.test.js b/frontend/src/lib/websocketManager.test.js index 1ddacdee..7dc95042 100644 --- a/frontend/src/lib/websocketManager.test.js +++ b/frontend/src/lib/websocketManager.test.js @@ -1,10 +1,13 @@ import assert from "node:assert/strict"; import test from "node:test"; +import { apiClient } from "@/lib/api"; + test("multiple realtime subscribers share one WebSocket connection", async () => { const originalWindow = globalThis.window; const originalLocalStorage = globalThis.localStorage; const originalWebSocket = globalThis.WebSocket; + const originalRequest = apiClient.request; class FakeWebSocket { static CONNECTING = 0; @@ -29,6 +32,10 @@ test("multiple realtime subscribers share one WebSocket connection", async () => this.sent.push(JSON.parse(payload)); } + receive(payload) { + this.onmessage?.({ data: JSON.stringify(payload) }); + } + close() { this.readyState = FakeWebSocket.CLOSED; this.onclose?.({ code: 1000, reason: "", wasClean: true }); @@ -42,13 +49,21 @@ test("multiple realtime subscribers share one WebSocket connection", async () => }, }; globalThis.WebSocket = FakeWebSocket; + apiClient.request = async (endpoint) => { + assert.equal(endpoint, "/api/auth/ws-ticket/"); + return { ticket: "single-use-ticket", expires_in: 30 }; + }; try { const { default: manager } = await import("./websocketManager.ts"); const unsubscribeMessages = manager.subscribe(() => {}); const unsubscribeStatus = manager.subscribeStatus(() => {}); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(FakeWebSocket.instances.length, 1); + assert.match(FakeWebSocket.instances[0].url, /ticket=single-use-ticket/); + assert.doesNotMatch(FakeWebSocket.instances[0].url, /token=/); const socket = FakeWebSocket.instances[0]; socket.open(); @@ -61,11 +76,116 @@ test("multiple realtime subscribers share one WebSocket connection", async () => load_history: false, }); + socket.receive({ type: "ping", timestamp: "2026-08-17T12:00:00Z" }); + assert.deepEqual(socket.sent.at(-1), { + action: "pong", + timestamp: "2026-08-17T12:00:00Z", + }); + unsubscribeMessages(); unsubscribeStatus(); } finally { globalThis.window = originalWindow; globalThis.localStorage = originalLocalStorage; globalThis.WebSocket = originalWebSocket; + apiClient.request = originalRequest; + } +}); + +test("connection recovers immediately when browser comes back online", async () => { + const originalWindow = globalThis.window; + const originalDocument = globalThis.document; + const originalLocalStorage = globalThis.localStorage; + const originalWebSocket = globalThis.WebSocket; + const originalRequest = apiClient.request; + const originalSetTimeout = globalThis.setTimeout; + const originalClearTimeout = globalThis.clearTimeout; + const originalSetInterval = globalThis.setInterval; + const originalClearInterval = globalThis.clearInterval; + const listeners = new Map(); + const scheduledTimeouts = []; + + class FakeWebSocket { + static OPEN = 1; + static instances = []; + + constructor(url) { + this.url = url; + this.readyState = 0; + FakeWebSocket.instances.push(this); + } + + open() { + this.readyState = FakeWebSocket.OPEN; + this.onopen?.(); + } + + send() {} + + fail() { + this.readyState = 3; + this.onclose?.({ code: 1006, reason: "network", wasClean: false }); + } + + close() { + this.readyState = 3; + } + } + + globalThis.window = { + addEventListener(type, handler) { + listeners.set(type, handler); + }, + }; + globalThis.document = { + visibilityState: "visible", + addEventListener(type, handler) { + listeners.set(type, handler); + }, + }; + globalThis.localStorage = { + getItem(key) { + return key === "access_token" ? "access" : null; + }, + }; + globalThis.WebSocket = FakeWebSocket; + globalThis.setTimeout = (handler, delay) => { + const timer = { handler, delay, cleared: false }; + scheduledTimeouts.push(timer); + return timer; + }; + globalThis.clearTimeout = (timer) => { timer.cleared = true; }; + globalThis.setInterval = () => ({ interval: true }); + globalThis.clearInterval = () => {}; + apiClient.request = async () => ({ ticket: "recovery-ticket", expires_in: 30 }); + + try { + const { WebSocketManager } = await import("./websocketManager.ts"); + const manager = new WebSocketManager(); + const unsubscribe = manager.subscribe(() => {}); + await new Promise((resolve) => originalSetTimeout(resolve, 0)); + assert.equal(FakeWebSocket.instances.length, 1); + + FakeWebSocket.instances[0].open(); + FakeWebSocket.instances[0].fail(); + assert.equal(scheduledTimeouts.length, 1); + assert.ok(scheduledTimeouts[0].delay >= 1600); + assert.ok(scheduledTimeouts[0].delay <= 2400); + + listeners.get("online")?.(); + await new Promise((resolve) => originalSetTimeout(resolve, 0)); + assert.equal(scheduledTimeouts[0].cleared, true); + assert.equal(FakeWebSocket.instances.length, 2); + unsubscribe(); + } finally { + globalThis.window = originalWindow; + globalThis.document = originalDocument; + globalThis.localStorage = originalLocalStorage; + globalThis.WebSocket = originalWebSocket; + globalThis.setTimeout = originalSetTimeout; + globalThis.clearTimeout = originalClearTimeout; + globalThis.setInterval = originalSetInterval; + globalThis.clearInterval = originalClearInterval; + apiClient.request = originalRequest; } }); diff --git a/frontend/src/lib/websocketManager.ts b/frontend/src/lib/websocketManager.ts index 1a5a2a33..55f3917f 100644 --- a/frontend/src/lib/websocketManager.ts +++ b/frontend/src/lib/websocketManager.ts @@ -1,8 +1,6 @@ -/** - * Singleton WebSocket Manager - * Управляет единственным WebSocket соединением для всего приложения - */ +/** Shared, authenticated and self-healing WebSocket transport. */ +import { apiClient } from '@/lib/api'; import { AUTH_SESSION_EVENT, type AuthSessionEventDetail } from '@/lib/api/client'; import { getWebSocketUrl } from '@/lib/url'; @@ -18,28 +16,48 @@ export type ConnectionStatus = { maxReconnectAttempts: number; }; type StatusHandler = (status: ConnectionStatus) => void; +type WebSocketTicketResponse = { ticket: string; expires_in: number }; + +const RAPID_RECONNECT_LIMIT = 5; +const MAX_RECONNECT_DELAY_MS = 60_000; +const HEARTBEAT_TIMEOUT_MS = 70_000; +const HEARTBEAT_CHECK_MS = 15_000; export class WebSocketManager { private static instance: WebSocketManager | null = null; private ws: WebSocket | null = null; - private reconnectTimeout: NodeJS.Timeout | null = null; + private reconnectTimeout: ReturnType | null = null; + private heartbeatTimer: ReturnType | null = null; private reconnectAttempts = 0; - private maxReconnectAttempts = 5; private activeChatId: number | null = null; - private messageHandlers: Set = new Set(); - private statusHandlers: Set = new Set(); + private readonly messageHandlers = new Set(); + private readonly statusHandlers = new Set(); private isConnecting = false; private shouldReconnect = false; + private connectionGeneration = 0; + private lastServerMessageAt = 0; private constructor() { - if (typeof window !== 'undefined') { - window.addEventListener(AUTH_SESSION_EVENT, (event) => { - const detail = (event as CustomEvent).detail; - if (detail?.type === 'token-refreshed') this.reconnectWithLatestToken(); - if (detail?.type === 'session-cleared') this.disconnect(); - }); - window.addEventListener('storage', (event) => { - if (event.key === 'access_token') this.reconnectWithLatestToken(); + if (typeof window === 'undefined') return; + + window.addEventListener(AUTH_SESSION_EVENT, (event) => { + const detail = (event as CustomEvent).detail; + if (detail?.type === 'session-cleared') { + this.disconnect(); + } else if (detail?.type === 'token-refreshed' && !this.ws && this.hasSubscribers()) { + this.reconnectAttempts = 0; + this.connect(); + } + }); + window.addEventListener('storage', (event) => { + if (event.key !== 'access_token') return; + if (!event.newValue) this.disconnect(); + else if (!this.ws && this.hasSubscribers()) this.connectImmediately(); + }); + window.addEventListener('online', () => this.connectImmediately()); + if (typeof document !== 'undefined') { + document.addEventListener('visibilitychange', () => { + if (document.visibilityState === 'visible') this.connectImmediately(); }); } } @@ -60,7 +78,7 @@ export class WebSocketManager { isConnected: this.ws?.readyState === WebSocket.OPEN, isConnecting: this.isConnecting, reconnectAttempts: this.reconnectAttempts, - maxReconnectAttempts: this.maxReconnectAttempts, + maxReconnectAttempts: RAPID_RECONNECT_LIMIT, }; } @@ -75,20 +93,14 @@ export class WebSocketManager { }); } - send(payload: Record) { - if (this.ws?.readyState !== WebSocket.OPEN) { - return false; - } - + send(payload: Record): boolean { + if (this.ws?.readyState !== WebSocket.OPEN) return false; this.ws.send(JSON.stringify(payload)); return true; } private syncActiveChatOnSocket() { - if (this.activeChatId === null) { - return; - } - + if (this.activeChatId === null) return; this.send({ action: 'open_chat', chat_id: this.activeChatId, @@ -96,85 +108,72 @@ export class WebSocketManager { }); } - private reconnectWithLatestToken() { - if (!this.hasSubscribers() || typeof window === 'undefined') return; - if (!localStorage.getItem('access_token')) { - this.disconnect(); - return; - } - - if (this.reconnectTimeout) { - clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; - } - const previousSocket = this.ws; - this.ws = null; - if (previousSocket) { - previousSocket.onclose = null; - previousSocket.close(); - } - this.isConnecting = false; - this.reconnectAttempts = 0; - this.shouldReconnect = true; - this.connect(); + private async requestTicket(): Promise { + const response = await apiClient.request( + '/api/auth/ws-ticket/', + { method: 'POST' }, + ); + if (!response.ticket) throw new Error('WebSocket ticket is missing'); + return response.ticket; } connect() { - // Если уже подключены или подключаемся, не создаем новое соединение - if (this.ws?.readyState === WebSocket.OPEN || this.isConnecting) { - console.log('[WS Manager] Already connected or connecting'); - return; - } - + if (this.ws?.readyState === WebSocket.OPEN || this.isConnecting) return; if (!this.hasSubscribers()) { this.shouldReconnect = false; return; } - - // Проверяем что мы в браузере и пользователь авторизован if (typeof window === 'undefined' || !localStorage.getItem('access_token')) { - console.log('[WS Manager] Not in browser or not authenticated'); this.notifyStatus(); return; } - if (this.reconnectTimeout) { - clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; - } - + this.clearReconnectTimer(); this.shouldReconnect = true; this.isConnecting = true; + const generation = ++this.connectionGeneration; this.notifyStatus(); + void this.openConnection(generation); + } + private async openConnection(generation: number) { try { - let wsUrl = getWebSocketUrl(); + const ticket = await this.requestTicket(); + if ( + generation !== this.connectionGeneration + || !this.shouldReconnect + || !this.hasSubscribers() + ) return; - const token = localStorage.getItem('access_token'); - if (token) { - wsUrl += `${wsUrl.includes('?') ? '&' : '?'}token=${token}`; - } - - this.ws = new WebSocket(wsUrl); + const baseUrl = getWebSocketUrl(); + const wsUrl = `${baseUrl}${baseUrl.includes('?') ? '&' : '?'}ticket=${encodeURIComponent(ticket)}`; + const socket = new WebSocket(wsUrl); + this.ws = socket; - this.ws.onopen = () => { - console.log('[WS Manager] Connected'); + socket.onopen = () => { + if (this.ws !== socket) return; this.isConnecting = false; this.reconnectAttempts = 0; + this.lastServerMessageAt = Date.now(); + this.startHeartbeatWatchdog(socket); this.syncActiveChatOnSocket(); this.notifyStatus(); }; - this.ws.onmessage = (event) => { + socket.onmessage = (event) => { + if (this.ws !== socket) return; + this.lastServerMessageAt = Date.now(); try { - const data = JSON.parse(event.data); - - // Отправляем сообщение всем подписчикам - this.messageHandlers.forEach(handler => { + const data = JSON.parse(event.data) as RealtimeEvent; + if (data.type === 'ping') { + this.send({ action: 'pong', timestamp: data.timestamp }); + return; + } + this.messageHandlers.forEach((handler) => { try { handler(data); - } catch (err) { - console.error('[WS Manager] Handler error:', err); + } catch (error) { + console.error('[WS Manager] Handler error:', error); } }); } catch (error) { @@ -182,114 +181,118 @@ export class WebSocketManager { } }; - this.ws.onerror = (error) => { - console.error('[WS Manager] Error event:', error); - console.error('[WS Manager] WebSocket state:', this.ws?.readyState); + socket.onerror = () => { + if (this.ws !== socket) return; this.isConnecting = false; this.notifyStatus(); }; - this.ws.onclose = (event) => { - console.log('[WS Manager] Disconnected'); - console.log('[WS Manager] Close event code:', event.code); - console.log('[WS Manager] Close event reason:', event.reason); - console.log('[WS Manager] Close event wasClean:', event.wasClean); - this.isConnecting = false; + socket.onclose = () => { + if (this.ws !== socket) return; + this.stopHeartbeatWatchdog(); this.ws = null; + this.isConnecting = false; this.notifyStatus(); - - // Автоматический reconnect - if (this.shouldReconnect && this.hasSubscribers() && this.reconnectAttempts < this.maxReconnectAttempts) { - this.reconnectAttempts++; - this.notifyStatus(); - const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000); - console.log(`[WS Manager] Reconnecting in ${delay}ms... (${this.reconnectAttempts}/${this.maxReconnectAttempts})`); - this.reconnectTimeout = setTimeout(() => this.connect(), delay); - } + this.scheduleReconnect(); }; } catch (error) { - console.error('[WS Manager] Connection failed:', error); + if (generation !== this.connectionGeneration) return; + console.error('[WS Manager] Connection preparation failed:', error); this.isConnecting = false; this.notifyStatus(); + this.scheduleReconnect(); } } + private scheduleReconnect() { + if (!this.shouldReconnect || !this.hasSubscribers() || this.reconnectTimeout) return; + this.reconnectAttempts += 1; + this.notifyStatus(); + + const exponentialAttempt = Math.min(this.reconnectAttempts, RAPID_RECONNECT_LIMIT); + const baseDelay = this.reconnectAttempts > RAPID_RECONNECT_LIMIT + ? MAX_RECONNECT_DELAY_MS + : Math.min(1000 * 2 ** exponentialAttempt, 30_000); + const jitteredDelay = Math.round(baseDelay * (0.8 + Math.random() * 0.4)); + this.reconnectTimeout = setTimeout(() => { + this.reconnectTimeout = null; + this.connect(); + }, jitteredDelay); + } + + private connectImmediately() { + if (!this.shouldReconnect || !this.hasSubscribers() || this.ws || this.isConnecting) return; + this.clearReconnectTimer(); + this.reconnectAttempts = 0; + this.connect(); + } + + private startHeartbeatWatchdog(socket: WebSocket) { + this.stopHeartbeatWatchdog(); + this.heartbeatTimer = setInterval(() => { + if ( + this.ws === socket + && socket.readyState === WebSocket.OPEN + && Date.now() - this.lastServerMessageAt > HEARTBEAT_TIMEOUT_MS + ) { + socket.close(4408, 'Heartbeat timeout'); + } + }, HEARTBEAT_CHECK_MS); + } + + private stopHeartbeatWatchdog() { + if (!this.heartbeatTimer) return; + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + + private clearReconnectTimer() { + if (!this.reconnectTimeout) return; + clearTimeout(this.reconnectTimeout); + this.reconnectTimeout = null; + } + disconnect() { this.shouldReconnect = false; - - if (this.reconnectTimeout) { - clearTimeout(this.reconnectTimeout); - this.reconnectTimeout = null; + this.connectionGeneration += 1; + this.clearReconnectTimer(); + this.stopHeartbeatWatchdog(); + const socket = this.ws; + this.ws = null; + if (socket) { + socket.onclose = null; + socket.close(1000, 'Client disconnect'); } - - if (this.ws) { - this.ws.close(); - this.ws = null; - } - this.isConnecting = false; this.reconnectAttempts = 0; this.notifyStatus(); } setActiveChat(chatId: number) { - if (!Number.isFinite(chatId) || chatId <= 0) { - return; - } - - if (this.activeChatId === chatId) { - return; - } - + if (!Number.isFinite(chatId) || chatId <= 0 || this.activeChatId === chatId) return; const previousChatId = this.activeChatId; this.activeChatId = chatId; - if (previousChatId !== null) { - this.send({ - action: 'close_chat', - chat_id: previousChatId, - }); + this.send({ action: 'close_chat', chat_id: previousChatId }); } - this.syncActiveChatOnSocket(); } clearActiveChat(chatId?: number) { - if (this.activeChatId === null) { - return; - } - - if (typeof chatId === 'number' && this.activeChatId !== chatId) { - return; - } - + if (this.activeChatId === null) return; + if (typeof chatId === 'number' && this.activeChatId !== chatId) return; const chatToClose = this.activeChatId; this.activeChatId = null; - - this.send({ - action: 'close_chat', - chat_id: chatToClose, - }); + this.send({ action: 'close_chat', chat_id: chatToClose }); } subscribe(handler: MessageHandler) { this.messageHandlers.add(handler); this.shouldReconnect = true; - - // Если есть подписчик и нет соединения, подключаемся - if (this.hasSubscribers() && !this.ws && !this.isConnecting) { - this.connect(); - } - - // Возвращаем функцию отписки + if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect(); return () => { this.messageHandlers.delete(handler); - - // Если больше нет подписчиков, отключаемся - if (!this.hasSubscribers()) { - console.log('[WS Manager] No more subscribers, disconnecting'); - this.disconnect(); - } + if (!this.hasSubscribers()) this.disconnect(); }; } @@ -297,18 +300,10 @@ export class WebSocketManager { this.statusHandlers.add(handler); handler(this.getConnectionStatus()); this.shouldReconnect = true; - - if (this.hasSubscribers() && !this.ws && !this.isConnecting) { - this.connect(); - } - + if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect(); return () => { this.statusHandlers.delete(handler); - - if (!this.hasSubscribers()) { - console.log('[WS Manager] No more subscribers, disconnecting'); - this.disconnect(); - } + if (!this.hasSubscribers()) this.disconnect(); }; } diff --git a/frontend/src/services/calendarService.test.js b/frontend/src/services/calendarService.test.js index fabd57f3..4a500d1f 100644 --- a/frontend/src/services/calendarService.test.js +++ b/frontend/src/services/calendarService.test.js @@ -56,6 +56,7 @@ test("calendar service receives range snapshots through the shared WebSocket", a const originalLocalStorage = globalThis.localStorage; const originalWebSocket = globalThis.WebSocket; const originalLoadEvents = calendarService.loadEvents; + const originalRequest = apiClient.request; class FakeWebSocket { static CONNECTING = 0; @@ -97,6 +98,10 @@ test("calendar service receives range snapshots through the shared WebSocket", a }, }; globalThis.WebSocket = FakeWebSocket; + apiClient.request = async (endpoint) => { + assert.equal(endpoint, "/api/auth/ws-ticket/"); + return { ticket: "calendar-ws-ticket", expires_in: 30 }; + }; const received = []; const loading = []; @@ -113,6 +118,8 @@ test("calendar service receives range snapshots through the shared WebSocket", a onLoading: (value) => loading.push(value), }); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(FakeWebSocket.instances.length, 1); const socket = FakeWebSocket.instances[0]; socket.open(); @@ -170,5 +177,6 @@ test("calendar service receives range snapshots through the shared WebSocket", a globalThis.window = originalWindow; globalThis.localStorage = originalLocalStorage; globalThis.WebSocket = originalWebSocket; + apiClient.request = originalRequest; } });