auth/routers.py - master - developer_atatanoff/simple_auth_fastapi

auth/routers.py
174 строки5 KB
Alexey N. Tatanoff
Исправлены ошибки csrf аутентификации
05 янв 2026, 21:01
05 янв 2026, 21:015cfee19
from random import randint
from typing import Optional
import uuid

from sqlalchemy.exc import IntegrityError

from fastapi import APIRouter, HTTPException, status, Request, Response, Depends, Query
from fastapi.templating import Jinja2Templates

from fastapi_pagination import Page, Params
from fastapi_pagination.ext.sqlalchemy import paginate

from config.logger import logger
from config.settings import settings, VERSION

from auth.schemas import (
    UserLogin,
    CurrentUser,
    UserOut,
    UserIn,
    UsersOut,
    UserInUpdate,
    UserDelIn,
)
from auth.repository import AuthUserRepository
from auth.authenticat import AuthenticatLogin
from auth.verify import set_cookie
from auth.models import Users
from auth.database import get_db, AsyncSession

router = APIRouter(tags=["authenticat"])

templates = Jinja2Templates(directory="auth/templates")


@router.get("/enterm")
async def get_login_page(request: Request):
    csrf_token = str(uuid.uuid4())
    response = templates.TemplateResponse(
        "login.html",
        {"VERSION": randint(1, 10) if settings.DEBUG else VERSION, "request": request},
    )
    response.set_cookie(
        key="csrftoken",
        value=csrf_token,
        max_age=3600,
        secure=not settings.DEBUG,  # True для HTTPS в production
        samesite="strict"  # Защита от CSRF
    )

    return response

@router.post("/enterm")
async def login(request: Request, user_data: UserLogin):
    if not user_data.username or not user_data.password:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username and password are required",
        )
    auth = AuthenticatLogin(AuthUserRepository())
    user = await auth.authenticate(user_data, request)

    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    response = Response(status_code=status.HTTP_200_OK)
    is_secure = request.url.scheme == "https"
    response = set_cookie(response, is_secure, user)
    return response

@router.get("/api/v2/get-user/")
async def get_user(request: Request):
    # userout  нужно отдавать, вот оно где косяк
    return {"currentUser": UserOut(**request.state.user.model_dump())}


class GetDataParams(Params):
    page: int = Query(1, ge=1, description="Номер страницы")
    size: int = Query(10, ge=1, le=100, description="Размер страницы")
    sort_by: Optional[str] = Query("-dt_time", description="Поле для сортировки")


@router.get("/api/v2/get-users/", response_model=Page[UsersOut])
async def get_users(
    request: Request,
    db: AsyncSession = Depends(get_db),
    params: GetDataParams = Depends(),
):
    if request.state.user.is_superuser:
        stmt = AuthUserRepository().get_users_stmt()
        res = await paginate(db, stmt, params=params)
        return res
    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)


@router.post("/api/v2/set-user/", response_model=Page[UsersOut])
async def create_user(
    request: Request,
    user: UserIn,
    db: AsyncSession = Depends(get_db),
    params: GetDataParams = Depends(),
):
    if not request.state.user.is_superuser:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    rep = AuthUserRepository()
    create_stmt_user = rep.create_user_stmt(user)
    try:
        db.add(create_stmt_user)
        await db.commit()
    except IntegrityError as e:
        await db.rollback()  # Откат изменений
        raise HTTPException(
            status_code=400, detail=f"Пользователь с указанными данными уже существует."
        )

    stmt = rep.get_users_stmt()
    res = await paginate(db, stmt, params=params)
    return res


@router.post("/api/v2/update-user/", response_model=Page[UsersOut])
async def update_user(
    request: Request,
    user: UserInUpdate,
    db: AsyncSession = Depends(get_db),
    params: GetDataParams = Depends(),
):

    if not request.state.user.is_superuser:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    rep = AuthUserRepository()
    update_stmt_user = rep.update_user_stmt(user)
    try:
        await db.execute(update_stmt_user)
        await db.commit()
    except Exception as e:
        await db.rollback()
        raise HTTPException(status_code=400, detail=f"Произошла ошибка {e}.")

    stmt = rep.get_users_stmt()
    res = await paginate(db, stmt, params=params)
    return res


@router.post("/api/v2/del-user/", response_model=Page[UsersOut])
async def del_user(
    request: Request,
    user: UserDelIn,
    db: AsyncSession = Depends(get_db),
    params: GetDataParams = Depends(),
):

    rep = AuthUserRepository()
    del_user_stmt = rep.del_user_stmt(user)
    try:
        await db.execute(del_user_stmt)
        await db.commit()

    except IntegrityError as e:
        await db.rollback()
        raise HTTPException(status_code=400, detail=f"Произошла ошибка {e}.")
    stmt = rep.get_users_stmt()
    res = await paginate(db, stmt, params=params)
    return res


@router.post("/api/v2/sing-out/")
async def log_out(request: Request, user: CurrentUser):
    repo = AuthUserRepository()
    access_token = request.cookies.get("access_token")
    await repo.token.delete_access_token(access_token)
    refresh_token = request.cookies.get("refresh_token")
    logger.debug(user)
    await repo.delete_refresh_token(refresh_token)
    return {"ok": "ok"}