Support evaluations for remote child objects

Author: AlanCoding

ansible_base/rbac/models/role.py

@@ -227,7 +227,7 @@ def give_permission(self, actor, content_object):
     def remove_permission(self, actor, content_object):
         return self.give_or_remove_permission(actor, content_object, giving=False)
 
-    def get_or_create_object_role(self, **kwargs):
+    def get_or_create_object_role(self, kwargs, defaults):
         """Transaction-safe method to create ObjectRole
 
         The UI will assign many permissions concurrently.
@@ -238,13 +238,13 @@ def get_or_create_object_role(self, **kwargs):
         if transaction.get_connection().in_atomic_block:
             try:
                 with transaction.atomic():
-                    object_role = ObjectRole.objects.create(**kwargs)
+                    object_role = ObjectRole.objects.create(**kwargs, **defaults)
                     return (object_role, True)
             except IntegrityError:
                 object_role = ObjectRole.objects.get(**kwargs)
                 return (object_role, False)
         else:
-            object_role = ObjectRole.objects.create(**kwargs)
+            object_role = ObjectRole.objects.create(**kwargs, **defaults)
             return (object_role, True)
 
     def give_or_remove_permission(self, actor, content_object, giving=True, sync_action=False):
@@ -260,13 +260,19 @@ def give_or_remove_permission(self, actor, content_object, giving=True, sync_act
             object_id = content_object._meta.pk.get_db_prep_value(content_object.pk, connection)
 
         kwargs = dict(role_definition=self, content_type=obj_ct, object_id=object_id)
+        defaults = {}
+
+        # For remote objects, add parent reference so we can do evaluations if needed
+        if isinstance(content_object, RemoteObject):
+            if content_object.parent_reference:
+                defaults['parent_reference'] = content_object.parent_reference
 
         created = False
         object_role = ObjectRole.objects.filter(**kwargs).first()
         if object_role is None:
             if not giving:
                 return  # nothing to do
-            object_role, created = self.get_or_create_object_role(**kwargs)
+            object_role, created = self.get_or_create_object_role(kwargs, defaults)
 
         from ansible_base.rbac.triggers import needed_updates_on_assignment, update_after_assignment
 
@@ -564,28 +570,38 @@ def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int
             object_id = role_model._meta.pk.to_python(self.object_id)
         for permission in types_prefetch.permissions_for_object_role(self):
             permission_content_type = types_prefetch.get_content_type(permission.content_type_id)
+            permission_model = permission_content_type.model_class()
 
             # direct object permission
             if permission.content_type_id == self.content_type_id:
                 expected_evaluations.add((permission.codename, self.content_type_id, object_id))
                 continue
 
-            # add child permission on the parent object, usually only for add permission
+            # Add evaluation for the parent object, usually only for add permission
             if is_add_perm(permission.codename) or settings.ANSIBLE_BASE_CACHE_PARENT_PERMISSIONS:
                 expected_evaluations.add((permission.codename, self.content_type_id, object_id))
 
-            # add child object permission on child objects
-            # Only propogate add permission to children which are parents of the permission model
+            # Add evaluations for child objects, where this role gives permission to child objects
             filter_path = None
             child_model = None
             if is_add_perm(permission.codename):
-                for path, model in permission_registry.get_child_models(role_model):
-                    if '__' in path and model._meta.model_name == permission_content_type.model:
-                        path_to_parent, filter_path = path.split('__', 1)
-                        child_model = permission_content_type.model_class()._meta.get_field(path_to_parent).related_model
-                        eval_ct = DABContentType.objects.get_for_model(child_model).id
-                if not child_model:
-                    continue
+                # Add evaluations for add permission to children which are parents of the permission model
+                # this only matters when for multi-layer inheritance (grandchildren)
+                if issubclass(permission_model, RemoteObject):
+                    eval_ct = permission_content_type.parent_content_type
+                    if eval_ct == role_content_type:
+                        continue  # this permission is for a direct child model
+                else:
+                    for path, model in permission_registry.get_child_models(role_model):
+                        if '__' in path and model._meta.model_name == permission_content_type.model:
+                            path_to_parent, filter_path = path.split('__', 1)
+                            child_model = permission_model._meta.get_field(path_to_parent).related_model
+                            eval_ct = DABContentType.objects.get_for_model(child_model).id
+                            break
+                    if not child_model:
+                        continue  # this permission is for a direct child model
+            elif issubclass(permission_model, RemoteObject):
+                eval_ct = permission.content_type_id
             else:
                 for path, model in permission_registry.get_child_models(role_model):
                     if model._meta.model_name == permission_content_type.model:
@@ -603,8 +619,15 @@ def expected_direct_permissions(self, types_prefetch=None) -> set[tuple[str, int
             if eval_ct in cached_id_lists:
                 id_list = cached_id_lists[eval_ct]
             else:
-                # TODO: build id_list from ObjectRole objects it is remote object
-                id_list = child_model.objects.filter(**{filter_path: object_id}).values_list('pk', flat=True)
+                if issubclass(permission_model, RemoteObject):
+                    # Build id_list from ObjectRole objects if it is remote object
+                    id_list = (
+                        ObjectRole.objects.filter(parent_reference=object_id, content_type=eval_ct)
+                        .values_list(Cast('object_id', output_field=permission_model._meta.pk.django_field()), flat=True)
+                        .distinct()
+                    )
+                else:
+                    id_list = child_model.objects.filter(**{filter_path: object_id}).values_list('pk', flat=True)
                 cached_id_lists[eval_ct] = list(id_list)
 
             for id in id_list:

ansible_base/rbac/remote.py

@@ -38,6 +38,12 @@ def to_python(self, value: Union[str, int, uuid.UUID]) -> Union[int, uuid.UUID]:
             return uuid.UUID(value)
         return int(value)
 
+    def django_field(self):
+        "This gives a mock Django field like what it mimics"
+        if self.pk_field_type == "uuid":
+            return models.UUIDField()
+        return models.IntegerField()
+
 
 class StandinMeta:
     def __init__(self, ct: models.Model, abstract=False):
@@ -52,9 +58,11 @@ def __init__(self, ct: models.Model, abstract=False):
 class RemoteObject:
     """Placeholder for objects that live in another project."""
 
-    def __init__(self, content_type: models.Model, object_id: Union[int, str]):
+    def __init__(self, content_type: models.Model, object_id: Union[int, str], parent_reference=None):
         self.content_type = content_type
         self.object_id = object_id
+        # Since object is remote, we do not have its properties here, so a pointer to the parent can be specified here
+        self.parent_reference = parent_reference
         if not hasattr(self, '_meta'):
             # If object is created without a type-specific subclass, do the best we can
             self._meta = StandinMeta(content_type, abstract=True)

test_app/tests/rbac/remote/test_remote_assignment.py

@@ -1,6 +1,6 @@
 import pytest
 
-from ansible_base.rbac.models import RoleUserAssignment
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleUserAssignment
 from ansible_base.rbac.remote import RemoteObject
 from test_app.models import User
 
@@ -46,3 +46,29 @@ def test_prefetch_related_objects(foo_type, foo_rd, inv_rd, inventory):
     assert {assignment.content_object for assignment in RoleUserAssignment.objects.all()} == {a_foo, inventory}
     assert {assignment.content_object for assignment in RoleUserAssignment.objects.all()} == {a_foo, inventory}
     assert {assignment.content_object for assignment in RoleUserAssignment.objects.prefetch_related('content_object')} == {a_foo, inventory}
+
+
+@pytest.mark.django_db
+def test_organization_permission_remote_object(rando, foo_type, organization):
+    """If the remote object is a child of a shared organization object, org roles should evaluate that users have permission
+
+    This is supported by loading a reference to the parent in the RemoteObject.
+    """
+    permissions = []
+    for codename in ('view_foo', 'change_foo', 'foo_foo'):
+        permissions.append(DABPermission.objects.create(codename=codename, content_type=foo_type))
+    view_foo_rd = RoleDefinition.objects.create_from_permissions(
+        name='Foo fooers for the foos in foo service', permissions=[permissions[0].api_slug], content_type=foo_type
+    )
+    a_foo = RemoteObject(content_type=foo_type, object_id=42, parent_reference=organization.pk)
+    assignment = view_foo_rd.give_permission(rando, a_foo)
+    assert str(assignment.object_role.parent_reference) == str(organization.pk)
+    assert not rando.has_obj_perm(a_foo, 'change_foo')
+
+    org_ct = DABContentType.objects.get_for_model(organization)
+    org_foo_rd = RoleDefinition.objects.create_from_permissions(
+        name='Org level foo role', permissions=['view_foo', 'change_foo', 'foo_foo', 'shared.view_organization'], content_type=org_ct
+    )
+    org_foo_rd.give_permission(rando, organization)
+
+    assert rando.has_obj_perm(a_foo, 'foo')