Швырнул архитектуру, чтобы позже объединить контроллеры веба и сокета в одно, а также разделить процессоры

src/tamtam/controller.py

@@ -0,0 +1,23 @@
+import asyncio
+from tamtam.socket import TTMobileServer
+from classes.controllerbase import ControllerBase
+from common.config import ServerConfig
+
+class TTMobileController(ControllerBase):
+    def __init__(self):
+        self.config = ServerConfig()
+
+    def launch(self, api):
+        async def _start_all():
+            await asyncio.gather(
+                TTMobileServer(
+                    host=self.config.host,
+                    port=self.config.tamtam_tcp_port,
+                    ssl_context=api['ssl'],
+                    db_pool=api['db'],
+                    clients=api['clients'],
+                    send_event=api['event']
+                ).start()
+            )
+
+        return _start_all()

src/tamtam/models.py

@@ -0,0 +1,34 @@
+import pydantic
+
+class UserAgentModel(pydantic.BaseModel):
+    deviceType: str
+    appVersion: str
+    osVersion: str
+    timezone: str
+    screen: str
+    pushDeviceType: str
+    locale: str
+    deviceName: str
+    deviceLocale: str
+
+class HelloPayloadModel(pydantic.BaseModel):
+    userAgent: UserAgentModel
+    deviceId: str
+
+class RequestCodePayloadModel(pydantic.BaseModel):
+    phone: str
+
+class VerifyCodePayloadModel(pydantic.BaseModel):
+    verifyCode: str
+    authTokenType: str
+    token: str
+
+class FinalAuthPayloadModel(pydantic.BaseModel):
+    deviceType: str
+    tokenType: str
+    deviceId: str
+    token: str
+
+class LoginPayloadModel(pydantic.BaseModel):
+    interactive: bool
+    token: str

src/tamtam/processors.py

