[AAP-48392] Models and APIs for tracking remote permissions in DAB RBAC (#749)

## Description
This modifies the RBAC app so that the models can store permissions for
remote objects - objects that don't actually exist in the local server.
To know which are which, a `service` field is added to our type-tracking
model, which is also new as of this work. Importantly, permission
_evaluations_ can be done for both local items and remote items.

Why? Just as we have synchronization to a "resource server" via the
resource registry app, this allows you to appoint a single service to be
the gatekeeper for RBAC. This still requires synchronization, making it
different from other approaches. Several new endpoints under
`/service-index/` are introduced to help facilitate that
synchronization.

EDITing some snapshots of the progress state

- As of opening, the core code is not finished being written, just want
to get CI output continuously.
- As of July 14 - relatively stable with current AWX, but integration
work for rest of components is still WIP

Fixes #80

Also request review from @dleehr @TheRealHaoLiu 

## Type of Change
<!-- Mandatory: Check one or more boxes that apply -->
- [ ] 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
- [ ] Test update
- [ ] Refactoring (no functional changes)
- [ ] Development environment change
- [ ] Configuration change

## Self-Review Checklist
<!-- These items help ensure quality - they complement our automated CI
checks -->
- [x] I have performed a self-review of my code
- [x] I have added relevant comments to complex code sections
- [x] 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

## Testing Instructions
<!-- Optional for test-only changes. Mandatory for all other changes -->
<!-- Must be detailed enough for reviewers to reproduce -->
### Prerequisites
<!-- List any specific setup required -->

### Steps to Test
Some tests show how you can use this in isolation for types &
permissions. That means, you can call
`DABContentType.objects.load_remote_objects` to import some remote
types, and then make roles using these types, and assign permissions.

But that doesn't do much unless you set up a resource server and another
service for it to track, and synchronize permissions between them. For
this, you pretty much need aap-dev. Watching the demos also might be
good to see test cases. Post-install those are mainly:

 - modify a role definition
 - assign a permission
- repeat with the API from the resource server & the other server (AWX
probably)

### Expected Results
Role definitions & assignments synchronized using the endpoint system
added here.

## Additional Context
See backlinks for PRs that adopt this PR

### Required Actions
<!-- Check if changes require work in other areas -->
<!-- Remove section if no external actions needed -->
- [ ] Requires documentation updates
  <!-- API docs, feature docs, deployment guides -->
- [x] Requires downstream repository changes
  <!-- Specify repos: django-ansible-base, eda-server, etc. -->
- [ ] Requires infrastructure/deployment changes
  <!-- CI/CD, installer updates, new services -->
- [x] Requires coordination with other teams
  <!-- UI team, platform services, infrastructure -->
- [ ] Blocked by PR/MR: #XXX
  <!-- Reference blocking PRs/MRs with brief context -->

### Screenshots/Logs
<!-- Add if relevant to demonstrate the changes -->

---------

Co-authored-by: Zack Kayyali <[email protected]>

Author: AlanCoding

ansible_base/authentication/utils/claims.py

@@ -8,7 +8,6 @@
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import AbstractUser
-from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import IntegrityError, models
 from django.utils.timezone import now
@@ -21,6 +20,7 @@
 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 .trigger_definition import TRIGGER_DEFINITION
 
@@ -645,7 +645,7 @@ 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 ContentType.objects.get_for_models(Organization, Team).values()}
+        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):

ansible_base/jwt_consumer/hub/auth.py

@@ -65,7 +65,10 @@ def process_permissions(self):
         for roledef_name, teams in [('Team Admin', admin_teams), ('Team Member', member_teams)]:
 
             # the "shared" "non-local" definition ...
-            roledef = RoleDefinition.objects.get(name=roledef_name)
+            try:
+                roledef = RoleDefinition.objects.get(name=roledef_name)
+            except RoleDefinition.DoesNotExist:
+                raise RoleDefinition.DoesNotExist(f'Expected JWT role {roledef_name} does not exist locally')
 
             # pks for filtering ...
             team_pks = [team.pk for team in teams]

