Handle sync of system roles

Author: AlanCoding

ansible_base/rbac/models/permission.py

@@ -7,6 +7,15 @@
 
 class DABPermissionManager(models.Manager):
     def load_remote_objects(self, remote_data: list[dict], update_managed=False):
+        """Load the permission list from a remote system, requires types are loaded first
+
+        The remote_data should be the results from the /service-index/role-permissions/ endpoint
+        of another system.
+        This will save those remote permissions into the local database so we can track remote RBAC.
+
+        update_managed being True will refresh the managed roles like Organization Admin
+        so that if the algorithm includes any new permissions, those are added.
+        """
         for remote_type in remote_data:
             codename = remote_type.pop('codename')
             ct_slug = remote_type.pop('content_type')

ansible_base/rbac/service_api/serializers.py

@@ -87,11 +87,16 @@ def validate(self, attrs):
         has_oid = bool(self.initial_data.get('object_id'))
         has_oaid = bool(self.initial_data.get('object_ansible_id'))
 
-        if not self.partial and not has_oid and not has_oaid:
-            raise serializers.ValidationError("You must provide either 'object_id' or 'object_ansible_id'.")
-        elif not has_oaid:
-            # need to remove blank and null fields or else it can overwrite the non-null non-blank field
-            attrs['object_id'] = self.initial_data['object_id']
+        rd = attrs['role_definition']
+        if rd.content_type_id:
+            if not self.partial and not has_oid and not has_oaid:
+                raise serializers.ValidationError("You must provide either 'object_id' or 'object_ansible_id'.")
+            elif not has_oaid:
+                # need to remove blank and null fields or else it can overwrite the non-null non-blank field
+                attrs['object_id'] = self.initial_data['object_id']
+        else:
+            if has_oaid or has_oid:
+                raise serializers.ValidationError("Can not provide either 'object_id' or 'object_ansible_id' for system role")
 
         # NOTE: right now not enforcing the case you provide both, could check for consistency later
 
@@ -100,9 +105,12 @@ def validate(self, attrs):
     def find_existing_assignment(self, queryset):
         actor = self.validated_data[self.actor_field]
         role_definition = self.validated_data['role_definition']
-        object_id = self.validated_data['object_id']
-        filter_kwargs = {self.actor_field: actor}
-        return queryset.filter(role_definition=role_definition, object_id=object_id, **filter_kwargs).first()
+        filter_kwargs = {self.actor_field: actor, 'role_definition': role_definition}
+        if role_definition.content_type_id:
+            filter_kwargs['object_id'] = self.validated_data['object_id']
+        else:
+            filter_kwargs['object_id'] = None
+        return queryset.filter(**filter_kwargs).first()
 
     def create(self, validated_data):
         rd = validated_data['role_definition']
@@ -115,7 +123,7 @@ def create(self, validated_data):
         # Unlike the public view, the action is attributed to the specified user in data
         with impersonate(as_user):
 
-            object_id = validated_data['object_id']
+            object_id = validated_data.get('object_id')
             obj = None
             if object_id:
                 model = rd.content_type.model_class()

test_app/tests/rbac/remote/test_service_api.py

@@ -3,7 +3,8 @@
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
-from ansible_base.rbac.models import DABContentType, DABPermission
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition
+from test_app.models import Team, User
 
 
 @pytest.mark.django_db
@@ -196,6 +197,44 @@ def test_unassign_endpoint_for_team(team, org_inv_rd, inventory, admin_api_clien
     assert not rando.has_obj_perm(inventory, 'change')
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize('actor_type', ['user', 'team'])
+def test_assign_and_unassign_system_role(inventory, admin_api_client, actor_type, organization, member_rd):
+    if actor_type == 'user':
+        actor = User.objects.create(username='user1')
+        user = actor
+    else:
+        actor = Team.objects.create(name='random_team', organization=organization)
+        user = User.objects.create(username='user1')
+        member_rd.give_permission(user, actor)
+
+    rd = RoleDefinition.objects.managed.sys_auditor
+    assert 'view_inventory' in set(rd.permissions.values_list('codename', flat=True))
+    assert not user.has_obj_perm(inventory, 'view')
+
+    url = get_relative_url(f'service{actor_type}assignment-assign')
+    data = {"role_definition": rd.name, f"{actor_type}_ansible_id": str(actor.resource.ansible_id)}
+    response = admin_api_client.post(url, data)
+    assert response.status_code == 201, response.data
+    if hasattr(actor, '_singleton_permissions'):
+        delattr(actor, '_singleton_permissions')
+    assert user.has_obj_perm(inventory, 'view')  # gave system wide view permission
+
+    # Second try, response code indicates global assignment already exists
+    response = admin_api_client.post(url, data=data)
+    assert response.status_code == 200, response.data
+
+    unassign_url = get_relative_url(f'service{actor_type}assignment-unassign')
+    response = admin_api_client.post(unassign_url, data)
+    assert response.status_code == 204, response.data
+    if hasattr(actor, '_singleton_permissions'):
+        delattr(actor, '_singleton_permissions')
+    assert not user.has_obj_perm(inventory, 'view')  # permission removed
+
+    response = admin_api_client.post(unassign_url, data)
+    assert response.status_code == 200, response.data
+
+
 @pytest.mark.django_db
 def test_filter_assignment_list(admin_api_client, rando, inv_rd, view_inv_rd, org_inv_rd, inventory):
     inv_rd.give_permission(rando, inventory)