fix: 后台停用/禁用真正生效（重复集团任职与超管绕过）

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/utils.py

@@ -171,25 +171,46 @@ def verify_kefu_permission(request, username_from_frontend=None):
 
     # ---------- 1. 必须是客服或管理员 ----------
     from jituan.constants import SUPER_ADMIN_PHONES
-    from jituan.services.admin_context import is_kefu_backend_account
+    from jituan.services.admin_context import (
-    is_super_phone = getattr(current_user, 'Phone', '') in SUPER_ADMIN_PHONES
+        is_kefu_backend_account,
+        is_kefu_account_disabled,
+        is_hardcoded_super_phone,
+        is_system_super_admin,
+    )
+    is_super_phone = is_hardcoded_super_phone(current_user)
     is_admin = current_user.UserType == 'admin' or current_user.IsSuperuser or is_super_phone
+
+    # 已禁用客服：非白名单即使带 IsSuperuser 也不能继续调接口
+    if is_kefu_account_disabled(current_user):
+        return None, Response({'code': 403, 'msg': '客服账号已被禁用'}, status=403)
+
     if not is_kefu_backend_account(current_user):
         return None, Response({'code': 403, 'msg': '身份错误，非客服账号'}, status=403)
 
-    # ---------- 2. 客服扩展表存在且状态正常（管理员跳过） ----------
+    # ---------- 2. 客服扩展表存在且状态正常（仅白名单超管可跳过） ----------
     kefu = None
-    if not is_admin:
+    if is_admin and is_super_phone:
+        from types import SimpleNamespace
+        try:
+            kefu = current_user.KefuProfile
+        except ObjectDoesNotExist:
+            kefu = SimpleNamespace(user=current_user)
+    elif is_admin:
+        # 非白名单「超管」仍必须客服状态正常
+        try:
+            kefu = current_user.KefuProfile
+            if kefu.zhuangtai != 1:
+                return None, Response({'code': 403, 'msg': '客服账号已被禁用'}, status=403)
+        except ObjectDoesNotExist:
+            from types import SimpleNamespace
+            kefu = SimpleNamespace(user=current_user)
+    else:
         try:
             kefu = current_user.KefuProfile
             if kefu.zhuangtai != 1:
                 return None, Response({'code': 403, 'msg': '客服账号已被禁用'}, status=403)
         except ObjectDoesNotExist:
             return None, Response({'code': 403, 'msg': '客服账号不存在'}, status=403)
-    else:
-        # 管理员无需客服扩展表，用占位对象以通过 kefu_obj is None 检查
-        from types import SimpleNamespace
-        kefu = SimpleNamespace(user=current_user)
 
     # ---------- 3. 越权检测（只记录前端篡改username的情况） ----------
     if username_from_frontend and current_user.Phone != username_from_frontend:

backend/views/roles.py

@@ -764,7 +764,17 @@ class ModifyAdminUserView(APIView):
                 if nicheng is not None:
                     target_kefu.nicheng = nicheng
                 if status is not None:
-                    target_kefu.zhuangtai = int(status)
+                    new_status = int(status)
+                    target_kefu.zhuangtai = new_status
+                    if new_status != 1:
+                        # 禁用账号：停用全部集团/俱乐部任职，并摘掉非白名单的 IsSuperuser
+                        from jituan.constants import SUPER_ADMIN_PHONES
+                        from jituan.services.admin_assignments import deactivate_all_assignments_for_user
+                        deactivate_all_assignments_for_user(target_user.UserUID)
+                        if (target_user.Phone or '') not in SUPER_ADMIN_PHONES:
+                            if target_user.IsSuperuser:
+                                target_user.IsSuperuser = False
+                                target_user.save(update_fields=['IsSuperuser'])
                 if password:
                     if len(password) < 6:
                         return Response({'code': 400, 'msg': '密码至少6位'})

jituan/services/admin_assignments.py

@@ -7,19 +7,19 @@ from jituan.models import AdminAssignment
 
 
 def _club_filter(qs, club_id):
