Add callback to validate role assignment

Some dab-based apps like AWX may wish to add
exceptions to the user and team role assignments.

This PR adds logic to the serializer
.create method to execute a callback method
that is optionally defined on the model.

This callback signature looks like:

validate_role_assignment(self, actor, role_definition)

This callback should raise exceptions if necessary.

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

ansible_base/rbac/api/serializers.py

@@ -4,6 +4,8 @@
 from django.db.utils import IntegrityError
 from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
+from django.core.exceptions import ValidationError as DjangoValidationError, PermissionDenied as DjangoPermissionDenied
+
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.fields import flatten_choices_dict, to_choices_dict
@@ -242,6 +244,13 @@ def create(self, validated_data):
         # Resolve object
         obj = self.get_object_from_data(validated_data, rd, requesting_user)
 
+        # model-level callback to further validate the assignment
+        # can be optionally implemented by the model
+        # the callback should raise DRF exceptions directly if
+        # necessary
+        if getattr(obj, 'validate_role_assignment', None):
+            obj.validate_role_assignment(actor, rd)
+
         if rd.content_type:
             # Object role assignment
             if not obj:

docs/lib/validation.md

@@ -10,4 +10,17 @@ Must be a valid string
 
 `ansible_base.lib.utils.validation.validate_url` this is similar to the validate_url in django but has a parameter for `allow_plain_hostname: bool = False` which means you can have a url like `https://something:443/testing`.
 
-`ansible_base.lib.utils.validation.validate_url_list` this is a convince method which takes an array of urls and validates each of them using its own validate_url method. 
+`ansible_base.lib.utils.validation.validate_url_list` this is a convince method which takes an array of urls and validates each of them using its own validate_url method.
+
+
+# Validation callback for role assignment
+
+Apps that utilize django-ansible-base may wish to add extra validation when assigning roles to actors (users or teams).
+
+For this, django-ansible-base will call out to `validate_role_assignment` method that defined on the object that being assigned.
+
+The signature of this callback is
+
+`validate_role_assignment(self, actor, role_definition)`
+
+This method is reponsible for raising the appropriate exception if necessary (e.g. DRF ValidationError or DRF PermissionDenied).

test_app/models.py

@@ -5,6 +5,8 @@
 from django.db import models
 from django.db.models import JSONField
 
+from rest_framework.exceptions import ValidationError as DRFValidationError, PermissionDenied as DRFPermissionDenied
+
 from ansible_base.activitystream.models import AuditableModel
 from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel, ImmutableCommonModel, ImmutableModel, NamedCommonModel
 from ansible_base.lib.utils.models import prevent_search, user_summary_fields
@@ -137,6 +139,16 @@ class Meta:
     def summary_fields(self):
         return {"id": self.id, "name": self.name}
 
+    def validate_role_assignment(self, actor, role_definition):
+        if isinstance(actor, User):
+            name = actor.username
+        if isinstance(actor, Team):
+            name = actor.name
+        if name == 'test-400':
+            raise DRFValidationError({'detail': 'Role assignment not allowed 400'})
+        if name == 'test-403':
+            raise DRFPermissionDenied('Role assignment not allowed 403')
+
 
 class Credential(models.Model):
     "Example of a model that gets used by other models"

test_app/tests/rbac/api/test_rbac_validation.py

@@ -1,9 +1,14 @@
 import pytest
+from django.contrib.auth import get_user_model
 from django.test.utils import override_settings
 from django.urls import reverse
 
+from ansible_base.lib.utils.auth import get_team_model
 from ansible_base.rbac.models import RoleDefinition
 
+Team = get_team_model()
+User = get_user_model()
+
 
 @pytest.mark.django_db
 class TestSharedAssignmentsDisabled:
@@ -86,3 +91,39 @@ def test_can_not_make_global_role_with_member_permission(admin_api_client):
     )
     assert response.status_code == 400
     assert 'member_team permission can not be used in global roles' in str(response.data['permissions'])
+
+
+@pytest.mark.django_db
+def test_callback_validate_role_user_assignment(admin_api_client, inventory, inv_rd):
+    url = reverse('roleuserassignment-list')
+    user = User.objects.create(username='user-allowed')
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'user': user.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 201
+
+    user = User.objects.create(username='test-400')
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'user': user.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 400
+    assert "Role assignment not allowed 400" in str(response.data)
+
+    user = User.objects.create(username='test-403')
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'user': user.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 403
+    assert "Role assignment not allowed 403" in str(response.data)
+
+
+@pytest.mark.django_db
+def test_callback_validate_role_team_assignment(admin_api_client, inventory, organization, inv_rd):
+    url = reverse('roleteamassignment-list')
+    team = Team.objects.create(name='team-allowed', organization=organization)
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'team': team.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 201
+
+    team = Team.objects.create(name='test-400', organization=organization)
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'team': team.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 400
+    assert "Role assignment not allowed 400" in str(response.data)
+
+    team = Team.objects.create(name='test-403', organization=organization)
+    response = admin_api_client.post(url, data={'object_id': inventory.id, 'team': team.id, 'role_definition': inv_rd.id})
+    assert response.status_code == 403
+    assert "Role assignment not allowed 403" in str(response.data)