Compare commits

..

1 Commits

Author SHA1 Message Date
Peter Braun
ba411c1501 fix: update pyasn1 to 0.6.4 2026-08-11 14:49:44 +02:00
4 changed files with 38 additions and 134 deletions

View File

@@ -12,7 +12,6 @@ from django.conf import settings
from django.db.models import Q, Prefetch from django.db.models import Q, Prefetch
from django.contrib.auth.models import User from django.contrib.auth.models import User
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist
# Django REST Framework # Django REST Framework
@@ -20,7 +19,7 @@ from rest_framework.exceptions import ParseError, PermissionDenied
# django-ansible-base # django-ansible-base
from ansible_base.lib.utils.validation import to_python_boolean 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.policies import visible_users
from ansible_base.rbac import permission_registry from ansible_base.rbac import permission_registry
@@ -2508,76 +2507,40 @@ class UnifiedJobAccess(BaseAccess):
# ) # )
def filtered_queryset(self): def filtered_queryset(self):
# AAP-87084 / AAP-87087: Pre-compute role IDs once to avoid 4x redundant inv_pk_qs = Inventory.access_ids_qs(self.user, 'view')
# 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() by_template = (
self.model.objects.filter(unified_job_template_id__in=UnifiedJobTemplate.accessible_pk_qs(self.user, 'read_role'))
role_subclasses = UnifiedJobTemplate._submodels_with_roles() .order_by()
ujt_codenames = [f'view_{cls._meta.model_name}' for cls in role_subclasses] .values_list('pk', flat=True)
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: by_inventory_update = (
inv_accessible = Inventory.objects.values_list('id', flat=True) InventoryUpdate.objects.filter(
else: inventory_source__inventory__id__in=inv_pk_qs,
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') .order_by()
.distinct() .values_list('pk', flat=True)
) )
if 'audit_organization' in user_singletons: by_adhoc = (
org_accessible = Organization.objects.values_list('id', flat=True) AdHocCommand.objects.filter(
else: inventory__id__in=inv_pk_qs,
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') .order_by()
.distinct() .values_list('pk', flat=True)
) )
return self.model.objects.filter( by_org_auditor = (
Q(unified_job_template_id__in=ujt_accessible) self.model.objects.filter(
| Q( organization__in=Organization.access_ids_qs(self.user, 'audit_organization'),
pk__in=InventoryUpdate.objects.filter(
inventory_source__inventory__id__in=inv_accessible,
).values('pk')
) )
| Q( .order_by()
pk__in=AdHocCommand.objects.filter( .values_list('pk', flat=True)
inventory__id__in=inv_accessible,
).values('pk')
)
| Q(organization__in=org_accessible)
) )
return self.model.objects.filter(pk__in=by_template.union(by_inventory_update, by_adhoc, by_org_auditor))
def get_queryset(self): def get_queryset(self):
return super(UnifiedJobAccess, self).get_queryset().filter(workflowapproval__isnull=True) return super(UnifiedJobAccess, self).get_queryset().filter(workflowapproval__isnull=True)

View File

@@ -13,14 +13,13 @@ from awx.main.models import (
JobTemplate, JobTemplate,
Organization, Organization,
Project, Project,
Team,
UnifiedJob, UnifiedJob,
) )
@pytest.mark.django_db @pytest.mark.django_db
def test_unified_job_list_uses_or_not_union(user, organization, inventory, setup_managed_roles, get): def test_unified_job_list_uses_union(user, organization, inventory, setup_managed_roles, get):
"""The unified job list RBAC query uses OR-based filtering, not UNION.""" """The unified job list RBAC query uses UNION instead of OR to allow per-branch query planning."""
org_admin = user('uj-org-admin') org_admin = user('uj-org-admin')
RoleDefinition.objects.get(name='Organization Admin').give_permission(org_admin, organization) RoleDefinition.objects.get(name='Organization Admin').give_permission(org_admin, organization)
@@ -39,10 +38,8 @@ def test_unified_job_list_uses_or_not_union(user, organization, inventory, setup
assert response.status_code == 200 assert response.status_code == 200
assert response.data['count'] >= 3 assert response.data['count'] >= 3
uj_rbac_queries = [q['sql'] for q in ctx.captured_queries if 'main_unifiedjob' in q['sql'] and 'dab_rbac_roleevaluation' in q['sql']] 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 a unified-job RBAC query" assert uj_rbac_queries, "Expected at least one query using UNION for unified job RBAC filtering"
for sql in uj_rbac_queries:
assert 'UNION' not in sql, "RBAC query should use OR, not UNION"
@pytest.mark.django_db @pytest.mark.django_db
@@ -80,62 +77,6 @@ def test_unified_job_list_inventory_viewer_sees_inventory_updates(user, setup_ma
assert inv_update.pk in result_ids 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 @pytest.mark.django_db
def test_unified_job_list_rando_sees_nothing(rando, setup_managed_roles, get): def test_unified_job_list_rando_sees_nothing(rando, setup_managed_roles, get):
"""Unprivileged user sees no unified jobs.""" """Unprivileged user sees no unified jobs."""

View File

@@ -46,7 +46,7 @@ pexpect
prometheus-client prometheus-client
psycopg psycopg
psutil psutil
pyasn1>=0.6.2 # CVE-2026-2349 pyasn1>=0.6.4 # CVE-2026-59886
pygerduty pygerduty
PyGithub PyGithub
pyopenssl pyopenssl

View File

@@ -365,7 +365,7 @@ psycopg==3.2.10
# dispatcherd # dispatcherd
ptyprocess==0.7.0 ptyprocess==0.7.0
# via pexpect # via pexpect
pyasn1==0.6.2 pyasn1==0.6.4
# via # via
# -r /awx_devel/requirements/requirements.in # -r /awx_devel/requirements/requirements.in
# pyasn1-modules # pyasn1-modules