75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""Зависимости FastAPI: аутентификация и авторизация."""
|
|
from fastapi import Depends, Header, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session as DBSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.security import decode_access_token
|
|
from app.db.session import get_db
|
|
from app.models.enums import RoleCode
|
|
from app.models.user import Role, User, UserRole
|
|
from app.models.user import Session as UserSession
|
|
|
|
bearer = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_current_session(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
|
|
db: DBSession = Depends(get_db),
|
|
) -> UserSession:
|
|
"""Текущая сессия по access-токену."""
|
|
if not credentials:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Требуется авторизация")
|
|
try:
|
|
payload = decode_access_token(credentials.credentials)
|
|
except Exception as exc: # noqa: BLE001
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Недействительный токен") from exc
|
|
|
|
session = db.get(UserSession, payload.get("sid"))
|
|
if not session or session.revoked_at:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Сессия завершена")
|
|
return session
|
|
|
|
|
|
def get_current_user(
|
|
session: UserSession = Depends(get_current_session),
|
|
db: DBSession = Depends(get_db),
|
|
) -> User:
|
|
"""Текущий пользователь."""
|
|
user = db.get(User, session.user_id)
|
|
if not user:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Пользователь не найден")
|
|
return user
|
|
|
|
|
|
def get_user_role_codes(db: DBSession, user_id: int) -> set[RoleCode]:
|
|
"""Множество ролей пользователя."""
|
|
rows = db.scalars(select(UserRole).where(UserRole.user_id == user_id)).all()
|
|
codes: set[RoleCode] = set()
|
|
for row in rows:
|
|
role = db.get(Role, row.role_id)
|
|
if role:
|
|
codes.add(role.code)
|
|
return codes
|
|
|
|
|
|
def require_roles(*codes: RoleCode):
|
|
"""Фабрика зависимости: доступ только для указанных ролей."""
|
|
|
|
def checker(
|
|
user: User = Depends(get_current_user),
|
|
db: DBSession = Depends(get_db),
|
|
) -> User:
|
|
if not get_user_role_codes(db, user.id).intersection(set(codes)):
|
|
raise HTTPException(status.HTTP_403_FORBIDDEN, "Недостаточно прав")
|
|
return user
|
|
|
|
return checker
|
|
|
|
|
|
def require_service_token(x_service_token: str | None = Header(default=None)) -> None:
|
|
"""Проверка сервисного токена бота."""
|
|
if not settings.bot_service_token or x_service_token != settings.bot_service_token:
|
|
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Недействительный сервисный токен")
|