100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
"""
|
|
认证路由：登录 / 刷新令牌 / 修改密码 / 当前用户信息。
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session, select
|
|
|
|
from app.auth import (
|
|
create_access_token,
|
|
create_refresh_token,
|
|
decode_token,
|
|
get_current_user,
|
|
hash_password,
|
|
verify_password,
|
|
)
|
|
from app.db import User, get_session
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class TokenResponse(BaseModel):
|
|
access_token: str
|
|
refresh_token: str
|
|
token_type: str = "bearer"
|
|
user: dict
|
|
|
|
|
|
class RefreshRequest(BaseModel):
|
|
refresh_token: str
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
old_password: str
|
|
new_password: str
|
|
|
|
|
|
def _user_dict(u: User) -> dict:
|
|
return {
|
|
"id": u.id,
|
|
"username": u.username,
|
|
"display_name": u.display_name,
|
|
"is_admin": u.is_admin,
|
|
}
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(body: LoginRequest, session: Session = Depends(get_session)):
|
|
user = session.exec(select(User).where(User.username == body.username)).first()
|
|
if not user or not verify_password(body.password, user.hashed_password):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误")
|
|
if not user.is_active:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="账号已禁用")
|
|
return TokenResponse(
|
|
access_token=create_access_token(user.id),
|
|
refresh_token=create_refresh_token(user.id),
|
|
user=_user_dict(user),
|
|
)
|
|
|
|
|
|
@router.post("/refresh")
|
|
def refresh(body: RefreshRequest, session: Session = Depends(get_session)):
|
|
user_id = decode_token(body.refresh_token, expected_type="refresh")
|
|
user = session.exec(select(User).where(User.id == user_id)).first()
|
|
if not user or not user.is_active:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在或已禁用")
|
|
return {
|
|
"access_token": create_access_token(user.id),
|
|
"token_type": "bearer",
|
|
}
|
|
|
|
|
|
@router.post("/change-password")
|
|
def change_password(
|
|
body: ChangePasswordRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session),
|
|
):
|
|
if not verify_password(body.old_password, current_user.hashed_password):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="旧密码错误")
|
|
if len(body.new_password) < 6:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="新密码至少 6 位")
|
|
# 重新获取以确保在同一 session 中
|
|
user = session.exec(select(User).where(User.id == current_user.id)).first()
|
|
if user:
|
|
user.hashed_password = hash_password(body.new_password)
|
|
session.add(user)
|
|
session.commit()
|
|
return {"message": "密码已修改"}
|
|
|
|
|
|
@router.get("/me")
|
|
def me(current_user: User = Depends(get_current_user)):
|
|
return _user_dict(current_user)
|