62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
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
|