@@ -0,0 +1,400 @@
+import hashlib
+import secrets
+import time
+import logging
+import json
+import re
+from common.static import Static
+from common.tools import Tools
+
+from common.proto_tcp import MobileProto
+from common.proto_web import WebProto
+from common.opcodes import Opcodes
+
+from tamtam.models import *
+
+
+class Processors:
+    def __init__(self, db_pool=None, clients=None, send_event=None, type="socket"):
+        if clients is None:
+            clients = {}  # Более правильная логика
+        self.static = Static()
+        self.tools = Tools()
+        self.opcodes = Opcodes()
+        self.error_types = self.static.ErrorTypes()
+        self.db_pool = db_pool
+        self.logger = logging.getLogger(__name__)
+
+        if type == "socket":
+            self.proto = MobileProto()
+        elif type == "web":
+            self.proto = WebProto()
+
+    async def _send(self, writer, packet):
+        try:
+            writer.write(packet)
+            await writer.drain()
+        except:
+            pass
+
+    async def _send_error(self, seq, opcode, type, writer):
+        payload = self.static.ERROR_TYPES.get(type, {
+            "localizedMessage": "Неизвестная ошибка",
+            "error": "unknown.error",
+            "message": "Unknown error",
+            "title": "Неизвестная ошибка"
+        })
+
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_ERR, seq=seq, opcode=opcode, payload=payload
+        )
+
+        await self._send(writer, packet)
+
+    async def session_init(self, payload, seq, writer):
+        """Обработчик приветствия"""
+        # Валидируем данные пакета
+        try:
+            HelloPayloadModel.model_validate(payload)
+        except Exception as e:
+            await self._send_error(seq, self.opcodes.SESSION_INIT, self.error_types.INVALID_PAYLOAD, writer)
+            return None, None
+
+        # Получаем данные из пакета
+        device_type = payload.get("userAgent").get("deviceType")
+        device_name = payload.get("userAgent").get("deviceName")
+
+        # Данные пакета
+        payload = {
+            "proxy": "",
+            "logs-enabled": False,
+            "proxy-domains": [],
+            "location": "RU",
+            "libh-enabled": False,
+            "phone-auto-complete-enabled": False
+        }
+
+        # Собираем пакет
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_OK, seq=seq, opcode=self.opcodes.SESSION_INIT, payload=payload
+        )
+
+        # Отправляем
+        await self._send(writer, packet)
+        return device_type, device_name
+
+    async def auth_request(self, payload, seq, writer):
+        """Обработчик запроса кода"""
+        # Валидируем данные пакета
+        try:
+            RequestCodePayloadModel.model_validate(payload)
+        except Exception as e:
+            await self._send_error(seq, self.opcodes.AUTH_REQUEST, self.error_types.INVALID_PAYLOAD, writer)
+            return
+
+        # Извлекаем телефон из пакета
+        phone = re.sub(r'\D', '', payload.get("phone", ""))  # Не хардкодим, через регулярки
+
+        # Генерируем токен с кодом
+        code = f"{secrets.randbelow(1_000_000):06d}"  # Старая версия ненадежна, могла отбросить ведущие нули или вообще интерпритировать как систему счисления с основанием 8
+        token = secrets.token_urlsafe(128)
+
+        # Хешируем
+        code_hash = hashlib.sha256(code.encode()).hexdigest()
+        token_hash = hashlib.sha256(token.encode()).hexdigest()
+
+        # Срок жизни токена (5 минут)
+        expires = int(time.time()) + 300
+
+        # Ищем пользователя, и если он существует, сохраняем токен
+        async with self.db_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute("SELECT * FROM users WHERE phone = %s", (phone,))
+                user = await cursor.fetchone()
+
+                # Если пользователь существует, сохраняем токен
+                if user:
+                    # Сохраняем токен
+                    await cursor.execute("INSERT INTO auth_tokens (phone, token_hash, code_hash, expires, state) VALUES (%s, %s, %s, %s, %s)", (phone, token_hash, code_hash, expires, "started",))
+
+        # Данные пакета
+        payload = {
+            "verifyToken": token,
+            "retries": 5,
+            "codeDelay": 60,
+            "codeLength": 6,
+            "callDelay": 0,
+            "requestType": "SMS"
+        }
+
+        # Собираем пакет
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_OK, seq=seq, opcode=self.opcodes.AUTH_REQUEST, payload=payload
+        )
+
+        # Отправляем
+        await self._send(writer, packet)
+        self.logger.debug(f"Код для {phone}: {code}")
+
+    async def auth(self, payload, seq, writer):
+        """Обработчик проверки кода"""
+        # Валидируем данные пакета
+        try:
+            VerifyCodePayloadModel.model_validate(payload)
+        except Exception as e:
+            await self._send_error(seq, self.opcodes.AUTH, self.error_types.INVALID_PAYLOAD, writer)
+            return
+
+        # Извлекаем данные из пакета
+        code = payload.get("verifyCode")
+        token = payload.get("token")
+
+        # Хешируем токен с кодом
+        hashed_code = hashlib.sha256(code.encode()).hexdigest()
+        hashed_token = hashlib.sha256(token.encode()).hexdigest()
+
+        # Ищем токен с кодом
+        async with self.db_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                # Ищем токен
+                await cursor.execute("SELECT * FROM auth_tokens WHERE token_hash = %s AND expires > UNIX_TIMESTAMP()",
+                                     (hashed_token,))
+                stored_token = await cursor.fetchone()
+
+                if not stored_token:
+                    await self._send_error(seq, self.opcodes.AUTH, self.error_types.CODE_EXPIRED, writer)
+                    return
+
+                # Проверяем код
+                if stored_token.get("code_hash") != hashed_code:
+                    await self._send_error(seq, self.opcodes.AUTH, self.error_types.INVALID_CODE, writer)
+                    return
+
+                # Ищем аккаунт
+                await cursor.execute("SELECT * FROM users WHERE phone = %s", (stored_token.get("phone"),))
+                account = await cursor.fetchone()
+
+                # Обновляем состояние токена
+                await cursor.execute("UPDATE auth_tokens set state = %s WHERE token_hash = %s", ("verified", hashed_token,))
+
+        # Генерируем профиль
+        # Аватарка с биографией
+        photo_id = int(account["avatar_id"]) if account.get("avatar_id") else None
+        avatar_url = f"{self.config.avatar_base_url}{photo_id}" if photo_id else None
+        description = account.get("description")
+
+        # Собираем данные пакета
+        payload = {
+            "profile": self.tools.generate_profile_tt(
+                id=account.get("id"),
+                phone=int(account.get("phone")),
+                avatarUrl=avatar_url,
+                photoId=photo_id,
+                updateTime=int(account.get("updatetime")),
+                firstName=account.get("firstname"),
+                lastName=account.get("lastname"),
+                options=json.loads(account.get("options")),
+                description=description,
+                username=account.get("username")
+            ),
+            "tokenAttrs": {
+                "AUTH": {
+                    "token": token
+                }
+            },
+            "tokenTypes": {
+                "AUTH": token
+            }
+        }
+
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_OK, seq=seq, opcode=self.opcodes.AUTH, payload=payload
+        )
+
+        await self._send(writer, packet)
+
+    async def auth_confirm(self, payload, seq, writer, deviceType, deviceName):
+        """Обработчик финальной аутентификации"""
+        # Валидируем данные пакета
+        try:
+            FinalAuthPayloadModel.model_validate(payload)
+        except Exception as e:
+            await self._send_error(seq, self.opcodes.AUTH_CONFIRM, self.error_types.INVALID_PAYLOAD, writer)
+            return
+
+        # Извлекаем данные из пакета
+        token = payload.get("token")
+
+        if not deviceType:
+            deviceType = payload.get("deviceType")
+
+        # Хешируем токен
+        hashed_token = hashlib.sha256(token.encode()).hexdigest()
+
+        # Генерируем постоянный токен
+        login = secrets.token_urlsafe(128)
+        hashed_login = hashlib.sha256(login.encode()).hexdigest()
+
+        # Ищем токен с кодом
+        async with self.db_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                # Ищем токен
+                await cursor.execute("SELECT * FROM auth_tokens WHERE token_hash = %s AND expires > UNIX_TIMESTAMP()",
+                                     (hashed_token,))
+                stored_token = await cursor.fetchone()
+
+                if stored_token is None:
+                    await self._send_error(seq, self.opcodes.AUTH_CONFIRM, self.error_types.INVALID_TOKEN, writer)
+                    return
+
+                # Если авторизация только началась - отдаем ошибку
+                if stored_token.get("state") == "started":
+                    await self._send_error(seq, self.opcodes.AUTH_CONFIRM, self.error_types.INVALID_TOKEN, writer)
+                    return
+
+                # Ищем аккаунт
+                await cursor.execute("SELECT * FROM users WHERE phone = %s", (stored_token.get("phone"),))
+                account = await cursor.fetchone()
+
+                # Удаляем токен
+                await cursor.execute("DELETE FROM auth_tokens WHERE token_hash = %s", (hashed_token,))
+
+                # Создаем сессию
+                await cursor.execute(
+                    "INSERT INTO tokens (phone, token_hash, device_type, device_name, location, time) VALUES (%s, %s, %s, %s, %s, %s)",
+                    (stored_token.get("phone"), hashed_login, deviceType, deviceName, "Epstein Island",
+                     int(time.time()),)
+                )
+
+        # Аватарка с биографией
+        photo_id = None if not account.get("avatar_id") else int(account.get("avatar_id"))
+        avatar_url = None if not photo_id else self.config.avatar_base_url + photo_id
+        description = None if not account.get("description") else account.get("description")
+
+        # Собираем данные пакета
+        payload = {
+            "userToken": "0", # Пока как заглушка
+            "profile": self.tools.generate_profile_tt(
+                id=account.get("id"),
+                phone=int(account.get("phone")),
+                avatarUrl=avatar_url,
+                photoId=photo_id,
+                updateTime=int(account.get("updatetime")),
+                firstName=account.get("firstname"),
+                lastName=account.get("lastname"),
+                options=json.loads(account.get("options")),
+                description=description,
+                username=account.get("username")
+            ),
+            "tokenType": "LOGIN",
+            "token": login
+        }
+
+        # Создаем пакет
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_OK, seq=seq, opcode=self.opcodes.AUTH_CONFIRM, payload=payload
+        )
+
+        # Отправялем
+        await self._send(writer, packet)
+
+    async def login(self, payload, seq, writer):
+        """Обработчик авторизации клиента на сервере"""
+        # Валидируем данные пакета
+        try:
+            LoginPayloadModel.model_validate(payload)
+        except pydantic.ValidationError as error:
+            self.logger.error(f"Возникли ошибки при валидации пакета: {error}")
+            await self._send_error(seq, self.opcodes.LOGIN, self.error_types.INVALID_PAYLOAD, writer)
+            return
+        
+        # Получаем данные из пакета
+        token = payload.get("token")
+
+        # Хешируем токен
+        hashed_token = hashlib.sha256(token.encode()).hexdigest()
+
+        # Ищем токен в бд
+        async with self.db_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute("SELECT * FROM tokens WHERE token_hash = %s", (hashed_token,))
+                token_data = await cursor.fetchone()
+
+                # Если токен не найден, отправляем ошибку
+                if token_data is None:
+                    await self._send_error(seq, self.opcodes.LOGIN, self.error_types.INVALID_TOKEN, writer)
+                    return
+
+                # Ищем аккаунт пользователя в бд
+                await cursor.execute("SELECT * FROM users WHERE phone = %s", (token_data.get("phone"),))
+                user = await cursor.fetchone()
+
+                # Ищем данные пользователя в бд
+                await cursor.execute("SELECT * FROM user_data WHERE phone = %s", (token_data.get("phone"),))
+                user_data = await cursor.fetchone()
+
+        # Аватарка с биографией
+        photo_id = None if not user.get("avatar_id") else int(user.get("avatar_id"))
+        avatar_url = None if not photo_id else self.config.avatar_base_url + photo_id
+        description = None if not user.get("description") else user.get("description")
+
+        # Генерируем профиль
+        profile = self.tools.generate_profile_tt(
+            id=user.get("id"),
+            phone=int(user.get("phone")),
+            avatarUrl=avatar_url,
+            photoId=photo_id,
+            updateTime=int(user.get("updatetime")),
+            firstName=user.get("firstname"),
+            lastName=user.get("lastname"),
+            options=json.loads(user.get("options")),
+            description=description,
+            username=user.get("username")
+        )
+
+        chats = await self.tools.generate_chats(
+            json.loads(user_data.get("chats")),
+            self.db_pool, user.get("id")
+        )
+
+        # Формируем данные пакета
+        payload = {
+            "profile": profile,
+            "chats": chats,
+            "chatMarker": 0,
+            "messages": {},
+            "contacts": [],
+            "presence": {},
+            "config": {
+                "hash": "0",
+                "server": {},
+                "user": json.loads(user_data.get("user_config")),
+                "chatFolders": {
+                    "FOLDERS": [],
+                    "ALL_FILTER_EXCLUDE": []
+                }
+            },
+            "token": token,
+            "calls": [],
+            "videoChatHistory": False,
+            "drafts": {
+            "chats": {
+                "discarded": {},
+                "saved": {}
+            },
+            "users": {
+                "discarded": {},
+                "saved": {}
+            }
+            },
+            "time": int(time.time() * 1000)
+        }
+
+        # Собираем пакет
+        packet = self.proto.pack_packet(
+            cmd=self.proto.CMD_OK, seq=seq, opcode=self.opcodes.LOGIN, payload=payload
+        )
+
+        # Отправляем
+        await self._send(writer, packet)
+        return int(user.get("phone")), int(user.get("id")), hashed_token

