AAP-50626 AAP-50627 Use ansible_id references in access lists (#783)

This is needed to complete the UI connection of the access lists.

Background:

Due to technical issues, we have asked the UI to use the component-based
access list for resources owned by a certain component. So if there's an
AWX resource, get the access list served from AWX for that item, like an
inventory or job template.

However, our endpoints were still using the local team or user id. This
is a problem, because all other RBAC management is centralized around
the resource server. So a translation is needed, and right now there's
no easy way to do that translation. So the solution (this) is to
formulate all the access list related endpoints around the ansible_id of
the actor, that being the user or the team.

Corresponding to this, we need a fix for the bug
#534, so that the
client has the ansible_id as a starting point when navigating to the
component endpoints

---------

Co-authored-by: John Westcott IV <[email protected]>

Author: AlanCoding

ansible_base/rbac/api/fields.py

@@ -0,0 +1,41 @@
+from django.apps import apps
+from django.core.exceptions import ObjectDoesNotExist
+from rest_framework import serializers
+
+
+class ActorAnsibleIdField(serializers.UUIDField):
+    """
+    UUID field that serializes actor objects to their ansible_id and accepts ansible_id for deserialization.
+
+    Always resolves ansible_id input to the corresponding actor object.
+    Uses the source parameter to determine which field to populate.
+    """
+
+    def to_representation(self, actor):
+        """Convert actor object to its ansible_id UUID"""
+        if actor is None:
+            return None
+        try:
+            if hasattr(actor, 'resource') and actor.resource:
+                return super().to_representation(actor.resource.ansible_id)
+        except ObjectDoesNotExist:
+            # Resource doesn't exist, return None
+            pass
+        return None
+
+    def to_internal_value(self, data):
+        """Convert ansible_id UUID to actor object"""
+        if data is None:
+            return None
+
+        # Let UUIDField handle validation and conversion
+        uuid_value = super().to_internal_value(data)
+
+        # Always resolve to actor object
+        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
+        try:
+            resource = resource_cls.objects.get(ansible_id=uuid_value)
+        except resource_cls.DoesNotExist:
+            source_name = getattr(self, 'source', 'actor')
+            raise serializers.ValidationError(f"No {source_name} found with ansible_id={uuid_value}")
+        return resource.content_object

ansible_base/rbac/api/serializers.py

@@ -1,3 +1,5 @@
+import logging
+
 from django.apps import apps
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import transaction
@@ -18,8 +20,11 @@
 
 from ..models import DABContentType, DABPermission
 from ..remote import RemoteObject
+from .fields import ActorAnsibleIdField
 from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
 
+logger = logging.getLogger(__name__)
+
 
 class RoleDefinitionSerializer(CommonModelSerializer):
     permissions = serializers.SlugRelatedField(
@@ -112,17 +117,19 @@ def get_by_ansible_id(self, ansible_id, requesting_user, for_field):
             raise ValidationError({for_field: msg.format(pk_value=ansible_id)})
         return resource.content_object
 
-    def get_actor_from_data(self, validated_data, requesting_user):
+    def validate(self, attrs):
+        """Validate that exactly one of actor or actor_ansible_id is provided"""
         actor_aid_field = f'{self.actor_field}_ansible_id'
-        if validated_data.get(self.actor_field) and validated_data.get(actor_aid_field):
+
+        # Check what was actually provided in the request
+        has_actor_in_request = self.actor_field in self.initial_data
+        has_actor_aid_in_request = actor_aid_field in self.initial_data
+
+        # If both actor and actor_ansible_id are present or both not present than we error out
+        if has_actor_in_request == has_actor_aid_in_request:
             self.raise_id_fields_error(self.actor_field, actor_aid_field)
-        elif validated_data.get(self.actor_field):
-            actor = validated_data[self.actor_field]
-        elif ansible_id := validated_data.get(actor_aid_field):
-            actor = self.get_by_ansible_id(ansible_id, requesting_user, for_field=actor_aid_field)
-        else:
-            self.raise_id_fields_error(self.actor_field, f'{self.actor_field}_ansible_id')
-        return actor
+
+        return super().validate(attrs)
 
     def get_object_from_data(self, validated_data, role_definition, requesting_user):
         obj = None
@@ -145,10 +152,11 @@ def get_object_from_data(self, validated_data, role_definition, requesting_user)
         elif validated_data.get('object_ansible_id'):
             obj = self.get_by_ansible_id(validated_data.get('object_ansible_id'), requesting_user, for_field='object_ansible_id')
             if permission_registry.content_type_model.objects.get_for_model(obj) != role_definition.content_type:
+                model_name = getattr(role_definition.content_type, 'model', 'global')
                 raise ValidationError(
                     {
                         'object_ansible_id': _('Object type of %(model_name)s does not match role type of %(role_definition)s')
-                        % {'model_name': obj._meta.model_name, 'role_definition': role_definition.content_type.model}
+                        % {'model_name': obj._meta.model_name, 'role_definition': model_name}
                     }
                 )
         return obj
@@ -158,7 +166,7 @@ def create(self, validated_data):
         requesting_user = self.context['view'].request.user
 
         # Resolve actor - team or user
-        actor = self.get_actor_from_data(validated_data, requesting_user)
+        actor = validated_data[self.actor_field]
 
         # Resolve object
         obj = self.get_object_from_data(validated_data, rd, requesting_user)
@@ -216,7 +224,8 @@ def _get_summary_fields(self, obj) -> dict[str, dict]:
 
 class RoleUserAssignmentSerializer(BaseAssignmentSerializer):
     actor_field = 'user'
-    user_ansible_id = serializers.UUIDField(
+    user_ansible_id = ActorAnsibleIdField(
+        source='user',
         required=False,
         help_text=_('The resource ID of the user who will receive permissions from this assignment. An alternative to user field.'),
         allow_null=True,  # for ease of use of the browseable API
@@ -232,7 +241,8 @@ def get_actor_queryset(self, requesting_user):
 
 class RoleTeamAssignmentSerializer(BaseAssignmentSerializer):
     actor_field = 'team'
-    team_ansible_id = serializers.UUIDField(
+    team_ansible_id = ActorAnsibleIdField(
+        source='team',
         required=False,
         help_text=_('The resource ID of the team who will receive permissions from this assignment. An alternative to team field.'),
         allow_null=True,
@@ -257,9 +267,21 @@ def _get_related(self, obj) -> dict[str, str]:
             return {}
         related_fields = {}
         actor_cls = self.Meta.model
+
+        # Use ansible_id if available, otherwise fall back to pk
+        actor_identifier = obj.pk
+        try:
+            if hasattr(obj, 'resource') and obj.resource:
+                actor_identifier = str(obj.resource.ansible_id)
+        except ObjectDoesNotExist:
+            # Resource doesn't exist, stick with pk
+            logger.warning(
+                f"No resource for {self.Meta.model} {obj.pk} due to internal error. Linking role-{actor_cls._meta.model_name}-access-assignments as pk."
+            )
+
         related_fields['details'] = get_relative_url(
             f'role-{actor_cls._meta.model_name}-access-assignments',
-            kwargs={'model_name': self.context.get("content_type").api_slug, 'pk': self.context.get("related_object").pk, 'actor_pk': obj.pk},
+            kwargs={'model_name': self.context.get("content_type").api_slug, 'pk': self.context.get("related_object").pk, 'actor_pk': actor_identifier},
         )
         return related_fields
 

ansible_base/rbac/api/views.py

@@ -1,6 +1,9 @@
+import uuid
 from collections import OrderedDict
 from typing import Type
 
+from django.apps import apps
+from django.core.exceptions import ObjectDoesNotExist
 from django.db import transaction
 from django.db.models import Model
 from django.utils.translation import gettext_lazy as _
@@ -379,6 +382,28 @@ class UserAccessAssignmentViewSet(
     def get_url_actor(self):
         actor_pk = self.kwargs.get("actor_pk")
         actor_cls = self.get_actor_model()
+
+        # First, try to parse as UUID for ansible_id lookup
+        try:
+            parsed_uuid = uuid.UUID(actor_pk)
+            # It's a valid UUID, try resource lookup first
+            try:
+                resource_cls = apps.get_model('dab_resource_registry', 'Resource')
+                resource = resource_cls.objects.get(ansible_id=parsed_uuid)
+                actor = resource.content_object
+                # Verify the content object is the correct type
+                if isinstance(actor, actor_cls):
+                    return actor
+                else:
+                    raise NotFound(f'Resource with ansible_id {parsed_uuid} is not a {actor_cls._meta.model_name}')
+            except (LookupError, ObjectDoesNotExist):
+                # Resource registry not available or resource not found with this UUID
+                raise NotFound(f'The {actor_cls._meta.model_name} with ansible_id={actor_pk} can not be found')
+        except ValueError:
+            # Not a valid UUID, continue with primary key lookup
+            pass
+
+        # Fallback to primary key lookup (only for non-UUID values)
         try:
             return actor_cls.objects.get(pk=actor_pk)
         except actor_cls.DoesNotExist:

ansible_base/rbac/service_api/serializers.py

@@ -4,6 +4,7 @@
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 
+from ..api.fields import ActorAnsibleIdField
 from ..models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ..remote import RemoteObject
 
@@ -24,25 +25,6 @@ class Meta:
         fields = ['api_slug', 'codename', 'content_type', 'name']
 
 
-class ActorAnsibleIDField(serializers.Field):
-    def to_representation(self, actor):
-        if actor is None:
-            return None
-        resource = actor.resource
-        if resource is None:
-            return None
-        return str(resource.ansible_id)
-
-    def to_internal_value(self, data):
-        resource_cls = apps.get_model('dab_resource_registry', 'Resource')
-        try:
-            resource = resource_cls.objects.get(ansible_id=data)
-        except resource_cls.DoesNotExist:
-            raise serializers.ValidationError(f"No {self.source} found with ansible_id={data}")
-
-        return resource.content_object
-
-
 class ObjectIDAnsibleIDField(serializers.Field):
     "This is an ansible_id field intended to be used with source pointing to object_id, so, does conversion"
 
@@ -71,7 +53,7 @@ def to_internal_value(self, value):
 class BaseAssignmentSerializer(serializers.ModelSerializer):
     content_type = serializers.SlugRelatedField(read_only=True, slug_field='api_slug')
     role_definition = serializers.SlugRelatedField(slug_field='name', queryset=RoleDefinition.objects.all())
-    created_by_ansible_id = ActorAnsibleIDField(source='created_by', required=False)
+    created_by_ansible_id = ActorAnsibleIdField(source='created_by', required=False)
     object_ansible_id = ObjectIDAnsibleIDField(source='object_id', required=False, allow_null=True)
     object_id = serializers.CharField(allow_blank=True, required=False, allow_null=True)
     from_service = serializers.CharField(write_only=True)
@@ -158,7 +140,7 @@ def create(self, validated_data):
 
 
 class ServiceRoleUserAssignmentSerializer(BaseAssignmentSerializer):
-    user_ansible_id = ActorAnsibleIDField(source='user', required=True)
+    user_ansible_id = ActorAnsibleIdField(source='user', required=True)
     actor_field = 'user'
 
     class Meta:
@@ -167,7 +149,7 @@ class Meta:
 
 
 class ServiceRoleTeamAssignmentSerializer(BaseAssignmentSerializer):
-    team_ansible_id = ActorAnsibleIDField(source='team', required=True)
+    team_ansible_id = ActorAnsibleIdField(source='team', required=True)
     actor_field = 'team'
 
     class Meta:

test_app/management/commands/create_demo_data.py

@@ -7,6 +7,7 @@
 
 from ansible_base.authentication.models import Authenticator, AuthenticatorUser
 from ansible_base.oauth2_provider.models import OAuth2Application
+from ansible_base.rbac import permission_registry
 from ansible_base.rbac.models import DABContentType, RoleDefinition
 from test_app.models import EncryptionModel, InstanceGroup, Inventory, Organization, Team, User
 
@@ -41,6 +42,7 @@ def handle(self, *args, **kwargs):
         (galaxy, _) = Organization.objects.get_or_create(name='Galaxy_community')
 
         (spud, _) = User.objects.get_or_create(username='angry_spud')
+        (team_member, _) = User.objects.get_or_create(username='team_member')
         (bull_bot, _) = User.objects.get_or_create(username='ansibullbot')
         (admin, _) = User.objects.get_or_create(username='admin')
         spud.set_password('password')
@@ -72,8 +74,8 @@ def handle(self, *args, **kwargs):
 
             # Inventory objects exist inside of an organization
             Inventory.objects.create(name='K8S clusters', organization=operator_stuff)
-            Inventory.objects.create(name='Galaxy Host', organization=galaxy)
-            Inventory.objects.create(name='AWX deployment', organization=awx)
+            galaxy_inv = Inventory.objects.create(name='Galaxy Host', organization=galaxy)
+            awx_inv = Inventory.objects.create(name='AWX deployment', organization=awx)
             # Objects that have no associated organization
             InstanceGroup.objects.create(name='Default')
             isolated_group = InstanceGroup.objects.create(name='Isolated Network')
@@ -95,7 +97,18 @@ def handle(self, *args, **kwargs):
             user.set_password('password')
             user.save()
 
-        RoleDefinition.objects.managed.team_member.give_permission(spud, awx_devs)
+        # Give some users team member and give that team some inventory object permissions
+        for user in (spud, team_member):
+            RoleDefinition.objects.managed.team_member.give_permission(spud, awx_devs)
+
+        with impersonate(bull_bot):
+            inv_admin, _ = RoleDefinition.objects.get_or_create(
+                name='Inventory Admin',
+                permissions=['change_inventory', 'view_inventory'],
+                defaults={'content_type': permission_registry.content_type_model.objects.get_for_model(Inventory)},
+            )
+        for inv in (awx_inv, galaxy_inv):
+            inv_admin.give_permission(awx_devs, inv)
 
         OAuth2Application.objects.get_or_create(
             name="Demo OAuth2 Application",