From 643dd0d4a79247f8686746a8accb35e84efe5458 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 01:50:30 +0200 Subject: [PATCH] [v3-3-test] Mask nested entities when writing bulk audit-log entries (#70890) The audit-log maskers for Variables and Connections dispatch on top-level key names. A bulk request body has exactly one top-level key, `actions`, and the entities sit two levels down in `actions[].entities[]` -- so neither `val`/`value` nor `extra` was ever seen, and the payload was recorded as supplied. The masker also runs on the raw request body before validation, so `extra` can arrive as any JSON type. `json.loads` raises `TypeError` rather than `JSONDecodeError` for a non-string, which escaped the audit-log path entirely. Bulk bodies newly reach this branch, so the shape is now reachable where it previously was not. (cherry picked from commit fc8d6d8b6194674f4043fb4f632e7700876c0fd7) --- .../airflow/api_fastapi/logging/decorators.py | 59 ++++++- .../api_fastapi/logging/test_decorators.py | 150 ++++++++++++++++++ 2 files changed, 207 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/logging/decorators.py b/airflow-core/src/airflow/api_fastapi/logging/decorators.py index f32a2bffc707f..71942cafe41b2 100644 --- a/airflow-core/src/airflow/api_fastapi/logging/decorators.py +++ b/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: @@ -59,7 +102,11 @@ def _mask_connection_fields(extra_fields): result[k] = {ek: "***" for ek in parsed_extra} else: result[k] = "Expected JSON object in `extra` field, got non-dict JSON" - except json.JSONDecodeError: + except (json.JSONDecodeError, TypeError): + # ``extra`` is declared as a string, but this runs on the raw body before + # validation, so it can arrive as any JSON type -- a number or an already-decoded + # object makes ``json.loads`` raise TypeError rather than JSONDecodeError. Both + # are recorded without the value, instead of raising out of the audit-log path. result[k] = "Encountered non-JSON in `extra` field" else: result[k] = secrets_masker.redact(v, k) @@ -67,6 +114,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. diff --git a/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py b/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py index 41b9883d6d313..ab9218e1dd287 100644 --- a/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py +++ b/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py @@ -122,3 +122,153 @@ def test_masks_val_alias(self): def test_value_without_key_is_still_masked(self): result = _mask_variable_fields({"value": "secretval"}) assert result == {"value": "***"} + + +class TestMaskBulkFields: + """The bulk endpoints nest their entities, and the masking has to reach them.""" + + 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) + + @pytest.mark.parametrize( + "extra", + [ + pytest.param({"token": "CONN_LEAK_token"}, id="already-decoded-object"), + pytest.param(["CONN_LEAK_token"], id="already-decoded-array"), + pytest.param(123, id="number"), + pytest.param(True, id="bool"), + ], + ) + def test_non_string_extra_does_not_raise_and_does_not_leak(self, extra): + """``extra`` reaches this before validation, so it need not be a string. + + ``json.loads`` raises ``TypeError`` rather than ``JSONDecodeError`` for a non-string, which + would otherwise escape the audit-log decorator. + """ + body = {"actions": [{"action": "create", "entities": [{"connection_id": "c1", "extra": extra}]}]} + + result = _mask_connection_fields(body) + + assert "CONN_LEAK_token" not in json.dumps(result) + assert result["actions"][0]["entities"][0]["connection_id"] == "c1" + + def test_flat_body_still_takes_the_single_entity_path(self): + assert _mask_variable_fields({"key": "k", "value": "secret"}) == {"key": "k", "value": "***"}