From 7242fe89ae247ef874f187d245b5925e570ae46f Mon Sep 17 00:00:00 2001 From: Dirk Julich Date: Tue, 11 Aug 2026 21:40:01 +0200 Subject: [PATCH] =?UTF-8?q?AAP-87084=20=E2=80=94=20Replace=20UNION=20with?= =?UTF-8?q?=20OR=20+=20pre-computed=20role=20set=20in=20unified=20job=20RB?= =?UTF-8?q?AC=20query=20(#16577)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * AAP-87084 — Replace UNION with OR + pre-computed role set in unified job RBAC query The UNION ALL approach introduced by AAP-81173 forces PostgreSQL to materialize all accessible job IDs across four RBAC branches before any filtering or LIMIT can apply. This causes a 10x per-call regression on the unified jobs list (36ms → 366ms) and a 3x regression on the dashboard date-bucketed aggregation (88.9ms → 261.5ms). Pre-compute the user's role IDs once as a Python list and pass them as literal parameters to OR-based RoleEvaluation filters. This eliminates the 4x redundant roleuserassignment subquery scans and restores single-pass filtering with early LIMIT exit. Resolves: AAP-87084, AAP-87087 * Assert RBAC query list is non-empty before checking for UNION Ensures the test does not pass vacuously if no RBAC query is captured. * Add test for singleton permission shortcut branches Exercises the UJT, inventory, and org auditor singleton permission paths in filtered_queryset() to increase coverage on new code. * Add test for team-granted permissions in unified job list Verifies that a user who can view a job template only through a team assignment (not direct) sees the corresponding unified jobs. --------- Co-authored-by: Claude Opus 4.6 --- awx/main/access.py | 91 +++++++++++++------ .../dab_rbac/test_unified_job_access.py | 67 +++++++++++++- 2 files changed, 127 insertions(+), 31 deletions(-) diff --git a/awx/main/access.py b/awx/main/access.py index 0c228357e8..e3410b2441 100644 --- a/awx/main/access.py +++ b/awx/main/access.py @@ -12,6 +12,7 @@ from django.conf import settings from django.db.models import Q, Prefetch from django.contrib.auth.models import User from django.utils.translation import gettext_lazy as _ +from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist # Django REST Framework @@ -19,7 +20,7 @@ from rest_framework.exceptions import ParseError, PermissionDenied # django-ansible-base from ansible_base.lib.utils.validation import to_python_boolean -from ansible_base.rbac.models import RoleEvaluation +from ansible_base.rbac.models import RoleEvaluation, RoleUserAssignment from ansible_base.rbac.policies import visible_users from ansible_base.rbac import permission_registry @@ -2507,39 +2508,75 @@ class UnifiedJobAccess(BaseAccess): # ) def filtered_queryset(self): - inv_pk_qs = Inventory.access_ids_qs(self.user, 'view') + # AAP-87084 / AAP-87087: Pre-compute role IDs once to avoid 4x redundant + # roleuserassignment scans, and use OR (not UNION) to allow single-pass + # filtering with early LIMIT exit. + user_role_ids = list(RoleUserAssignment.objects.filter(user_id=self.user.id).values_list('object_role_id', flat=True)) + if not user_role_ids: + return self.model.objects.none() - by_template = ( - self.model.objects.filter(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role')) - .order_by() - .values_list('pk', flat=True) - ) + user_singletons = self.user.singleton_permissions() - by_inventory_update = ( - InventoryUpdate.objects.filter( - inventory_source__inventory__id__in=inv_pk_qs, + role_subclasses = UnifiedJobTemplate._submodels_with_roles() + ujt_codenames = [f'view_{cls._meta.model_name}' for cls in role_subclasses] + if not (set(ujt_codenames) - user_singletons): + ujt_accessible = UnifiedJobTemplate.objects.filter(polymorphic_ctype__in=ContentType.objects.get_for_models(*role_subclasses).values()).values_list( + 'id', flat=True ) - .order_by() - .values_list('pk', flat=True) - ) - - by_adhoc = ( - AdHocCommand.objects.filter( - inventory__id__in=inv_pk_qs, + else: + dab_role_cts = permission_registry.content_type_model.objects.get_for_models(*role_subclasses).values() + ujt_accessible = ( + RoleEvaluation.objects.filter( + role_id__in=user_role_ids, + codename__in=ujt_codenames, + content_type_id__in=[ct.id for ct in dab_role_cts], + ) + .values_list('object_id') + .distinct() ) - .order_by() - .values_list('pk', flat=True) - ) - by_org_auditor = ( - self.model.objects.filter( - organization__in=Organization.access_ids_qs(self.user, 'audit_organization'), + if 'view_inventory' in user_singletons: + inv_accessible = Inventory.objects.values_list('id', flat=True) + else: + inv_ct_id = permission_registry.content_type_model.objects.get_for_model(Inventory).id + inv_accessible = ( + RoleEvaluation.objects.filter( + role_id__in=user_role_ids, + codename='view_inventory', + content_type_id=inv_ct_id, + ) + .values_list('object_id') + .distinct() ) - .order_by() - .values_list('pk', flat=True) - ) - return self.model.objects.filter(pk__in=by_template.union(by_inventory_update, by_adhoc, by_org_auditor)) + if 'audit_organization' in user_singletons: + org_accessible = Organization.objects.values_list('id', flat=True) + else: + org_ct_id = permission_registry.content_type_model.objects.get_for_model(Organization).id + org_accessible = ( + RoleEvaluation.objects.filter( + role_id__in=user_role_ids, + codename='audit_organization', + content_type_id=org_ct_id, + ) + .values_list('object_id') + .distinct() + ) + + return self.model.objects.filter( + Q(unified_job_template_id__in=ujt_accessible) + | Q( + pk__in=InventoryUpdate.objects.filter( + inventory_source__inventory__id__in=inv_accessible, + ).values('pk') + ) + | Q( + pk__in=AdHocCommand.objects.filter( + inventory__id__in=inv_accessible, + ).values('pk') + ) + | Q(organization__in=org_accessible) + ) def get_queryset(self): return super(UnifiedJobAccess, self).get_queryset().filter(workflowapproval__isnull=True) diff --git a/awx/main/tests/functional/dab_rbac/test_unified_job_access.py b/awx/main/tests/functional/dab_rbac/test_unified_job_access.py index eaec23de4c..ee8df9828a 100644 --- a/awx/main/tests/functional/dab_rbac/test_unified_job_access.py +++ b/awx/main/tests/functional/dab_rbac/test_unified_job_access.py @@ -13,13 +13,14 @@ from awx.main.models import ( JobTemplate, Organization, Project, + Team, UnifiedJob, ) @pytest.mark.django_db -def test_unified_job_list_uses_union(user, organization, inventory, setup_managed_roles, get): - """The unified job list RBAC query uses UNION instead of OR to allow per-branch query planning.""" +def test_unified_job_list_uses_or_not_union(user, organization, inventory, setup_managed_roles, get): + """The unified job list RBAC query uses OR-based filtering, not UNION.""" org_admin = user('uj-org-admin') RoleDefinition.objects.get(name='Organization Admin').give_permission(org_admin, organization) @@ -38,8 +39,10 @@ def test_unified_job_list_uses_union(user, organization, inventory, setup_manage assert response.status_code == 200 assert response.data['count'] >= 3 - uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'UNION' in q['sql'] and 'main_unifiedjob' in q['sql']] - assert uj_rbac_queries, "Expected at least one query using UNION for unified job RBAC filtering" + uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'main_unifiedjob' in q['sql'] and 'dab_rbac_roleevaluation' in q['sql']] + assert uj_rbac_queries, "Expected a unified-job RBAC query" + for sql in uj_rbac_queries: + assert 'UNION' not in sql, "RBAC query should use OR, not UNION" @pytest.mark.django_db @@ -77,6 +80,62 @@ def test_unified_job_list_inventory_viewer_sees_inventory_updates(user, setup_ma assert inv_update.pk in result_ids +@pytest.mark.django_db +def test_unified_job_list_singleton_permissions(user, organization, inventory, setup_managed_roles, get): + """Users with global (singleton) view permissions see jobs via shortcut paths + that bypass RoleEvaluation queries entirely.""" + singleton_user = user('uj-singleton') + RoleDefinition.objects.get(name='Organization Admin').give_permission(singleton_user, organization) + + project = Project.objects.create(name='uj-singleton-project', organization=organization) + jt = JobTemplate.objects.create(name='uj-singleton-jt', project=project, inventory=inventory, organization=organization) + job = jt.create_unified_job() + + inv_src = InventorySource.objects.create(name='uj-singleton-invsrc', inventory=inventory, source='ec2') + inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source) + + adhoc = AdHocCommand.objects.create(name='uj-singleton-adhoc', inventory=inventory) + + # Inject singleton permissions to exercise the shortcut branches in + # filtered_queryset() without needing a global RoleDefinition. + singleton_user._singleton_permissions = { + 'view_jobtemplate', + 'view_project', + 'view_workflowjobtemplate', + 'view_inventory', + 'audit_organization', + } + + response = get(reverse('api:unified_job_list'), singleton_user) + assert response.status_code == 200 + result_ids = [r['id'] for r in response.data['results']] + assert job.pk in result_ids + assert inv_update.pk in result_ids + assert adhoc.pk in result_ids + + +@pytest.mark.django_db +def test_unified_job_list_team_member_sees_team_granted_jobs(user, setup_managed_roles, get): + """A user who can view a JT only through a team assignment must see + the corresponding unified jobs in the list.""" + org = Organization.objects.create(name='uj-team-org') + team = Team.objects.create(name='uj-test-team', organization=org) + team_user = user('uj-team-member') + + RoleDefinition.objects.get(name='Team Member').give_permission(team_user, team) + + inventory = org.inventories.create(name='uj-team-inv') + project = Project.objects.create(name='uj-team-project', organization=org) + jt = JobTemplate.objects.create(name='uj-team-jt', project=project, inventory=inventory, organization=org) + RoleDefinition.objects.get(name='JobTemplate Execute').give_permission(team, jt) + job = jt.create_unified_job() + + response = get(reverse('api:unified_job_list'), team_user) + assert response.status_code == 200 + result_ids = [r['id'] for r in response.data['results']] + assert job.pk in result_ids, f"Team member should see job {job.pk} via team-granted JT execute permission, " f"but got result IDs: {result_ids}" + + @pytest.mark.django_db def test_unified_job_list_rando_sees_nothing(rando, setup_managed_roles, get): """Unprivileged user sees no unified jobs."""