Test and fix creating assignments through the API

Author: AlanCoding

ansible_base/rbac/api/serializers.py

@@ -16,6 +16,7 @@
 from ansible_base.rbac.validators import check_locally_managed, validate_permissions_for_model
 
 from ..models import DABContentType, DABPermission, get_evaluation_model
+from ..remote import RemoteObject
 
 
 class RoleDefinitionSerializer(CommonModelSerializer):
@@ -129,6 +130,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:

ansible_base/rbac/models/role.py

@@ -24,7 +24,7 @@
 from ansible_base.rbac.validators import validate_assignment, validate_permissions_for_model
 from ansible_base.resource_registry.fields import AnsibleResourceField
 
-from ..remote import RemoteObject
+from ..remote import RemoteObject, StandInPK
 from .content_type import DABContentType
 from .fields import FederatedForeignKey
 from .permission import DABPermission
@@ -531,7 +531,7 @@ def descendent_roles(self):
             descendents.update(set(target_team.has_roles.all()))
         return descendents
 
-    def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int, Union[int, str, UUID]]]:
+    def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int, Union[int, UUID]]]:
         """The expected permissions that holding this ObjectRole confers to the holder
 
         This is given in the form of tuples, which represent RoleEvaluation entries.
@@ -546,7 +546,8 @@ def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int
         role_content_type = types_prefetch.get_content_type(self.content_type_id)
         role_model = role_content_type.model_class()
         if role_content_type.is_remote:
-            object_id = self.object_id
+            pk_field = StandInPK(role_content_type)  # remote, mock, field
+            object_id = pk_field.to_python(self.object_id)
         else:
             # ObjectRole.object_id is stored as text, we convert it to the model pk native type
             object_id = role_model._meta.pk.to_python(self.object_id)

ansible_base/rbac/remote.py

@@ -1,4 +1,5 @@
 import inspect
+import uuid
 from typing import Type, Union
 
 from django.apps import apps
@@ -19,26 +20,33 @@
 """
 
 
+class StandInPK:
+    def __init__(self, ct: models.Model):
+        self.pk_field_type = ct.pk_field_type
+
+    def get_prep_value(self, value: Union[str, int, uuid.UUID]) -> Union[str, int]:
+        if self.pk_field_type == "uuid":
+            if isinstance(value, uuid.UUID):
+                return str(value)
+            return str(uuid.UUID(value))
+        return int(value)
+
+    def to_python(self, value: Union[str, int, uuid.UUID]) -> Union[int, uuid.UUID]:
+        if self.pk_field_type == "uuid":
+            if isinstance(value, uuid.UUID):
+                return value
+            return uuid.UUID(value)
+        return int(value)
+
+
 class StandinMeta:
     def __init__(self, ct: models.Model, abstract=False):
         self.service = ct.service
         self.model_name = ct.model
         self.app_label = ct.app_label
         self.abstract = abstract
 
-        class PK:
-            def __init__(self, pk_field_type):
-                self.pk_field_type = pk_field_type
-
-            def get_prep_value(self, value):
-                # TODO: handle uuid fields based on
-                # if self.pk_field_type:
-                return int(value)
-
-            def to_python(self, value):
-                return int(value)
-
-        self.pk = PK(ct.pk_field_type)
+        self.pk = StandInPK(ct)
 
 
 class RemoteObject:
@@ -54,6 +62,12 @@ def __init__(self, content_type: models.Model, object_id: Union[int, str]):
             if content_type.model != self._meta.model_name:
                 raise RuntimeError(f'RemoteObject created with type {content_type} but with type for {self._meta.model_name}')
 
+        # Raise an early error if the primary key is obviously not valid for the model type
+        try:
+            self._meta.pk.to_python(object_id)
+        except (ValueError, TypeError, AttributeError) as e:
+            raise ValueError(f"Invalid primary key value {object_id} for type {content_type.pk_field_type}, error: {e}")
+
     def __repr__(self):
         return f"<RemoteObject {self.content_type} id={self.object_id}>"
 

test_app/tests/rbac/remote/conftest.py

@@ -19,3 +19,18 @@ def foo_rd(foo_type, foo_permission):
     return RoleDefinition.objects.create_from_permissions(
         name='Foo fooers for the foos in foo service', permissions=[foo_permission.api_slug], content_type=foo_type
     )
+
+
+@pytest.fixture
+def foo_type_uuid():
+    return DABContentType.objects.create(service='foo', model='foo_uuid', app_label='foo', pk_field_type='uuid')
+
+
+@pytest.fixture
+def foo_permission_uuid(foo_type_uuid):
+    return DABPermission.objects.create(codename='foo_foo_uuid', content_type=foo_type_uuid)
+
+
+@pytest.fixture
+def foo_rd_uuid(foo_type_uuid, foo_permission_uuid):
+    return RoleDefinition.objects.create_from_permissions(name='UUID foo role thing', permissions=[foo_permission_uuid.api_slug], content_type=foo_type_uuid)

test_app/tests/rbac/remote/test_public_api_compat.py

@@ -1,3 +1,5 @@
+import uuid
+
 import pytest
 
 from ansible_base.lib.utils.response import get_relative_url
@@ -53,3 +55,31 @@ def test_user_role_assignment_remote_and_local(admin_api_client, rando, foo_type
     sf = item['summary_fields']
     assert 'content_object' in sf
     assert sf['content_object'] == {'<remote_object_placeholder>': True, 'model_name': 'foo', 'service': 'foo', 'pk': 42}
+
+
+@pytest.mark.django_db
+def test_give_permission_to_remote_object(admin_api_client, rando, foo_type, foo_rd):
+    a_foo = RemoteObject(content_type=foo_type, object_id=42)
+    assert not rando.has_obj_perm(a_foo, 'foo')
+
+    url = get_relative_url('roleuserassignment-list')
+    # NOTE: at this point the object_id is made up, cross-server coordination is not running here
+    data = {"role_definition": foo_rd.id, "user": rando.pk, "object_id": 42}
+    response = admin_api_client.post(path=url, data=data)
+    assert response.status_code == 201, response.data
+
+    assert rando.has_obj_perm(a_foo, 'foo')
+
+
+@pytest.mark.django_db
+def test_give_permission_to_remote_object_uuid(admin_api_client, rando, foo_type_uuid, foo_rd_uuid):
+    pk_value = str(uuid.uuid4())
+    a_foo = RemoteObject(content_type=foo_type_uuid, object_id=pk_value)
+    assert not rando.has_obj_perm(a_foo, 'foo')
+
+    url = get_relative_url('roleuserassignment-list')
+    data = {"role_definition": foo_rd_uuid.id, "user": rando.pk, "object_id": pk_value}
+    response = admin_api_client.post(path=url, data=data)
+    assert response.status_code == 201, response.data
+
+    assert rando.has_obj_perm(a_foo, 'foo')

test_app/tests/rbac/remote/test_remote_types.py

@@ -0,0 +1,18 @@
+import uuid
+
+import pytest
+
+from ansible_base.rbac.remote import RemoteObject
+
+
+@pytest.mark.django_db
+def test_validate_object_id_uuid(foo_type_uuid):
+    "The 42 primary key is invalid for uuid type objects"
+    with pytest.raises(ValueError):
+        RemoteObject(content_type=foo_type_uuid, object_id=42)
+
+
+@pytest.mark.django_db
+def test_validate_object_id_int(foo_type):
+    with pytest.raises(ValueError):
+        RemoteObject(content_type=foo_type, object_id=str(uuid.uuid4()))