151 lines
5.0 KiB
Python
151 lines
5.0 KiB
Python
"""菜单 code 权限守卫（V0.3：通过 sys_user_role → sys_role_menu → sys_menu.perms）"""
|
||
from typing import Any, Callable, Dict, Iterable, Set
|
||
|
||
from fastapi import Depends, HTTPException
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.auth import get_current_user
|
||
from app.core.database import get_db
|
||
from app.models.models import SysUser
|
||
from app.services.web_auth_roles import iter_user_menu_perm_codes
|
||
|
||
ADMIN_ROLES = {"admin", "super_admin"}
|
||
SUPER_ADMIN_ROLE = "super_admin"
|
||
|
||
|
||
def require_admin_or_super_admin(current_user: dict = Depends(get_current_user)):
|
||
"""运营端角色守卫：admin / super_admin 可访问。"""
|
||
role = current_user.get("role")
|
||
if role not in ADMIN_ROLES:
|
||
raise HTTPException(status_code=403, detail="需要运营人员权限")
|
||
return current_user
|
||
|
||
|
||
def get_user_menu_codes(db: Session, user_id: str) -> Set[str]:
|
||
"""读取用户授权的菜单权限标识集合（与前端 meta.menuCode / perms 对齐）。"""
|
||
return iter_user_menu_perm_codes(db, user_id or "")
|
||
|
||
|
||
def user_has_menu_perm(db: Session, user: dict, perm_code: str) -> bool:
|
||
"""判断用户是否拥有指定权限码（不拦截请求，用于字段/按钮级能力控制）。"""
|
||
code = (perm_code or "").strip()
|
||
if not code:
|
||
return False
|
||
if user.get("role") == SUPER_ADMIN_ROLE:
|
||
return True
|
||
user_id = (user.get("user_id") or "").strip()
|
||
if not user_id:
|
||
return False
|
||
return code in get_user_menu_codes(db, user_id)
|
||
|
||
|
||
def strip_row_fields(rows: Iterable[Dict[str, Any]], *field_names: str) -> None:
|
||
"""从响应行中移除敏感字段（原地修改）。"""
|
||
keys = [k for k in field_names if k]
|
||
if not keys:
|
||
return
|
||
for row in rows:
|
||
for key in keys:
|
||
row.pop(key, None)
|
||
|
||
|
||
def require_menu_code(menu_code: str) -> Callable:
|
||
"""接口菜单权限守卫（已统一放行 menuCode 校验，仅保留运营端登录）。"""
|
||
_ = menu_code # 保留参数以兼容各路由 Depends(require_menu_code("..."))
|
||
return require_admin_or_super_admin
|
||
|
||
|
||
def require_any_menu_code(*menu_codes: str) -> Callable:
|
||
"""接口菜单权限守卫（已统一放行 menuCode 校验，仅保留运营端登录）。"""
|
||
_ = menu_codes
|
||
return require_admin_or_super_admin
|
||
|
||
|
||
def _strict_user_identity(db: Session, user: dict) -> tuple[str, str]:
|
||
"""从用户表读取当前租户，拒绝信任 JWT/请求中的租户覆盖。"""
|
||
user_id = str((user or {}).get("user_id") or "").strip()
|
||
if not user_id:
|
||
return "", ""
|
||
row = (
|
||
db.query(SysUser.id, SysUser.tenant_no)
|
||
.filter(
|
||
SysUser.id == user_id,
|
||
SysUser.del_flag == 0,
|
||
SysUser.status == "1",
|
||
)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return "", ""
|
||
return str(row[0] or "").strip(), str(row[1] or "").strip()
|
||
|
||
|
||
def strict_user_has_any_menu_perm(
|
||
db: Session,
|
||
user: dict,
|
||
menu_codes: Iterable[str],
|
||
) -> bool:
|
||
"""按用户真实租户严格判断任一菜单权限，含直接与职位继承角色。"""
|
||
required = {
|
||
str(code or "").strip() for code in menu_codes if str(code or "").strip()
|
||
}
|
||
if not required:
|
||
return False
|
||
user_id, tenant_no = _strict_user_identity(db, user)
|
||
if not user_id or not tenant_no:
|
||
return False
|
||
granted = iter_user_menu_perm_codes(db, user_id, tenant_no=tenant_no)
|
||
return bool(required.intersection(granted))
|
||
|
||
|
||
def strict_user_has_menu_perm(db: Session, user: dict, menu_code: str) -> bool:
|
||
"""按用户真实租户严格判断单个菜单权限。"""
|
||
return strict_user_has_any_menu_perm(db, user, (menu_code,))
|
||
|
||
|
||
def assert_strict_menu_code(
|
||
db: Session,
|
||
user: dict,
|
||
menu_code: str,
|
||
) -> None:
|
||
"""路由内部字段级权限断言。"""
|
||
if not strict_user_has_menu_perm(db, user, menu_code):
|
||
raise HTTPException(status_code=403, detail="无对应菜单权限")
|
||
|
||
|
||
def require_strict_menu_code(menu_code: str) -> Callable:
|
||
"""租户化的精确菜单权限依赖；与历史宽松守卫并存。"""
|
||
required = str(menu_code or "").strip()
|
||
|
||
async def dependency(
|
||
user: dict = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
assert_strict_menu_code(db, user, required)
|
||
return user
|
||
|
||
dependency.strict_menu_codes = (required,)
|
||
return dependency
|
||
|
||
|
||
def require_strict_any_menu_code(*menu_codes: str) -> Callable:
|
||
"""租户化的任一菜单权限依赖。"""
|
||
required = tuple(
|
||
dict.fromkeys(
|
||
str(code or "").strip()
|
||
for code in menu_codes
|
||
if str(code or "").strip()
|
||
)
|
||
)
|
||
|
||
async def dependency(
|
||
user: dict = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
if not strict_user_has_any_menu_perm(db, user, required):
|
||
raise HTTPException(status_code=403, detail="无对应菜单权限")
|
||
return user
|
||
|
||
dependency.strict_menu_codes = required
|
||
return dependency
|