"""FastAPI-зависимости: аутентификация по API-ключу.""" from __future__ import annotations import hashlib from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.models import User security_scheme = HTTPBearer(auto_error=False) def hash_api_key(api_key: str) -> str: return hashlib.sha256(api_key.encode()).hexdigest() async def get_current_user_from_header( credentials: HTTPAuthorizationCredentials | None = Depends(security_scheme), db: AsyncSession = Depends(get_db), ) -> User: """Аутентификация по заголовку Authorization: Bearer .""" if credentials is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing Authorization header. Use: Authorization: Bearer ", ) api_key = credentials.credentials if not api_key.startswith("md2vk_"): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key format", ) api_key_hash = hash_api_key(api_key) result = await db.execute( select(User).where(User.api_key_hash == api_key_hash, User.is_active == True) ) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key") return user async def get_user_by_api_key(api_key: str, db: AsyncSession) -> User: """Проверяет API-ключ из тела запроса. Используется в POST-эндпоинтах.""" if not api_key.startswith("md2vk_"): raise HTTPException(status_code=401, detail="Invalid API key format") api_key_hash = hash_api_key(api_key) result = await db.execute( select(User).where(User.api_key_hash == api_key_hash, User.is_active == True) ) user = result.scalar_one_or_none() if user is None: raise HTTPException(status_code=401, detail="Invalid API key") return user