src/tamtam/proto.py

@@ -0,0 +1,113 @@
+import lz4.block, msgpack, logging, json
+
+class Proto:
+    def __init__(self) -> None:
+        self.logger = logging.getLogger(__name__)
+
+    # TODO узнать какие должны быть лимиты и поменять,
+    # сейчас это больше заглушка
+    MAX_PAYLOAD_SIZE = 1048576 # 1 MB
+    MAX_DECOMPRESSED_SIZE = 1048576 # 1 MB
+    HEADER_SIZE = 10 # 1+2+1+2+4
+
+    ### Работа с протоколом
+    def unpack_packet(self, data: bytes) -> dict | None:
+        # Проверяем минимальный размер пакета
+        if len(data) < self.HEADER_SIZE:
+            self.logger.warning(f"Пакет слишком маленький: {len(data)} байт")
+            return None
+
+        # Распаковываем заголовок
+        ver = int.from_bytes(data[0:1], "big")
+        cmd = int.from_bytes(data[1:3], "big")
+        seq = int.from_bytes(data[3:4], "big")
+        opcode = int.from_bytes(data[4:6], "big")
+        packed_len = int.from_bytes(data[6:10], "big")
+
+        # Флаг упаковки
+        comp_flag = packed_len >> 24
+
+        # Парсим данные пакета
+        payload_length = packed_len & 0xFFFFFF
+
+        # Проверяем размер payload
+        if payload_length > self.MAX_PAYLOAD_SIZE:
+            self.logger.warning(f"Payload слишком большой: {payload_length} B (лимит {self.MAX_PAYLOAD_SIZE})")
+            return None
+
+        # Проверяем длину пакета
+        if len(data) < self.HEADER_SIZE + payload_length:
+            self.logger.warning(f"Пакет неполный: требуется {self.HEADER_SIZE + payload_length} B, получено {len(data)}")
+            return None
+
+        payload_bytes = data[10 : 10 + payload_length]
+        payload = None
+
+        # Декодируем данные пакета
+        if payload_bytes:
+            # Разжимаем данные пакета, если требуется
+            if comp_flag != 0:
+                compressed_data = payload_bytes
+                try:
+                    payload_bytes = lz4.block.decompress(
+                        compressed_data,
+                        uncompressed_size=self.MAX_DECOMPRESSED_SIZE,
+                    )
+                except lz4.block.LZ4BlockError:
+                    self.logger.warning("Ошибка декомпрессии LZ4")
+                    return None
+
+            # Распаковываем msgpack
+            payload = msgpack.unpackb(payload_bytes, raw=False, strict_map_key=False)
+
+        self.logger.debug(f"Распаковал - ver={ver} cmd={cmd} seq={seq} opcode={opcode} payload={payload}")
+
+        # Возвращаем
+        return {
+            "ver": ver,
+            "cmd": cmd,
+            "seq": seq,
+            "opcode": opcode,
+            "payload": payload,
+        }
+
+    def pack_packet(self, ver: int = 10, cmd: int = 1, seq: int = 1, opcode: int = 6, payload: dict = None) -> bytes:
+        # Запаковываем заголовок
+        ver_b = ver.to_bytes(1, "big")
+        cmd_b = cmd.to_bytes(2, "big")
+        seq_b = seq.to_bytes(1, "big")
+        opcode_b = opcode.to_bytes(2, "big")
+
+        # Запаковываем данные пакета
+        payload_bytes: bytes | None = msgpack.packb(payload)
+        if payload_bytes is None:
+            payload_bytes = b""
+        payload_len = len(payload_bytes) & 0xFFFFFF
+        payload_len_b = payload_len.to_bytes(4, 'big')
+
+        self.logger.debug(f"Упаковал - ver={ver} cmd={cmd} seq={seq} opcode={opcode} payload={payload}")
+
+        # Возвращаем пакет
+        return ver_b + cmd_b + seq_b + opcode_b + payload_len_b + payload_bytes
+    
+    ### Констаты протокола
+    CMD_OK = 0x100
+    CMD_NOF = 0x200
+    CMD_ERR = 0x300
+    PROTO_VER = 10
+
+    HELLO = 6
+    REQUEST_CODE = 17
+    VERIFY_CODE = 18
+    FINAL_AUTH = 23
+    LOGIN = 19
+    PING = 1
+    TELEMETRY = 5
+    GET_ASSETS = 27
+    GET_CALL_HISTORY = 79
+    SEND_MESSAGE = 64
+    GET_FOLDERS = 272
+    GET_SESSIONS = 96
+    LOGOUT = 20
+    SEARCH_CHATS = 48
+    SEARCH_BY_PHONE = 46

