AAP-54064 Delete DABPermission entries if model is not registered and not used (ansible#851)

This is specifically needed for a prior behavior of `Team` model
auto-registration. This was deleted a few months ago, but old versions
may still create permissions for the team model.

Re-stating, that leaves a scenario where an app might _enable RBAC_ but
_register no models_ and then fail to upgrade. This is written
specifically to address that situation. Previously, we just threw an
error out of an abundance of caution (which was correct, because we
didn't have a specific action clarified).

For this specific case, the remediation action is clear - delete the
permission entries. We still have to be really specific to avoid
deleting unintended things. The criteria should be well-formed in this
patch.

Author: AlanCoding

ansible_base/rbac/migrations/0005_remote_permissions_data.py

@@ -1,62 +1,8 @@
 # Generated by Django 4.2.23 on 2025-06-30 12:48
 
-import logging
-
 from django.db import migrations
 
-
-logger = logging.getLogger(__name__)
-
-
-def create_types_if_needed(apps, schema_editor):
-    """Before we can migrate to the new DABContentType, entries in that table must be created.
-
-    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
-    Only needed in the upgrade case, otherwise better to run at true post-migrate.
-    """
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
-    if permission_cls.objects.exists() or rd_cls.objects.exists():
-        logger.info('Running DABContentType creation script as part of 0005 migration')
-        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
-
-        create_DAB_contenttypes(apps=apps)
-
-
-def migrate_content_type(apps, schema_editor):
-    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
-    ct_cls.objects.clear_cache()
-    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
-        cls = apps.get_model('dab_rbac', model_name)
-        update_ct = 0
-        for obj in cls.objects.all():
-            old_ct = obj.content_type
-            if old_ct:
-                try:
-                    # NOTE: could give duplicate normally, but that is impossible in migration path
-                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
-                except Exception as e:
-                    raise RuntimeError(
-                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
-                    ) from e
-                obj.save()
-                update_ct += 1
-        if update_ct:
-            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
-    for model_name in ('roleevaluation', 'roleevaluationuuid'):
-        cls = apps.get_model('dab_rbac', model_name)
-        cls.objects.all().delete()
-
-    # DABPermission model had api_slug added in last migration
-    # if records existed before this point, it needs to be filled in
-    mod_ct = 0
-    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
-    for permission in permission_cls.objects.all():
-        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
-        permission.save()
-        mod_ct += 1
-    if mod_ct:
-        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+from . import _utils
 
 
 class Migration(migrations.Migration):
@@ -66,6 +12,6 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.RunPython(create_types_if_needed, migrations.RunPython.noop),
-        migrations.RunPython(migrate_content_type, migrations.RunPython.noop),
+        migrations.RunPython(_utils.create_types_if_needed, migrations.RunPython.noop),
+        migrations.RunPython(_utils.migrate_content_type, migrations.RunPython.noop),
     ]

ansible_base/rbac/migrations/_utils.py

@@ -1,5 +1,10 @@
+# Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+import logging
+
 from django.db import models
 
+logger = logging.getLogger(__name__)
+
 # This method has moved, and this is put here temporarily to make branch management easier
 from ansible_base.rbac.management import create_dab_permissions as create_custom_permissions  # noqa
 
@@ -45,3 +50,127 @@ def give_permissions(apps, rd, users=(), teams=(), object_id=None, content_type_
                 for team_id in teams
             ]
         RoleTeamAssignment.objects.bulk_create(team_assignments, ignore_conflicts=True)
+
+
+def cleanup_orphaned_permissions(apps):
+    """
+    Delete orphaned DABPermission objects for models no longer in the permission registry.
+
+    This is used during migrations to clean up permissions for any previously-registered model
+    that are no longer tracked by RBAC, but only if they are not referenced by any RoleDefinition.
+
+    Args:
+        apps: Django apps registry (from migration context)
+
+    Returns:
+        int: Number of permissions deleted
+    """
+    # Get model classes from apps registry
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    role_definition_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+
+    # Get permission registry to check which models are registered
+    from ansible_base.rbac import permission_registry
+    registered_model_keys = set()
+    for model in permission_registry._registry:
+        registered_model_keys.add((model._meta.app_label, model._meta.model_name))
+
+    # Find orphaned permissions
+    orphaned_permissions = []
+    for permission in permission_cls.objects.all():
+        if permission.content_type:
+            model_key = (permission.content_type.app_label, permission.content_type.model)
+            if model_key not in registered_model_keys:
+                # Check if this permission is referenced by any RoleDefinition
+                referencing_roles = role_definition_cls.objects.filter(permissions=permission)
+                if referencing_roles.exists():
+                    # Log warning for unregistered model still referenced by role definitions
+                    role_names = list(referencing_roles.values_list('name', flat=True))
+                    logger.warning(
+                        f'Permission {permission.codename} for unregistered model '
+                        f'{permission.content_type.app_label}.{permission.content_type.model} '
+                        f'is still referenced by role definitions: {role_names}'
+                    )
+                else:
+                    logger.info(f'Deleting orphaned permission {permission.codename} for unregistered model {model_key}')
+                    orphaned_permissions.append(permission)
+
+    # Delete orphaned permissions
+    deleted_count = 0
+    if orphaned_permissions:
+        deleted_count = len(orphaned_permissions)
+        permission_cls.objects.filter(id__in=[p.id for p in orphaned_permissions]).delete()
+        logger.info(f'Deleted {deleted_count} orphaned DABPermission objects for unregistered models')
+
+    return deleted_count
+
+
+def migrate_content_type(apps, schema_editor):
+    """
+    Migrate content type references from Django ContentType to DABContentType.
+
+    This function handles the migration of content type references across all RBAC models
+    from the old Django ContentType to the new DABContentType system.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+    """
+    # Pre-check: Delete orphaned DABPermission objects before migration
+    cleanup_orphaned_permissions(apps)
+
+    ct_cls = apps.get_model('dab_rbac', 'DABContentType')
+    ct_cls.objects.clear_cache()
+
+    for model_name in ('dabpermission', 'objectrole', 'roledefinition', 'roleuserassignment', 'roleteamassignment'):
+        cls = apps.get_model('dab_rbac', model_name)
+        update_ct = 0
+        for obj in cls.objects.all():
+            old_ct = obj.content_type
+            if old_ct:
+                try:
+                    # NOTE: could give duplicate normally, but that is impossible in migration path
+                    obj.new_content_type = ct_cls.objects.get_by_natural_key(old_ct.app_label, old_ct.model)
+                except Exception as e:
+                    raise RuntimeError(
+                        f"Failed to get new content type for a {model_name} pk={obj.pk}, obj={obj.__dict__}"
+                    ) from e
+                obj.save()
+                update_ct += 1
+        if update_ct:
+            logger.info(f'Updated content_type reference to new model for {model_name} for {update_ct} entries')
+    for model_name in ('roleevaluation', 'roleevaluationuuid'):
+        cls = apps.get_model('dab_rbac', model_name)
+        cls.objects.all().delete()
+
+    # DABPermission model had api_slug added in last migration
+    # if records existed before this point, it needs to be filled in
+    mod_ct = 0
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    for permission in permission_cls.objects.all():
+        permission.api_slug = f'{permission.new_content_type.service}.{permission.codename}'
+        permission.save()
+        mod_ct += 1
+    if mod_ct:
+        logger.info(f'Set new field DABPermission.api_slug for {mod_ct} existing permissions')
+
+
+def create_types_if_needed(apps, schema_editor):
+    """
+    Create DABContentType entries if needed before migration.
+
+    Before we can migrate to the new DABContentType, entries in that table must be created.
+    This method runs what is ordinarily the post_migrate logic, but in the migration case here.
+    Only needed in the upgrade case, otherwise better to run at true post-migrate.
+
+    Args:
+        apps: Django apps registry (from migration context)
+        schema_editor: Django schema editor (unused but required for migration signature)
+    """
+    permission_cls = apps.get_model('dab_rbac', 'DABPermission')
+    rd_cls = apps.get_model('dab_rbac', 'RoleDefinition')
+    if permission_cls.objects.exists() or rd_cls.objects.exists():
+        logger.info('Running DABContentType creation script as part of 0005 migration')
+        from ansible_base.rbac.management.create_types import create_DAB_contenttypes
+
+        create_DAB_contenttypes(apps=apps)

test_app/tests/rbac/test_migrations.py

@@ -1,8 +1,8 @@
 import pytest
 from django.apps import apps
 
-from ansible_base.rbac.migrations._utils import give_permissions
-from ansible_base.rbac.models import DABContentType, DABPermission, RoleTeamAssignment, RoleUserAssignment
+from ansible_base.rbac.migrations._utils import cleanup_orphaned_permissions, give_permissions
+from ansible_base.rbac.models import DABContentType, DABPermission, RoleDefinition, RoleTeamAssignment, RoleUserAssignment
 from ansible_base.rbac.permission_registry import permission_registry
 from test_app.models import Team, User
 
@@ -33,3 +33,53 @@ def test_give_permissions_by_id(organization, inventory, inv_rd):
 def test_permission_migration():
     "These are expected to be created via a post_migrate signal just like auth.Permission"
     assert len(DABPermission.objects.order_by('content_type').values_list('content_type').distinct()) == len(permission_registry.all_registered_models)
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_deletes_unregistered_unreferenced():
+    """Test that cleanup_orphaned_permissions deletes permissions for unregistered models that are not referenced by any RoleDefinition.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    # Create a DABContentType for a model not in the permission registry
+    dab_fake_ct = DABContentType.objects.create(app_label='fake_app', model='fakemodel', service='local')
+
+    # Create a permission for this fake model (should be deleted)
+    orphaned_permission = DABPermission.objects.create(name='Can view fake model', content_type=dab_fake_ct, codename='view_fakemodel')
+
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify the orphaned permission was deleted
+    assert deleted_count == 1
+    assert DABPermission.objects.count() == initial_count - 1
+    assert not DABPermission.objects.filter(id=orphaned_permission.id).exists()
+
+
+@pytest.mark.django_db
+def test_cleanup_orphaned_permissions_preserves_referenced():
+    """Test that cleanup_orphaned_permissions preserves permissions for unregistered models that are referenced by RoleDefinitions.
+
+    Generated by Claude Sonnet 4 (claude-sonnet-4@20250514)
+    """
+    # Create a DABContentType for a model not in the permission registry
+    dab_fake_ct = DABContentType.objects.create(app_label='fake_app', model='fakemodel', service='local')
+
+    # Create a permission for this fake model
+    referenced_permission = DABPermission.objects.create(name='Can view fake model', content_type=dab_fake_ct, codename='view_fakemodel')
+
+    # Create a RoleDefinition that references this permission
+    role_def = RoleDefinition.objects.create(name='Fake Role', content_type=dab_fake_ct)
+    role_def.permissions.add(referenced_permission)
+
+    initial_count = DABPermission.objects.count()
+
+    # Run cleanup
+    deleted_count = cleanup_orphaned_permissions(apps)
+
+    # Verify the referenced permission was NOT deleted
+    assert deleted_count == 0
+    assert DABPermission.objects.count() == initial_count
+    assert DABPermission.objects.filter(id=referenced_permission.id).exists()