External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-10-29 22:03:20 -02:30
Code Issues Packages Projects Releases Wiki Activity

AAP-87084 — Revert UnifiedJob queryset to DAB APIs, scope actor role filter (#16599)

Browse Source 
* AAP-87084 — Revert UnifiedJob queryset to DAB APIs, scope actor role filter

The previous patch (7242fe89) inlined all DAB RBAC internals into
UnifiedJobAccess.filtered_queryset to pre-compute role IDs and avoid
repeated subqueries.  While it fixed the UNION regression from c84575c,
it created 70 lines of duplicated logic inconsistent with the
UnifiedJobTemplate path and fetched ALL user roles regardless of
relevance.

This commit:

- Reverts filtered_queryset to the clean post-f8fa690 form that
  delegates to DAB's access_ids_qs / accessible_pk_qs APIs
- Adds _scoped_actor_role_filter() as a temporary shim that pre-filters
  RoleUserAssignment by content_type_id, auto-including Org and Team CTs
  for inherited permissions and NULL for global roles
- Wires the scoped filter into UnifiedJobTemplate.accessible_pk_qs
- Adds tests for custom global roles, mixed permissions, and unrelated
  role exclusion

The shim duplicates logic planned for DAB's _actor_role_filter
content_type_ids parameter; once that merges, replace and delete.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Remove unused ContentType and RoleUserAssignment imports to fix flake8

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* AAP-87084 — Remove _scoped_actor_role_filter shim, use DAB _actor_role_filter directly

The scoped content_type_ids parameter now lives in DAB's
_actor_role_filter. The AWX shim and its hardcoded Org/Team CT
inclusions are no longer needed. The unified job accessible_pk_qs
uses the unscoped _actor_role_filter for correctness.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Alan Rominger
2026-08-18 15:29:54 -04:00
committed by GitHub
co-authored by Claude Opus 4.6
parent d6675e67e0
commit 70f42d2b7d
2 changed files with 135 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
awx/main/access.py
+6 -61
View File
@@ -12,7 +12,6 @@ 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
@@ -20,7 +19,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, RoleUserAssignment
from ansible_base.rbac.models import RoleEvaluation
from ansible_base.rbac.policies import visible_users
from ansible_base.rbac import permission_registry
@@ -2508,74 +2507,20 @@ class UnifiedJobAccess(BaseAccess):
# )
def filtered_queryset(self):
# 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()
user_singletons = self.user.singleton_permissions()
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
)
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()
)
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()
)
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()
)
inv_pk_qs = Inventory.access_ids_qs(self.user, 'view')
return self.model.objects.filter(
Q(unified_job_template_id__in=ujt_accessible)
Q(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
| Q(
pk__in=InventoryUpdate.objects.filter(
inventory_source__inventory__id__in=inv_accessible,
inventory_source__inventory__id__in=inv_pk_qs,
).values('pk')
)
| Q(
pk__in=AdHocCommand.objects.filter(
inventory__id__in=inv_accessible,
inventory__id__in=inv_pk_qs,
).values('pk')
)
| Q(organization__in=org_accessible)
| Q(organization__in=Organization.access_ids_qs(self.user, 'audit_organization'))
)
def get_queryset(self):
awx/main/tests/functional/dab_rbac/test_unified_job_access.py
+129
View File
@@ -8,6 +8,8 @@ from ansible_base.rbac.models import RoleDefinition
from awx.api.versioning import reverse
from awx.main.models import (
AdHocCommand,
Credential,
CredentialType,
InventorySource,
InventoryUpdate,
JobTemplate,
@@ -169,3 +171,130 @@ def test_unified_job_list_pagination_uses_unfiltered_count(rando, setup_managed_
assert response.status_code == 200
assert len(response.data['results']) == 0
assert response.data['count'] == total_jobs
@pytest.mark.django_db
def test_direct_jt_permission_only_sees_jt_jobs(user, setup_managed_roles, get):
"""A user with a direct JT permission (but no inventory or org permissions)
sees only jobs from that JT — inventory updates are not visible."""
org = Organization.objects.create(name='uj-direct-jt-org')
inventory = org.inventories.create(name='uj-direct-jt-inv')
project = Project.objects.create(name='uj-direct-jt-project', organization=org)
jt = JobTemplate.objects.create(name='uj-direct-jt', project=project, inventory=inventory, organization=org)
job = jt.create_unified_job()
inv_src = InventorySource.objects.create(name='uj-direct-jt-invsrc', inventory=inventory, source='ec2')
inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source)
jt_viewer = user('uj-direct-jt-viewer')
RoleDefinition.objects.get(name='JobTemplate Execute').give_permission(jt_viewer, jt)
response = get(reverse('api:unified_job_list'), jt_viewer)
assert response.status_code == 200
result_ids = [r['id'] for r in response.data['results']]
assert job.pk in result_ids, "JT permission holder should see JT jobs"
assert inv_update.pk not in result_ids, "JT permission holder should NOT see inventory updates"
@pytest.mark.django_db
def test_global_ujt_view_singleton_sees_all_template_jobs(user, setup_managed_roles, get):
"""A custom global role granting view on all UJT subclasses triggers
the singleton shortcut and surfaces all template-based jobs."""
org = Organization.objects.create(name='uj-global-ujt-org')
inventory = org.inventories.create(name='uj-global-ujt-inv')
project = Project.objects.create(name='uj-global-ujt-project', organization=org)
jt = JobTemplate.objects.create(name='uj-global-ujt-jt', project=project, inventory=inventory, organization=org)
job = jt.create_unified_job()
global_viewer = user('uj-global-ujt-viewer')
rd = RoleDefinition.objects.create_from_permissions(
name='global-ujt-viewer-test',
permissions=['view_jobtemplate', 'view_project', 'view_workflowjobtemplate'],
content_type=None,
managed=True,
)
rd.give_global_permission(global_viewer)
response = get(reverse('api:unified_job_list'), global_viewer)
assert response.status_code == 200
result_ids = [r['id'] for r in response.data['results']]
assert job.pk in result_ids, "Global UJT viewer should see JT jobs via singleton shortcut"
@pytest.mark.django_db
def test_global_view_inventory_sees_inventory_updates_and_adhoc(user, setup_managed_roles, get):
"""A custom global role granting only view_inventory lets the user see
inventory updates and ad hoc commands but not JT-based jobs."""
org = Organization.objects.create(name='uj-global-inv-org')
inventory = org.inventories.create(name='uj-global-inv-test')
project = Project.objects.create(name='uj-global-inv-project', organization=org)
jt = JobTemplate.objects.create(name='uj-global-inv-jt', project=project, inventory=inventory, organization=org)
job = jt.create_unified_job()
inv_src = InventorySource.objects.create(name='uj-global-inv-src', inventory=inventory, source='ec2')
inv_update = InventoryUpdate.objects.create(inventory_source=inv_src, source=inv_src.source)
adhoc = AdHocCommand.objects.create(name='uj-global-inv-adhoc', inventory=inventory)
global_inv_viewer = user('uj-global-inv-viewer')
rd = RoleDefinition.objects.create_from_permissions(
name='global-inv-viewer-test',
permissions=['view_inventory'],
content_type=None,
managed=True,
)
rd.give_global_permission(global_inv_viewer)
response = get(reverse('api:unified_job_list'), global_inv_viewer)
assert response.status_code == 200
result_ids = [r['id'] for r in response.data['results']]
assert inv_update.pk in result_ids, "Global inventory viewer should see inventory updates"
assert adhoc.pk in result_ids, "Global inventory viewer should see ad hoc commands"
assert job.pk not in result_ids, "Global inventory viewer should NOT see JT jobs"
@pytest.mark.django_db
def test_unrelated_credential_role_sees_no_unified_jobs(user, setup_managed_roles, get):
"""A user with only credential permissions should see nothing in the
unified job list — credential roles are unrelated to UJ visibility."""
org = Organization.objects.create(name='uj-cred-org')
inventory = org.inventories.create(name='uj-cred-inv')
project = Project.objects.create(name='uj-cred-project', organization=org)
jt = JobTemplate.objects.create(name='uj-cred-jt', project=project, inventory=inventory, organization=org)
jt.create_unified_job()
ct = CredentialType.defaults['ssh']()
ct.save()
cred = Credential.objects.create(name='uj-test-cred', credential_type=ct, organization=org)
cred_user = user('uj-cred-only')
RoleDefinition.objects.get(name='Credential Admin').give_permission(cred_user, cred)
response = get(reverse('api:unified_job_list'), cred_user)
assert response.status_code == 200
assert len(response.data['results']) == 0, "User with only credential perms should see no unified jobs"
@pytest.mark.django_db
def test_jt_role_plus_credential_role_only_shows_jt_jobs(user, setup_managed_roles, get):
"""A user with both a direct JT role and an unrelated credential role
should only see jobs from the JT — the credential role must not
pollute or interfere."""
org = Organization.objects.create(name='uj-mixed-org')
inventory = org.inventories.create(name='uj-mixed-inv')
project = Project.objects.create(name='uj-mixed-project', organization=org)
jt = JobTemplate.objects.create(name='uj-mixed-jt', project=project, inventory=inventory, organization=org)
job = jt.create_unified_job()
ct = CredentialType.defaults['ssh']()
ct.save()
cred = Credential.objects.create(name='uj-mixed-cred', credential_type=ct, organization=org)
mixed_user = user('uj-mixed-user')
RoleDefinition.objects.get(name='JobTemplate Execute').give_permission(mixed_user, jt)
RoleDefinition.objects.get(name='Credential Admin').give_permission(mixed_user, cred)
response = get(reverse('api:unified_job_list'), mixed_user)
assert response.status_code == 200
result_ids = [r['id'] for r in response.data['results']]
assert job.pk in result_ids, "User should see JT job via direct JT permission"
assert len(result_ids) == 1, f"User should only see the one JT job, got: {result_ids}"