Disable type creation and test unregistered model

Author: AlanCoding

ansible_base/rbac/management/__init__.py

@@ -1,7 +1,7 @@
 import logging
 
 from django.apps import apps as global_apps
-from django.db import DEFAULT_DB_ALIAS, router, connection
+from django.db import DEFAULT_DB_ALIAS, connection, router
 
 from ansible_base.rbac import permission_registry
 from ansible_base.rbac.remote import get_resource_prefix
@@ -102,9 +102,12 @@ def sync_DAB_permissions(verbosity=2, using=DEFAULT_DB_ALIAS, apps=global_apps):
         table = DABContentType._meta.db_table
         pk_column = DABContentType._meta.pk.column
         with connection.cursor() as cursor:
-            cursor.execute(f"""
+            cursor.execute(
+                f"""
                 SELECT setval(
                     pg_get_serial_sequence(%s, %s),
                     (SELECT MAX({pk_column}) FROM {table})
                 )
-            """, [table, pk_column])
+            """,
+                [table, pk_column],
+            )

ansible_base/rbac/models/content_type.py

@@ -1,18 +1,14 @@
 import inspect
-import logging
 from collections import defaultdict
 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 _
 
 from ..remote import RemoteObject, get_local_resource_prefix, get_resource_prefix
 
-logger = logging.getLogger(__name__)
-
 
 class DABContentTypeManager(django_models.Manager["DABContentType"]):
     """Manager storing DABContentType objects in a local cache like original ContentType.
@@ -76,12 +72,9 @@ def get_for_model(
         try:
             ct = self.get(service=service, app_label=opts.app_label, model=opts.model_name)
         except self.model.DoesNotExist:
-            logger.warning(f'Could not find content type for {(service, opts.app_label, opts.model_name)}, so creating new')
-            ct, _ = self.get_or_create(
-                service=service,
-                app_label=opts.app_label,
-                model=opts.model_name,
-                defaults=dict(api_slug=f'{service}.{opts.model_name}', pk_field_type=model._meta.pk.db_type(connection)),
+            raise RuntimeError(
+                f'Could not find content type for {(service, opts.app_label, opts.model_name)}, '
+                'and creating new objects via get_for_model is not allowed for DAB RBAC'
             )
         self._add_to_cache(self.db, ct)
         return ct
@@ -135,19 +128,11 @@ def get_for_models(
                 for model in opts_models:
                     results[model] = ct
                 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():
-                if opts_models:
-                    pk_field_type = opts_models[0]._meta.pk.db_type(connection)
-                else:
-                    pk_field_type = 'integer'
-                logger.warning(f'Could not find content type for {(service_create, app_label, model_name)}, so creating new, out of:\n{needed_models.keys()}')
-                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
+            if needed_opts:
+                raise RuntimeError(
+                    f'Could not find content type for any of {needed_opts.keys()}, '
+                    f'and creating new objects via get_for_models is not enabled for DAB RBAC, looked in:\n{needed_models.keys()}'
                 )
-                self._add_to_cache(self.db, ct)
-                for model in opts_models:
-                    results[model] = ct
         return results
 
     def get_by_natural_key(self, *args: str) -> "DABContentType":

ansible_base/rbac/models/role.py

@@ -69,6 +69,9 @@ def contribute_to_class(self, cls: Type[models.Model], name: str) -> None:
         self.managed = ManagedRoleManager(self.model._meta.apps)
 
     def give_creator_permissions(self, user, obj) -> Optional['RoleUserAssignment']:
+        if not permission_registry.is_registered(obj):
+            return  # Exit before getting content type, which will not exist
+
         # If the user is a superuser, no need to bother giving the creator permissions
         for super_flag in settings.ANSIBLE_BASE_BYPASS_SUPERUSER_FLAGS:
             if getattr(user, super_flag):

test_app/tests/rbac/features/test_creator_permission.py

@@ -1,7 +1,8 @@
 import pytest
+from django.apps import apps
 from django.test.utils import override_settings
 
-from ansible_base.rbac.models import DABContentType, RoleDefinition, RoleEvaluation
+from ansible_base.rbac.models import DABContentType, RoleDefinition, RoleEvaluation, RoleUserAssignment
 from test_app.models import User
 
 INVENTORY_OBJ_PERMS = ('change_inventory', 'update_inventory', 'view_inventory', 'delete_inventory')
@@ -60,3 +61,18 @@ def test_custom_creation_perms(rando, inventory):
     with override_settings(ANSIBLE_BASE_CREATOR_DEFAULTS=['change', 'view']):
         RoleDefinition.objects.give_creator_permissions(rando, inventory)
         assert set(perm_name.split('_', 1)[0] for perm_name in RoleEvaluation.get_permissions(rando, inventory)) == {'change', 'view'}
+
+
+@pytest.mark.django_db
+def test_creator_permission_for_unregistered_model(admin_api_client, inventory, inv_rd, user):
+    prior_ct = DABContentType.objects.count()
+    prior_assignments = RoleUserAssignment.objects.count()
+    prior_rds = RoleDefinition.objects.count()
+
+    cls = apps.get_model('test_app.secretcolor')
+    obj = cls.objects.create()
+    RoleDefinition.objects.give_creator_permissions(user, obj)  # should do nothing
+
+    assert DABContentType.objects.count() == prior_ct  # did not create anything
+    assert RoleUserAssignment.objects.count() == prior_assignments
+    assert RoleDefinition.objects.count() == prior_rds

test_app/tests/rbac/remote/test_public_api_compat.py

@@ -20,7 +20,6 @@ def test_role_definition_list_remote_and_local(admin_api_client, inv_rd, foo_rd)
 
 
 @pytest.mark.django_db
-@pytest.mark.skip
 def test_create_remote_role_definition(admin_api_client, foo_type, foo_permission):
     """
     Test creation of a custom, remote role definition.