52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
from __future__ import annotations
|
||
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, HTTPException, Request, Response, status
|
||
|
||
from server import config, db
|
||
from server.models import LoginRequest, LoginResponse
|
||
|
||
|
||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||
|
||
|
||
async def _parse_body(request: Request) -> dict:
|
||
"""兼容 JSON 和 form-data。"""
|
||
ctype = (request.headers.get("content-type") or "").lower()
|
||
if "application/json" in ctype:
|
||
data = await request.json()
|
||
if not isinstance(data, dict):
|
||
raise HTTPException(status_code=422, detail="JSON body 必须是对象")
|
||
return data
|
||
form = await request.form()
|
||
return dict(form)
|
||
|
||
|
||
@router.post("/login", response_model=LoginResponse)
|
||
async def login(request: Request, response: Response):
|
||
"""
|
||
登录接口（支持 JSON 和 form-data）：
|
||
- 校验用户名/密码
|
||
- 生成 token，写入数据库
|
||
- 通过 Set-Cookie 返回 auth_token，前端后续请求自动携带
|
||
- 下一次登录会生成新 token，旧 token 自动失效
|
||
"""
|
||
req = LoginRequest(**(await _parse_body(request)))
|
||
|
||
if req.username != config.ADMIN_USERNAME or req.password != config.ADMIN_PASSWORD:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
|
||
|
||
token = uuid.uuid4().hex
|
||
db.set_auth_token(req.username, token)
|
||
|
||
response.set_cookie(
|
||
key="auth_token",
|
||
value=token,
|
||
httponly=True,
|
||
max_age=365 * 24 * 60 * 60,
|
||
samesite="lax",
|
||
)
|
||
|
||
return LoginResponse(token=token)
|