[AAP-86051] Audit bulk_update paths and add bypass observability tests

Add field-scoped audit_bulk_update_instances for bulk_update_sorted_by_id
and scheduler job_explanation batches. Expand functional and unit tests for
bulk host/workflow create and bulk update bypass logging.

Co-authored-by: Cursor <cursoragent@cursor.com>

awx/main/scheduler/task_manager.py

@@ -28,6 +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 awx.main.models import (
     Instance,
     InventorySource,
@@ -666,6 +667,7 @@ 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'])
         UnifiedJob.objects.bulk_update(tasks_to_update_job_explanation, ['job_explanation'])
 
     def task_needs_capacity(self, task, tasks_to_update_job_explanation):

awx/main/tests/functional/test_bulk.py

@@ -632,6 +632,32 @@ def test_bulk_host_create_performance_large_inventory(organization, inventory, p
     assert Host.objects.filter(inventory=inventory).count() == 10010
 
 
+@pytest.mark.django_db
+def test_bulk_host_create_logs_bypass_for_unsafe_description_when_enforcement_disabled(organization, inventory, post, user, caplog):
+    """Bulk host create skips post_save; audit before bulk_create must log Tier 2 violations."""
+    import logging
+
+    inventory.organization = organization
+    inv_admin = user('inventory_admin', False)
+    organization.member_role.members.add(inv_admin)
+    inventory.admin_role.members.add(inv_admin)
+
+    with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=False):
+        with caplog.at_level(logging.WARNING, logger='ansible_base.lib.utils.validation_signals'):
+            post(
+                reverse('api:bulk_host_create'),
+                {
+                    'inventory': inventory.id,
+                    'hosts': [{'name': f'bulk-host-bypass-{uuid4().hex[:8]}', 'description': '<script>x</script>'}],
+                },
+                inv_admin,
+                expect=201,
+            )
+    assert 'ORM bypass (bulk_create)' in caplog.text
+    assert 'description' in caplog.text
+    assert 'main.Host' in caplog.text
+
+
 @pytest.mark.django_db
 def test_bulk_job_launch_logs_bypass_for_unsafe_limit_when_enforcement_disabled(organization, inventory, project, post, user, caplog):
     """Bulk workflow nodes skip post_save; audit before bulk_create must log Tier 2 violations."""

awx/main/tests/unit/utils/test_validation_bypass_observability.py

@@ -0,0 +1,130 @@
+import logging
+from pathlib import Path
+
+import pytest
+
+import awx.api.serializers  # noqa: F401 — register CleanTextMixin models
+
+from ansible_base.lib.utils.bulk_validation_audit import audit_bulk_model_instances
+
+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,
+)
+
+LOGGER = 'ansible_base.lib.utils.validation_signals'
+
+
+@pytest.mark.django_db
+def test_configure_validation_bypass_observability_wires_dab(mocker):
+    mocker.patch('awx.main.utils.validation_bypass_observability.register_validation_signals')
+    allow = mocker.patch('awx.main.utils.validation_bypass_observability.extend_caller_allowlist_prefixes')
+    deny = mocker.patch('awx.main.utils.validation_bypass_observability.extend_internal_caller_prefixes')
+
+    configure_validation_bypass_observability()
+
+    allow.assert_called_once()
+    assert 'awx.api.serializers' in allow.call_args[0][0]
+    deny.assert_called_once()
+    assert 'awx.main.models' in deny.call_args[0][0]
+
+
+@pytest.mark.django_db
+def test_audit_bulk_model_instances_logs_host_description_on_bulk_create(caplog):
+    org = Organization.objects.create(name='org-bulk-host-audit')
+    inv = Inventory.objects.create(name='inv-bulk-host-audit', organization=org)
+    host = Host(name='host-bulk', description='<script>x</script>', inventory=inv)
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_bulk_model_instances([host], operation='bulk_create')
+    assert 'ORM bypass (bulk_create)' in caplog.text
+    assert 'description' in caplog.text
+    assert 'main.Host' in caplog.text
+
+
+@pytest.mark.django_db
+def test_audit_workflow_job_nodes_logs_limit_pseudo_field(caplog):
+    wfj = WorkflowJob.objects.create(name='wf-audit-limit')
+    node = WorkflowJobNode(workflow_job=wfj, identifier='wf-node-1')
+    node.limit = '<script>x</script>'
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_workflow_job_nodes_for_bulk_create([node])
+    assert 'ORM bypass (bulk_create)' in caplog.text
+    assert 'limit' in caplog.text
+    assert 'WorkflowJobNode' in caplog.text
+
+
+@pytest.mark.django_db
+def test_audit_workflow_job_nodes_no_op_for_empty_list(caplog):
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_workflow_job_nodes_for_bulk_create([])
+    assert 'ORM bypass' not in caplog.text
+
+
+@pytest.mark.django_db
+def test_audit_bulk_update_instances_logs_registered_text_field(caplog):
+    job = Job.objects.create(job_explanation='<script>x</script>')
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_bulk_update_instances([job], ['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):
+    wj = WorkflowJob.objects.create(name='wf-job', job_explanation='<script>x</script>')
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_bulk_update_instances([wj], ['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):
+    job = Job.objects.create(job_explanation='<script>x</script>')
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        audit_bulk_update_instances([job], [])
+    assert 'ORM bypass (bulk_update)' not in caplog.text
+
+
+@pytest.mark.django_db
+def test_bulk_update_sorted_by_id_skips_audit_when_fields_are_not_text(caplog):
+    org = Organization.objects.create(name='org-bulk-update-audit')
+    inv = Inventory.objects.create(name='inv-bulk-update-audit', organization=org)
+    host = Host.objects.create(name='host-1', description='<script>x</script>', inventory=inv)
+    host.ansible_facts = {'k': 'v'}
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        bulk_update_sorted_by_id(Host, [host], fields=['ansible_facts'])
+    assert 'ORM bypass (bulk_update)' not in caplog.text
+
+
+@pytest.mark.django_db
+def test_bulk_update_sorted_by_id_audits_text_fields_in_fields_list(caplog):
+    org = Organization.objects.create(name='org-bulk-update-desc')
+    inv = Inventory.objects.create(name='inv-bulk-update-desc', organization=org)
+    host = Host.objects.create(name='host-desc', description='clean', inventory=inv)
+    host.description = '<script>x</script>'
+    with caplog.at_level(logging.WARNING, logger=LOGGER):
+        bulk_update_sorted_by_id(Host, [host], fields=['description'])
+    assert 'ORM bypass (bulk_update)' in caplog.text
+    assert 'description' in caplog.text
+
+
+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'
+    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)
+
+
+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')

awx/main/utils/db.py

@@ -10,6 +10,8 @@ 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
+
 
 def set_connection_name(function):
     set_application_name(settings.DATABASES, settings.CLUSTER_HOST_ID, function=function)
@@ -36,6 +38,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)
     return model.objects.bulk_update(sorted_objects, fields, batch_size=batch_size)
 
 

awx/main/utils/validation_bypass_observability.py

@@ -10,6 +10,7 @@ from ansible_base.lib.utils.bulk_validation_audit import (
 )
 from ansible_base.lib.utils.validation_signals import (
     _get_caller_info,
+    _get_text_fields,
     _protected_models,
     _validate_field,
     extend_caller_allowlist_prefixes,
@@ -71,3 +72,44 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None:
             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)