Allow making evaluations as querysets on remote models

Author: AlanCoding

ansible_base/rbac/evaluations.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Iterable, Optional, Type
 
 from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
@@ -10,6 +10,8 @@
 from ansible_base.rbac.models import DABPermission, RoleDefinition, get_evaluation_model
 from ansible_base.rbac.validators import validate_codename_for_model
 
+from .remote import RemoteObject
+
 """
 RoleEvaluation or RoleEvaluationUUID models are the authority for permission evaluations,
 meaning, determining whether a user has a permission to an object.
@@ -86,7 +88,7 @@ def __call__(self, actor, codename: str = 'view', queryset: Optional[QuerySet] =
 
 
 class AccessibleIdsDescriptor(BaseEvaluationDescriptor):
-    def __call__(self, actor, codename: str = 'view', content_types=None, cast_field=None) -> QuerySet:
+    def __call__(self, actor, codename: str = 'view', content_types: Optional[Iterable[int]] = None, cast_field=None) -> QuerySet:
         full_codename = validate_codename_for_model(codename, self.cls)
         if isinstance(actor, AnonymousUser):
             return self.cls.objects.none().values_list()
@@ -98,6 +100,25 @@ def __call__(self, actor, codename: str = 'view', content_types=None, cast_field
         return get_evaluation_model(self.cls).accessible_ids(self.cls, actor, full_codename, content_types=content_types, cast_field=cast_field)
 
 
+def remote_obj_id_qs(actor, remote_cls: Type[RemoteObject], codename: str = 'view', content_types: Optional[Iterable[int]] = None, cast_field=None) -> QuerySet:
+    """Returns a queryset of ids for a remote object"""
+    full_codename = validate_codename_for_model(codename, remote_cls)
+    evaluation_model = get_evaluation_model(remote_cls)
+    if isinstance(actor, AnonymousUser):
+        return evaluation_model.objects.none().values_list('object_id')
+    if actor._meta.model_name == 'user' and has_super_permission(actor, full_codename):
+        # Return all known objects on this server, some objects on remote server may not be known
+        if content_types:
+            filter_kwargs = {'content_type_id__in': content_types}
+        else:
+            filter_kwargs = {'content_type': remote_cls.get_ct_from_type()}
+        if cast_field is None:
+            return evaluation_model.objects.filter(**filter_kwargs).values_list('object_id').distinct()
+        else:
+            return evaluation_model.objects.filter(**filter_kwargs).values_list(Cast('object_id', output_field=cast_field)).distinct()
+    return evaluation_model.accessible_ids(remote_cls, actor, full_codename, content_types=content_types, cast_field=cast_field)
+
+
 def bound_has_obj_perm(self, obj, codename) -> bool:
     if not permission_registry.is_registered(obj):
         raise ValidationError(f'Object of {obj._meta.model_name} type is not registered with DAB RBAC')

ansible_base/rbac/management/__init__.py

@@ -82,7 +82,6 @@ def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
     # a list of the ones we're going to create.
     all_perms = set(Permission.objects.using(using).filter(content_type__in=ctypes).values_list("content_type", "codename"))
 
-    # TODO: add api_slug field when added
     perms = []
     for ct, (codename, name) in searched_perms:
         if (ct.pk, codename) not in all_perms:
@@ -91,6 +90,7 @@ def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
             permission.codename = codename
             permission.name = name
             permission.content_type = ct
+            permission.api_slug = f'{ct.service}.{codename}'
             perms.append(permission)
 
     Permission.objects.using(using).bulk_create(perms)

ansible_base/rbac/management/create_types.py

@@ -2,7 +2,7 @@
 from typing import Type
 
 from django.apps import apps as global_apps
-from django.db import DEFAULT_DB_ALIAS, models
+from django.db import DEFAULT_DB_ALIAS, connection, models
 
 from ansible_base.rbac import permission_registry
 from ansible_base.rbac.remote import get_resource_prefix
@@ -34,7 +34,6 @@ def create_DAB_contenttypes(
 
     content_types = get_local_DAB_contenttypes(using, DABContentType)
 
-    # TODO: add api_slug field when added
     ct_data = []
     for model in permission_registry.all_registered_models:
         service = get_resource_prefix(model)
@@ -44,6 +43,8 @@ def create_DAB_contenttypes(
                 service=service,
                 app_label=model._meta.app_label,
                 model=model._meta.model_name,
+                api_slug=f'{service}.{model._meta.model_name}',
+                pk_field_type=model._meta.pk.db_type(connection),
             )
             # To make usage earier in a transitional period, we will set the content type
             # of any new entries created here to the id of its corresponding ContentType

ansible_base/rbac/migrations/0004_remote_permissions_additions.py

@@ -159,4 +159,20 @@ class Migration(migrations.Migration):
             model_name='roleevaluationuuid',
             name='dab_rbac_ro_role_id_4fe905_idx',
         ),
+        # Fields unique to DAB RBAC and not generally shared with ContentType or Permission
+        migrations.AddField(
+            model_name='dabcontenttype',
+            name='api_slug',
+            field=models.CharField(default='', help_text='String to use for references to this type from other models in the API.', max_length=201),
+        ),
+        migrations.AddField(
+            model_name='dabcontenttype',
+            name='pk_field_type',
+            field=models.CharField(default='integer', help_text='Database field type of the primary key field of the model, relevant for interal logic tracking permissions.', max_length=100),
+        ),
+        migrations.AddField(
+            model_name='dabpermission',
+            name='api_slug',
+            field=models.CharField(default='', help_text='String to use for references to this type from other models in the API.', max_length=201),
+        ),
     ]

ansible_base/rbac/models/__init__.py

@@ -1,5 +1,7 @@
 from django.db import connection
+import inspect
 
+from ..remote import RemoteObject
 from .content_type import DABContentType
 from .permission import DABPermission
 from .role import ObjectRole, RoleDefinition, RoleEvaluation, RoleEvaluationUUID, RoleTeamAssignment, RoleUserAssignment
@@ -18,10 +20,14 @@
 
 
 def get_evaluation_model(cls):
-    pk_field = cls._meta.pk
-    # For proxy models, including django-polymorphic, use the id field from parent table
-    # we accomplish this by inspecting the raw database type of the field
-    pk_db_type = pk_field.db_type(connection)
+    if (inspect.isclass(cls) and issubclass(cls, RemoteObject)) or isinstance(cls, RemoteObject):
+        # For remote models, we save the pk type in the database specifically for use here
+        pk_db_type = cls.get_ct_from_type().pk_field_type
+    else:
+        pk_field = cls._meta.pk
+        # For proxy models, including django-polymorphic, use the id field from parent table
+        # we accomplish this by inspecting the raw database type of the field
+        pk_db_type = pk_field.db_type(connection)
     for eval_cls in (RoleEvaluation, RoleEvaluationUUID):
         if pk_db_type == eval_cls._meta.get_field('object_id').db_type(connection):
             return eval_cls

ansible_base/rbac/models/content_type.py

@@ -3,6 +3,7 @@
 from typing import Any, Dict, Optional, Sequence, Tuple, Type, Union
 
 from django.apps import apps
+from django.db import connection
 from django.db import models as django_models
 from django.db.models.options import Options
 from django.utils.translation import gettext_lazy as _
@@ -26,7 +27,6 @@ def clear_cache(self) -> None:
         self._cache.clear()
 
     def create(self, *args: Any, **kwargs: Any) -> "DABContentType":
-        # TODO: set api_slug field
         obj = super().create(*args, **kwargs)
         self._add_to_cache(self.db, obj)
         return obj
@@ -58,7 +58,7 @@ def get_for_model(
             self._add_to_cache(self.db, ct)
             return ct
         elif inspect.isclass(model) and issubclass(model, RemoteObject):
-            ct = self.get_by_natural_key(model.type_data)
+            ct = self.get_by_natural_key(model._meta.service, model._meta.app_label, model._meta.model_name)
             self._add_to_cache(self.db, ct)
             return ct
 
@@ -77,6 +77,8 @@ def get_for_model(
                 service=service,
                 app_label=opts.app_label,
                 model=opts.model_name,
+                api_slug=f'{service}.{opts.model_name}',
+                pk_field_type=model._meta.pk.db_type(connection),
             )
         self._add_to_cache(self.db, ct)
         return ct
@@ -132,7 +134,13 @@ def get_for_models(
                 self._add_to_cache(self.db, ct)
             # Named it service_create to not shadown variable from prior loop
             for (service_create, app_label, model_name), opts_models in needed_opts.items():
-                ct = self.create(service=service_create, app_label=app_label, model=model_name)
+                if opts_models:
+                    pk_field_type = opts_models[0]._meta.pk.db_type(connection)
+                else:
+                    pk_field_type = 'integer'
+                ct = self.create(
+                    service=service_create, app_label=app_label, model=model_name, api_slug=f'{service_create}.{model_name}', pk_field_type=pk_field_type
+                )
                 self._add_to_cache(self.db, ct)
                 for model in opts_models:
                     results[model] = ct
@@ -188,6 +196,16 @@ class DABContentType(django_models.Model):
         on_delete=django_models.SET_NULL,
         related_name='child_content_types',
     )
+    api_slug = django_models.CharField(
+        max_length=201,  # combines service and model fields with a period in-between
+        default='',  # will be set by the saving or creation logic
+        help_text=_("String to use for references to this type from other models in the API."),
+    )
+    pk_field_type = django_models.CharField(
+        max_length=100,
+        default='integer',
+        help_text=_("Database field type of the primary key field of the model, relevant for interal logic tracking permissions."),
+    )
 
     objects = DABContentTypeManager()
 
@@ -216,6 +234,15 @@ def app_labeled_name(self) -> str:
             return self.model
         return f"{model._meta.app_config.verbose_name} | {model._meta.verbose_name}"
 
+    def save(self, *args, **kwargs):
+        # Set the api_slug field if it is not synchronized to other fields
+        api_slug = f'{self.service}.{self.model}'
+        if api_slug != self.api_slug:
+            self.api_slug = api_slug
+            if update_fields := kwargs.get('update_fields', []):
+                update_fields.append('api_slug')
+        return super().save(*args, **kwargs)
+
     def model_class(self) -> Union[Type[django_models.Model], Type[RemoteObject]]:
         """Return the model class or a stand-in.
 