ansible_base/lib/abstract_models/common.py

@@ -167,9 +167,11 @@ def from_db(self, db, field_names, values):
     def get_summary_fields(self):
         response = {}
         for field in self._meta.fields:
+            if field.name in self.ignore_relations:
+                continue  # This check is beneficial before getattr to prevent some error cases
             if isinstance(field, models.ForeignObject) and getattr(self, field.name):
                 # ignore relations on inherited django models
-                if field.name.endswith("_ptr") or (field.name in self.ignore_relations):
+                if field.name.endswith("_ptr"):
                     continue
                 if hasattr(getattr(self, field.name), 'summary_fields'):
                     response[field.name] = getattr(self, field.name).summary_fields()

ansible_base/lib/dynamic_config/dynamic_urls.py

@@ -13,6 +13,8 @@
 installed_apps = getattr(settings, 'INSTALLED_APPS', [])
 for app in installed_apps:
     if app.startswith('ansible_base.'):
+        if app in getattr(settings, 'ANSIBLE_BASE_APPS_EXCLUDE_VIEW_LIST', []):
+            continue
         if not importlib.util.find_spec(f'{app}.urls'):
             logger.debug(f'Module {app} does not specify urls.py')
             continue

ansible_base/lib/dynamic_config/settings_logic.py

@@ -108,6 +108,7 @@ def get_mergeable_dab_settings(settings: dict) -> dict:  # NOSONAR
             'user_ansible_id',
             'team_ansible_id',
             'object_ansible_id',
+            'assignment',  # for RoleAssignmentFilterBackend, assignment filtering
         )
 
     # SPECTACULAR SETTINGS

ansible_base/rbac/api/queries.py

@@ -0,0 +1,35 @@
+from typing import Union
+
+from django.db.models import Model
+
+from ..models import DABContentType, get_evaluation_model
+from ..remote import RemoteObject
+
+
+def assignment_qs_user_to_obj(actor: Model, obj: Union[Model, RemoteObject]):
+    """Queryset of assignments (team or user) that grants the actor any form of permission to obj"""
+    evaluation_cls = get_evaluation_model(obj)
+    ct = DABContentType.objects.get_for_model(obj)
+    reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
+
+    # All relevant assignments for the object
+    obj_eval_qs = evaluation_cls.objects.filter(object_id=obj.pk, content_type_id=ct.id)
+    obj_assignment_qs = actor.role_assignments.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+
+    global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions__content_type=ct)
+
+    return (global_assignment_qs | obj_assignment_qs).distinct()
+
+
+def assignment_qs_user_to_obj_perm(actor: Model, obj: Union[Model, RemoteObject], permission: Model):
+    """Queryset of assignments that grants this specific permission to this specific object"""
+    evaluation_cls = get_evaluation_model(obj)
+    ct = DABContentType.objects.get_for_model(obj)
+    reverse_name = evaluation_cls._meta.get_field('role').remote_field.name
+
+    obj_eval_qs = evaluation_cls.objects.filter(codename=permission.codename, object_id=obj.pk, content_type_id=ct.id)
+    obj_assignment_qs = actor.role_assignments.filter(**{f'object_role__{reverse_name}__in': obj_eval_qs})
+
+    global_assignment_qs = actor.role_assignments.filter(content_type=None, role_definition__permissions=permission)
+
+    return (global_assignment_qs | obj_assignment_qs).distinct()

ansible_base/rbac/api/serializers.py

@@ -2,135 +2,45 @@
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import transaction
 from django.db.utils import IntegrityError
-from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
-from rest_framework.fields import flatten_choices_dict, to_choices_dict
 from rest_framework.serializers import ValidationError
 
 from ansible_base.lib.abstract_models.common import get_url_for_object