-    if club_id is None:
+    if club_id is None or club_id == '':
         return qs.filter(Q(club_id__isnull=True) | Q(club_id=''))
     return qs.filter(club_id=club_id)
 
 
 def find_assignment(yonghuid, club_id, role_code):
     qs = AdminAssignment.query.filter(yonghuid=yonghuid, role_code=role_code)
-    return _club_filter(qs, club_id).first()
+    return _club_filter(qs, club_id).order_by('-id').first()
 
 
 def find_active_assignment(yonghuid, club_id, role_code):
     qs = AdminAssignment.query.filter(yonghuid=yonghuid, role_code=role_code, status=1)
-    return _club_filter(qs, club_id).first()
+    return _club_filter(qs, club_id).order_by('-id').first()
 
 
 def clear_other_primary_assignments(yonghuid, except_id=None):
@@ -53,10 +53,17 @@ def create_or_reactivate_assignment(
     if is_primary:
         _clear_other_primary(yonghuid)
 
-    inactive = find_assignment(yonghuid, club_id, role_code)
+    # MySQL 下 club_id=NULL 唯一约束不生效，可能有多条停用重复行，全部复活改只复活最新一条
-    if inactive and inactive.status != 1:
+    inactive_qs = _club_filter(
+        AdminAssignment.objects.filter(yonghuid=yonghuid, role_code=role_code).exclude(status=1),
+        club_id,
+    ).order_by('-id')
+    inactive = inactive_qs.first()
+    if inactive:
         if is_primary:
             _clear_other_primary(yonghuid)
+        # 同键其它停用重复行保持停用，只启用一条
+        inactive_qs.exclude(pk=inactive.pk).update(status=0, UpdateTime=timezone.now())
         AdminAssignment.objects.filter(pk=inactive.id).update(
             status=1,
             data_scope=data_scope,
@@ -79,22 +86,41 @@ def create_or_reactivate_assignment(
 
 
 def deactivate_assignment(assignment_id):
+    """
+    停用任职。必须连同「同人+同角色+同俱乐部」的全部有效行一起停用：
+    MySQL 对 club_id=NULL 的 unique 不生效，集团任职常会积多条，只停一条等于没停。
+    """
     try:
         pk = int(assignment_id)
     except (TypeError, ValueError):
         return False, '任职 id 无效'
 
-    updated = AdminAssignment.objects.filter(pk=pk, status=1).update(
+    row = AdminAssignment.objects.filter(pk=pk).first()
+    if not row:
+        return False, '任职记录不存在'
+
+    qs = AdminAssignment.objects.filter(
+        yonghuid=row.yonghuid,
+        role_code=row.role_code,
+        status=1,
+    )
+    qs = _club_filter(qs, row.club_id)
+    updated = qs.update(status=0, UpdateTime=timezone.now())
+    # 目标行若已是停用，仍视为成功（幂等）
+    if updated or int(row.status or 0) != 1:
+        return True, None
+    return True, None
+
+
+def deactivate_all_assignments_for_user(yonghuid):
+    """账号禁用时：停用该用户全部有效任职。"""
+    yonghuid = (yonghuid or '').strip()
+    if not yonghuid:
+        return 0
+    return AdminAssignment.objects.filter(yonghuid=yonghuid, status=1).update(
         status=0,
         UpdateTime=timezone.now(),
     )
-    if updated:
-        return True, None
-
-    exists = AdminAssignment.objects.filter(pk=pk).exists()
-    if exists:
-        return True, None
-    return False, '任职记录不存在'
 
 
 def dedupe_active_assignments():

jituan/services/admin_context.py

@@ -157,8 +157,29 @@ def is_system_super_admin(user):
     )
 
 
+def is_hardcoded_super_phone(user):
+    """仅代码白名单手机号（不可被后台「禁用」锁死本人）。"""
+    return getattr(user, 'Phone', '') in SUPER_ADMIN_PHONES
+
+
+def is_kefu_account_disabled(user):
+    """
+    客服扩展表已禁用。
+    非白名单账号：即使挂了 IsSuperuser 也必须认 zhuangtai（否则后台点禁用等于摆设）。
+    """
+    if is_hardcoded_super_phone(user):
+        return False
+    try:
+        kefu = user.KefuProfile
+    except Exception:
+        return False
+    return int(getattr(kefu, 'zhuangtai', 1) or 0) != 1
+
+
 def is_kefu_backend_account(user):
     """可登录客服后台：超管 / UserType=kefu / 有正常 KefuProfile。"""
+    if is_kefu_account_disabled(user):
+        return False
     if is_system_super_admin(user):
         return True
     if getattr(user, 'UserType', '') == 'kefu':

jituan/views.py

@@ -148,12 +148,27 @@ class ClubKefuLoginView(APIView):
             if not user.CheckPassword(password):
                 continue
             password_matched = True
+
+            from jituan.services.admin_context import is_hardcoded_super_phone
+            try:
+                kefu_probe = user.KefuProfile
+            except Exception:
+                kefu_probe = None
+            if kefu_probe is not None and int(getattr(kefu_probe, 'zhuangtai', 1) or 0) != 1:
+                if not is_hardcoded_super_phone(user):
+                    if target_user is None:
+                        target_user = user
+                        target_kefu = kefu_probe
+                    continue
+
             is_admin = user.IsSuperuser or user.UserType == 'admin' or user.Phone in SUPER_ADMIN_PHONES
             if is_admin:
                 target_user = user
-                target_kefu = None
+                target_kefu = kefu_probe
                 break
-            kefu = user.KefuProfile
+            kefu = kefu_probe
+            if kefu is None:
+                continue
             stored_erji = kefu.erjimima or ''
             if stored_erji.startswith(('$2b$', '$2a$')) and len(stored_erji) == 60:
                 import bcrypt as _bcrypt
@@ -170,7 +185,7 @@ class ClubKefuLoginView(APIView):
             target_kefu = kefu
             break
         else:
-            if target_user and target_kefu:
+            if target_user and target_kefu and int(getattr(target_kefu, 'zhuangtai', 1) or 0) != 1:
                 return Response({'code': 4, 'msg': '账号已被封禁', 'data': None},
                                 status=status.HTTP_403_FORBIDDEN)
             if password_matched:

users/views/kefu_base.py

@@ -168,15 +168,29 @@ class KefuLoginView(APIView):
                 continue
             password_matched = True
 
-            # 管理员：跳过二级密码和客服扩展表检查
+            # 管理员：跳过二级密码；但非白名单若客服已禁用仍拦截
             is_admin = user.IsSuperuser or user.UserType == 'admin'
-            if is_admin:
+            from jituan.constants import SUPER_ADMIN_PHONES
+            is_whitelist = (user.Phone or '') in SUPER_ADMIN_PHONES
+            try:
+                kefu_probe = user.KefuProfile
+            except Exception:
+                kefu_probe = None
+            if kefu_probe is not None and int(getattr(kefu_probe, 'zhuangtai', 1) or 0) != 1 and not is_whitelist:
+                if target_user is None:
+                    target_user = user
+                    target_kefu = kefu_probe
+                continue
+
+            if is_admin or is_whitelist:
                 target_user = user
-                target_kefu = None
+                target_kefu = kefu_probe
                 break
 
             # 客服：验证二级密码
-            kefu = user.KefuProfile
+            kefu = kefu_probe
+            if kefu is None:
+                continue
             stored_erji = kefu.erjimima or ''
             if stored_erji.startswith(('$2b$', '$2a$')) and len(stored_erji) == 60:
                 import bcrypt as _bcrypt
@@ -198,7 +212,7 @@ class KefuLoginView(APIView):
             break
         else:
             # 循环结束未找到正常账号
-            if target_user and target_kefu:
+            if target_user and target_kefu and int(getattr(target_kefu, 'zhuangtai', 1) or 0) != 1:
                 # 只有被禁用的账号匹配，返回封禁提示
                 logger.warning(f"登录失败，客服账号已禁用: {phone}, IP: {client_ip}")
                 return Response({