mirror of
https://github.com/ansible/awx.git
synced 2026-08-11 23:31:43 -02:30
Compare commits
1 Commits
devel
...
bump-pyasn
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ba411c1501 |
@@ -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:
|
|
||||||
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()
|
|
||||||
)
|
|
||||||
|
|
||||||
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)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
by_inventory_update = (
|
||||||
|
InventoryUpdate.objects.filter(
|
||||||
|
inventory_source__inventory__id__in=inv_pk_qs,
|
||||||
|
)
|
||||||
|
.order_by()
|
||||||
|
.values_list('pk', flat=True)
|
||||||
|
)
|
||||||
|
|
||||||
|
by_adhoc = (
|
||||||
|
AdHocCommand.objects.filter(
|
||||||
|
inventory__id__in=inv_pk_qs,
|
||||||
|
)
|
||||||
|
.order_by()
|
||||||
|
.values_list('pk', flat=True)
|
||||||
|
)
|
||||||
|
|
||||||
|
by_org_auditor = (
|
||||||
|
self.model.objects.filter(
|
||||||
|
organization__in=Organization.access_ids_qs(self.user, 'audit_organization'),
|
||||||
|
)
|
||||||
|
.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))
|
||||||
|
|
||||||
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)
|
||||||
|
|
||||||
|
|||||||
@@ -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."""
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user