From 2ac80e887e82e74e70a3c26388e16143dbcb80da Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 29 Sep 2025 13:41:46 -0400 Subject: [PATCH 1/2] Implement my review comments --- ansible_base/authentication/utils/claims.py | 10 ++++------ ansible_base/lib/utils/apps.py | 4 ++++ .../models/service_identifier.py | 8 +------- ansible_base/resource_registry/rest_client.py | 7 ++----- ansible_base/resource_registry/shared_types.py | 6 ++---- ansible_base/resource_registry/tasks/sync.py | 17 +++++++---------- 6 files changed, 20 insertions(+), 32 deletions(-) create mode 100644 ansible_base/lib/utils/apps.py diff --git a/ansible_base/authentication/utils/claims.py b/ansible_base/authentication/utils/claims.py index 1690b582b..92aea3d3b 100644 --- a/ansible_base/authentication/utils/claims.py +++ b/ansible_base/authentication/utils/claims.py @@ -19,6 +19,7 @@ from ansible_base.authentication.models import Authenticator, AuthenticatorMap, AuthenticatorUser from ansible_base.authentication.utils.authenticator_map import check_role_type, expand_syntax from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.lib.utils.auth import get_organization_model, get_team_model from ansible_base.lib.utils.string import is_empty @@ -30,9 +31,6 @@ User = get_user_model() -is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS - - class TriggerResult(Enum): ALLOW = auto() DENY = auto() @@ -723,7 +721,7 @@ def reconcile_user_claims(cls, user: AbstractUser, authenticator_user: Authentic claims = getattr(user, 'claims', authenticator_user.claims) - if is_rbac_installed: + 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")) @@ -878,7 +876,7 @@ def __init__(self): self.cache = {} # NOTE(cutwater): We may probably execute this query once and cache the query results. self.content_types = {} - if is_rbac_installed: + 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()} @@ -962,7 +960,7 @@ def cache_existing(self, role_assignments: Iterable[models.Model]) -> None: - Role definitions are also cached separately in self.role_definitions """ local_resource_prefixes = ["shared"] - if is_rbac_installed: + if is_rbac_installed(): from ansible_base.rbac.remote import get_local_resource_prefix local_resource_prefixes.append(get_local_resource_prefix()) diff --git a/ansible_base/lib/utils/apps.py b/ansible_base/lib/utils/apps.py new file mode 100644 index 000000000..1ebf44401 --- /dev/null +++ b/ansible_base/lib/utils/apps.py @@ -0,0 +1,4 @@ +def is_rbac_installed() -> bool: + from django.conf import settings + + return bool('ansible_base.rbac' in settings.INSTALLED_APPS) diff --git a/ansible_base/resource_registry/models/service_identifier.py b/ansible_base/resource_registry/models/service_identifier.py index 35d768063..739499f9e 100644 --- a/ansible_base/resource_registry/models/service_identifier.py +++ b/ansible_base/resource_registry/models/service_identifier.py @@ -23,11 +23,5 @@ def save(self, *args, **kwargs): def service_id(): global _service_id if not _service_id: - service_obj = ServiceID.objects.first() - if service_obj: - _service_id = str(service_obj.pk) - else: - # Create a ServiceID if none exists - service_obj = ServiceID.objects.create() - _service_id = str(service_obj.pk) + _service_id = str(ServiceID.objects.first().pk) return _service_id diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index e45130f79..d8024d1bd 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -8,12 +8,13 @@ from django.apps import apps from django.conf import settings +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.resource_server import get_resource_server_config, get_service_token def _check_rbac_installed(): """Check if ansible_base.rbac is installed and raise RuntimeError if not.""" - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + if is_rbac_installed(): raise RuntimeError("This operation requires ansible_base.rbac to be installed") @@ -174,16 +175,13 @@ def get_resource_type_manifest(self, name, filters: Optional[dict] = None): # RBAC related methods def list_role_types(self, filters: Optional[dict] = None): - _check_rbac_installed() return self._make_request("get", "role-types/", params=filters) def list_role_permissions(self, filters: Optional[dict] = None): - _check_rbac_installed() return self._make_request("get", "role-permissions/", params=filters) def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List user role assignments.""" - _check_rbac_installed() params = (filters or {}).copy() if user_ansible_id is not None: params['user_ansible_id'] = user_ansible_id @@ -191,7 +189,6 @@ def list_user_assignments(self, user_ansible_id: Optional[str] = None, filters: def list_team_assignments(self, team_ansible_id: Optional[str] = None, filters: Optional[dict] = None): """List team role assignments.""" - _check_rbac_installed() params = (filters or {}).copy() if team_ansible_id is not None: params['team_ansible_id'] = team_ansible_id diff --git a/ansible_base/resource_registry/shared_types.py b/ansible_base/resource_registry/shared_types.py index 66cad375c..bcef8956f 100644 --- a/ansible_base/resource_registry/shared_types.py +++ b/ansible_base/resource_registry/shared_types.py @@ -1,7 +1,7 @@ -from django.conf import settings from rest_framework import serializers from rest_framework.exceptions import ValidationError +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.utils.resource_type_serializers import AnsibleResourceForeignKeyField, SharedResourceTypeSerializer from ansible_base.resource_registry.utils.sso_provider import get_sso_provider_server @@ -84,8 +84,6 @@ class LenientPermissionSlugListField(serializers.ListField): child = serializers.CharField() def to_internal_value(self, data): - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: - raise RuntimeError("LenientPermissionSlugListField requires ansible_base.rbac to be installed") from ansible_base.rbac.models import DABPermission data = super().to_internal_value(data) @@ -105,7 +103,7 @@ class RoleDefinitionType(SharedResourceTypeSerializer): permissions = LenientPermissionSlugListField() def __init__(self, *args, **kwargs): - if 'ansible_base.rbac' not in settings.INSTALLED_APPS: + if not is_rbac_installed(): raise RuntimeError("RoleDefinitionType requires ansible_base.rbac to be installed") super().__init__(*args, **kwargs) diff --git a/ansible_base/resource_registry/tasks/sync.py b/ansible_base/resource_registry/tasks/sync.py index 7a099a392..4e2d60741 100644 --- a/ansible_base/resource_registry/tasks/sync.py +++ b/ansible_base/resource_registry/tasks/sync.py @@ -18,6 +18,7 @@ from django.db.utils import Error, IntegrityError from requests import HTTPError +from ansible_base.lib.utils.apps import is_rbac_installed from ansible_base.resource_registry.models import Resource, ResourceType from ansible_base.resource_registry.models.service_identifier import service_id from ansible_base.resource_registry.registry import get_registry @@ -25,8 +26,6 @@ logger = logging.getLogger('ansible_base.resources_api.tasks.sync') -_is_rbac_installed = 'ansible_base.rbac' in settings.INSTALLED_APPS - class ManifestNotFound(HTTPError): """Raise when server returns 404 for a manifest""" @@ -141,7 +140,7 @@ def fetch_manifest( def get_ansible_id_or_pk(assignment) -> str: - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_ansible_id_or_pk requires ansible_base.rbac to be installed") # For object-scoped assignments, try to get the object's ansible_id if assignment.content_type.model in ('organization', 'team'): @@ -157,7 +156,7 @@ def get_ansible_id_or_pk(assignment) -> str: def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> Any: - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_content_object requires ansible_base.rbac to be installed") content_object = None if role_definition.content_type.model in ('organization', 'team'): @@ -172,8 +171,6 @@ def get_content_object(role_definition, assignment_tuple: AssignmentTuple) -> An def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple]: """Fetch remote assignments from the resource server and convert to tuples.""" - if not _is_rbac_installed: - raise RuntimeError("get_remote_assignments requires ansible_base.rbac to be installed") assignments = set() # Fetch user assignments with pagination @@ -245,7 +242,7 @@ def get_remote_assignments(api_client: ResourceAPIClient) -> set[AssignmentTuple def get_local_assignments() -> set[AssignmentTuple]: """Get local assignments and convert to tuples.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("get_local_assignments requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleTeamAssignment, RoleUserAssignment @@ -305,7 +302,7 @@ def get_local_assignments() -> set[AssignmentTuple]: def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Delete a local assignment based on the tuple.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("delete_local_assignment requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleDefinition @@ -335,7 +332,7 @@ def delete_local_assignment(assignment_tuple: AssignmentTuple) -> bool: def create_local_assignment(assignment_tuple: AssignmentTuple) -> bool: """Create a local assignment based on the tuple.""" - if not _is_rbac_installed: + if not is_rbac_installed(): raise RuntimeError("create_local_assignment requires ansible_base.rbac to be installed") from ansible_base.rbac.models.role import RoleDefinition @@ -713,7 +710,7 @@ def _sync_assignments(self): if not self.sync_assignments: return - if not _is_rbac_installed: + if not is_rbac_installed(): self.write(">>> Skipping role assignments sync (rbac not installed)") return From aabec8c7cd2af13a9ef13c235197b445bd4905f3 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Mon, 29 Sep 2025 15:30:36 -0400 Subject: [PATCH 2/2] Add back missing negative --- ansible_base/resource_registry/rest_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ansible_base/resource_registry/rest_client.py b/ansible_base/resource_registry/rest_client.py index d8024d1bd..0dbaa74f7 100644 --- a/ansible_base/resource_registry/rest_client.py +++ b/ansible_base/resource_registry/rest_client.py @@ -14,7 +14,7 @@ def _check_rbac_installed(): """Check if ansible_base.rbac is installed and raise RuntimeError if not.""" - if is_rbac_installed(): + if not is_rbac_installed(): raise RuntimeError("This operation requires ansible_base.rbac to be installed")