import hashlib import hmac import uuid from typing import Annotated from fastapi import Depends, Header, HTTPException, Request, Response from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.db import get_db from app.models import WebSession COOKIE = "shop_session" SessionDep = Annotated[AsyncSession, Depends(get_db, scope="function")] def bot_secret_ok(got: str) -> bool: return hmac.compare_digest( hashlib.sha256(got.encode()).digest(), hashlib.sha256(settings.bot_secret.encode()).digest(), ) async def require_bot(x_bot_secret: Annotated[str, Header()]) -> None: if not bot_secret_ok(x_bot_secret): raise HTTPException(status_code=401, detail="Неверный ключ бота") def web_owner(session_id: uuid.UUID) -> str: return f"web:{session_id}" def telegram_owner(telegram_user_id: int) -> str: return f"tg:{telegram_user_id}" async def ensure_web_session(request: Request, response: Response, db: AsyncSession) -> uuid.UUID: raw = request.cookies.get(COOKIE) if raw: try: session_id = uuid.UUID(raw) except ValueError: session_id = None else: found = await db.get(WebSession, session_id) if found is not None: return found.id session_id = uuid.uuid4() db.add(WebSession(id=session_id)) await db.flush() response.set_cookie( COOKIE, str(session_id), httponly=True, samesite="lax", secure=settings.cookie_secure, path="/", max_age=60 * 60 * 24 * 30, ) return session_id