[AAP-86051] Audit bulk workflow job launch before node bulk_create

Log Tier 1/2 bypass violations for WorkflowJobNode prompt fields stored in
char_prompts, which post_save and bulk CharField scans do not cover alone.

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

awx/api/serializers.py

@@ -124,6 +124,7 @@ from awx.main.utils.filters import SmartFilter
 from awx.main.utils.plugins import load_combined_inventory_source_options
 from awx.main.utils.named_url_graph import reset_counters
 from awx.main.utils.inventory_vars import update_group_variables
+from awx.main.utils.validation_bypass_observability import audit_workflow_job_nodes_for_bulk_create
 from awx.main.scheduler.task_manager_models import TaskManagerModels
 from awx.main.redact import UriCleaner, REPLACE_STR
 from awx.main.tasks.system import update_inventory_computed_fields
@@ -5335,6 +5336,7 @@ class BulkJobLaunchSerializer(PromptFieldCleanTextMixin, serializers.Serializer)
             # we'll need this later when we do the m2m through model bulk create
             node_m2m_objects[node_attrs['identifier']]['node'] = node_obj
 
+        audit_workflow_job_nodes_for_bulk_create(nodes)
         WorkflowJobNode.objects.bulk_create(nodes)
 
         # Deal with the m2m objects we have to create once the node exists

awx/main/tests/functional/test_bulk.py

@@ -632,6 +632,27 @@ 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_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."""
+    import logging
+
+    normal_user = user('normal_user', False)
+    organization.member_role.members.add(normal_user)
+    jt = JobTemplate.objects.create(name='my-jt', inventory=inventory, project=project, playbook='helloworld.yml')
+    jt.execute_role.members.add(normal_user)
+    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_job_launch'),
+                {'name': 'Bulk Job Launch', 'jobs': [{'unified_job_template': jt.id, 'limit': '<script>x</script>'}]},
+                normal_user,
+                expect=201,
+            )
+    assert 'ORM bypass (bulk_create)' in caplog.text
+    assert 'limit' in caplog.text
+
+
 @pytest.mark.django_db
 def test_bulk_job_launch_rejects_unsafe_limit(organization, inventory, project, post, user):
     """PromptFieldCleanTextMixin._run_clean_text_validation must raise when

awx/main/utils/validation_bypass_observability.py

@@ -1,11 +1,25 @@
 """Register django-ansible-base ORM bypass caller attribution for AWX."""
 
+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.validation_signals import (
+    _get_caller_info,
+    _protected_models,
+    _validate_field,
     extend_caller_allowlist_prefixes,
     extend_internal_caller_prefixes,
     register_validation_signals,
 )
 
+# Tier 2 prompt fields on LaunchTimeConfigBase models; not Django CharFields (AAP-78694).
+_WORKFLOW_JOB_NODE_PROMPT_PSEUDO_FIELDS = frozenset({'scm_branch', 'limit', 'job_tags', 'skip_tags'})
+
 
 def configure_validation_bypass_observability() -> None:
     """Call from AppConfig.ready() so ORM bypass logs show AWX entry points."""
@@ -26,3 +40,34 @@ def configure_validation_bypass_observability() -> None:
             "awx.main.dispatch",
         ]
     )
+
+
+def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None:
+    """Log Tier 1/2 violations for WorkflowJobNode rows before ``bulk_create``.
+
+    ``audit_bulk_model_instances`` only inspects model ``CharField`` / ``TextField``
+    columns. Node prompt values (limit, job_tags, etc.) live in ``char_prompts`` via
+    ``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:
+        return
+    sample = next(iter(nodes))
+    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:
+        for field_name in _WORKFLOW_JOB_NODE_PROMPT_PSEUDO_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_create', field_name, resource_type, tier, caller_info, reason)