-from ansible_base.lib.serializers.common import CommonModelSerializer, ImmutableCommonModelSerializer
+from ansible_base.lib.serializers.common import AbstractCommonModelSerializer, CommonModelSerializer, ImmutableCommonModelSerializer
+from ansible_base.lib.utils.auth import get_team_model
+from ansible_base.lib.utils.response import get_relative_url
 from ansible_base.rbac.models import RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ansible_base.rbac.permission_registry import permission_registry  # careful for circular imports
 from ansible_base.rbac.policies import check_content_obj_permission, visible_users
 from ansible_base.rbac.validators import check_locally_managed, validate_permissions_for_model
 
-
-class ChoiceLikeMixin(serializers.ChoiceField):
-    """
-    This uses a ForeignKey to populate the choices of a choice field.
-    This also manages some string manipulation, right now, adding the local service name.
-    """
-
-    default_error_messages = serializers.PrimaryKeyRelatedField.default_error_messages
-
-    def get_dynamic_choices(self):
-        raise NotImplementedError
-
-    def get_dynamic_object(self, data):
-        raise NotImplementedError
-
-    def to_representation(self, value):
-        raise NotImplementedError
-
-    def __init__(self, **kwargs):
-        # Workaround so that the parent class does not resolve the choices right away
-        self.html_cutoff = kwargs.pop('html_cutoff', self.html_cutoff)
-        self.html_cutoff_text = kwargs.pop('html_cutoff_text', self.html_cutoff_text)
-
-        self.allow_blank = kwargs.pop('allow_blank', False)
-        super(serializers.ChoiceField, self).__init__(**kwargs)
-
-    def _initialize_choices(self):
-        choices = self.get_dynamic_choices()
-        self._grouped_choices = to_choices_dict(choices)
-        self._choices = flatten_choices_dict(self._grouped_choices)
-        self.choice_strings_to_values = {str(k): k for k in self._choices}
-
-    @cached_property
-    def grouped_choices(self):
-        self._initialize_choices()
-        return self._grouped_choices
-
-    @cached_property
-    def choices(self):
-        self._initialize_choices()
-        return self._choices
-
-    def to_internal_value(self, data):
-        try:
-            return self.get_dynamic_object(data)
-        except ObjectDoesNotExist:
-            self.fail('does_not_exist', pk_value=data)
-        except (TypeError, ValueError):
-            self.fail('incorrect_type', data_type=type(data).__name__)
-
-
-class ContentTypeField(ChoiceLikeMixin):
-    def __init__(self, **kwargs):
-        kwargs['help_text'] = _('The type of resource this applies to.')
-        super().__init__(**kwargs)
-
-    def get_resource_type_name(self, cls) -> str:
-        return f"{permission_registry.get_resource_prefix(cls)}.{cls._meta.model_name}"
-
-    def get_dynamic_choices(self):
-        return list(sorted((self.get_resource_type_name(cls), cls._meta.verbose_name.title()) for cls in permission_registry.all_registered_models))
-
-    def get_dynamic_object(self, data):
-        model = data.rsplit('.')[-1]
-        cls = permission_registry.get_model_by_name(model)
-        if cls is None:
-            return permission_registry.content_type_model.objects.none().get()  # raises correct DoesNotExist
-        return permission_registry.content_type_model.objects.get_for_model(cls)
-
-    def to_representation(self, value):
-        if isinstance(value, str):
-            return value  # slight hack to work to AWX schema tests
-        return self.get_resource_type_name(value.model_class())
-
-
-class PermissionField(ChoiceLikeMixin):
-    @property
-    def service_prefix(self):
-        if registry := permission_registry.get_resource_registry():
-            return registry.api_config.service_type
-        return 'local'
-
-    def get_dynamic_choices(self):
-        perms = []
-        for cls in permission_registry.all_registered_models:
-            cls_name = cls._meta.model_name
-            for action in cls._meta.default_permissions:
-                perms.append(f'{permission_registry.get_resource_prefix(cls)}.{action}_{cls_name}')
-            for perm_name, description in cls._meta.permissions:
-                perms.append(f'{permission_registry.get_resource_prefix(cls)}.{perm_name}')
-        return list(sorted(perms))
-
-    def get_dynamic_object(self, data):
-        codename = data.rsplit('.')[-1]
-        return permission_registry.permission_qs.get(codename=codename)
-
-    def to_representation(self, value):
-        if isinstance(value, str):
-            return value  # slight hack to work to AWX schema tests
-        ct = permission_registry.content_type_model.objects.get_for_id(value.content_type_id)  # optimization
-        return f'{permission_registry.get_resource_prefix(ct.model_class())}.{value.codename}'
-
-
-class ManyRelatedListField(serializers.ListField):
-    def to_representation(self, data):
-        "Adds the .all() to treat the value as a queryset"
-        return [self.child.to_representation(item) if item is not None else None for item in data.all()]
+from ..models import DABContentType, DABPermission
+from ..remote import RemoteObject
+from .queries import assignment_qs_user_to_obj, assignment_qs_user_to_obj_perm
 
 
 class RoleDefinitionSerializer(CommonModelSerializer):