ansible_base/rbac/models/permission.py

@@ -27,6 +27,11 @@ class DABPermission(models.Model):
             )
         ),
     )
+    api_slug = models.CharField(
+        max_length=201,  # combines content_type.service and codename fields with a period in-between
+        default='',
+        help_text=_("String to use for references to this type from other models in the API."),
+    )
 
     class Meta:
         app_label = 'dab_rbac'
@@ -37,3 +42,12 @@ class Meta:
 
     def __str__(self):
         return f"<{self.__class__.__name__}: {self.codename}>"
+
+    def save(self, *args, **kwargs):
+        # Set the api_slug field if it is not synchronized to other fields
+        api_slug = f'{self.content_type.service}.{self.codename}'
+        if api_slug != self.api_slug:
+            self.api_slug = api_slug
+            if update_fields := kwargs.get('update_fields', []):
+                update_fields.append('api_slug')
+        return super().save(*args, **kwargs)

ansible_base/rbac/remote.py

@@ -36,6 +36,18 @@ def get_ct_from_type(cls):
         ct_model = apps.get_model('dab_rbac', 'DABContentType')
         return ct_model.objects.get_by_natural_key(cls._meta.service, cls._meta.app_label, cls._meta.model_name)
 
+    @classmethod
+    def access_ids_qs(cls, actor, codename: str = 'view', content_types=None, cast_field=None):
+        """Returns a values_list type queryset of ids
+
+        Remote objects do not exist locally, so we can not get a queryset of them,
+        but we can still do this, giving a queryset of ids.
+        You could use the materialized list to filter API endpoints on the remote server?
+        """
+        from .evaluations import remote_obj_id_qs
+
+        return remote_obj_id_qs(actor, remote_cls=cls, codename=codename, content_types=content_types, cast_field=cast_field)
+
 
 def get_remote_base_class() -> Type[RemoteObject]:
     """Return the class which represents remote objects.
@@ -80,8 +92,7 @@ def get_resource_prefix(model: Union[Type[models.Model], models.Model, Type[Remo
     """
     if isinstance(model, RemoteObject) or (inspect.isclass(model) and issubclass(model, RemoteObject)):
         # If it is a remote object, it was only ever created from this to begin with
