[AAP-45443] Allow filtering by ansible_id (#734)

### What is being changed?
Allows API query filtering by ansible id
`user_ansible_id`
`team_ansible_id`
`object_ansible_id`

Note, filtering with object_ansible_id accrues an additional database
query for the Resource model.

Example API calls

```
/v1/role_user_assignments/?user_ansible_id=<uuid>
/v1/role_user_assignments/?object_ansible_id=<uuid>
/v1/role_team_assignments/?team_ansible_id=<uuid>
```
query params can be combined where appropriate

#### Validation

Adds some validation if the provided ansible_id is not a UUID
```
[
    "Invalid UUID format for user_ansible_id: 80c7e291-b121-48fc-"
]
```

### Why is this change needed?
Currently there is no way to filter on these fields, so working with the
API becomes cumbersome and inefficient

### How does this change address the issue?
Adds a django viewset filter backend to alias the query param to the
appropriate queryset call.
e.g.
`user_ansible_id` applies a `user__resource__ansible_id` query

## Type of Change
- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Documentation update
- [x] Test update
- [ ] Refactoring (no functional changes)
- [ ] Development environment change
- [ ] Configuration change

## Self-Review Checklist
- [x] I have performed a self-review of my code
- [x] I have added relevant comments to complex code sections
- [ ] I have updated documentation where needed
- [x] I have considered the security impact of these changes
- [x] I have considered performance implications
- [x] I have thought about error handling and edge cases
- [x] I have tested the changes in my local environment

---------

Signed-off-by: Seth Foster <[email protected]>

Author: fosterseth

ansible_base/rbac/api/filter_backends.py

@@ -0,0 +1,83 @@
+import uuid
+
+from rest_framework.exceptions import ValidationError
+from rest_framework.filters import BaseFilterBackend
+
+from ansible_base.resource_registry.models import Resource
+
+
+class AnsibleIdAliasFilterBackend(BaseFilterBackend):
+    '''
+    Filter backend for object_ansible_id.
+    Note that this accrues an additional query to the Resource model.
+
+    Example:
+    /api/v1/role_user_assignments/?object_ansible_id=da0488f5-013b-460c-8a62-c3c10a1d0fad
+    '''
+
+    def filter_queryset(self, request, queryset, view):
+        object_ansible_id = request.query_params.get('object_ansible_id')
+        if object_ansible_id:
+            try:
+                # Validate if the provided ansible_id is a valid UUID
+                uuid.UUID(object_ansible_id)
+            except ValueError:
+                raise ValidationError(f"Invalid UUID format for object_ansible_id: {object_ansible_id}")
+
+            try:
+                # Find the Resource object by its ansible_id
+                resource_obj = Resource.objects.get(ansible_id=object_ansible_id)
+
+                # Filter the queryset based on the resource's content_type and object_id
+                queryset = queryset.filter(
+                    object_role__content_type=resource_obj.content_type, object_role__object_id=str(resource_obj.object_id)  # Ensure object_id is string
+                )
+            except Resource.DoesNotExist:
+                # If the resource is not found, return an empty queryset
+                return queryset.none()
+
+        return queryset
+
+
+class UserAnsibleIdAliasFilterBackend(AnsibleIdAliasFilterBackend):
+    """
+    Filter backend for user_ansible_id and object_ansible_id.
+
+    Example:
+    /api/v1/role_user_assignments/?user_ansible_id=80c7e291-b121-48fc-8fb1-174aac6f57a6
+    /api/v1/role_user_assignments/?object_ansible_id=da0488f5-013b-460c-8a62-c3c10a1d0fad
+    """
+
+    def filter_queryset(self, request, queryset, view):
+        user_ansible_id = request.query_params.get('user_ansible_id')
+        if user_ansible_id:
+            try:
+                # Validate if the provided ansible_id is a valid UUID
+                uuid.UUID(user_ansible_id)
+            except ValueError:
+                raise ValidationError(f"Invalid UUID format for user_ansible_id: {user_ansible_id}")
+            # Filter the queryset based on the user's ansible_id
+            queryset = queryset.filter(user__resource__ansible_id=user_ansible_id)
+        return super().filter_queryset(request, queryset, view)
+
+
+class TeamAnsibleIdAliasFilterBackend(AnsibleIdAliasFilterBackend):
+    """
+    Filter backend for team_ansible_id and object_ansible_id.
+
+    Example:
+    /api/v1/role_team_assignments/?team_ansible_id=c2b59b42-a874-43ca-9e1f-abe410864f65
+    /api/v1/role_team_assignments/?object_ansible_id=da0488f5-013b-460c-8a62-c3c10a1d0fad
+    """
+
+    def filter_queryset(self, request, queryset, view):
+        team_ansible_id = request.query_params.get('team_ansible_id')
+        if team_ansible_id:
+            try:
+                # Validate if the provided ansible_id is a valid UUID
+                uuid.UUID(team_ansible_id)
+            except ValueError:
+                raise ValidationError(f"Invalid UUID format for team_ansible_id: {team_ansible_id}")
+            # Filter the queryset based on the team's ansible_id
+            queryset = queryset.filter(team__resource__ansible_id=team_ansible_id)
+        return super().filter_queryset(request, queryset, view)

ansible_base/rbac/api/views.py

@@ -12,6 +12,7 @@
 
 from ansible_base.lib.utils.views.django_app_api import AnsibleBaseDjangoAppApiView
 from ansible_base.lib.utils.views.permissions import try_add_oauth2_scope_permission
+from ansible_base.rbac.api.filter_backends import TeamAnsibleIdAliasFilterBackend, UserAnsibleIdAliasFilterBackend
 from ansible_base.rbac.api.permissions import RoleDefinitionPermissions
 from ansible_base.rbac.api.serializers import (
     RoleDefinitionDetailSerializer,
@@ -159,6 +160,7 @@ class RoleTeamAssignmentViewSet(BaseAssignmentViewSet):
 
     serializer_class = RoleTeamAssignmentSerializer
     prefetch_related = ('team',)
+    filter_backends = (TeamAnsibleIdAliasFilterBackend,)
 
 
 class RoleUserAssignmentViewSet(BaseAssignmentViewSet):
@@ -175,3 +177,4 @@ class RoleUserAssignmentViewSet(BaseAssignmentViewSet):
 
     serializer_class = RoleUserAssignmentSerializer
     prefetch_related = ('user',)
+    filter_backends = (UserAnsibleIdAliasFilterBackend,)

test_app/tests/rbac/test_ansible_id_alias_filter.py

@@ -0,0 +1,82 @@
+import pytest
+from django.contrib.contenttypes.models import ContentType
+from django.utils.http import urlencode
+
+from ansible_base.lib.utils.response import get_relative_url
+from ansible_base.resource_registry.models import Resource
+
+
+class TestAnsibleIdAliasFilterBackend:
+
+    @pytest.mark.django_db
+    def test_filter_user_ansible_id(self, admin_api_client, org_inv_rd, inv_rd, inventory, rando, organization):
+        '''
+        Test filtering RoleUserAssignment by user_ansible_id and object_ansible_id.
+        '''
+        # user - org assigment
+        user_resource = Resource.objects.get(object_id=rando.pk, content_type=ContentType.objects.get_for_model(rando).pk)
+        organization_resource = Resource.objects.get(object_id=organization.pk, content_type=ContentType.objects.get_for_model(organization).pk)
+        url = get_relative_url('roleuserassignment-list')
+        data = dict(role_definition=org_inv_rd.id, content_type='shared.organization', user_ansible_id=user_resource.ansible_id, object_id=organization.id)
+        response = admin_api_client.post(url, data=data, format="json")
+        assert response.status_code == 201, response.data
+
+        # user - inventory assignment (just a random assignment to make total count > 1)
+        data = dict(role_definition=inv_rd.id, content_type='shared.inventory', user_ansible_id=user_resource.ansible_id, object_id=inventory.id)
+        response = admin_api_client.post(url, data=data, format="json")
+        assert response.status_code == 201, response.data
+
+        # make sure > 1 assignments total to ensure filtering is not returning undesired results
+        response = admin_api_client.get(url)
+        assert response.data["count"] > 1, response.data
+
+        # filter by user_ansible_id
+        query_params = {'user_ansible_id': user_resource.ansible_id}
+        response = admin_api_client.get(url + '?' + urlencode(query_params))
+        assert response.status_code == 200, response.data
+        assert response.data["count"] == 2, response.data
+
+        # filter by object_ansible_id
+        query_params = {'object_ansible_id': organization_resource.ansible_id}
+        response = admin_api_client.get(url + '?' + urlencode(query_params))
+        assert response.status_code == 200, response.data
+        assert response.data["count"] == 1, response.data
+
+        # filter by both user_ansible_id and object_ansible_id
+        query_params = {'user_ansible_id': user_resource.ansible_id, 'object_ansible_id': organization_resource.ansible_id}
+        response = admin_api_client.get(url + '?' + urlencode(query_params))
+        assert response.status_code == 200, response.data
+        assert response.data["count"] == 1, response.data
+
+    @pytest.mark.django_db
+    def test_filter_team_ansible_id(self, admin_api_client, team, inv_rd, inventory):
+        '''
+        Test filtering RoleTeamAssignment by team_ansible_id.
+        '''
+        team_resource = Resource.objects.get(object_id=team.pk, content_type=ContentType.objects.get_for_model(team).pk)
+        url = get_relative_url('roleteamassignment-list')
+        data = dict(role_definition=inv_rd.id, content_type='shared.organization', team_ansible_id=team_resource.ansible_id, object_id=inventory.id)
+        response = admin_api_client.post(url, data=data, format="json")
+        assert response.status_code == 201, response.data
+
+        # filter by team_ansible_id
+        query_params = {'team_ansible_id': team_resource.ansible_id}
+        response = admin_api_client.get(url + '?' + urlencode(query_params))
+        assert response.status_code == 200, response.data
+        assert response.data["count"] == 1, response.data
+
+    @pytest.mark.parametrize("ansible_id_type", ["user", "team", "object"])
+    @pytest.mark.django_db
+    def test_invalid_ansible_id_format(self, admin_api_client, ansible_id_type):
+        '''
+        Test that invalid UUID formats for ansible_id raise a 400 error.
+        '''
+        if ansible_id_type == "team":
+            actor = "team"
+        else:
+            actor = "user"
+        url = get_relative_url(f'role{actor}assignment-list')
+        query_params = {f'{ansible_id_type}_ansible_id': 'invalid-uuid-format'}
+        response = admin_api_client.get(url + '?' + urlencode(query_params))
+        assert response.status_code == 400, response.data
+        assert "Invalid UUID format for" in str(response.data), response.data