Rewrite authorization through Telegram.

Author: Umiacha

auth_backend/auth_plugins/telegram.py

@@ -24,61 +24,55 @@
 logger = logging.getLogger(__name__)
 
 
-class TelegramSettings(Settings):
+class TelegramSettings(Settings):  # TODO: переписать раздел про ТГ в README.md
     TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth"
-    TELEGRAM_BOT_TOKEN: str | None = None
+    TELEGRAM_BOT_TOKEN: str
 
 
 class TelegramAuth(OauthMeta):
+    """Вход в приложение 'Твой ФФ' через Telegram Login Widget."""
     prefix = '/telegram'
     tags = ['Telegram']
     settings = TelegramSettings()
 
-    class OauthResponseSchema(BaseModel):
-        id_token: str | None = Field(default=None, help="Telegram JWT token identifier")
-        id: str | None = None
-        first_name: str | None = None
+    class TGAuthResponseSchema(BaseModel):
+        id: str
+        first_name: str
         last_name: str | None = None
         username: str | None = None
         photo_url: str | None = None
-        auth_date: str | None = None
-        hash: str | None = None
-        scopes: list[Scope] | None = None
-        session_name: str | None = None
+        auth_date: str
+        hash: str
+        scopes: list[Scope] | None = None  # Телеграм не передает это поле. Осталось "исторически".
+        session_name: str | None = None  # Телеграм не передает это поле. Осталось "исторически".
 
     @classmethod
     async def _register(
         cls,
-        user_inp: OauthResponseSchema,
+        user_inp: TGAuthResponseSchema,
         background_tasks: BackgroundTasks,
         user_session: UserSession = Depends(UnionAuth(auto_error=True, scopes=[], allow_none=True)),
     ) -> Session:
+        """Добавление метода аутентификации через (виджет) Телеграма."""
         old_user = None
         new_user = {}
-        telegram_user_id = None
-        userinfo = None
 
-        if user_inp.id_token is None:
-            userinfo = await cls._check(user_inp)
-            telegram_user_id = user_inp.id
-            logger.debug(userinfo)
-        else:
-            userinfo = jwt.decode(user_inp.id_token, cls.settings.ENCRYPTION_KEY, algorithms=["HS256"])
-            telegram_user_id = userinfo['id']
-            logger.debug(userinfo)
-
-        user = await cls._get_user('user_id', telegram_user_id, db_session=db.session)
+        # Проверяем получение корректных данных
+        userinfo = await cls._check(user_inp)
+        tg_user_id = userinfo['id']
 
+        user = await cls._get_user('user_id', tg_user_id, db_session=db.session)
         if user is not None:
             raise AlreadyExists(User, user.id)
+
         if user_session is None:
-            user = await cls._create_user(db_session=db.session) if user_session is None else user_session.user
+            user = await cls._create_user(db_session=db.session)
         else:
             user = user_session.user
             old_user = {'user_id': user.id}
         new_user["user_id"] = user.id