-        service, _, _ = model.type_data
-        return service
+        return model._meta.model_name
 
     if registry := get_resource_registry():
         # duplicates logic in ansible_base/resource_registry/apps.py

ansible_base/rbac/validators.py

@@ -1,3 +1,4 @@
+import inspect
 import re
 from collections import defaultdict
 from typing import Optional, Type, Union
@@ -27,8 +28,16 @@ def prnt_codenames(codename_set: set[str]) -> str:
     return ', '.join(codename_set)
 
 
-def codenames_for_cls(cls) -> list[str]:
+def codenames_for_remote_cls(cls: Union[Type[RemoteObject], RemoteObject]) -> list[str]:
+    """For remote objects, we have to use the database to get its known permissions"""
+    ct = cls.get_ct_from_type()
+    return [permission.codename for permission in ct.dab_permissions.all()]
+
+
+def codenames_for_cls(cls: Union[Model, Type[Model], Type[RemoteObject], RemoteObject]) -> list[str]:
     "Helper method that gives the Django permission codenames for a given class"
+    if (inspect.isclass(cls) and issubclass(cls, RemoteObject)) or isinstance(cls, RemoteObject):
+        return codenames_for_remote_cls(cls)
     return [t[0] for t in cls._meta.permissions] + [f'{act}_{cls._meta.model_name}' for act in cls._meta.default_permissions]
 
 