src/tamtam/socket.py

@@ -0,0 +1,162 @@
+import asyncio, logging, traceback
+from common.proto_tcp import MobileProto
+from tamtam.processors import Processors
+from common.rate_limiter import RateLimiter
+from common.opcodes import Opcodes
+
+class TTMobileServer:
+    def __init__(self, host="0.0.0.0", port=443, ssl_context=None, db_pool=None, clients={}, send_event=None):
+        self.host = host
+        self.port = port
+        self.ssl_context = ssl_context
+        self.server = None
+        self.logger = logging.getLogger(__name__)
+        self.db_pool = db_pool
+        self.clients = clients
+
+        self.opcodes = Opcodes()
+
+        self.proto = MobileProto()
+        self.processors = Processors(db_pool=db_pool, clients=clients, send_event=send_event)
+
+        # rate limiter
+        self.auth_rate_limiter = RateLimiter(max_attempts=5, window_seconds=60)
+
+        self.read_timeout = 300 # Таймаут чтения из сокета (секунды)
+        self.max_read_size = 65536 # Максимальный размер данных из сокета
+
+    async def handle_client(self, reader, writer):
+        """Функция для обработки подключений"""
+        # IP-адрес клиента
+        address = writer.get_extra_info("peername")
+        self.logger.info(f"Работаю с клиентом {address[0]}:{address[1]}")
+
+        deviceType = None
+        deviceName = None
+
+        userPhone = None
+        userId = None
+        hashedToken = None
+
+        try:
+            while True:
+                # Читаем новые данные из сокета (с таймаутом!)
+                try:
+                    data = await asyncio.wait_for(
+                        reader.read(self.max_read_size),
+                        timeout=self.read_timeout
+                    )
+                except asyncio.TimeoutError:
+                    self.logger.info(f"Таймаут соединения для {address[0]}:{address[1]}")
+                    break
+
+                # Если сокет закрыт - выходим из цикла
+                if not data:
+                    break
+
+                # Проверяем размер данных
+                if len(data) > self.max_read_size:
+                    self.logger.warning(f"Пакет от {address[0]}:{address[1]} превышает лимит ({len(data)} байт)")
+                    break
+
+                # Распаковываем данные
+                packet = self.proto.unpack_packet(data)
+
+                # Если пакет невалидный — пропускаем
+                if packet is None:
+                    self.logger.warning(f"Невалидный пакет от {address[0]}:{address[1]}")
+                    continue
+
+                opcode = packet.get("opcode")
+                seq = packet.get("seq")
+                payload = packet.get("payload")
+
+                match opcode:
+                    case self.opcodes.SESSION_INIT:
+                        deviceType, deviceName = await self.processors.session_init(payload, seq, writer)
+                    case self.opcodes.AUTH_REQUEST:
+                        if not self.auth_rate_limiter.is_allowed(address[0]):
+                            await self.processors._send_error(seq, self.opcodes.AUTH_REQUEST, self.processors.error_types.RATE_LIMITED, writer)
+                        else:
+                            await self.processors.auth_request(payload, seq, writer)
+                    case self.opcodes.AUTH:
+                        if not self.auth_rate_limiter.is_allowed(address[0]):
+                            await self.processors._send_error(seq, self.opcodes.AUTH, self.processors.error_types.RATE_LIMITED, writer)
+                        else:
+                            await self.processors.auth(payload, seq, writer)
+                    case self.opcodes.AUTH_CONFIRM:
+                        if not self.auth_rate_limiter.is_allowed(address[0]):
+                            await self.processors._send_error(seq, self.opcodes.AUTH_CONFIRM, self.processors.error_types.RATE_LIMITED, writer)
+                        else:
+                            await self.processors.auth_confirm(payload, seq, writer, deviceType, deviceName)
+                    case self.opcodes.LOGIN:
+                        if not self.auth_rate_limiter.is_allowed(address[0]):
+                            await self.processors._send_error(seq, self.opcodes.LOGIN, self.processors.error_types.RATE_LIMITED, writer)
+                        else:
+                            userPhone, userId, hashedToken = await self.processors.login(payload, seq, writer)
+
+                            if userPhone:
+                                await self._finish_auth(writer, address, userPhone, userId)
+                    case _:
+                        self.logger.warning(f"Неизвестный опкод {opcode}")
+        except Exception as e:
+            self.logger.error(f"Произошла ошибка при работе с клиентом {address[0]}:{address[1]}: {e}")
+            traceback.print_exc()
+
+        writer.close()
+        self.logger.info(f"Прекратил работать работать с клиентом {address[0]}:{address[1]}")
+
+    async def _finish_auth(self, writer, addr, phone, id):
+        """Завершение открытия сессии"""
+        # Ищем пользователя в словаре
+        user = self.clients.get(id)
+
+        # Добавляем новое подключение в словарь
+        if user:
+            user["clients"].append(
+                {
+                    "writer": writer,
+                    "ip": addr[0],
+                    "port": addr[1],
+                    "protocol": "oneme_mobile"
+                }
+            )
+        else:
+            self.clients[id] = {
+                "phone": phone,
+                "id": id,
+                "clients": [
+                    {
+                        "writer": writer,
+                        "ip": addr[0],
+                        "port": addr[1],
+                        "protocol": "oneme_mobile"
+                    }
+                ]
+            }
+
+    async def _end_session(self, id, ip, port):
+        """Завершение сессии"""
+        # Получаем пользователя в списке
+        user = self.clients.get(id)
+        if not user:
+            return
+
+        # Получаем подключения пользователя
+        clients = user.get("clients", [])
+
+        # Удаляем нужное подключение из словаря
+        for i, client in enumerate(clients):
+            if (client.get("ip"), client.get("port")) == (ip, port):
+                clients.pop(i)
+
+    async def start(self):
+        """Функция для запуска сервера"""
+        self.server = await asyncio.start_server(
+            self.handle_client, self.host, self.port, ssl=self.ssl_context
+        )
+
+        self.logger.info(f"Сокет запущен на порту {self.port}")
+
+        async with self.server:
+            await self.server.serve_forever()