-    # Relational versions - we may switch to these if custom permission and type models are exposed but out of scope here
-    # permissions = serializers.SlugRelatedField(many=True, slug_field='codename', queryset=DABPermission.objects.all())
-    # content_type = ContentTypeField(slug_field='model', queryset=permission_registry.content_type_model.objects.all(), allow_null=True, default=None)
-    permissions = ManyRelatedListField(child=PermissionField())
-    content_type = ContentTypeField(allow_null=True, default=None)
+    permissions = serializers.SlugRelatedField(
+        slug_field='api_slug',
+        queryset=DABPermission.objects.all(),
+        many=True,
+        error_messages={
+            'does_not_exist': "Cannot use permission with api_slug '{value}', object does not exist",
+            'invalid': "Each content type must be a valid slug string",
+        },
+    )
+    content_type = serializers.SlugRelatedField(
+        slug_field='api_slug',
+        queryset=DABContentType.objects.all(),
+        allow_null=True,  # for global roles
+        default=None,
+        error_messages={
+            'does_not_exist': "Cannot use type with api_slug '{value}', object does not exist",
+            'invalid': "Each content type must be a valid slug string",
+        },
+    )
 
     class Meta:
         model = RoleDefinition
@@ -145,7 +55,7 @@ def validate(self, validated_data):
             permissions = list(self.instance.permissions.all())
         if 'content_type' in validated_data:
             content_type = validated_data['content_type']
-        else:
+        elif self.instance:
             content_type = self.instance.content_type
         validate_permissions_for_model(permissions, content_type)
         if getattr(self, 'instance', None):
@@ -154,11 +64,11 @@ def validate(self, validated_data):
 
 
 class RoleDefinitionDetailSerializer(RoleDefinitionSerializer):
-    content_type = ContentTypeField(read_only=True)
+    content_type = serializers.SlugRelatedField(slug_field='api_slug', read_only=True)
 
 
 class BaseAssignmentSerializer(CommonModelSerializer):
