162 lines
6.5 KiB
Python
162 lines
6.5 KiB
Python
"""权限管理：角色（菜单权限矩阵）与用户（分配角色 / 启停 / 重置密码）。
|
||
|
||
只有拿到「权限管理」菜单授权的用户能进（写操作还要 rw），由 security.guard 拦。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from sqlalchemy.orm import Session
|
||
|
||
from .. import security
|
||
from ..db import get_db
|
||
from ..models import Role, User
|
||
from ..schemas import RoleIn, RolePatch, UserIn, UserPatch
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/rbac/meta")
|
||
def meta():
|
||
return {
|
||
"menus": [{"key": k, "label": v} for k, v in security.MENUS],
|
||
"levels": [{"value": "rw", "label": "可编辑"}, {"value": "ro", "label": "只读"},
|
||
{"value": "", "label": "无权限"}],
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------- 角色
|
||
@router.get("/rbac/roles")
|
||
def list_roles(db: Session = Depends(get_db)):
|
||
roles = db.query(Role).order_by(Role.id).all()
|
||
out = []
|
||
for r in roles:
|
||
d = r.to_dict()
|
||
d["user_count"] = db.query(User).filter(User.role_id == r.id).count()
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
@router.post("/rbac/roles")
|
||
def create_role(body: RoleIn, db: Session = Depends(get_db)):
|
||
name = (body.name or "").strip()
|
||
if not name:
|
||
raise HTTPException(status_code=400, detail="角色名不能为空")
|
||
if db.query(Role).filter(Role.name == name).count():
|
||
raise HTTPException(status_code=400, detail=f"角色「{name}」已存在")
|
||
role = Role(name=name[:64], description=(body.description or "").strip() or None,
|
||
menus=security.menus_json(body.menus))
|
||
db.add(role)
|
||
db.commit()
|
||
return role.to_dict()
|
||
|
||
|
||
@router.patch("/rbac/roles/{rid}")
|
||
def update_role(rid: int, body: RolePatch, db: Session = Depends(get_db)):
|
||
role = db.get(Role, rid)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
if body.name is not None:
|
||
name = body.name.strip()
|
||
if not name:
|
||
raise HTTPException(status_code=400, detail="角色名不能为空")
|
||
dup = db.query(Role).filter(Role.name == name, Role.id != rid).count()
|
||
if dup:
|
||
raise HTTPException(status_code=400, detail=f"角色「{name}」已存在")
|
||
role.name = name[:64]
|
||
if body.description is not None:
|
||
role.description = body.description.strip() or None
|
||
if body.menus is not None:
|
||
role.menus = security.menus_json(body.menus)
|
||
db.commit()
|
||
return role.to_dict()
|
||
|
||
|
||
@router.delete("/rbac/roles/{rid}")
|
||
def delete_role(rid: int, me: User = Depends(security.current_user), db: Session = Depends(get_db)):
|
||
role = db.get(Role, rid)
|
||
if role is None:
|
||
raise HTTPException(status_code=404, detail="角色不存在")
|
||
if role.is_builtin:
|
||
raise HTTPException(status_code=400, detail="内置角色不能删除，可以改它的菜单权限")
|
||
if me.role_id == rid:
|
||
raise HTTPException(status_code=400, detail="不能删除自己正在使用的角色")
|
||
if db.query(User).filter(User.role_id == rid).count():
|
||
raise HTTPException(status_code=400, detail="该角色下还有用户，请先给他们换角色")
|
||
db.delete(role)
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
# ---------------------------------------------------------------------- 用户
|
||
@router.get("/rbac/users")
|
||
def list_users(db: Session = Depends(get_db)):
|
||
return [u.to_dict() for u in db.query(User).order_by(User.id).all()]
|
||
|
||
|
||
@router.post("/rbac/users")
|
||
def create_user(body: UserIn, db: Session = Depends(get_db)):
|
||
name = (body.name or "").strip()
|
||
username = (body.username or "").strip()
|
||
password = (body.password or "").strip()
|
||
if not name or not username:
|
||
raise HTTPException(status_code=400, detail="姓名和登录账号都要填")
|
||
if len(password) < 6:
|
||
raise HTTPException(status_code=400, detail="初始密码至少 6 位")
|
||
if db.query(User).filter(User.username == username).count():
|
||
raise HTTPException(status_code=400, detail=f"登录账号「{username}」已存在")
|
||
pw_hash, salt = security.hash_password(password)
|
||
user = User(name=name[:64], username=username[:64], password_hash=pw_hash, password_salt=salt,
|
||
role_id=body.role_id, active=bool(body.active))
|
||
db.add(user)
|
||
db.commit()
|
||
return user.to_dict()
|
||
|
||
|
||
@router.patch("/rbac/users/{uid}")
|
||
def update_user(uid: int, body: UserPatch, me: User = Depends(security.current_user),
|
||
db: Session = Depends(get_db)):
|
||
user = db.get(User, uid)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
if body.name is not None:
|
||
name = body.name.strip()
|
||
if not name:
|
||
raise HTTPException(status_code=400, detail="姓名不能为空")
|
||
user.name = name[:64]
|
||
if body.role_id is not None and body.role_id != user.role_id:
|
||
if db.get(Role, body.role_id) is None:
|
||
raise HTTPException(status_code=400, detail="角色不存在")
|
||
if user.id == me.id:
|
||
raise HTTPException(status_code=400, detail="不能修改自己的角色")
|
||
user.role_id = body.role_id
|
||
security.revoke_all(db, user.id) # 权限变了，强制重新登录
|
||
if body.password is not None and body.password.strip():
|
||
if len(body.password.strip()) < 6:
|
||
raise HTTPException(status_code=400, detail="新密码至少 6 位")
|
||
user.password_hash, user.password_salt = security.hash_password(body.password.strip())
|
||
security.revoke_all(db, user.id)
|
||
if body.active is not None:
|
||
if user.id == me.id and not body.active:
|
||
raise HTTPException(status_code=400, detail="不能停用自己的账号")
|
||
user.active = bool(body.active)
|
||
if not user.active:
|
||
security.revoke_all(db, user.id)
|
||
db.commit()
|
||
return user.to_dict()
|
||
|
||
|
||
@router.delete("/rbac/users/{uid}")
|
||
def delete_user(uid: int, me: User = Depends(security.current_user), db: Session = Depends(get_db)):
|
||
user = db.get(User, uid)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
if user.id == me.id:
|
||
raise HTTPException(status_code=400, detail="不能删除自己，可以换一个角色或让别人删")
|
||
remaining = db.query(User).filter(User.active.is_(True), User.id != uid).count()
|
||
if not remaining:
|
||
raise HTTPException(status_code=400, detail="至少要保留一个可用账号")
|
||
security.revoke_all(db, user.id)
|
||
db.delete(user)
|
||
db.commit()
|
||
return {"ok": True}
|