diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py index 93ce5dcc76..f355554f61 100644 --- a/awx/main/scheduler/task_manager.py +++ b/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): diff --git a/awx/main/tests/functional/test_bulk.py b/awx/main/tests/functional/test_bulk.py index 437ad9ee51..2c59c76db5 100644 --- a/awx/main/tests/functional/test_bulk.py +++ b/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': ''}], + }, + 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.""" diff --git a/awx/main/tests/unit/utils/test_validation_bypass_observability.py b/awx/main/tests/unit/utils/test_validation_bypass_observability.py new file mode 100644 index 0000000000..c3581e651d --- /dev/null +++ b/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='', 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 = '' + 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='') + 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='') + 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='') + 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='', 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 = '' + 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') diff --git a/awx/main/utils/db.py b/awx/main/utils/db.py index 2ce3619535..a4b0741888 100644 --- a/awx/main/utils/db.py +++ b/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) diff --git a/awx/main/utils/validation_bypass_observability.py b/awx/main/utils/validation_bypass_observability.py index 5965d53983..cb91ea880f 100644 --- a/awx/main/utils/validation_bypass_observability.py +++ b/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)