-    content_type = ContentTypeField(read_only=True, allow_null=True)
+    content_type = serializers.SlugRelatedField(slug_field='api_slug', read_only=True)
     object_ansible_id = serializers.UUIDField(
         required=False,
         help_text=_('The resource id of the object this role applies to. An alternative to the object_id field.'),
@@ -222,6 +132,8 @@ def get_object_from_data(self, validated_data, role_definition, requesting_user)
             if not role_definition.content_type:
                 raise ValidationError({'object_id': _('System role does not allow for object assignment')})
             model = role_definition.content_type.model_class()
+            if issubclass(model, RemoteObject):
+                return model(content_type=role_definition.content_type, object_id=validated_data['object_id'])
             try:
                 obj = serializers.PrimaryKeyRelatedField(queryset=model.access_qs(requesting_user)).to_internal_value(validated_data['object_id'])
             except ValidationError as exc:
@@ -336,3 +248,96 @@ def get_actor_queryset(self, requesting_user):
 
 class RoleMetadataSerializer(serializers.Serializer):
     allowed_permissions = serializers.DictField(help_text=_('A List of permissions allowed for a role definition, given its content type.'))
+
+
+class AccessListMixin:
+
+    def _get_related(self, obj) -> dict[str, str]:
+        if obj is None:
+            return {}
+        related_fields = {}
+        actor_cls = self.Meta.model
+        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},
+        )
+        return related_fields
+
+    @staticmethod
+    def summarize_role_definition(role_definition):
+        return {"name": role_definition.name, "url": get_url_for_object(role_definition)}
+
+    @staticmethod
+    def summarize_assignment_list(assignment_qs, obj_ct):
+        assignment_list = []
+        team_ct = DABContentType.objects.get_for_model(get_team_model())
+        for assignment in assignment_qs.distinct():
+            if assignment.content_type_id is None:
+                perm_type = "global"
+            elif assignment.content_type_id == team_ct.pk:
+                perm_type = "team"
+            elif assignment.content_type_id == obj_ct.pk:
+                perm_type = "direct"
+            else:
+                perm_type = "indirect"
+            assignment_list.append({"type": perm_type, "role_definition": AccessListMixin.summarize_role_definition(assignment.role_definition)})
+
+        return assignment_list
+
+    def get_object_role_assignments(self, actor):
+        obj = self.context.get("related_object")
+        permission = self.context.get("permission")
+        ct = self.context.get("content_type")
+
+        if permission:
+            assignment_qs = assignment_qs_user_to_obj_perm(actor, obj, permission)
+        else:
+            assignment_qs = assignment_qs_user_to_obj(actor, obj)
+
+        return self.summarize_assignment_list(assignment_qs, ct)
+
+    def get_url(self, obj) -> str:
+        return get_url_for_object(obj)
+
+
+class UserAccessListMixin(AccessListMixin, serializers.ModelSerializer):
+    "controller uses auth.User model so this needs to be as compatible as possible, thus ModelSerializer"
+
+    object_role_assignments = serializers.SerializerMethodField()
+    url = serializers.SerializerMethodField()
+    related = serializers.SerializerMethodField('_get_related')
+    _expected_fields = ['id', 'url', 'related', 'username', 'is_superuser', 'object_role_assignments']
+
+
+class TeamAccessListMixin(AccessListMixin, AbstractCommonModelSerializer):
+    object_role_assignments = serializers.SerializerMethodField()
+    url = serializers.SerializerMethodField()
+    related = serializers.SerializerMethodField('_get_related')
+    _expected_fields = ['id', 'url', 'related', 'name', 'organization', 'object_role_assignments']
+
+
+class UserAccessAssignmentSerializer(RoleUserAssignmentSerializer):
+    intermediary_roles = serializers.SerializerMethodField()
+
+    class Meta(RoleUserAssignmentSerializer.Meta):
+        fields = RoleUserAssignmentSerializer.Meta.fields + ['intermediary_roles']
+
+    def get_intermediary_roles(self, assignment):
+        team_ct = DABContentType.objects.get_for_model(get_team_model())
+
+        permission = self.context.get("permission")
+        if assignment.content_type != team_ct:
+            return []
+        team = assignment.content_object
+        obj = self.context.get("related_object")
+
+        if permission:
+            assignment_qs = assignment_qs_user_to_obj_perm(team, obj, permission)
+        else:
+            assignment_qs = assignment_qs_user_to_obj(team, obj)
+
+        return AccessListMixin.summarize_assignment_list(assignment_qs, self.context.get("content_type"))
+
+
+class TeamAccessAssignmentSerializer(RoleTeamAssignmentSerializer):
+    pass