SupplyAgent/api/routers/auth.py

from __future__ import annotations

from fastapi import APIRouter, HTTPException, Request, Response, status

from api.schemas.auth import CreateAuthUserBody, LoginBody, UpdateAuthUserBody
from api.services.auth import (
    SESSION_COOKIE_NAME,
    DuplicateUsernameError,
    InvalidCredentialsError,
    authenticate,
    create_user,
    delete_user,
    list_users,
    revoke_session,
    update_user,
)
from supply_infra.config import get_infra_settings

router = APIRouter(prefix="/api", tags=["authentication"])


@router.post("/auth/login")
def login(body: LoginBody, request: Request, response: Response) -> dict:
    try:
        user, token = authenticate(
            username=body.username,
            password=body.password,
            ip_address=request.client.host if request.client else None,
            user_agent=request.headers.get("user-agent"),
        )
    except InvalidCredentialsError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
        ) from None

    settings = get_infra_settings()
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=token,
        max_age=settings.auth_session_hours * 60 * 60,
        httponly=True,
        secure=settings.auth_cookie_secure,
        samesite="lax",
        path="/",
    )
    return {"user": user}


@router.get("/auth/me")
def me(request: Request) -> dict:
    return {"user": request.state.current_user}


@router.post("/auth/logout")
def logout(request: Request, response: Response) -> dict[str, bool]:
    token = request.cookies.get(SESSION_COOKIE_NAME)
    if token:
        revoke_session(token)
    response.delete_cookie(
        key=SESSION_COOKIE_NAME,
        path="/",
        secure=get_infra_settings().auth_cookie_secure,
        httponly=True,
        samesite="lax",
    )
    return {"ok": True}


@router.get("/admin/users")
def admin_users() -> dict:
    return {"items": list_users()}


@router.post("/admin/users", status_code=201)
def admin_create_user(body: CreateAuthUserBody) -> dict:
    try:
        return create_user(
            username=body.username,
            password=body.password,
            display_name=body.display_name,
            role=body.role,
        )
    except DuplicateUsernameError:
        raise HTTPException(status_code=409, detail="用户名已存在") from None


@router.patch("/admin/users/{user_id}")
def admin_update_user(
    user_id: int,
    body: UpdateAuthUserBody,
    request: Request,
) -> dict:
    current_user = request.state.current_user
    if current_user["id"] == user_id and (
        (body.role is not None and body.role != current_user["role"])
        or body.status == "disabled"
        or body.password is not None
    ):
        raise HTTPException(status_code=409, detail="不能降级、禁用或重置当前登录账号")

    result = update_user(
        user_id,
        display_name=body.display_name,
        role=body.role,
        status=body.status,
        password=body.password,
    )
    if result is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return result


@router.delete("/admin/users/{user_id}", status_code=204)
def admin_delete_user(user_id: int, request: Request) -> Response:
    if request.state.current_user["id"] == user_id:
        raise HTTPException(status_code=409, detail="不能删除当前登录账号")
    if not delete_user(user_id):
        raise HTTPException(status_code=404, detail="用户不存在")
    return Response(status_code=204)