backend/app/core/security.py - master - DaniilSk/DesignLab

backend/app/core/security.py
58 строк2 KB
Даниил
feat(admin-auth): add protected admin api
13 июл 2026, 23:41
13 июл 2026, 23:41406f05e
from datetime import datetime, timedelta, timezone
from uuid import UUID

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import get_settings
from app.db.session import get_session
from app.repositories.users import UserRepository

bearer = HTTPBearer(auto_error=False)


def create_access_token(user_id: UUID) -> str:
    settings = get_settings()
    expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    return jwt.encode({"sub": str(user_id), "exp": expires_at}, settings.auth_secret_key, algorithm="HS256")


def create_admin_token(username: str) -> str:
    settings = get_settings()
    expires_at = datetime.now(timezone.utc) + timedelta(minutes=60)
    return jwt.encode({"sub": username, "scope": "admin", "exp": expires_at}, settings.auth_secret_key, algorithm="HS256")


def decode_admin_token(token: str) -> str:
    payload = jwt.decode(token, get_settings().auth_secret_key, algorithms=["HS256"])
    if payload.get("scope") != "admin":
        raise ValueError("Invalid scope")
    return str(payload["sub"])


async def get_current_admin(credentials: HTTPAuthorizationCredentials | None = Depends(bearer)) -> str:
    if credentials is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    try:
        return decode_admin_token(credentials.credentials)
    except Exception as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin token") from exc


async def get_current_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
    session: AsyncSession = Depends(get_session),
):
    if credentials is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    try:
        payload = jwt.decode(credentials.credentials, get_settings().auth_secret_key, algorithms=["HS256"])
        user_id = UUID(payload["sub"])
    except Exception as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from exc
    user = await UserRepository(session).get_by_id(user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user