Fix/magic link bot auth

requirements.txt

@@ -9,6 +9,7 @@ annotated-doc==0.0.4
 annotated-types==0.7.0
 anyio==4.12.0
 APScheduler==3.11.2
+argcomplete==3.6.3
 argon2-cffi==25.1.0
 argon2-cffi-bindings==25.1.0
 async-lru==2.1.0
@@ -22,9 +23,11 @@ cffi==2.0.0
 charset-normalizer==3.4.4
 click==8.3.1
 colorama==0.4.6
+commitizen==4.13.9
 coverage==7.13.4
 cryptography==46.0.5
 cyclonedx-python-lib==11.6.0
+decli==0.6.3
 defusedxml==0.7.1
 Deprecated==1.3.1
 ecdsa==0.19.1
@@ -38,6 +41,7 @@ httpx==0.28.1
 idna==3.11
 iniconfig==2.3.0
 itsdangerous==2.2.0
+Jinja2==3.1.6
 librt==0.7.8
 license-expression==30.4.4
 limits==5.6.0
@@ -61,6 +65,7 @@ pip-requirements-parser==32.0.1
 pip_audit==2.10.0
 platformdirs==4.9.4
 pluggy==1.6.0
+prompt_toolkit==3.0.51
 propcache==0.4.1
 py-serializable==2.1.0
 pyasn1==0.6.2
@@ -81,6 +86,7 @@ python-multipart==0.0.22
 python-socks==2.8.0
 pytz==2025.2
 PyYAML==6.0.3
+questionary==2.1.1
 requests==2.32.5
 rich==14.3.3
 rsa==4.9.1
@@ -91,14 +97,17 @@ sortedcontainers==2.4.0
 SQLAlchemy==2.0.45
 starlette==0.50.0
 stevedore==5.7.0
+termcolor==3.3.0
 tomli==2.4.0
 tomli_w==1.2.0
+tomlkit==0.14.0
 tqdm==4.67.3
 typing-inspection==0.4.2
 typing_extensions==4.15.0
 tzdata==2025.3
 tzlocal==5.3.1
 urllib3==2.6.3
 uvicorn==0.40.0
+wcwidth==0.6.0
 wrapt==2.0.1
 yarl==1.22.0

src/config.py

@@ -4,6 +4,7 @@
 load_dotenv()
 
 TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
+BOT_SECRET = os.getenv("BOT_SECRET")
 
 DATABASE_URL = os.getenv("DATABASE_URL")
 

src/domain/services/tg_client.py

@@ -3,6 +3,7 @@
 
 from src.logger import logger
 from src.shared.schemas.bot.tg_link import TgLinkStatus
+from src.config import BOT_SECRET
 
 @alru_cache(maxsize=512, ttl=300) # кэширование, хранить максимум 512 результатов, удалсять через 300 секунд
 async def check_registration(user_id: str) -> TgLinkStatus:
@@ -52,8 +53,12 @@ async def generate_magic_token(tg_id: int) -> str | None:
     """Отправляет запрос к api на создание magic токена"""
     try:
         async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, read=10.0)) as client:
-            logger.info(f"Отвправка запроса на генерацию magic токна для пользователя: {tg_id}...")
-            response = await client.get(f"http://api:8000/api/v1/auth/telegram/generate-link/{tg_id}")
+            logger.info(f"Отправка запроса на генерацию magic токна для пользователя: {tg_id}...")
+
+            headers = {
+                "X-Bot-Secret": BOT_SECRET
+            }
+            response = await client.get(f"http://api:8000/api/v1/auth/telegram/generate-link/{tg_id}", headers=headers)
 
             if response.status_code == 200:
                 data = response.json()

src/domain/services/tg_link_service.py

@@ -1,11 +1,32 @@
+import hmac
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from datetime import datetime, timezone
 
 from src.infrastructure.db.models import User,  MagicToken
+from src.config import BOT_SECRET
 from src.logger import logger
 
 class TgLinkService:
+    @staticmethod
+    def verify_bot_secret_key(bot_secret_header: str):
+        """
+        Сравниваем секретный ключ с полученным ключом в заголовке.
+        Устранена возможность timing-атаки.
+        """
+        if not BOT_SECRET:
+            raise RuntimeError("Не задан секретный ключ X-Bot-Secret")
+
+        result =  hmac.compare_digest(
+            bot_secret_header.encode(),
+            BOT_SECRET.encode()
+        )
+        if not result:
+            logger.warning("Неверный ключ в X-Bot-Secret. Возможна попытка атаки!")
+
+        return result
+
     @staticmethod
     async def check_telegram_connection(tg_id: int, db: AsyncSession) -> User | None:
         """

src/presentation/api/v1/auth/tg_link.py

@@ -1,13 +1,14 @@
-from datetime import datetime, timezone, timedelta
 import secrets
+
+from datetime import datetime, timezone, timedelta
 from fastapi import Request, Depends, APIRouter
 from slowapi import Limiter
 from slowapi.util import get_remote_address
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.infrastructure.db.database import get_db
 from src.domain.services.tg_link_service import TgLinkService
-from src.presentation.api.v1.exceptions import NotCorrect
+from src.presentation.api.v1.exceptions import NotCorrect, NoAccess
 from src.logger import logger
 
 router = APIRouter(prefix="/auth", tags=["Магические ссылки"])
@@ -26,17 +27,35 @@ async def create_telegram_magic_link(
         db: AsyncSession
     -> dict
     """
-    logger.debug(f"Получен tg_id: {telegram_id}")
+    logger.debug(f"Получен запрос на привязку аккаунта к боту tg_id: {telegram_id}")
+
+    logger.info("Проверка заголовка X-Bot-Secret...")
+    bot_secret_header = request.headers.get("X-Bot-Secret")
 
+    if not bot_secret_header:
+        logger.info("В заголовке отсутствует секретный ключ")
+        raise NoAccess("Недостаточно прав для доступа к этому ресурсу")
+
+    logger.info("Сравнение ключей...")
+    verify_key = TgLinkService.verify_bot_secret_key(bot_secret_header)
+
+    if not verify_key:
+        logger.info("Предоставленный ключ не совпадает с секретным ключом")
+        raise NoAccess("Недостаточно прав для доступа к этому ресурсу")
+
+    logger.info("Проверка наличия привязки аккаунта к платформе...")
     if await TgLinkService.check_telegram_connection(telegram_id, db):
         logger.warning(f"Этот аккаунт телеграм уже привязан к платформе: {telegram_id}")
-        raise NotCorrect("Уже привязан")
+        raise NotCorrect("Этот аккаунт телеграм уже привязан к платформе")
 
     token = secrets.token_urlsafe(32)
     expires = datetime.now(timezone.utc) + timedelta(minutes=10)
 
+    #! TODO: Добавить изоляцию на уровне docker
+
     logger.info("Сохранение magic токена...")
     await TgLinkService.save_link_token(token=token, expires_at=expires, db=db, telegram_id=telegram_id)
+    logger.info("Токен успешно сохранен!")
 
     return {"token": token}