fix(backend): 权限漏洞修复 + 业务逻辑审查修复

1. 宏观状态越权修复 (products.py + product_service.py):
   - update_overall_status 增加权限校验：仅 SUPER_ADMIN 或当前操作该产品主线任务的人可修改，否则 403

2. 任务撤回越权修复 (task_service.py + tasks.py):
   - recall_task 增加校验：操作人必须等于上游任务负责人（谁发出的谁撤回），否则 403
   - 管理员 (SUPER_ADMIN/SUPERVISOR) 直接放行

3. 驳回上游溯源修复 (task_service.py - reject_task):
   - 将 TaskLog (action_type=create) 提升为优先溯源方式，解决协助分支转交后驳回找不到正确发起人的 Bug
   - 保留 parent_task.assignee_id + task.assignee_id 两级兜底

4. complete_task 父节点继承修复:
   - 修复 task_type == 'MAIN' 永不匹配的 Bug（模型无此值）
   - 改为 not parent_task_id or task_type in (TRANSFER, RECOVERY)

backend/app/api/v1/endpoints/products.py

@@ -102,6 +102,7 @@ async def update_product_endpoint(
     product_id: str,
     data: ProductUpdate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """更新产品"""
     import uuid
@@ -112,6 +113,7 @@ async def update_product_endpoint(
 async def delete_product_endpoint(
     product_id: str,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """删除产品及其关联任务"""
     import uuid
@@ -131,13 +133,18 @@ async def update_product_overall_status(
     serial_number: str,
     data: OverallStatusUpdate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """
     更新产品宏观流转状态。
     移动端首次扫码或手动切换时调用。
     合法值: 备货 | 生产 | 测试 | 维修 | 在库
+
+    权限：仅 SUPER_ADMIN 或当前操作该产品主线任务的人可以修改。
     """
-    return await product_service.update_overall_status(db, serial_number, data.status)
+    return await product_service.update_overall_status(
+        db, serial_number, data.status, current_user,
+    )
 
 
 # ============================================================

backend/app/api/v1/endpoints/tasks.py

@@ -60,6 +60,7 @@ async def get_task(
 async def create_task_endpoint(
     data: TaskCreate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """创建任务"""
     return await task_service.create_task(db, data)
@@ -70,6 +71,7 @@ async def update_task_endpoint(
     task_id: str,
     data: TaskUpdate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """更新任务"""
     return await task_service.update_task(db, uuid.UUID(task_id), data)
@@ -137,12 +139,16 @@ async def recall_task_endpoint(
     task_id: str,
     operator_id: str | None = Query(None),
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """
     **撤回转交：删除 PENDING 子任务，恢复父任务为 WIP。**
     适用场景：转交后发现选错人，在对方接收前撤回。
     """
-    return await task_service.recall_task(db, uuid.UUID(task_id), operator_id)
+    return await task_service.recall_task(
+        db, uuid.UUID(task_id), operator_id,
+        operator_role=current_user.get("role"),
+    )
 
 
 # ============================================================
@@ -161,6 +167,7 @@ async def spawn_subtask_endpoint(
     data: SpawnRequest,
     operator_id: str | None = Query(None),
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """
     **派发协助分支：在当前任务下创建并行子任务，父任务状态保持不变。**
@@ -263,6 +270,7 @@ async def create_subtask_endpoint(
     task_id: str,
     data: SubtaskCreate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """
     **创建子任务：支持无限层级嵌套。**
@@ -297,6 +305,7 @@ async def add_task_record_endpoint(
     task_id: str,
     data: TaskRecordCreate,
     db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
 ):
     """追加进度记录（备注+图片），不改变任务状态"""
     return await task_service.add_task_record(db, uuid.UUID(task_id), data)

backend/app/services/product_service.py

@@ -275,8 +275,17 @@ async def update_product(db: AsyncSession, product_id: uuid.UUID, data: ProductU
 VALID_OVERALL_STATUS = {"备货", "生产", "测试", "维修", "在库"}
 
 
-async def update_overall_status(db: AsyncSession, serial_number: str, status_value: str) -> ProductScanResponse:
+async def update_overall_status(
-    """更新产品宏观状态"""
+    db: AsyncSession, serial_number: str, status_value: str,
+    current_user: dict | None = None,
+) -> ProductScanResponse:
+    """更新产品宏观状态
+
+    权限校验：
+    - SUPER_ADMIN 角色：直接放行
+    - 当前操作该产品主线任务（WIP/PENDING 状态主干任务）的人：放行
+    - 其他：403
+    """
     if status_value not in VALID_OVERALL_STATUS:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -292,6 +301,38 @@ async def update_overall_status(db: AsyncSession, serial_number: str, status_val
     if not product:
         raise HTTPException(status_code=404, detail=f"未找到序列号 {serial_number} 的产品")
 
+    # ── 权限校验 ──
+    if current_user:
+        user_role = current_user.get("role", "")
+        user_username = current_user.get("username", "")
+
+        # SUPER_ADMIN 直接放行
+        if user_role == "SUPER_ADMIN":
+            pass
+        else:
+            # 检查当前用户是否是该产品主线任务的负责人
+            from sqlalchemy import or_
+            main_task_result = await db.execute(
+                select(Task).where(
+                    Task.product_id == product.id,
+                    Task.status.in_(["WIP", "PENDING"]),
+                    or_(
+                        Task.parent_task_id.is_(None),
+                        Task.task_type.in_(["TRANSFER", "RECOVERY"]),
+                    ),
+                ).order_by(Task.created_at.desc()).limit(1)
+            )
+            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
     await db.commit()
     await db.refresh(product)

backend/app/services/task_service.py

@@ -395,14 +395,50 @@ async def end_task(
 # ============================================================
 
 async def recall_task(
-    db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None
+    db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None,
+    operator_role: str | None = None,
 ) -> TaskResponse:
-    """撤回 PENDING 转交：标记为 CANCELED，以被撤回节点为父生成接力新任务给操作人。"""
+    """撤回 PENDING 转交：标记为 CANCELED，以被撤回节点为父生成接力新任务给操作人。
+
+    权限校验：
+    - 管理员（SUPER_ADMIN / SUPERVISOR）：直接放行
+    - 操作人 必须等于 上游任务的负责人（谁发出的谁才能撤回），否则 403
+    """
     task = await _get_task_or_404(db, task_id)
 
     if task.status != TASK_STATUS_PENDING:
         raise HTTPException(status_code=409, detail="只有待接收(PENDING)的任务可以撤回")
 
+    # ── 权限校验：谁发出的谁才能撤回 ──
+    if not (operator_role and operator_role in ADMIN_ROLES):
+        # 查找上游任务的负责人（发出者）
+        upstream_assignee: str | None = None
+        if task.parent_task_id:
+            parent_result = await db.execute(
+                select(Task).where(Task.id == task.parent_task_id)
+            )
+            parent_task = parent_result.scalar_one_or_none()
+            if parent_task:
+                upstream_assignee = parent_task.assignee_id
+        else:
+            # 无父任务：从任务日志追溯创建人
+            log_result = await db.execute(
+                select(TaskLog).where(
+                    TaskLog.task_id == task_id,
+                    TaskLog.action_type == "create",
+                ).order_by(TaskLog.created_at.asc()).limit(1)
+            )
+            create_log = log_result.scalar_one_or_none()
+            if create_log:
+                upstream_assignee = create_log.operator_id
+
+        # 如果找不到上游负责人，或者操作人不等于上游负责人 → 403
+        if not upstream_assignee or operator_id != upstream_assignee:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="您不是该任务的发起人，无法撤回",
+            )
+
     now = get_beijing_time()
 
     # 1. 废掉当前待接收任务
@@ -608,8 +644,20 @@ async def reject_task(
 
     # --- 2. 确定返工任务的负责人（追溯上一道工序的转交人） ---
     rework_assignee_id: str | None = None
-    if task.parent_task_id:
+
-        # 有父任务：返工给父任务的负责人（即上一环的转交人 A）
+    # 🚀 优先方式：从任务日志追溯创建人（准确记录是谁发起的转交）
+    log_result = await db.execute(
+        select(TaskLog).where(
+            TaskLog.task_id == task_id,
+            TaskLog.action_type == "create",
+        ).order_by(TaskLog.created_at.asc()).limit(1)
+    )
+    create_log = log_result.scalar_one_or_none()
+    if create_log and create_log.operator_id:
+        rework_assignee_id = create_log.operator_id
+
+    # 兜底方式1：有父任务 → 返工给父任务的负责人
+    if not rework_assignee_id and task.parent_task_id:
         parent_result = await db.execute(
             select(Task).where(Task.id == task.parent_task_id)
         )
@@ -617,20 +665,8 @@ async def reject_task(
         if parent_task:
             rework_assignee_id = parent_task.assignee_id
 
+    # 兜底方式2：用当前任务的负责人
     if not rework_assignee_id:
-        # 无父任务（顶层转交）：从任务日志追溯创建人（转交发起者 A）
-        log_result = await db.execute(
-            select(TaskLog).where(
-                TaskLog.task_id == task_id,
-                TaskLog.action_type == "create",
-            ).order_by(TaskLog.created_at.asc()).limit(1)
-        )
-        create_log = log_result.scalar_one_or_none()
-        if create_log and create_log.operator_id:
-            rework_assignee_id = create_log.operator_id
-
-    if not rework_assignee_id:
-        # 最后兜底：用当前任务的负责人（通常不应该走到这里）
         rework_assignee_id = task.assignee_id
 
     # --- 3. 创建返工任务 ---
@@ -913,7 +949,12 @@ async def complete_task(
         # 🚀 智能父节点继承算法
         # 主线任务转交 → 保持平级继承（主分支永远在一维主干上）
         # 协助分支转交 → 认当前任务为父（形成向外无限延伸的孙子节点树枝）
-        new_parent_id = task.parent_task_id if task.task_type == "MAIN" else task.id
+        # 注：TASK_TYPE 实际值为 TRANSFER/RECOVERY/SPAWN，不存在 "MAIN"
+        is_main_line = (
+            not task.parent_task_id
+            or task.task_type in ("TRANSFER", "RECOVERY")
+        )
+        new_parent_id = task.parent_task_id if is_main_line else task.id
 
         next_task = Task(
             product_id=task.product_id,