This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a26eb122498 Mask XCom values in the audit log (#73791)
a26eb122498 is described below
commit a26eb1224984dd8afa29390c125c4b05c5325a0f
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 16:56:21 2026 +0200
Mask XCom values in the audit log (#73791)
* Mask XCom values in the audit log
The audit log masks the value of Variable and Connection writes
unconditionally, because it records that something was written rather
than what. XCom create and update requests fell through to per-key
name redaction, so their values were stored in the audit log as sent.
Mask the XCom value the same way; the key and map index are kept.
Generated-by: Claude Opus 5
* Add newsfragment
Generated-by: Claude Opus 5
* Keep the targeted XCom in the audit log when an update has a body
Replacing the path/query context with the request body dropped xcom_key, so
an update_xcom_entry row no longer said which XCom was changed.
Generated-by: Claude Opus 5
---
airflow-core/newsfragments/73791.improvement.rst | 1 +
.../src/airflow/api_fastapi/logging/decorators.py | 16 +++++++++
.../core_api/routes/public/test_xcom.py | 41 ++++++++++++++++++++++
.../unit/api_fastapi/logging/test_decorators.py | 15 ++++++++
4 files changed, 73 insertions(+)
diff --git a/airflow-core/newsfragments/73791.improvement.rst
b/airflow-core/newsfragments/73791.improvement.rst
new file mode 100644
index 00000000000..642b67114b3
--- /dev/null
+++ b/airflow-core/newsfragments/73791.improvement.rst
@@ -0,0 +1 @@
+Mask XCom values in the audit log. Creating or updating an XCom entry through
the REST API now records the XCom key and map index in the audit log with the
value masked as ``***``, the same way Variable values are recorded, instead of
storing the value as sent.
diff --git a/airflow-core/src/airflow/api_fastapi/logging/decorators.py
b/airflow-core/src/airflow/api_fastapi/logging/decorators.py
index a011c3bb121..03b01f87d75 100644
--- a/airflow-core/src/airflow/api_fastapi/logging/decorators.py
+++ b/airflow-core/src/airflow/api_fastapi/logging/decorators.py
@@ -185,6 +185,18 @@ def _find_addressed_attempt(
return attempt_ids[0] if len(attempt_ids) == 1 else None
+def _mask_xcom_fields(extra_fields):
+ """
+ Mask the XCom value.
+
+ The XCom create and update requests carry the value under ``value``. As
with Variables, the
+ audit log records that an XCom was written, not what it holds, so the
value is masked
+ regardless of its key: a task can write any payload to XCom, and the audit
log is readable
+ under a different permission than the XCom itself.
+ """
+ return {k: "***" if k == "value" else v for k, v in extra_fields.items()}
+
+
def _resolve_team_name(params: dict, *, dag_id: str | None, session: Session)
-> str | None:
"""
Return the team the audited action belongs to, for the resources that own
no Dag.
@@ -279,6 +291,10 @@ def action_logging(event: str | None = None):
extra_fields = _mask_variable_fields(request_body or extra_fields)
elif "connection" in event_name:
extra_fields = _mask_connection_fields(request_body or
extra_fields)
+ elif "xcom" in event_name:
+ # Unlike Variables and Connections, an XCom update names its
target only in the path
+ # (``xcom_key``) and query, so the body is merged into that
context, not swapped for it.
+ extra_fields = _mask_xcom_fields({**extra_fields,
**masked_body_json})
elif has_json_body:
extra_fields = {**extra_fields, **masked_body_json}
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py
index e24a41f1cbf..809789655f4 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py
@@ -92,6 +92,19 @@ run_id = DagRun.generate_run_id(
)
+def _last_audit_extra(session: Session, event: str) -> dict:
+ from airflow.models import Log
+
+ extra = session.scalar(
+ select(Log.extra)
+ .where(Log.dag_id == TEST_DAG_ID, Log.event == event)
+ .order_by(Log.dttm.desc())
+ .limit(1)
+ )
+ assert extra is not None
+ return json.loads(extra)
+
+
@provide_session
def _create_xcom(key, value, backend, *, session: Session = NEW_SESSION) ->
None:
XComModel.set(
@@ -1101,6 +1114,19 @@ class TestCreateXComEntry(TestXComEndpoint):
assert current_data["map_index"] == request_body.map_index
check_last_log(session, dag_id=TEST_DAG_ID, event="create_xcom_entry",
logical_date=None)
+ def test_create_xcom_entry_audit_log_masks_value(self, test_client,
session):
+ """The audit log records that an XCom was written, not its value,
whatever the key name."""
+ response = test_client.post(
+
f"/dags/{TEST_DAG_ID}/dagRuns/{run_id}/taskInstances/{TEST_TASK_ID}/xcomEntries",
+ json=XComCreateBody(key="report_rows",
value="xcom-payload-canary").model_dump(),
+ )
+
+ assert response.status_code == 201
+ extra = _last_audit_extra(session, "create_xcom_entry")
+ assert extra["value"] == "***"
+ assert extra["key"] == "report_rows"
+ assert "xcom-payload-canary" not in json.dumps(extra)
+
def test_create_xcom_entry_duplicate_check_is_bounded(self, test_client):
"""Checking for an existing XCom before inserting must ask the
database for one row."""
with capture_orm_selects("xcom_v2") as statements:
@@ -1330,6 +1356,21 @@ class TestPatchXComEntry(TestXComEndpoint):
assert response.json()["detail"] == expected_detail
check_last_log(session, dag_id=TEST_DAG_ID, event="update_xcom_entry",
logical_date=None)
+ def test_patch_xcom_entry_audit_log_masks_value(self, test_client,
session):
+ """An XCom update is audited without its new value."""
+ self._create_xcom(TEST_XCOM_KEY, TEST_XCOM_VALUE)
+
+ response = test_client.patch(
+
f"/dags/{TEST_DAG_ID}/dagRuns/{run_id}/taskInstances/{TEST_TASK_ID}/xcomEntries/{TEST_XCOM_KEY}",
+ json={"value": {"nested": "xcom-payload-canary"}},
+ )
+
+ assert response.status_code == 200
+ extra = _last_audit_extra(session, "update_xcom_entry")
+ assert extra["value"] == "***"
+ assert extra["xcom_key"] == TEST_XCOM_KEY
+ assert "xcom-payload-canary" not in json.dumps(extra)
+
@conf_vars({("core", "multi_team"): "True"})
def test_patch_xcom_entry_with_team_name(self, test_client):
self._create_xcom(TEST_XCOM_KEY, TEST_XCOM_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 98b3b822594..66fb2fe3773 100644
--- a/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py
+++ b/airflow-core/tests/unit/api_fastapi/logging/test_decorators.py
@@ -32,6 +32,7 @@ from airflow.api_fastapi.auth.managers.models.base_user
import BaseUser
from airflow.api_fastapi.logging.decorators import (
_mask_connection_fields,
_mask_variable_fields,
+ _mask_xcom_fields,
_sanitize_for_stdlib_log,
action_logging,
)
@@ -152,6 +153,20 @@ class TestMaskVariableFields:
assert result == {"value": "***"}
+class TestMaskXComFields:
+ """The XCom value is masked unconditionally in the audit log; the key and
map index are kept."""
+
+ def test_masks_value(self):
+ result = _mask_xcom_fields({"key": "report_rows", "value": "payload",
"map_index": -1})
+ assert result == {"key": "report_rows", "value": "***", "map_index":
-1}
+
+ def test_masks_structured_value(self):
+ assert _mask_xcom_fields({"value": {"nested": ["payload"]}}) ==
{"value": "***"}
+
+ def test_body_without_value_is_unchanged(self):
+ assert _mask_xcom_fields({"xcom_key": "k", "map_index": 0}) ==
{"xcom_key": "k", "map_index": 0}
+
+
class TestMaskBulkFields:
"""The bulk endpoints nest their entities, and the masking has to reach
them."""
