新增

backend/auth/roles.py

@@ -7,12 +7,15 @@ from typing import Iterable
 
 from .permissions import (
     ALL_PERMISSIONS,
+    DATA_SCOPE_ALL,
     OPERATOR_PERMISSIONS,
     VIEWER_PERMISSIONS,
     WRITE_PERMISSIONS,
-    normalize_permissions,
+    expand_legacy_bundles,
+    permission_implies,
 )
 
+
 ROLE_ADMIN = "admin"
 ROLE_OPERATOR = "operator"
 ROLE_VIEWER = "viewer"
@@ -44,7 +47,7 @@ def default_role_seeds() -> list[RoleRecord]:
         RoleRecord(
             code=ROLE_ADMIN,
             label=ROLE_LABELS[ROLE_ADMIN],
-            description="拥有全部菜单与操作权限，可管理全局数据",
+            description="拥有全部菜单、按钮与全局数据权限",
             is_system=True,
             is_admin=True,
             permissions=list(ALL_PERMISSIONS),
@@ -52,7 +55,7 @@ def default_role_seeds() -> list[RoleRecord]:
         RoleRecord(
             code=ROLE_OPERATOR,
             label=ROLE_LABELS[ROLE_OPERATOR],
-            description="管理自己的账号、规则与私信",
+            description="管理自己的账号、规则与私信（仅本人数据）",
             is_system=True,
             is_admin=False,
             permissions=list(OPERATOR_PERMISSIONS),
@@ -91,20 +94,27 @@ def role_label(code: str | None) -> str:
 
 
 def is_admin(role: str | None) -> bool:
-    """True when the role has global data scope (built-in admin)."""
+    """True for the built-in admin role (global admin flag)."""
     record = get_cached_role(role)
     if record is not None:
         return bool(record.is_admin)
-    # Fallback before cache is warm / for unit tests.
     return str(role or "") == ROLE_ADMIN
 
 
+def has_global_scope(role: str | None) -> bool:
+    """True when the role may see all users' data (admin or data.scope_all)."""
+    if is_admin(role):
+        return True
+    return has_permission(role, DATA_SCOPE_ALL)
+
+
 def can_write(role: str | None) -> bool:
     record = get_cached_role(role)
     if record is not None:
         if record.is_admin:
             return True
-        return any(code in WRITE_PERMISSIONS for code in record.permissions)
+        held = expand_legacy_bundles(set(record.permissions))
+        return any(code in WRITE_PERMISSIONS for code in held)
     return str(role or "") in (ROLE_ADMIN, ROLE_OPERATOR)
 
 
@@ -112,6 +122,10 @@ def can_manage_users(role: str | None) -> bool:
     return has_permission(role, "users.manage")
 
 
+def can_manage_roles(role: str | None) -> bool:
+    return has_permission(role, "roles.manage")
+
+
 def has_permission(role: str | None, permission: str) -> bool:
     code = str(permission or "").strip()
     if not code:
@@ -121,13 +135,14 @@ def has_permission(role: str | None, permission: str) -> bool:
         if str(role or "") == ROLE_ADMIN:
             return True
         if str(role or "") == ROLE_OPERATOR:
-            return code in OPERATOR_PERMISSIONS
+            return permission_implies(expand_legacy_bundles(set(OPERATOR_PERMISSIONS)), code)
         if str(role or "") == ROLE_VIEWER:
-            return code in VIEWER_PERMISSIONS
+            return permission_implies(set(VIEWER_PERMISSIONS), code)
         return False
     if record.is_admin:
         return True
-    return code in record.permissions
+    held = expand_legacy_bundles(set(record.permissions))
+    return permission_implies(held, code)
 
 
 def permissions_for_role(role: str | None) -> list[str]:
@@ -136,7 +151,9 @@ def permissions_for_role(role: str | None) -> list[str]:
         if str(role or "") == ROLE_ADMIN:
             return list(ALL_PERMISSIONS)
         if str(role or "") == ROLE_OPERATOR:
-            return list(OPERATOR_PERMISSIONS)
+            from .permissions import expand_paired_permissions
+
+            return expand_paired_permissions(OPERATOR_PERMISSIONS)
         if str(role or "") == ROLE_VIEWER:
             return list(VIEWER_PERMISSIONS)
         return []
@@ -164,4 +181,6 @@ def sanitize_role_permissions(
 ) -> list[str]:
     if force_all:
         return list(ALL_PERMISSIONS)
-    return normalize_permissions(codes)
+    from .permissions import expand_paired_permissions
+
+    return expand_paired_permissions(codes)