1) Рефакторинг.

2) Авторизация через телеграм.

Author: ProFastCode

.env.dist

@@ -10,4 +10,5 @@ POSTGRES_USER=
 POSTGRES_PASSWORD=
 POSTGRES_DATABASE=
 
-TELEGRAM_BOT_TOKEN=
+TELEGRAM_BOT_TOKEN=
+TELEGRAM_BOT_USERNAME=

app/__init__.py

@@ -11,7 +11,9 @@
     title=settings.APP_TITLE,
     version=settings.APP_VERSION,
     description=settings.APP_DESCRIPTION,
+    docs_url=f'{settings.APP_API_PREFIX}/docs',
     openapi_url=f'{settings.APP_API_PREFIX}/openapi.json',
+    swagger_ui_oauth2_redirect_url=f'{settings.APP_API_PREFIX}/docs/oauth2-redirect',
 )
 
 app.include_router(api.api_router, prefix=settings.APP_API_PREFIX)

app/api/deps.py

@@ -8,24 +8,30 @@
 from app import models
 from app.core import exps, settings
 from app.core.db import Database, SessionLocal
-from app.core.security import JWTTokenManager
+from app.core.security import JWTManager, TelegramAuth
 
 
 async def get_db() -> Database:
     async with SessionLocal() as session:
         yield Database(session)
 
 
-async def get_tkn_manager() -> JWTTokenManager:
-    return JWTTokenManager(settings.APP_SECRET_KEY)
+async def get_jwt_manager() -> JWTManager:
+    return JWTManager(settings.APP_SECRET_KEY)
+
+
+async def get_telegram_auth() -> TelegramAuth:
+    return TelegramAuth(
+        settings.TELEGRAM_BOT_TOKEN, settings.TELEGRAM_BOT_USERNAME
+    )
 
 
 async def get_current_user(
-    access_token: Annotated[str, Header()],
-    db: Annotated[Database, Depends(get_db)],
-    tkn_manager: Annotated[JWTTokenManager, Depends(get_tkn_manager)],
+        access_token: Annotated[str, Header()],
+        db: Annotated[Database, Depends(get_db)],
+        jwt_manager: Annotated[JWTManager, Depends(get_jwt_manager)],
 ) -> models.User:
-    payload = tkn_manager.decode_token(access_token)
+    payload = jwt_manager.decode_token(access_token)
     if payload.get('type') != 'access':
         raise exps.TOKEN_INVALID
     if not (user := await db.user.read(payload.get('id'))):

app/api/endpoints/tokens/auth.py

@@ -1,50 +1,41 @@
 """
 User Endpoints Module
 """
-import hmac
 
 from fastapi import APIRouter, Depends
 from typing_extensions import Annotated
 
 from app import models
 from app.api import deps
-from app.core import exps, settings
+from app.core import exps
 from app.core.db import Database
-from app.core.security import JWTTokenManager
+from app.core.security import JWTManager, TelegramAuth
 
