diff --git a/awx/main/migrations/0208_fix_system_auditor_migration.py b/awx/main/migrations/0208_fix_system_auditor_migration.py new file mode 100644 index 0000000000..cfb207a016 --- /dev/null +++ b/awx/main/migrations/0208_fix_system_auditor_migration.py @@ -0,0 +1,45 @@ +import logging + +from django.db import migrations + +logger = logging.getLogger('awx.main.migrations') + + +def fix_system_auditor_assignments(apps, schema_editor): + """ + Corrective migration for a bug in 0192 (migrate_to_new_rbac) where a stale + loop variable caused old system_auditor members to not receive Platform Auditor. + Only adds missing assignments — does not remove any, since we cannot safely + distinguish bug-granted assignments from legitimate ones (e.g. from 0202 or admin actions). + """ + Role = apps.get_model('main', 'Role') + RoleDefinition = apps.get_model('dab_rbac', 'RoleDefinition') + RoleUserAssignment = apps.get_model('dab_rbac', 'RoleUserAssignment') + + platform_auditor = RoleDefinition.objects.filter(name='Platform Auditor').first() + if not platform_auditor: + return + + old_system_auditor = Role.objects.filter(singleton_name='system_auditor').first() + if not old_system_auditor: + return + + expected_user_ids = set(old_system_auditor.members.values_list('id', flat=True)) + current_user_ids = set(RoleUserAssignment.objects.filter(role_definition=platform_auditor).values_list('user_id', flat=True)) + missing_user_ids = expected_user_ids - current_user_ids + + for user_id in missing_user_ids: + RoleUserAssignment.objects.create(user_id=user_id, role_definition=platform_auditor) + + if missing_user_ids: + logger.info(f'Fixed {len(missing_user_ids)} missing Platform Auditor assignments for old system_auditor members') + + +class Migration(migrations.Migration): + dependencies = [ + ('main', '0207_alter_skip_tags_to_textfield'), + ] + + operations = [ + migrations.RunPython(fix_system_auditor_assignments, migrations.RunPython.noop), + ] diff --git a/awx/main/migrations/_dab_rbac.py b/awx/main/migrations/_dab_rbac.py index 2ccb0a6782..3f8cabf6c3 100644 --- a/awx/main/migrations/_dab_rbac.py +++ b/awx/main/migrations/_dab_rbac.py @@ -268,7 +268,7 @@ def migrate_to_new_rbac(apps, schema_editor): if old_system_auditor: # if the system auditor role is not present, this is a new install and no users should exist ct = 0 - for user in role.members.all(): + for user in old_system_auditor.members.all(): RoleUserAssignment.objects.create(user=user, role_definition=new_system_auditor) ct += 1 if ct: diff --git a/awx/main/tests/functional/test_migrations.py b/awx/main/tests/functional/test_migrations.py index caa9579f6a..2081c32f56 100644 --- a/awx/main/tests/functional/test_migrations.py +++ b/awx/main/tests/functional/test_migrations.py @@ -224,3 +224,84 @@ class TestGithubAppBug: assert CredentialType.objects.filter( namespace='github_app_lookup', name='Legacy GitHub App Credential' ).exists(), "CredentialType should be updated to 'github_app_lookup' and retain its name." + + +@pytest.mark.django_db +class TestSystemAuditorMigration: + def test_system_auditor_migrated_to_platform_auditor(self, migrator): + """ + The bug in 0192 used a stale loop variable `role` instead of + `old_system_auditor`, so system_auditor members were not granted + Platform Auditor. With the fix applied, they should be. + """ + old_state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more')) + User = old_state.apps.get_model('auth', 'User') + Role = old_state.apps.get_model('main', 'Role') + + user = User.objects.create(username='sys-auditor-user') + system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'}) + system_auditor_role.members.add(user) + + new_state = migrator.apply_tested_migration(('main', '0192_custom_roles')) + RoleUserAssignment = new_state.apps.get_model('dab_rbac', 'RoleUserAssignment') + + assert RoleUserAssignment.objects.filter( + user=user.id, role_definition__name='Platform Auditor' + ).exists(), "system_auditor member should have been granted Platform Auditor" + + def test_fix_system_auditor_migration_backfills_missing(self, migrator): + """ + Migration 0208 should backfill Platform Auditor for any + system_auditor members who were missed by the original migration. + """ + state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more')) + User = state.apps.get_model('auth', 'User') + Role = state.apps.get_model('main', 'Role') + + user = User.objects.create(username='missed-auditor') + system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'}) + system_auditor_role.members.add(user) + + state = migrator.apply_tested_migration(('main', '0192_custom_roles')) + RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment') + RoleDefinition = state.apps.get_model('dab_rbac', 'RoleDefinition') + + # Simulate the pre-fix state: remove the Platform Auditor assignment + platform_auditor = RoleDefinition.objects.get(name='Platform Auditor') + RoleUserAssignment.objects.filter(user=user.id, role_definition=platform_auditor).delete() + assert not RoleUserAssignment.objects.filter( + user=user.id, role_definition=platform_auditor + ).exists(), "Setup: user should not have Platform Auditor before corrective migration" + + state = migrator.apply_tested_migration(('main', '0208_fix_system_auditor_migration')) + RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment') + + assert RoleUserAssignment.objects.filter( + user=user.id, role_definition__name='Platform Auditor' + ).exists(), "Corrective migration should have backfilled Platform Auditor" + + def test_fix_system_auditor_migration_no_duplicates(self, migrator): + """ + Migration 0208 must be idempotent — users who already have + Platform Auditor should not receive a duplicate assignment. + """ + state = migrator.apply_initial_migration(('main', '0190_alter_inventorysource_source_and_more')) + User = state.apps.get_model('auth', 'User') + Role = state.apps.get_model('main', 'Role') + + user = User.objects.create(username='already-auditor') + system_auditor_role, _ = Role.objects.get_or_create(singleton_name='system_auditor', defaults={'role_field': 'system_auditor'}) + system_auditor_role.members.add(user) + + state = migrator.apply_tested_migration(('main', '0192_custom_roles')) + RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment') + + # Confirm the user already has the assignment (from the fixed 0192) + assert RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Platform Auditor').exists() + + state = migrator.apply_tested_migration(('main', '0208_fix_system_auditor_migration')) + RoleUserAssignment = state.apps.get_model('dab_rbac', 'RoleUserAssignment') + + assert ( + RoleUserAssignment.objects.filter(user=user.id, role_definition__name='Platform Auditor').count() == 1 + ), "Should have exactly one Platform Auditor assignment, not duplicates"