-        tg_id = cls.create_auth_method_param('user_id', telegram_user_id, user.id, db_session=db.session)
-        new_user[cls.get_name()] = {"user_id": tg_id.value}
+        tg_auth = cls.create_auth_method_param('user_id', tg_user_id, user.id, db_session=db.session)
+        new_user[cls.get_name()] = {"user_id": tg_auth.value}
         userdata = await TelegramAuth._convert_data_to_userdata_format(userinfo)
         background_tasks.add_task(
             get_kafka_producer().produce,
@@ -93,12 +87,12 @@ async def _register(
             db_session=db.session,
             session_name=user_inp.session_name,
         )
-
+    
     @classmethod
-    async def _login(cls, user_inp: OauthResponseSchema, background_tasks: BackgroundTasks) -> Session:
-        """Вход в пользователя с помощью аккаунта https://lk.msu.ru
+    async def _login(cls, user_inp: TGAuthResponseSchema, background_tasks: BackgroundTasks) -> Session:
+        """Вход в пользователя с помощью аккаунта ТГ.
 
-        Производит вход, если находит пользователя по уникаотному идендификатору. Если аккаунт не
+        Производит вход, если находит пользователя по id (из Телеграма). Если аккаунт не
         найден, возвращает ошибка.
         """
 
@@ -128,24 +122,23 @@ async def _login(cls, user_inp: OauthResponseSchema, background_tasks: Backgroun
         )
 
     @classmethod
-    async def _redirect_url(cls):
+    async def _redirect_url(cls):  # А это вообще нужно, если мы используем виджет с атрибутом redirect_url (а не callback)?
         """URL на который происходит редирект после завершения входа на стороне провайдера"""
         return OauthMeta.UrlSchema(url=cls.settings.TELEGRAM_REDIRECT_URL)
-
+    
     @classmethod
-    async def _auth_url(cls):
-        """URL на который происходит редирект из приложения для авторизации на стороне провайдера"""
-
+    async def _auth_url(cls):  # А это вообще нужно, если в виджете ТГ уже прописан атрибут src и там скрипт?!
+        """URL на который происходит редирект из приложения, чтобы авторизоваться на стороне провайдера."""
         return OauthMeta.UrlSchema(
             url=f"https://oauth.telegram.org/auth?bot_id={cls.settings.TELEGRAM_BOT_TOKEN.split(':')[0]}&origin={quote(cls.settings.TELEGRAM_REDIRECT_URL)}&return_to={quote(cls.settings.TELEGRAM_REDIRECT_URL)}"
         )
-
+    
     @classmethod
     async def _check(cls, user_inp):
-        '''Проверка данных пользователя
+        """Проверка данных пользователя.
 
         https://core.telegram.org/widgets/login#checking-authorization
-        '''
+        """
         data_check = {
             'id': user_inp.id,
             'first_name': user_inp.first_name,
@@ -167,9 +160,10 @@ async def _check(cls, user_inp):
             return data_check
         else:
             raise OauthAuthFailed('Invalid user data from Telegram', 'Неправильные учетные данные')
-
+    
     @classmethod
     async def _convert_data_to_userdata_format(cls, data: dict[str, Any]) -> UserLogin:
+        """Конвертация данных в формат для userdata-api."""
         first_name, last_name = '', ''
         if 'first_name' in data.keys() and data['first_name'] is not None:
             first_name = data['first_name']
@@ -185,111 +179,3 @@ async def _convert_data_to_userdata_format(cls, data: dict[str, Any]) -> UserLog
         ]
         result = {"items": items, "source": cls.get_name()}
         return cls.userdata_process_empty_strings(UserLogin.model_validate(result))
-
-
-class NewTelegramSettings(Settings):
-    TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth"
-    TELEGRAM_BOT_TOKEN: str | None = None  # TODO: добавить сюда токен бота для тестов. В целом, брать его из .env.
-
-
-class NewTelegramAuth(OauthMeta):
-    """Вход в приложение 'Твой ФФ' через Telegram Login Widget."""
-    prefix = '/telegram'
-    tags = ['Telegram']
-    settings = NewTelegramSettings()
-
-    class TGAuthResponseSchema(BaseModel):
-        # id_token: str | None = Field(default=None, help="Telegram JWT token identifier")
-        id: str
-        first_name: str
-        last_name: str | None = None
-        username: str | None = None
-        photo_url: str | None = None
-        auth_date: str
-        hash: str
-        # scopes: list[Scope] | None = None
-        # session_name: str | None = None
-
-    @classmethod
-    async def _register(
-        cls,
-        user_inp: TGAuthResponseSchema,
-        background_tasks: BackgroundTasks,
-        user_session: UserSession = Depends(UnionAuth(auto_error=True, scopes=[], allow_none=True)),
-    ) -> Session:
-        old_user = None
-        new_user = {}
-
-        # Проверяем получение корректных данных
-        userinfo = await cls._check(user_inp)
-        tg_user_id = userinfo['id']
-
-        # if user_inp.id_token is None:
-        #     userinfo = await cls._check(user_inp)
-        #     telegram_user_id = user_inp.id
-        #     logger.debug(userinfo)
-        # else:
-        #     userinfo = jwt.decode(user_inp.id_token, cls.settings.ENCRYPTION_KEY, algorithms=["HS256"])
-        #     telegram_user_id = userinfo['id']
-        #     logger.debug(userinfo)
-
-        # Проверяем, что этот тг-аккаунт уже не привязан к существующему пользователю!  # TODO: уточнить, что это правда так работает...
-        user = await cls._get_user('user_id', tg_user_id, db_session=db.session)
-        if user is not None:
-            raise AlreadyExists(User, user.id)
-
-        if user_session is None:
-            user = await cls._create_user(db_session=db.session)  # if user_session is None else user_session.user
-        else:
-            user = user_session.user
-            old_user = {'user_id': user.id}
-        new_user["user_id"] = user.id
-        tg_auth = cls.create_auth_method_param('user_id', tg_user_id, user.id, db_session=db.session)
-        new_user[cls.get_name()] = {"user_id": tg_auth.value}
-        userdata = await TelegramAuth._convert_data_to_userdata_format(userinfo)
-        background_tasks.add_task(
-            get_kafka_producer().produce,
-            cls.settings.KAFKA_USER_LOGIN_TOPIC_NAME,
-            TelegramAuth.generate_kafka_key(user.id),
-            userdata,
-        )
-        await AuthPluginMeta.user_updated(new_user, old_user)
-        return await cls._create_session(
-            user,  # TODO: спросить, зачем (в google.py) отправляется новая сессия с user. user обновляется к этому моменту? Даже если меняется, то зачем именно здесь пользователю новая сессия?
-            user_inp.scopes,
-            db_session=db.session,
-            session_name=user_inp.session_name,
-        )
-
-    @classmethod
-    async def _redirect_url(cls):
-        """URL на который происходит редирект после завершения входа на стороне провайдера"""
-        return OauthMeta.UrlSchema(url=cls.settings.TELEGRAM_REDIRECT_URL)
-    
-    @classmethod
-    async def _check(cls, user_inp):  # TODO: перепроверить, что написанное реально поддерживает проверку по доке тг!
-        '''Проверка данных пользователя
-
-        https://core.telegram.org/widgets/login#checking-authorization
-        '''
-        data_check = {
-            'id': user_inp.id,
-            'first_name': user_inp.first_name,
-            'last_name': user_inp.last_name,
-            'username': user_inp.username,
-            'photo_url': user_inp.photo_url,
-            'auth_date': user_inp.auth_date,
-        }
-        check_hash = user_inp.hash
-        data_check_string = ''
-        for k, v in sorted(data_check.items()):
-            if v is None:
-                continue
-            data_check_string += f'{unquote(k)}={unquote(v)}\n'
-        data_check_string = data_check_string.rstrip('\n')
-        secret_key = hashlib.sha256(str.encode(cls.settings.TELEGRAM_BOT_TOKEN)).digest()
-        signing = hmac.new(secret_key, msg=str.encode(data_check_string), digestmod=hashlib.sha256).hexdigest()
-        if signing == check_hash:
-            return data_check
-        else:
-            raise OauthAuthFailed('Invalid user data from Telegram', 'Неправильные учетные данные')