212 lines
6.2 KiB
Python
212 lines
6.2 KiB
Python
"""认证 API
|
|
|
|
登录、密码修改、用户管理（管理员）
|
|
"""
|
|
|
|
import secrets
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, update
|
|
|
|
from app.core.auth import hash_password, verify_password, create_access_token
|
|
from app.core.deps import get_current_user, get_current_admin
|
|
from app.db.database import get_db
|
|
from app.db.tables import users_table
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
# ---------- Request/Response Models ----------
|
|
|
|
class LoginRequest(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
old_password: str
|
|
new_password: str
|
|
|
|
|
|
class CreateUserRequest(BaseModel):
|
|
username: str
|
|
role: str = "user"
|
|
|
|
|
|
# ---------- Public Endpoints ----------
|
|
|
|
@router.post("/login")
|
|
async def login(req: LoginRequest):
|
|
"""用户登录，返回 JWT token"""
|
|
async with get_db() as db:
|
|
result = await db.execute(
|
|
select(users_table).where(users_table.c.username == req.username)
|
|
)
|
|
user = result.mappings().first()
|
|
|
|
if user is None or not user["is_active"]:
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
if not verify_password(req.password, user["password_hash"]):
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
token = create_access_token({"sub": str(user["id"]), "role": user["role"]})
|
|
|
|
return {
|
|
"token": token,
|
|
"user": {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
"role": user["role"],
|
|
},
|
|
}
|
|
|
|
|
|
# ---------- Authenticated Endpoints ----------
|
|
|
|
@router.get("/me")
|
|
async def get_me(current_user: dict = Depends(get_current_user)):
|
|
"""获取当前用户信息"""
|
|
return {
|
|
"id": current_user["id"],
|
|
"username": current_user["username"],
|
|
"role": current_user["role"],
|
|
"is_active": current_user["is_active"],
|
|
}
|
|
|
|
|
|
@router.post("/change-password")
|
|
async def change_password(
|
|
req: ChangePasswordRequest,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
"""修改自己的密码"""
|
|
if not verify_password(req.old_password, current_user["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="旧密码错误")
|
|
|
|
new_hash = hash_password(req.new_password)
|
|
async with get_db() as db:
|
|
await db.execute(
|
|
update(users_table)
|
|
.where(users_table.c.id == current_user["id"])
|
|
.values(password_hash=new_hash)
|
|
)
|
|
await db.commit()
|
|
|
|
return {"message": "密码修改成功"}
|
|
|
|
|
|
# ---------- Admin Endpoints ----------
|
|
|
|
@router.get("/users")
|
|
async def list_users(admin: dict = Depends(get_current_admin)):
|
|
"""列出所有用户（管理员）"""
|
|
async with get_db() as db:
|
|
result = await db.execute(
|
|
select(users_table).order_by(users_table.c.id)
|
|
)
|
|
rows = result.mappings().all()
|
|
|
|
return [
|
|
{
|
|
"id": r["id"],
|
|
"username": r["username"],
|
|
"role": r["role"],
|
|
"is_active": r["is_active"],
|
|
"created_at": r["created_at"].isoformat() if r["created_at"] else None,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
@router.post("/users")
|
|
async def create_user(req: CreateUserRequest, admin: dict = Depends(get_current_admin)):
|
|
"""创建新用户（管理员），自动生成随机密码"""
|
|
# 检查用户名是否已存在
|
|
async with get_db() as db:
|
|
result = await db.execute(
|
|
select(users_table).where(users_table.c.username == req.username)
|
|
)
|
|
if result.first():
|
|
raise HTTPException(status_code=400, detail="用户名已存在")
|
|
|
|
if req.role not in ("admin", "user"):
|
|
raise HTTPException(status_code=400, detail="角色必须是 admin 或 user")
|
|
|
|
# 生成 12 位随机密码
|
|
raw_password = secrets.token_urlsafe(9)
|
|
password_hash = hash_password(raw_password)
|
|
|
|
await db.execute(
|
|
users_table.insert().values(
|
|
username=req.username,
|
|
password_hash=password_hash,
|
|
role=req.role,
|
|
)
|
|
)
|
|
await db.commit()
|
|
|
|
logger.info(f"管理员 {admin['username']} 创建了用户 {req.username} ({req.role})")
|
|
|
|
return {
|
|
"username": req.username,
|
|
"password": raw_password,
|
|
"role": req.role,
|
|
"message": "请妥善保管密码，此密码仅显示一次",
|
|
}
|
|
|
|
|
|
@router.delete("/users/{user_id}")
|
|
async def disable_user(user_id: int, admin: dict = Depends(get_current_admin)):
|
|
"""禁用用户（软删除）"""
|
|
if user_id == admin["id"]:
|
|
raise HTTPException(status_code=400, detail="不能禁用自己")
|
|
|
|
async with get_db() as db:
|
|
result = await db.execute(
|
|
select(users_table).where(users_table.c.id == user_id)
|
|
)
|
|
user = result.mappings().first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
|
|
await db.execute(
|
|
update(users_table)
|
|
.where(users_table.c.id == user_id)
|
|
.values(is_active=False)
|
|
)
|
|
await db.commit()
|
|
|
|
return {"message": f"用户 {user['username']} 已禁用"}
|
|
|
|
|
|
@router.post("/users/{user_id}/reset-password")
|
|
async def reset_password(user_id: int, admin: dict = Depends(get_current_admin)):
|
|
"""重置用户密码（管理员），生成新的随机密码"""
|
|
async with get_db() as db:
|
|
result = await db.execute(
|
|
select(users_table).where(users_table.c.id == user_id)
|
|
)
|
|
user = result.mappings().first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="用户不存在")
|
|
|
|
raw_password = secrets.token_urlsafe(9)
|
|
password_hash = hash_password(raw_password)
|
|
|
|
await db.execute(
|
|
update(users_table)
|
|
.where(users_table.c.id == user_id)
|
|
.values(password_hash=password_hash)
|
|
)
|
|
await db.commit()
|
|
|
|
return {
|
|
"username": user["username"],
|
|
"password": raw_password,
|
|
"message": "请妥善保管新密码，此密码仅显示一次",
|
|
}
|