@@ -115,11 +124,8 @@ def check_view_permission_criteria(codename_set: set[str], permissions_by_model:
     because being able to change a thing without the ability to see it makes no sense.
     """
     for cls, valid_model_permissions in permissions_by_model.items():
-        # if issubclass(cls, RemoteObject):
-        #     from .models.content_type import DABContentType
-
-        #     cls_ct = cls.get_ct_from_type()
-        #     if any(permission.codename.startswith('view') for permission in cls_ct.permissions.all()):
+        # NOTE: there is some concern about using valid_model_permissions here, as opposed to all model permissions
+        # however, no specific issue has yet been identified for this
         if any('view' in codename for codename in valid_model_permissions):
             model_permissions = set(valid_model_permissions) & codename_set
             local_codenames = {codename for codename in model_permissions if not is_add_perm(codename)}
@@ -193,7 +199,7 @@ def validate_permissions_for_model(permissions, content_type: Optional[Model], m
                 raise ValidationError({'permissions', 'Local custom roles can only include view permission for shared models'})
 
 
-def validate_codename_for_model(codename: str, model: Union[Model, Type[Model]]) -> str:
+def validate_codename_for_model(codename: str, model: Union[Model, Type[Model], Type[RemoteObject], RemoteObject]) -> str:
     """Shortcut method and validation to allow action name, codename, or app_name.codename
 
     This institutes a shortcut for easier use of the evaluation methods

test_app/tests/rbac/remote/test_remote_models.py

@@ -1,6 +1,6 @@
 import pytest
 
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleUserAssignment
 from ansible_base.rbac.remote import RemoteObject
 
 
@@ -11,4 +11,10 @@ def test_give_remote_permission(rando):
     DABPermission.objects.create(codename='foo_foo', content_type=foo_type)
     rd = RoleDefinition.objects.create_from_permissions(name='Foo fooers for the foos in foo service', permissions=['foo.foo_foo'], content_type=foo_type)
     a_foo = RemoteObject(content_type=foo_type, object_id=42)
-    rd.give_permission(rando, a_foo)
+    assignment = rd.give_permission(rando, a_foo)
+
+    assignment = RoleUserAssignment.objects.get(pk=assignment.pk)
+    assert isinstance(assignment.content_object, RemoteObject)
+
+    # We can do evaluation querysets, but these can not return objects, just id values
+    assert set(foo_type.model_class().access_ids_qs(actor=rando, codename='foo')) == {(int(assignment.object_id),)}