harness_zbt/backend/app/api/v1/admin_auth.py

"""后台管理员登录和当前用户接口。"""

from typing import Annotated, Any

from fastapi import APIRouter, Depends, Request, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, Field

from app.api.responses import success_response
from app.api.v1.h5_auth import RefreshRequest
from app.core.config import Settings
from app.core.errors import AppError
from app.domains.identity.models import AdminUser
from app.domains.identity.service import AdminAuthService

admin_bearer_scheme = HTTPBearer(auto_error=False)
ADMIN_REFRESH_COOKIE = "zbt_admin_refresh"
ADMIN_REFRESH_COOKIE_PATH = "/api/v1/admin/auth"


class AdminLoginRequest(BaseModel):
    """管理员登录请求体；Field 同时生成接口文档并执行长度校验。"""

    username: str = Field(min_length=3, max_length=64)
    password: str = Field(min_length=8, max_length=128)


def create_admin_router(auth_service: AdminAuthService, settings: Settings) -> APIRouter:
    """创建后台认证路由。"""
    router = APIRouter(tags=["admin-auth"])

    def current_admin(
        credentials: Annotated[
            HTTPAuthorizationCredentials | None,
            Depends(admin_bearer_scheme),
        ],
    ) -> AdminUser:
        # 该依赖校验 Bearer Token，成功后把管理员对象传给具体接口。
        if credentials is None:
            raise AppError("AUTH_REQUIRED", "请先登录后台", 401)
        return auth_service.authenticate_access(credentials.credentials)

    def set_refresh_cookie(response: Response, refresh_token: str) -> None:
        response.set_cookie(
            key=ADMIN_REFRESH_COOKIE,
            value=refresh_token,
            max_age=settings.refresh_token_days * 24 * 60 * 60,
            httponly=True,
            secure=settings.app_env == "production",
            samesite="lax",
            path=ADMIN_REFRESH_COOKIE_PATH,
        )

    @router.post("/admin/auth/login")
    def login(
        payload: AdminLoginRequest,
        request: Request,
        response: Response,
    ) -> dict[str, Any]:
        result = auth_service.login(payload.username, payload.password)
        set_refresh_cookie(response, result["tokens"]["refresh_token"])
        return success_response(request, result)

    @router.post("/admin/auth/refresh")
    def refresh(
        request: Request,
        response: Response,
        payload: RefreshRequest | None = None,
    ) -> dict[str, Any]:
        refresh_token = (
            payload.refresh_token
            if payload is not None
            else request.cookies.get(ADMIN_REFRESH_COOKIE)
        )
        if refresh_token is None:
            raise AppError("AUTH_REQUIRED", "缺少刷新凭证", 401)
        result = auth_service.refresh(refresh_token)
        set_refresh_cookie(response, result["tokens"]["refresh_token"])
        return success_response(request, result)

    @router.post("/admin/auth/logout")
    def logout(
        request: Request,
        response: Response,
        user: Annotated[AdminUser, Depends(current_admin)],
        credentials: Annotated[
            HTTPAuthorizationCredentials,
            Depends(admin_bearer_scheme),
        ],
    ) -> dict[str, Any]:
        del user
        auth_service.logout(credentials.credentials)
        response.delete_cookie(
            key=ADMIN_REFRESH_COOKIE,
            path=ADMIN_REFRESH_COOKIE_PATH,
            samesite="lax",
        )
        return success_response(request, {"logged_out": True})

    @router.get("/admin/me")
    def me(
        request: Request,
        user: Annotated[AdminUser, Depends(current_admin)],
    ) -> dict[str, Any]:
        return success_response(
            request,
            {
                "id": user.id,
                "username": user.username,
                "display_name": user.display_name,
                "status": user.status,
                "roles": sorted(user.roles),
                "permissions": sorted(user.permissions),
                "data_scope": user.data_scope,
            },
        )

    return router