Add callback to validate role assignment

ansible_base/rbac/api/serializers.py

@@ -242,6 +242,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/apps/rbac/for_app_developers.md

@@ -279,3 +279,10 @@ and `Team.tracked_parents` ManyToMany relationships, respectively.
 So if you have a team object, `team.users.add(user)` will also give that
 user _member permission_ to that team, where those permissions are defined by the
 role definition with the name "team-member".
+
+
+### Role assignment callback
+
+Apps that utilize django-ansible-base may wish to add extra validation when assigning roles to actors (users or teams).
+
+see [Validation callback for role assignment](../../lib/validation.md)

docs/lib/validation.md

@@ -10,4 +10,26 @@ 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, for example,
+
+```python
+from rest_framework.exceptions import ValidationError
+class MyDjangoModel:
+    def validate_role_assignment(self, actor, role_definition):
+        raise ValidationError({'detail': 'Role assignment not allowed.'})
+```
+
+Note, if you want the exception to result in a HTTP 400 or 403 response, you can raise django rest framework exceptions instead of django exceptions.

test_app/models.py

@@ -4,6 +4,8 @@
 from django.contrib.auth.models import AbstractUser
 from django.db import models
 from django.db.models import JSONField
+from rest_framework.exceptions import PermissionDenied as DRFPermissionDenied
+from rest_framework.exceptions import ValidationError as DRFValidationError
 
 from ansible_base.activitystream.models import AuditableModel
 from ansible_base.lib.abstract_models import AbstractOrganization, AbstractTeam, CommonModel, ImmutableCommonModel, ImmutableModel, NamedCommonModel
@@ -149,6 +151,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)