Mask nested entities when writing bulk audit-log entries

airflow-core/src/airflow/api_fastapi/logging/decorators.py

@@ -45,8 +45,51 @@ def _sanitize_for_stdlib_log(value: str) -> str:
     return value.replace("\r", " ").replace("\n", " ")
 
 
+def _mask_bulk_entities(extra_fields, mask_entity):
+    """
+    Apply per-entity masking to a bulk request body.
+
+    A ``BulkBody`` has exactly one top-level field, ``actions``; the entities carrying the
+    secrets sit two levels down, in ``actions[].entities[]``. The per-entity maskers below
+    inspect top-level key names, so handing them a bulk body means they see only the key
+    ``actions`` and pass its whole payload through untouched. Reach the entities first.
+
+    Returns ``None`` when the body is not bulk-shaped, so callers fall back to flat masking.
+    """
+    actions = extra_fields.get("actions")
+    if not isinstance(actions, list):
+        return None
+
+    masked_actions = []
+    for action in actions:
+        if not isinstance(action, dict):
+            masked_actions.append(action)
+            continue
+        entities = action.get("entities")
+        if not isinstance(entities, list):
+            masked_actions.append(action)
+            continue
+        # ``delete`` actions may list bare id/key strings rather than entity objects;
+        # those carry no secret and are left as they are.
+        masked_actions.append(
+            {
+                **action,
+                "entities": [mask_entity(e) if isinstance(e, dict) else e for e in entities],
+            }
+        )
+    return {**extra_fields, "actions": masked_actions}
+
+
 def _mask_connection_fields(extra_fields):
-    """Mask connection fields."""
+    """Mask connection fields, for either a single-entity or a bulk request body."""
+    bulk = _mask_bulk_entities(extra_fields, _mask_connection_entity)
+    if bulk is not None:
+        return bulk
+    return _mask_connection_entity(extra_fields)
+
+
+def _mask_connection_entity(extra_fields):
+    """Mask the fields of one connection."""
     result = {}
     for k, v in extra_fields.items():
         if k == "extra" and v:
@@ -67,6 +110,14 @@ def _mask_connection_fields(extra_fields):
 
 
 def _mask_variable_fields(extra_fields):