-router = APIRouter()
+router = APIRouter(prefix='/auth')
 
 
-@router.post('/auth/', response_model=models.AuthToken)
-async def auth(
+@router.post('/telegram/', response_model=models.AuthToken)
+async def telegram(
         user: models.UserCreate,
         db: Annotated[Database, Depends(deps.get_db)],
-        tkn_manager: Annotated[JWTTokenManager, Depends(deps.get_tkn_manager)],
+        auth: Annotated[TelegramAuth, Depends(deps.get_telegram_auth)],
+        jwt_manager: Annotated[JWTManager, Depends(deps.get_jwt_manager)],
 ):
     """
     Get auth token
     """
-    data_check_string = '\n'.join(
-        sorted(
-            f'{x}={y}'
-            for x, y in user.model_dump().items()
-            if x not in 'hash' and y is not None
-        )
-    )
-    computed_hash = hmac.new(
-        settings.telegram_bot_token_hash.digest(),
-        data_check_string.encode(),
-        'sha256',
-    ).hexdigest()
-    is_correct = hmac.compare_digest(computed_hash, user.hash)
-    if not is_correct:
+    user_dict = user.model_dump()
+    computed_hash = auth.generate_hash(user_dict)
+    if not auth.is_correct(computed_hash, user.hash):
         raise exps.USER_IS_CORRECT
 
     model_user = models.User(**user.model_dump())
-    if cur_user := await db.user.retrieve(user.id):
+    if cur_user := await db.user.retrieve_one(ident=user.id):
         await db.user.update(cur_user.id, **model_user.model_dump())
     else:
         await db.user.create(model_user)
     await db.session.commit()
-    auth_token = tkn_manager.encode_token({'id': user.id, 'type': 'auth'}, 15)
-    return models.AuthToken(auth_token=auth_token)
+    token = models.AuthToken(token=jwt_manager.encode_token(
+        {'id': user.id, 'type': 'auth'},15
+    ))
+    return token

app/api/endpoints/tokens/pair.py

@@ -4,27 +4,26 @@
 from app import models
 from app.api import deps
 from app.core import exps
-from app.core.security import JWTTokenManager
+from app.core.security import JWTManager
 
 router = APIRouter()
 
 
 @router.post('/pair/', response_model=models.PairTokens)
 async def new_pair_tokens(
     data: models.AuthToken,
-    tkn_manager: Annotated[JWTTokenManager, Depends(deps.get_tkn_manager)],
+    jwt_manager: Annotated[JWTManager, Depends(deps.get_jwt_manager)],
 ):
     """
     Get pair tokens
     """
 
-    payload = tkn_manager.decode_token(data.auth_token)
+    payload = jwt_manager.decode_token(data.auth_token)
     if payload.get('type') != 'auth':
         raise exps.TOKEN_INVALID
     payload['type'] = 'access'
-    access_token = tkn_manager.encode_token(payload, 120)
+    access_token = models.AccessToken(token=jwt_manager.encode_token(payload, 120))
     payload['type'] = 'refresh'
-    refresh_token = tkn_manager.encode_token(payload, 1440)
-    return models.PairTokens(
-        access_token=access_token, refresh_token=refresh_token
-    )
+    refresh_token = models.RefreshToken(token=jwt_manager.encode_token(payload, 1440))
+    tokens = models.PairTokens(access=access_token, refresh=refresh_token)
+    return tokens

app/api/endpoints/tokens/refresh.py

@@ -5,7 +5,7 @@
 from app.api import deps
 from app.core import exps
 from app.core.db import Database
-from app.core.security import JWTTokenManager
+from app.core.security import JWTManager
 
 router = APIRouter()
 
@@ -14,15 +14,14 @@
 async def refresh_access_token(
     data: models.RefreshToken,
     db: Annotated[Database, Depends(deps.get_db)],
-    tkn_manager: Annotated[JWTTokenManager, Depends(deps.get_tkn_manager)],
+    jwt_manager: Annotated[JWTManager, Depends(deps.get_jwt_manager)],
 ):
     """
     Get new access token
     """
-    payload = tkn_manager.decode_token(data.refresh_token)
+    payload = jwt_manager.decode_token(data.refresh_token)
     if payload.get('type') != 'refresh':
         raise exps.TOKEN_INVALID
     if not await db.user.read(payload.get('id')):
         raise exps.USER_NOT_FOUND
-    access_token = tkn_manager.encode_token(payload, 120)
-    return models.AccessToken(access_token=access_token)
+    return models.AccessToken(token=jwt_manager.encode_token(payload, 120))

app/core/security/__init__.py

@@ -1,3 +1,4 @@
-from .token_manager import JWTTokenManager
+from .auth import TelegramAuth
+from .jwt_manager import JWTManager
 
-__all__ = ['JWTTokenManager']
+__all__ = ['JWTManager', 'TelegramAuth']

app/core/security/auth/__init__.py

@@ -0,0 +1,3 @@
+from .telegram import TelegramAuth
+
+__all__ = ['TelegramAuth']

app/core/security/auth/telegram.py

@@ -0,0 +1,32 @@
+import hashlib
+import hmac
+
+from app.core import settings
+
+
+class TelegramAuth:
+    def __init__(
+        self,
+        token: str = settings.TELEGRAM_BOT_TOKEN,
+        username: str = settings.TELEGRAM_BOT_USERNAME,
+    ):
+        self.token = token
+        self.username = username
+
+    @property
+    def hash(self) -> bytes:
+        return hashlib.sha256(self.token.encode()).digest()
+
+    def generate_hash(self, user: dict) -> str:
+        user.pop('hash')
+        string = '\n'.join(f'{x}={y}' for x, y in sorted(user.items()))
+        computed_hash = hmac.new(
+            self.hash,
+            string.encode(),
+            hashlib.sha256,
+        ).hexdigest()
+        return computed_hash
+
+    @classmethod
+    def is_correct(cls, computed_hash: str, user_hash: str) -> bool:
+        return hmac.compare_digest(computed_hash, user_hash)

app/core/security/jwt_manager/__init__.py

@@ -0,0 +1,3 @@
+from .jwt_manager import JWTManager
+
+__all__ = ['JWTManager']