From 5b3e807da1780e477d04db658594ea184cfd9d75 Mon Sep 17 00:00:00 2001 From: Vidya Nambiar Date: Fri, 25 Sep 2026 09:36:31 -0400 Subject: [PATCH] [AAP-86051] Address bypass observability review on caller attribution Use DAB bulk_update auditing instead of a duplicate AWX helper, narrow caller allowlists so logs point at serializers and scheduler, and document wiring. Signed-off-by: Vidya Nambiar Co-authored-by: Cursor --- awx/main/scheduler/task_manager.py | 6 +- .../test_validation_bypass_observability.py | 31 ++++--- awx/main/utils/db.py | 7 +- .../utils/validation_bypass_observability.py | 83 +++++++------------ 4 files changed, 55 insertions(+), 72 deletions(-) diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py index f355554f61..70b2655eb5 100644 --- a/awx/main/scheduler/task_manager.py +++ b/awx/main/scheduler/task_manager.py @@ -28,7 +28,7 @@ from ansible_base.lib.utils.db import advisory_lock # AWX from awx.main.dispatch.reaper import reap_job -from awx.main.utils.validation_bypass_observability import audit_bulk_update_instances +from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances from awx.main.models import ( Instance, InventorySource, @@ -667,7 +667,9 @@ class TaskManager(TaskBase): ) if not found_acceptable_queue: self.task_needs_capacity(task, tasks_to_update_job_explanation) - audit_bulk_update_instances(tasks_to_update_job_explanation, ['job_explanation']) + audit_bulk_model_instances( + tasks_to_update_job_explanation, operation='bulk_update', update_fields=['job_explanation'] + ) UnifiedJob.objects.bulk_update(tasks_to_update_job_explanation, ['job_explanation']) def task_needs_capacity(self, task, tasks_to_update_job_explanation): diff --git a/awx/main/tests/unit/utils/test_validation_bypass_observability.py b/awx/main/tests/unit/utils/test_validation_bypass_observability.py index c3581e651d..7948469e61 100644 --- a/awx/main/tests/unit/utils/test_validation_bypass_observability.py +++ b/awx/main/tests/unit/utils/test_validation_bypass_observability.py @@ -10,7 +10,6 @@ from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instan from awx.main.models import Host, Inventory, Job, Organization, WorkflowJob, WorkflowJobNode from awx.main.utils.db import bulk_update_sorted_by_id from awx.main.utils.validation_bypass_observability import ( - audit_bulk_update_instances, audit_workflow_job_nodes_for_bulk_create, configure_validation_bypass_observability, ) @@ -27,9 +26,15 @@ def test_configure_validation_bypass_observability_wires_dab(mocker): configure_validation_bypass_observability() allow.assert_called_once() - assert 'awx.api.serializers' in allow.call_args[0][0] + prefixes = allow.call_args[0][0] + assert 'awx.api.serializers' in prefixes + assert 'awx.main.scheduler' in prefixes + assert 'awx.main.utils' not in prefixes deny.assert_called_once() - assert 'awx.main.models' in deny.call_args[0][0] + internal = deny.call_args[0][0] + assert 'awx.main.models' in internal + assert 'awx.main.utils.validation_bypass_observability' in internal + assert 'awx.main.utils.db' in internal @pytest.mark.django_db @@ -64,27 +69,27 @@ def test_audit_workflow_job_nodes_no_op_for_empty_list(caplog): @pytest.mark.django_db -def test_audit_bulk_update_instances_logs_registered_text_field(caplog): +def test_audit_bulk_model_instances_logs_registered_text_field_on_bulk_update(caplog): job = Job.objects.create(job_explanation='') with caplog.at_level(logging.WARNING, logger=LOGGER): - audit_bulk_update_instances([job], ['job_explanation']) + audit_bulk_model_instances([job], operation='bulk_update', update_fields=['job_explanation']) assert 'ORM bypass (bulk_update)' in caplog.text assert 'job_explanation' in caplog.text @pytest.mark.django_db -def test_audit_bulk_update_instances_skips_unregistered_workflow_job(caplog): +def test_audit_bulk_model_instances_skips_unregistered_workflow_job_on_bulk_update(caplog): wj = WorkflowJob.objects.create(name='wf-job', job_explanation='') with caplog.at_level(logging.WARNING, logger=LOGGER): - audit_bulk_update_instances([wj], ['job_explanation']) + audit_bulk_model_instances([wj], operation='bulk_update', update_fields=['job_explanation']) assert 'ORM bypass (bulk_update)' not in caplog.text @pytest.mark.django_db -def test_audit_bulk_update_instances_no_op_when_fields_empty(caplog): +def test_audit_bulk_model_instances_bulk_update_no_op_when_update_fields_empty(caplog): job = Job.objects.create(job_explanation='') with caplog.at_level(logging.WARNING, logger=LOGGER): - audit_bulk_update_instances([job], []) + audit_bulk_model_instances([job], operation='bulk_update', update_fields=[]) assert 'ORM bypass (bulk_update)' not in caplog.text @@ -115,16 +120,16 @@ def test_task_manager_audits_before_job_explanation_bulk_update(): """Regression: scheduler must not bulk_update job_explanation without audit.""" task_manager_py = Path(__file__).resolve().parents[3] / 'scheduler' / 'task_manager.py' source = task_manager_py.read_text() - audit_marker = 'audit_bulk_update_instances(tasks_to_update_job_explanation' + audit_marker = "audit_bulk_model_instances(\n tasks_to_update_job_explanation, operation='bulk_update', update_fields=['job_explanation']" bulk_marker = "UnifiedJob.objects.bulk_update(tasks_to_update_job_explanation, ['job_explanation'])" assert audit_marker in source assert bulk_marker in source - assert source.index(audit_marker) < source.index(bulk_marker) + assert source.index('audit_bulk_model_instances') < source.index(bulk_marker) def test_db_bulk_update_sorted_by_id_audits_before_orm_bulk_update(): """Regression: shared bulk_update helper must audit before Django bulk_update.""" db_py = Path(__file__).resolve().parents[3] / 'utils' / 'db.py' source = db_py.read_text() - assert 'audit_bulk_update_instances(sorted_objects, fields)' in source - assert source.index('audit_bulk_update_instances') < source.index('model.objects.bulk_update') + assert "audit_bulk_model_instances(sorted_objects, operation='bulk_update', update_fields=fields)" in source + assert source.index('audit_bulk_model_instances') < source.index('model.objects.bulk_update') diff --git a/awx/main/utils/db.py b/awx/main/utils/db.py index a4b0741888..893647059a 100644 --- a/awx/main/utils/db.py +++ b/awx/main/utils/db.py @@ -10,7 +10,7 @@ from awx import MODE from django.conf import settings from django.db import connection -from awx.main.utils.validation_bypass_observability import audit_bulk_update_instances +from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances def set_connection_name(function): @@ -30,6 +30,9 @@ def bulk_update_sorted_by_id(model, objects, fields, batch_size=1000): which helps avoid the row-level locking contention that can lead to deadlocks in PostgreSQL when multiple processes are involved. + When ``fields`` includes registered Char/Text columns, logs Tier 1/2 ORM bypass + violations (warnings only) via django-ansible-base before ``bulk_update``. + Returns: int: The number of rows affected by the update. """ @@ -38,7 +41,7 @@ def bulk_update_sorted_by_id(model, objects, fields, batch_size=1000): return 0 # Return 0 when nothing is updated sorted_objects = sorted(objects, key=lambda obj: obj.id) - audit_bulk_update_instances(sorted_objects, fields) + audit_bulk_model_instances(sorted_objects, operation='bulk_update', update_fields=fields) return model.objects.bulk_update(sorted_objects, fields, batch_size=batch_size) diff --git a/awx/main/utils/validation_bypass_observability.py b/awx/main/utils/validation_bypass_observability.py index cb91ea880f..7725eb3277 100644 --- a/awx/main/utils/validation_bypass_observability.py +++ b/awx/main/utils/validation_bypass_observability.py @@ -1,20 +1,22 @@ -"""Register django-ansible-base ORM bypass caller attribution for AWX.""" +"""Register django-ansible-base ORM bypass observability for AWX. + +Startup registers DAB ``post_save`` bypass signals and AWX caller prefixes. +Bulk ``bulk_create`` / ``bulk_update`` paths that skip ``post_save`` are audited at +call sites via DAB helpers; workflow job node prompt pseudo-fields are handled here. +""" from collections.abc import Iterable from django.db.models import Model -from ansible_base.lib.utils.bulk_validation_audit import ( - _log_bulk_violation, - audit_bulk_model_instances, -) +from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances from ansible_base.lib.utils.validation_signals import ( _get_caller_info, - _get_text_fields, _protected_models, _validate_field, extend_caller_allowlist_prefixes, extend_internal_caller_prefixes, + log_orm_bypass_violation, register_validation_signals, ) @@ -23,7 +25,14 @@ _WORKFLOW_JOB_NODE_PROMPT_PSEUDO_FIELDS = frozenset({'scm_branch', 'limit', 'job def configure_validation_bypass_observability() -> None: - """Call from AppConfig.ready() so ORM bypass logs show AWX entry points.""" + """Call from AppConfig.ready() so ORM bypass logs show AWX entry points. + + DAB resolves ``[caller: …]`` by walking the stack outward and returning the + first module on the caller allowlist (denylist is not applied in that phase). + Allowlist only product surfaces (API, serializers, tasks, management, scheduler). + List bulk-audit helpers under ``awx.main.utils`` on the internal denylist so + logs attribute to the serializer, task, or scheduler frame that invoked them. + """ register_validation_signals() extend_caller_allowlist_prefixes( [ @@ -31,7 +40,7 @@ def configure_validation_bypass_observability() -> None: "awx.api.serializers", "awx.main.tasks", "awx.main.management", - "awx.main.utils", + "awx.main.scheduler", ] ) extend_internal_caller_prefixes( @@ -39,6 +48,8 @@ def configure_validation_bypass_observability() -> None: "awx.main.models", "awx.main.signals", "awx.main.dispatch", + "awx.main.utils.validation_bypass_observability", + "awx.main.utils.db", ] ) @@ -51,17 +62,18 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None: ``NullablePromptPseudoField`` and must be read with ``getattr`` after deferred attrs are applied on the in-memory instances. """ - audit_bulk_model_instances(nodes, operation='bulk_create') - if not nodes: + materialized = list(nodes) + if not materialized: return - sample = next(iter(nodes)) + audit_bulk_model_instances(materialized, operation='bulk_create') + sample = materialized[0] protected = _protected_models.get(type(sample)) if protected is None: return name_fields, excluded_fields = protected - caller_info = _get_caller_info() resource_type = f"{sample._meta.app_label}.{sample._meta.object_name}" - for instance in nodes: + caller_info = None + for instance in materialized: for field_name in _WORKFLOW_JOB_NODE_PROMPT_PSEUDO_FIELDS: if field_name in excluded_fields: continue @@ -71,45 +83,6 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None: violation = _validate_field(field_name, value, name_fields) if violation: tier, reason = violation - _log_bulk_violation('bulk_create', field_name, resource_type, tier, caller_info, reason) - - -def _audited_text_field_names(model: type[Model]) -> frozenset[str]: - protected = _protected_models.get(model) - if protected is None: - return frozenset() - _, excluded_fields = protected - text_fields, _json_fields = _get_text_fields(model) - return frozenset(field_name for field_name in text_fields if field_name not in excluded_fields) - - -def audit_bulk_update_instances(instances: Iterable[Model], fields: Iterable[str]) -> None: - """Log Tier 1/2 violations for instances about to be bulk-updated (non-blocking). - - Only fields named in ``fields`` that are registered Char/Text columns are checked, - so callers updating JSON or non-text columns (e.g. Host ``ansible_facts``) are not - scanned for unrelated text on the in-memory instance. - """ - update_fields = frozenset(fields) - if not update_fields or not instances: - return - caller_info = _get_caller_info() - for instance in instances: - target_fields = update_fields & _audited_text_field_names(type(instance)) - if not target_fields: - continue - protected = _protected_models.get(type(instance)) - if protected is None: - continue - name_fields, excluded_fields = protected - resource_type = f"{instance._meta.app_label}.{instance._meta.object_name}" - for field_name in target_fields: - if field_name in excluded_fields: - continue - value = getattr(instance, field_name, None) - if value is None or not isinstance(value, str): - continue - violation = _validate_field(field_name, value, name_fields) - if violation: - tier, reason = violation - _log_bulk_violation('bulk_update', field_name, resource_type, tier, caller_info, reason) + if caller_info is None: + caller_info = _get_caller_info() + log_orm_bypass_violation('bulk_create', field_name, resource_type, tier, caller_info, reason)