+    """Mask variable values, for either a single-entity or a bulk request body."""
+    bulk = _mask_bulk_entities(extra_fields, _mask_variable_entity)
+    if bulk is not None:
+        return bulk
+    return _mask_variable_entity(extra_fields)
+
+
+def _mask_variable_entity(extra_fields):
     """
     Mask the variable value.
 

airflow-core/tests/unit/api_fastapi/logging/test_decorators.py

@@ -130,6 +130,141 @@ def test_value_without_key_is_still_masked(self):
         assert result == {"value": "***"}
 
 
+class TestMaskBulkFields:
+    """The bulk endpoints nest their entities, and the masking has to reach them.
+
+    A ``BulkBody`` has exactly one top-level field, ``actions``; the entities carrying the
+    secrets sit two levels down, in ``actions[].entities[]``. Both maskers dispatch on
+    top-level key names, so a bulk body previously presented them with the single key
+    ``actions`` -- neither ``val``/``value`` for variables, nor ``extra`` for connections --
+    and the payload was written to the audit log as supplied.
+    """
+
+    def test_bulk_variable_values_are_masked(self):
+        result = _mask_variable_fields(
+            {
+                "actions": [
+                    {
+                        "action": "create",
+                        "entities": [
+                            {"key": "campaign_signing_material", "value": "VARVAL_LEAK_token"},
+                            {"key": "other", "val": "VARVAL_LEAK_alias"},
+                        ],
+                        "action_on_existence": "overwrite",
+                    }
+                ]
+            }
+        )
+        assert result == {
+            "actions": [
+                {
+                    "action": "create",
+                    "entities": [
+                        {"key": "campaign_signing_material", "value": "***"},
+                        {"key": "other", "val": "***"},
+                    ],
+                    "action_on_existence": "overwrite",
+                }
+            ]
+        }
+
+    def test_bulk_connection_extra_is_masked(self):
+        """``extra`` is the load-bearing case for connections.
+
+        Key-name redaction already covered a nested ``password`` -- but only while
+        ``hide_sensitive_var_conn_fields`` is enabled, which is a deployment setting and is off
+        in this test environment. ``extra`` was never covered by it at all: the name is not a
+        recognised sensitive field, and the value is a JSON *string*, which ``redact`` returns
+        unchanged. Masking ``extra`` is structural here, so it does not depend on that setting.
+        """
+        result = _mask_connection_fields(
+            {
+                "actions": [
+                    {
+                        "action": "create",
+                        "entities": [
+                            {
+                                "connection_id": "c1",
+                                "conn_type": "http",
+                                "password": "CONN_LEAK_pw",
+                                "extra": json.dumps({"token": "CONN_LEAK_token", "region": "eu"}),
+                            }
+                        ],
+                    }
+                ]
+            }
+        )
+        entity = result["actions"][0]["entities"][0]
+        assert entity["extra"] == {"token": "***", "region": "***"}
+        assert entity["connection_id"] == "c1"
+        # every value is gone, only the key names of ``extra`` remain
+        assert "CONN_LEAK_token" not in json.dumps(result)
+
+    @pytest.mark.parametrize(
+        ("body", "masker"),
+        [
+            (
+                {"actions": [{"action": "create", "entities": [{"key": "k", "value": "LEAK_v"}]}]},
+                _mask_variable_fields,
+            ),
+            (
+                {
+                    "actions": [
+                        {
+                            "action": "update",
+                            "entities": [{"connection_id": "c", "extra": json.dumps({"t": "LEAK_v"})}],
+                        }
+                    ]
+                },
+                _mask_connection_fields,
+            ),
+        ],
+        ids=["variables", "connections"],
+    )
+    def test_no_secret_survives_in_the_serialized_entry(self, body, masker):
+        """The audit entry is serialized whole, so assert on the serialization, not one field."""
+        assert "LEAK_v" not in json.dumps(masker(body))
+
+    def test_multiple_actions_and_entities_are_all_masked(self):
+        result = _mask_variable_fields(
+            {
+                "actions": [
+                    {
+                        "action": "create",
+                        "entities": [{"key": "a", "value": "s1"}, {"key": "b", "value": "s2"}],
+                    },
+                    {"action": "update", "entities": [{"key": "c", "value": "s3"}]},
+                ]
+            }
+        )
+        values = [e["value"] for a in result["actions"] for e in a["entities"]]
+        assert values == ["***", "***", "***"]
+
+    def test_delete_by_key_entities_are_left_alone(self):
+        """``delete`` may list bare keys rather than entity objects; those carry no secret."""
+        body = {"actions": [{"action": "delete", "entities": ["key_one", "key_two"]}]}
+        assert _mask_variable_fields(body) == body
+
+    @pytest.mark.parametrize(
+        "body",
+        [
+            {"key": "k", "value": "secret"},
+            {"actions": "not-a-list"},
+            {"actions": [{"action": "create"}]},
+            {"actions": [{"action": "create", "entities": "not-a-list"}]},
+            {"actions": ["not-a-dict"]},
+        ],
+        ids=["flat-body", "actions-not-list", "no-entities", "entities-not-list", "action-not-dict"],
+    )
+    def test_non_bulk_and_malformed_shapes_do_not_raise(self, body):
+        """The masker runs on request bodies before validation, so it must not add a failure mode."""
+        _mask_variable_fields(body)
+        _mask_connection_fields(body)
+
+    def test_flat_body_still_takes_the_single_entity_path(self):
+        assert _mask_variable_fields({"key": "k", "value": "secret"}) == {"key": "k", "value": "***"}
+
+
 class TestActionLoggingUserFields:
     """The audit Log records the raw identifier in ``owner`` and the human-friendly
     ``get_display_name()`` in ``owner_display_name``."""