This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 8a42fefe7d7 [v3-4-test] Reject reserved loop keys when deleting an
XCom (#76892) (#76966)
8a42fefe7d7 is described below
commit 8a42fefe7d7cd8a0bc4d987b4688daca5faf14fc
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Fri Nov 13 13:08:52 2026 +0800
[v3-4-test] Reject reserved loop keys when deleting an XCom (#76892)
(#76966)
set_xcom already refuses _airflow_loop_* keys, so task code could delete
loop XComs it can never write.
The external clear target key check is left out, as v3-4-test does not have
that key.
(cherry picked from commit 54970db82f16d119d000118d11dc5b819ac11fe4)
---
.../src/airflow/api_fastapi/common/xcom.py | 27 ++++++++++++++++++++++
.../api_fastapi/core_api/routes/public/xcom.py | 11 +++------
.../api_fastapi/execution_api/routes/xcoms.py | 12 ++++++----
.../execution_api/versions/head/test_xcoms.py | 14 +++++++++++
4 files changed, 52 insertions(+), 12 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/xcom.py
b/airflow-core/src/airflow/api_fastapi/common/xcom.py
new file mode 100644
index 00000000000..4e36c744714
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/xcom.py
@@ -0,0 +1,27 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from fastapi import HTTPException, status
+
+from airflow.models.dynamic_region import LOOP_XCOM_PREFIX
+
+
+def reject_reserved_xcom_key(key: str) -> None:
+ """Reject an XCom key that only Airflow writes, from what the Execution
API validated."""
+ if key.startswith(LOOP_XCOM_PREFIX):
+ raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
index 5467e87750e..123c894d699 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/xcom.py
@@ -49,6 +49,7 @@ from airflow.api_fastapi.common.parameters import (
teams_filter_factory,
)
from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.common.xcom import reject_reserved_xcom_key
from airflow.api_fastapi.core_api.datamodels.xcom import (
XComCollectionResponse,
XComCreateBody,
@@ -314,12 +315,6 @@ def get_xcom_entries(
return XComCollectionResponse(xcom_entries=entries,
total_entries=total_entries)
-def _reject_reserved_loop_key(key: str) -> None:
- # Loop gates write these keys through the Execution API, which validates
their values.
- if key.startswith(LOOP_XCOM_PREFIX):
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
-
-
@xcom_router.post(
"",
status_code=status.HTTP_201_CREATED,
@@ -346,7 +341,7 @@ def create_xcom_entry(
"""Create an XCom entry."""
from airflow.models.dagrun import DagRun
- _reject_reserved_loop_key(request_body.key)
+ reject_reserved_xcom_key(request_body.key)
dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id))
# Validate Dag ID
@@ -451,7 +446,7 @@ def update_xcom_entry(
dag_bag: DagBagDep,
) -> XComResponseNative:
"""Update an existing XCom entry."""
- _reject_reserved_loop_key(xcom_key)
+ reject_reserved_xcom_key(xcom_key)
scope = resolve_task_scope(
dag_id=dag_id,
run_id=dag_run_id,
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
index 2ce88dc8d3c..a8f326d84be 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
@@ -32,6 +32,7 @@ from sqlalchemy.sql.selectable import Select
from airflow.api_fastapi.common.dagbag import DagBagDep
from airflow.api_fastapi.common.db.common import SessionDep
+from airflow.api_fastapi.common.xcom import reject_reserved_xcom_key
from airflow.api_fastapi.core_api.base import BaseModel
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.api_fastapi.execution_api.datamodels.token import TIToken
@@ -46,7 +47,6 @@ from airflow.api_fastapi.execution_api.versions.v2026_10_30
import IdentifyArchi
from airflow.exceptions import TaskNotFound
from airflow.models.dagrun import DagRun
from airflow.models.dynamic_region import (
- LOOP_XCOM_PREFIX,
SENTINEL_REGION_ID,
AmbiguousProducerError,
build_loop_sequence_order,
@@ -706,8 +706,7 @@ def set_xcom(
"""Set an Airflow XCom."""
from airflow.configuration import conf
- if key.startswith(LOOP_XCOM_PREFIX):
- raise HTTPException(status.HTTP_400_BAD_REQUEST, "Reserved loop XCom
key")
+ reject_reserved_xcom_key(key)
# Validate that the provided key is not empty
# XCom keys must be non-empty strings to ensure proper data retrieval and
avoid ambiguity.
@@ -788,7 +787,10 @@ def set_xcom(
responses={
**create_openapi_http_exception_doc(
[
- (status.HTTP_400_BAD_REQUEST, "The target is in a loop the
caller is not in"),
+ (
+ status.HTTP_400_BAD_REQUEST,
+ "The key is reserved, or the target is in a loop the
caller is not in",
+ ),
(status.HTTP_409_CONFLICT, "Several live task instances match
the producer"),
]
),
@@ -807,6 +809,8 @@ def delete_xcom(
token=_TriggerAwareToken,
):
"""Delete a single XCom Value."""
+ reject_reserved_xcom_key(key)
+
owner = _find_writer_id(
token,
dag_id=dag_id,
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
index 5117d5f651d..a9bf99e5c05 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py
@@ -1595,6 +1595,20 @@ class TestXComsDeleteEndpoint:
assert verify.get(XComModelV1, (ti.dag_run.id, ti.task_id,
ti.map_index, "xcom_1")) is None
assert XComModelV2.get_for_attempt(ti1.id, "xcom_1",
session=verify) is not None
+ def test_xcom_delete_rejects_reserved_key(self, client,
create_task_instance, authenticate_as, session):
+ key = "_airflow_loop_decision_reason"
+ ti = create_task_instance()
+ ti.xcom_push(key=key, value='"reason"', session=session)
+ session.commit()
+ authenticate_as(ti)
+
+ response =
client.delete(f"/execution/xcoms/{ti.dag_id}/{ti.run_id}/{ti.task_id}/{key}")
+
+ assert response.status_code == 400
+ assert response.json()["detail"] == "Reserved loop XCom key"
+ with create_session(scoped=False) as verify:
+ assert XComModelV2.get_for_attempt(ti.id, key, session=verify) is
not None
+
@pytest.mark.parametrize(
("target_task_id", "ti_map_index"),
[
