"""Авторизация админ-панели: проверка роли is_admin и секретного кода сессии.""" import secrets from datetime import datetime, timedelta, timezone from fastapi import Depends, HTTPException, Path, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import get_current_user from app.database import get_db from app.models.admin import AdminSession from app.models.user import User # Срок жизни секретного кода доступа к админке ADMIN_SESSION_TTL_HOURS = 8 async def get_admin_user(current_user: User = Depends(get_current_user)) -> User: """Dependency: пользователь должен быть аутентифицирован И иметь is_admin.""" if not getattr(current_user, "is_admin", False): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Доступ только для администраторов", ) return current_user async def create_admin_session(user: User, db: AsyncSession) -> str: """Создать новый секретный код доступа к админке, вернуть код.""" code = secrets.token_urlsafe(32) # Колонки TIMESTAMP WITHOUT TIME ZONE — храним naive UTC session = AdminSession( user_id=user.id, code=code, expires_at=datetime.utcnow() + timedelta(hours=ADMIN_SESSION_TTL_HOURS), ) db.add(session) await db.commit() return code async def verify_admin_code( code: str = Path(..., description="Секретный код сессии из ссылки /admin/"), admin: User = Depends(get_admin_user), db: AsyncSession = Depends(get_db), ) -> User: """Dependency: проверяет И роль администратора, И валидность кода из пути. Двухфакторная защита: ссылка /admin/ бесполезна без JWT админа, а JWT админа без верного кода — тоже. """ result = await db.execute(select(AdminSession).where(AdminSession.code == code)) session = result.scalar_one_or_none() if session is None or session.user_id != admin.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Неверный код доступа к админке", ) expires = session.expires_at if expires.tzinfo is not None: expires = expires.replace(tzinfo=None) if expires < datetime.utcnow(): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Код доступа истёк, сгенерируйте новый", ) session.last_used_at = datetime.utcnow() await db.commit() return admin