refactor(rbac): drop runtime legacy-format resolution from engine
Roles are stored in the new resource:action format and any legacy keys are converted once at startup by the RBAC v2 DB migration (migrate_rbac_v2), so the permission engine no longer needs to resolve legacy codes on every check. - get_user_permissions: store role permission codes as-is (no dual add). - has_permission: match new-format codes only; drop resolve_permission and the dotted-code fallbacks. - Remove now-unused resolve_permission import in middleware. - LEGACY_PERMISSION_MAP / convert_* are retained as one-shot DB-migration utilities (upgrade path), not a runtime bridge. - Tests: adapt fixtures to new-format role permissions; assert legacy keys are no longer honored at request time. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -9,12 +9,10 @@ from fastapi import HTTPException, status
|
||||
from loguru import logger
|
||||
|
||||
from .permissions import (
|
||||
Permission,
|
||||
PermissionRegistry,
|
||||
PERMISSIONS,
|
||||
Permission,
|
||||
PermissionRegistry,
|
||||
PERMISSIONS,
|
||||
ScopeType,
|
||||
resolve_permission,
|
||||
LEGACY_PERMISSION_MAP
|
||||
)
|
||||
|
||||
|
||||
@@ -36,19 +34,14 @@ class PermissionEngine:
|
||||
if getattr(user, 'is_admin', False):
|
||||
return {p.code for p in PERMISSIONS.all()}
|
||||
|
||||
# Get permissions from role
|
||||
# Get permissions from role. Roles are stored in the new
|
||||
# ``resource:action`` format (legacy keys are converted once at startup
|
||||
# by the RBAC v2 DB migration), so no runtime resolution is needed.
|
||||
if user.role and user.role.permissions:
|
||||
role_perms = user.role.permissions
|
||||
for perm_code, enabled in role_perms.items():
|
||||
for perm_code, enabled in user.role.permissions.items():
|
||||
if enabled:
|
||||
# Resolve legacy to new format
|
||||
resolved = resolve_permission(perm_code)
|
||||
permissions.add(resolved)
|
||||
|
||||
# Also keep legacy for backwards compatibility
|
||||
if perm_code != resolved:
|
||||
permissions.add(perm_code)
|
||||
|
||||
permissions.add(perm_code)
|
||||
|
||||
return permissions
|
||||
|
||||
@staticmethod
|
||||
@@ -80,29 +73,26 @@ class PermissionEngine:
|
||||
|
||||
# Get user permissions
|
||||
user_permissions = PermissionEngine.get_user_permissions(user)
|
||||
|
||||
# Resolve permission code
|
||||
resolved_perm = resolve_permission(permission)
|
||||
|
||||
# Check direct permission
|
||||
if resolved_perm in user_permissions or permission in user_permissions:
|
||||
|
||||
# Check direct permission (new resource:action format)
|
||||
if permission in user_permissions:
|
||||
return True
|
||||
|
||||
|
||||
# Check wildcard permissions (e.g., vm:* grants all vm permissions)
|
||||
resource = resolved_perm.split(':')[0] if ':' in resolved_perm else permission.split('.')[0]
|
||||
resource = permission.split(':')[0]
|
||||
wildcard = f"{resource}:*"
|
||||
if wildcard in user_permissions:
|
||||
return True
|
||||
|
||||
|
||||
# Check manage permission (implies most actions)
|
||||
manage_perm = f"{resource}:manage"
|
||||
if manage_perm in user_permissions:
|
||||
# manage implies view, update, and some actions
|
||||
action = resolved_perm.split(':')[1] if ':' in resolved_perm else None
|
||||
action = permission.split(':')[1] if ':' in permission else None
|
||||
implied_actions = {'view', 'list', 'update', 'start', 'stop', 'restart'}
|
||||
if action in implied_actions:
|
||||
return True
|
||||
|
||||
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -8,7 +8,7 @@ from fastapi import Depends, HTTPException, status, Request
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .engine import PermissionEngine
|
||||
from .permissions import ScopeType, resolve_permission
|
||||
from .permissions import ScopeType
|
||||
|
||||
|
||||
class PermissionChecker:
|
||||
|
||||
@@ -96,7 +96,7 @@ def clean_database():
|
||||
user_role = Role(
|
||||
name="user",
|
||||
display_name="User",
|
||||
permissions={"proxmox.view": True, "vms.view": True},
|
||||
permissions={"server:view": True, "vm:view": True},
|
||||
is_system=True
|
||||
)
|
||||
db.add_all([admin_role, user_role])
|
||||
|
||||
@@ -62,9 +62,10 @@ class TestAPIVirtualMachines:
|
||||
# First restrict user's permissions to 'vms.view.own'
|
||||
user_role = seed_users["user"].role
|
||||
old_perms = dict(user_role.permissions)
|
||||
# We need both 'vms.view' (to pass the endpoint route dependency)
|
||||
# and 'vms.view.own' (to fail the `require_vm_access` owner check).
|
||||
user_role.permissions = {"proxmox.view": True, "vms.view": True, "vms.view.own": True}
|
||||
# We need both 'vm:view' (to pass the endpoint route dependency)
|
||||
# and the 'vms.view.own' ownership flag (to fail the `require_vm_access`
|
||||
# owner check, which reads that key directly).
|
||||
user_role.permissions = {"server:view": True, "vm:view": True, "vms.view.own": True}
|
||||
db_session.commit()
|
||||
|
||||
response = client.get(
|
||||
|
||||
@@ -47,9 +47,15 @@ class TestEngineHasPermission:
|
||||
assert rbac.PermissionEngine.has_permission(user, "vm:view") is True
|
||||
assert rbac.PermissionEngine.has_permission(user, "vm:delete") is False
|
||||
|
||||
def test_legacy_grant_resolves_to_new(self, rbac, fake_user_factory):
|
||||
def test_legacy_keys_not_resolved_at_runtime(self, rbac, fake_user_factory):
|
||||
# Roles are migrated to the new ``resource:action`` format once at
|
||||
# startup (RBAC v2 DB migration). The engine no longer resolves legacy
|
||||
# dot-format keys at request time, so a role still holding a legacy key
|
||||
# grants nothing until it has been migrated.
|
||||
user = fake_user_factory(permissions={"vms.create": True})
|
||||
assert rbac.PermissionEngine.has_permission(user, "vm:create") is True
|
||||
assert rbac.PermissionEngine.has_permission(user, "vm:create") is False
|
||||
# The migration utility is what bridges the formats, not the engine.
|
||||
assert rbac.resolve_permission("vms.create") == "vm:create"
|
||||
|
||||
def test_wildcard_grants_all_actions_on_resource(self, rbac, fake_user_factory):
|
||||
user = fake_user_factory(permissions={"vm:*": True})
|
||||
|
||||
Reference in New Issue
Block a user