fix(backend): 宏观状态权限校验 — 根除 current_user 空值绕过漏洞

Bug: if current_user: 在 Python 中 None/空dict均为falsy,
      一旦 current_user 为空则整个权限块被跳过直接放行。

修复:
1. if not current_user → 直接 raise 401
2. 反转 SUPER_ADMIN 判断: if user_role != 'SUPER_ADMIN' 进入校验
   (避免 if/else/pass 空分支带来的逻辑歧义)
3. main_task is not None 显式判断 (替代隐式 truthy)
This commit is contained in:
2026-08-11 17:39:30 +08:00
parent 9064973a3f
commit b97dfaa95d

View File

@ -301,37 +301,40 @@ async def update_overall_status(
if not product: if not product:
raise HTTPException(status_code=404, detail=f"未找到序列号 {serial_number} 的产品") raise HTTPException(status_code=404, detail=f"未找到序列号 {serial_number} 的产品")
# ── 权限校验 ── # ── 权限校验(无 current_user 一律拒绝,杜绝空 dict 绕过)──
if current_user: if not current_user:
user_role = current_user.get("role", "") raise HTTPException(
user_username = current_user.get("username", "") status_code=status.HTTP_401_UNAUTHORIZED,
detail="请先登录",
)
# SUPER_ADMIN 直接放行 user_role = current_user.get("role", "")
if user_role == "SUPER_ADMIN": user_username = current_user.get("username", "")
pass
else: # SUPER_ADMIN 直接放行
# 检查当前用户是否是该产品主线任务的负责人 if user_role != "SUPER_ADMIN":
from sqlalchemy import or_ # 检查当前用户是否是该产品主线任务的负责人
main_task_result = await db.execute( from sqlalchemy import or_
select(Task).where( main_task_result = await db.execute(
Task.product_id == product.id, select(Task).where(
Task.status.in_(["WIP", "PENDING"]), Task.product_id == product.id,
or_( Task.status.in_(["WIP", "PENDING"]),
Task.parent_task_id.is_(None), or_(
Task.task_type.in_(["TRANSFER", "RECOVERY"]), Task.parent_task_id.is_(None),
), Task.task_type.in_(["TRANSFER", "RECOVERY"]),
).order_by(Task.created_at.desc()).limit(1) ),
).order_by(Task.created_at.desc()).limit(1)
)
main_task = main_task_result.scalar_one_or_none()
has_permission = (
main_task is not None
and main_task.assignee_id == user_username
)
if not has_permission:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="只有 SUPER_ADMIN 或当前操作该产品主线任务的人才能修改宏观状态",
) )
main_task = main_task_result.scalar_one_or_none()
has_permission = (
main_task
and main_task.assignee_id == user_username
)
if not has_permission:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="只有 SUPER_ADMIN 或当前操作该产品主线任务的人才能修改宏观状态",
)
product.overall_status = status_value product.overall_status = status_value
await db.commit() await db.commit()