[v3-3-test] Reject reserved XCom serialization keys submitted as JSON string literals (#69378)

airflow-core/src/airflow/api_fastapi/core_api/datamodels/xcom.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import json
 from collections.abc import Iterable
 from datetime import datetime
 from typing import Any
@@ -88,6 +89,19 @@ def _check_forbidden_xcom_keys(value: Any) -> Any:
     from airflow._shared.serialization import FORBIDDEN_XCOM_KEYS
 
     def _walk(obj: Any, path: str = "value") -> None:
+        if isinstance(obj, str):
+            # A value submitted as a JSON string literal (e.g. ``json.dumps({...})``)
+            # is stored verbatim and re-parsed into a dict/list on a
+            # ``deserialize=true`` read, which would otherwise smuggle reserved keys
+            # past the dict/list checks below. Re-parse and inspect the decoded
+            # structure the same way the read path does.
+            try:
+                decoded = json.loads(obj)
+            except (ValueError, TypeError):
+                return
+            if isinstance(decoded, (dict, list)):
+                _walk(decoded, path)
+            return
         if isinstance(obj, dict):
             found = FORBIDDEN_XCOM_KEYS & obj.keys()
             if found:

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_xcom.py

@@ -746,6 +746,52 @@ def test_create_xcom_entry_blocks_forbidden_keys(self, test_client, key, value):
         assert "reserved serialization keys" in detail
         assert key in detail
 
+    @pytest.mark.parametrize(
+        "value",
+        [
+            pytest.param(
+                json.dumps({"__classname__": "airflow.sdk.definitions.connection.Connection"}),
+                id="classname-in-json-string",
+            ),
+            pytest.param(
+                json.dumps(
+                    {"nested": {"__type": "airflow.sdk.definitions.connection.Connection", "__var": {}}}
+                ),
+                id="nested-forbidden-in-json-string",
+            ),
+        ],
+    )
+    def test_create_xcom_entry_blocks_forbidden_keys_in_json_string(self, test_client, value):
+        """A forbidden payload submitted as a JSON string literal is blocked too.
+
+        ``_check_forbidden_xcom_keys._walk`` previously descended dict/list/tuple but not
+        ``str``, so a value like ``json.dumps({"__classname__": ...})`` slipped past the
+        filter and was reconstructed into a dict on a ``deserialize=true`` read.
+        """
+        response = test_client.post(
+            f"/dags/{TEST_DAG_ID}/dagRuns/{run_id}/taskInstances/{TEST_TASK_ID}/xcomEntries",
+            json={"key": "test_key", "value": value, "map_index": -1},
+        )
+        assert response.status_code == 422
+        assert "reserved serialization keys" in str(response.json()["detail"])
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            pytest.param("just a plain string", id="plain-string"),
+            pytest.param(json.dumps({"safe": "data", "count": 3}), id="benign-json-object-string"),
+            pytest.param(json.dumps(["a", "b"]), id="benign-json-array-string"),
+            pytest.param('{"not valid json', id="not-json"),
+        ],
+    )
+    def test_create_xcom_entry_allows_benign_string_values(self, test_client, value):
+        """String values that do not decode to a reserved-key structure stay accepted."""
+        response = test_client.post(
+            f"/dags/{TEST_DAG_ID}/dagRuns/{run_id}/taskInstances/{TEST_TASK_ID}/xcomEntries",
+            json={"key": "test_key", "value": value, "map_index": -1},
+        )
+        assert response.status_code != 422
+
 
 class TestDeleteXComEntry(TestXComEndpoint):
     def test_delete_xcom_entry(self, test_client, session):
@@ -894,6 +940,19 @@ def test_patch_xcom_entry_blocks_forbidden_keys(self, test_client, key, value):
         assert "reserved serialization keys" in detail
         assert key in detail
 
+    def test_patch_xcom_entry_blocks_forbidden_keys_in_json_string(self, test_client):
+        """A forbidden payload submitted as a JSON string literal is blocked on PATCH too."""
+        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": json.dumps({"__classname__": "airflow.sdk.definitions.connection.Connection"}),
+                "map_index": -1,
+            },
+        )
+        assert response.status_code == 422
+        assert "reserved serialization keys" in str(response.json()["detail"])
+
     def test_patch_xcom_preserves_int_type(self, test_client, session):
         """Test scenario described in #59032: if existing XCom value type is int,
         after patching with different value, it should still be int in the API response.