Decoupling apps from ansible_base.rbac

Author: john-westcott-iv

ansible_base/authentication/utils/claims.py

@@ -21,8 +21,6 @@
 from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel
 from ansible_base.lib.utils.auth import get_organization_model, get_team_model
 from ansible_base.lib.utils.string import is_empty
-from ansible_base.rbac.models import DABContentType
-from ansible_base.rbac.remote import get_local_resource_prefix
 
 from .trigger_definition import TRIGGER_DEFINITION
 
@@ -32,6 +30,9 @@
 User = get_user_model()
 
 
+is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS
+
+
 class TriggerResult(Enum):
     ALLOW = auto()
     DENY = auto()
@@ -722,7 +723,7 @@ def reconcile_user_claims(cls, user: AbstractUser, authenticator_user: Authentic
 
         claims = getattr(user, 'claims', authenticator_user.claims)
 
-        if 'ansible_base.rbac' in settings.INSTALLED_APPS:
+        if is_rbac_installed:
             cls(claims, user, authenticator_user).manage_permissions()
         else:
             logger.info(_("Skipping user claims with RBAC roles, because RBAC app is not installed"))
@@ -876,7 +877,11 @@ class RoleUserAssignmentsCache:
     def __init__(self):
         self.cache = {}
         # NOTE(cutwater): We may probably execute this query once and cache the query results.
-        self.content_types = {content_type.model: content_type for content_type in DABContentType.objects.get_for_models(Organization, Team).values()}
+        self.content_types = {}
+        if is_rbac_installed:
+            from ansible_base.rbac.models import DABContentType
+
+            self.content_types = {content_type.model: content_type for content_type in DABContentType.objects.get_for_models(Organization, Team).values()}
         self.role_definitions = {}
 
     def items(self):
@@ -956,6 +961,12 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None:
             - All cached assignments are marked with STATUS_EXISTING status
             - Role definitions are also cached separately in self.role_definitions
         """
+        local_resource_prefixes = ["shared"]
+        if is_rbac_installed:
+            from ansible_base.rbac.remote import get_local_resource_prefix
+
+            local_resource_prefixes.append(get_local_resource_prefix())
+
         for role_assignment in role_assignments:
             # Cache role definition
             if (role_definition := self._rd_by_id(role_assignment)) is None:
@@ -965,7 +976,7 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None:
             # Skip role assignments that should not be cached
             if not (
                 role_assignment.content_type is None  # Global/system roles (e.g., System Auditor)
-                or role_assignment.content_type.service in [get_local_resource_prefix(), "shared"]
+                or role_assignment.content_type.service in local_resource_prefixes
             ):  # Local object roles
                 continue
 

ansible_base/lib/routers/association_resource_router.py

@@ -17,8 +17,6 @@
 from rest_framework.response import Response
 from rest_framework.viewsets import ViewSetMixin
 
-from ansible_base.rbac.permission_registry import permission_registry
-
 logger = logging.getLogger('ansible_base.lib.routers.association_resource_router')
 
 
@@ -119,10 +117,13 @@ def check_parent_object_permissions(self, request, parent_obj: Model) -> None:
         will not check "change" permissions to the parent object on POST
         this method checks parent change permission, view permission should be handled by filter_queryset
         """
-        if (request.method not in SAFE_METHODS) and 'ansible_base.rbac' in settings.INSTALLED_APPS and permission_registry.is_registered(parent_obj):
-            from ansible_base.rbac.policies import check_content_obj_permission
+        if (request.method not in SAFE_METHODS) and 'ansible_base.rbac' in settings.INSTALLED_APPS:
+            from ansible_base.rbac.permission_registry import permission_registry
+
+            if permission_registry.is_registered(parent_obj):
+                from ansible_base.rbac.policies import check_content_obj_permission
 
-            check_content_obj_permission(request.user, parent_obj)
+                check_content_obj_permission(request.user, parent_obj)
 
     def get_parent_object(self) -> Model:
         """Modeled mostly after DRF get_object, but for the parent model