From 63fcdb00dacccaeb51a128fc1502644cfc4448c8 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Wed, 29 Oct 2025 17:23:05 +0300 Subject: [PATCH 1/6] Add start settings. --- auth_backend/auth_plugins/telegram.py | 108 ++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index 79a691c5..e336d2f9 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/auth_backend/auth_plugins/telegram.py @@ -185,3 +185,111 @@ 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', 'Неправильные учетные данные') \ No newline at end of file From f1b43b194624f643d17fa760fa86eee764e65450 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Sat, 8 Nov 2025 00:50:56 +0300 Subject: [PATCH 2/6] Rewrite authorization through Telegram. --- auth_backend/auth_plugins/telegram.py | 180 +++++--------------------- 1 file changed, 33 insertions(+), 147 deletions(-) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index e336d2f9..114f06cb 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/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', 'Неправильные учетные данные') \ No newline at end of file From e849d6c51a47ae89d8f1054ff4ef1e442f82b6d2 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Thu, 20 Nov 2025 20:26:59 +0300 Subject: [PATCH 3/6] Edit auth_url and redirect_url methods. --- auth_backend/auth_plugins/telegram.py | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index 114f06cb..cf2497d8 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/auth_backend/auth_plugins/telegram.py @@ -24,7 +24,7 @@ logger = logging.getLogger(__name__) -class TelegramSettings(Settings): # TODO: переписать раздел про ТГ в README.md +class TelegramSettings(Settings): TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth" TELEGRAM_BOT_TOKEN: str @@ -122,16 +122,20 @@ async def _login(cls, user_inp: TGAuthResponseSchema, background_tasks: Backgrou ) @classmethod - async def _redirect_url(cls): # А это вообще нужно, если мы используем виджет с атрибутом redirect_url (а не callback)? - """URL на который происходит редирект после завершения входа на стороне провайдера""" + async def _redirect_url(cls): + """URL на который происходит редирект после завершения входа на стороне провайдера. + + В данном случае не предполагается к использованию, т.к. данный URL вшит в виджет. + """ return OauthMeta.UrlSchema(url=cls.settings.TELEGRAM_REDIRECT_URL) @classmethod - 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)}" - ) + async def _auth_url(cls): + """URL на который происходит редирект из приложения, чтобы авторизоваться на стороне провайдера. + + В данном случае не предполагается, т.к. URL вшит в виджет. Отдается атрибут src виджета. + """ + return OauthMeta.UrlSchema(url='https://telegram.org/js/telegram-widget.js?22') @classmethod async def _check(cls, user_inp): From bf78c3b5a9ad99b85e8362d5a1caf5a582435777 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Sat, 22 Nov 2025 14:56:15 +0300 Subject: [PATCH 4/6] Format code. --- auth_backend/auth_plugins/telegram.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index cf2497d8..409c71bb 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/auth_backend/auth_plugins/telegram.py @@ -2,14 +2,14 @@ import hmac import logging from typing import Any -from urllib.parse import quote, unquote +from urllib.parse import unquote import jwt from event_schema.auth import UserLogin from fastapi import Depends from fastapi.background import BackgroundTasks from fastapi_sqlalchemy import db -from pydantic import BaseModel, Field +from pydantic import BaseModel from auth_backend.auth_method import AuthPluginMeta, OauthMeta, Session from auth_backend.exceptions import AlreadyExists, OauthAuthFailed @@ -26,11 +26,12 @@ class TelegramSettings(Settings): TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth" - TELEGRAM_BOT_TOKEN: str + TELEGRAM_BOT_TOKEN: str = None class TelegramAuth(OauthMeta): """Вход в приложение 'Твой ФФ' через Telegram Login Widget.""" + prefix = '/telegram' tags = ['Telegram'] settings = TelegramSettings() @@ -87,7 +88,7 @@ async def _register( db_session=db.session, session_name=user_inp.session_name, ) - + @classmethod async def _login(cls, user_inp: TGAuthResponseSchema, background_tasks: BackgroundTasks) -> Session: """Вход в пользователя с помощью аккаунта ТГ. @@ -124,19 +125,19 @@ async def _login(cls, user_inp: TGAuthResponseSchema, background_tasks: Backgrou @classmethod async def _redirect_url(cls): """URL на который происходит редирект после завершения входа на стороне провайдера. - + В данном случае не предполагается к использованию, т.к. данный URL вшит в виджет. """ return OauthMeta.UrlSchema(url=cls.settings.TELEGRAM_REDIRECT_URL) - + @classmethod async def _auth_url(cls): """URL на который происходит редирект из приложения, чтобы авторизоваться на стороне провайдера. - + В данном случае не предполагается, т.к. URL вшит в виджет. Отдается атрибут src виджета. """ return OauthMeta.UrlSchema(url='https://telegram.org/js/telegram-widget.js?22') - + @classmethod async def _check(cls, user_inp): """Проверка данных пользователя. @@ -164,7 +165,7 @@ 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.""" From 5b13f6931ca141dd89211b5f674fc3bbdc574268 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Sat, 22 Nov 2025 15:13:41 +0300 Subject: [PATCH 5/6] Fix token key. --- auth_backend/auth_plugins/telegram.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index 409c71bb..a3f41c27 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/auth_backend/auth_plugins/telegram.py @@ -26,7 +26,7 @@ class TelegramSettings(Settings): TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth" - TELEGRAM_BOT_TOKEN: str = None + TELEGRAM_BOT_TOKEN: str = '' # Сделал так для тестов, однако токен нужен для работы авторизации! class TelegramAuth(OauthMeta): From e08fc28e5cbef36963391d0aacf8bee50af4f798 Mon Sep 17 00:00:00 2001 From: Umiacha Date: Thu, 27 Nov 2025 21:12:39 +0300 Subject: [PATCH 6/6] Change default token to None. --- auth_backend/auth_plugins/telegram.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/auth_backend/auth_plugins/telegram.py b/auth_backend/auth_plugins/telegram.py index a3f41c27..ad45be0b 100644 --- a/auth_backend/auth_plugins/telegram.py +++ b/auth_backend/auth_plugins/telegram.py @@ -26,7 +26,7 @@ class TelegramSettings(Settings): TELEGRAM_REDIRECT_URL: str = "https://app.test.profcomff.com/auth" - TELEGRAM_BOT_TOKEN: str = '' # Сделал так для тестов, однако токен нужен для работы авторизации! + TELEGRAM_BOT_TOKEN: str | None = None # Сделал так для тестов, однако токен нужен для работы авторизации! class TelegramAuth(OauthMeta):