✨ Use telegram bot direct login

src/telegram/plugins/start.py

@@ -1,13 +1,11 @@
-from httpx import URL
 from pyrogram import filters, Client
-from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton
+from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton, LoginUrl
 
 from src.bot import bot
 from src.config import settings
 from src.telegram.enums import Message
 from src.telegram.message import NO_ACCOUNT_MSG, LOGIN_MSG, LOGIN_BUTTON
 from src.users.crud import get_user_crud
-from src.utils.telegram import encode_telegram_auth_data
 
 
 async def login(message: Message):
@@ -17,15 +15,12 @@ async def login(message: Message):
     if not user:
         await message.reply(NO_ACCOUNT_MSG % uid, quote=True)
         return
-    token = await encode_telegram_auth_data(uid)
-    url = settings.PROJECT_URL + "/api/users/auth"
-    url = URL(url).copy_add_param("jwt", token)
-    url = str(url)
+    url = settings.PROJECT_URL + "/api/users/callback"
     await message.reply(
         LOGIN_MSG,
         quote=True,
         reply_markup=InlineKeyboardMarkup(
-            [[InlineKeyboardButton(LOGIN_BUTTON, url=url)]]
+            [[InlineKeyboardButton(LOGIN_BUTTON, login_url=LoginUrl(url=url))]]
         ),
     )
 

src/users/endpoints.py

@@ -1,7 +1,6 @@
 from http import HTTPStatus
 
 from fastapi import APIRouter, Depends, HTTPException
-from jose import JWTError
 from starlette.requests import Request
 
 from .crud import SQLAlchemyCRUD
@@ -11,7 +10,7 @@ from ..html import templates
 from ..storage.sqlalchemy import SQLAlchemyStorage, get_sqlalchemy_storage
 from ..utils.oauth import back_auth_request
 from ..utils.redirect import RedirectResponseBuilder
-from ..utils.telegram import decode_telegram_auth_data, verify_telegram_auth_data
+from ..utils.telegram import verify_telegram_auth_data
 
 router = APIRouter()
 
@@ -65,15 +64,3 @@ async def user_login(
 ):
     tg_id = await verify_telegram_auth_data(request.query_params)
     return await auth(tg_id, request, storage)
-
-
-@router.get("/auth", name="users:auth")
-async def user_auth(
-        request: Request,
-        storage: SQLAlchemyStorage = Depends(get_sqlalchemy_storage),
-):
-    try:
-        tg_id = await decode_telegram_auth_data(request.query_params)
-    except JWTError:
-        tg_id = None
-    return await auth(tg_id, request, storage)

src/utils/telegram.py

@@ -1,12 +1,10 @@
 import hashlib
 import hmac
-from datetime import datetime, timezone
 from typing import Optional
 
 from starlette.datastructures import QueryParams
 
 from src.config import settings
-from src.users.crypto import encode_jwt, decode_jwt
 
 
 async def verify_telegram_auth_data(params: QueryParams) -> Optional[int]:
@@ -24,22 +22,3 @@ async def verify_telegram_auth_data(params: QueryParams) -> Optional[int]:
     hmac_hash = hmac.new(secret_key, str.encode(check_string), hashlib.sha256).hexdigest()
 
     return int(params.get("id")) if hmac_hash == hash_str else None
-
-
-async def encode_telegram_auth_data(uid: int) -> str:
-    jwt = encode_jwt(settings.ACCESS_TOKEN_EXP, str(uid))
-    return jwt
-
-
-async def decode_telegram_auth_data(params: QueryParams) -> Optional[int]:
-    jwt = params.get("jwt")
-    if not jwt:
-        return None
-    if not jwt:
-        return None
-    data = decode_jwt(jwt)
-    now = datetime.now(timezone.utc)
-    uid, exp = data["sub"], data["exp"]
-    if exp < now.timestamp():
-        return None
-    return int(uid)