From 709d0384b06042d7820c87b6a84a9eadfd063809 Mon Sep 17 00:00:00 2001 From: haseebmalik18 Date: Tue, 4 Aug 2026 00:56:56 -0400 Subject: [PATCH 1/2] Reject serde-reserved keys in HITL params_input #71036 --- .../core_api/routes/public/hitl.py | 30 ++++++++ .../core_api/routes/public/test_hitl.py | 70 +++++++++++++++++++ 2 files changed, 100 insertions(+) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py index 301a3923dea8e..413c66f69f90c 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +from collections.abc import Mapping from typing import Annotated import structlog @@ -23,6 +24,7 @@ from sqlalchemy import select from sqlalchemy.orm import joinedload +from airflow._shared.serialization import CLASSNAME, SCHEMA_ID from airflow._shared.timezones import timezone from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity from airflow.api_fastapi.common.db.common import SessionDep, paginated_select @@ -80,6 +82,27 @@ log = structlog.get_logger(__name__) +def _find_serde_reserved_key(value: object) -> str | None: + """ + Return the first serde-reserved key found at any depth in ``value``, else ``None``. + + ``serialize`` refuses a dict holding these keys, so a ``params_input`` carrying one cannot be + packed into the resume event and the parked task could never restart. Reject it on submission. + """ + reserved = (CLASSNAME, SCHEMA_ID) + stack: list[object] = [value] + while stack: + current = stack.pop() + if isinstance(current, Mapping): + for key in reserved: + if key in current: + return key + stack.extend(current.values()) + elif isinstance(current, (list, tuple)): + stack.extend(current) + return None + + def _get_task_instance_with_hitl_detail( dag_id: str, dag_run_id: str, @@ -221,6 +244,13 @@ def update_hitl_detail( status.HTTP_400_BAD_REQUEST, "Multiple options chosen but this Human-in-the-loop task accepts only a single option.", ) + reserved_key = _find_serde_reserved_key(update_hitl_detail_payload.params_input) + if reserved_key is not None: + raise HTTPException( + status.HTTP_400_BAD_REQUEST, + f"params_input may not contain the reserved key {reserved_key!r}, which cannot be " + "serialized when the task resumes.", + ) hitl_detail_model.responded_by = hitl_user hitl_detail_model.responded_at = timezone.utcnow() diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py index ad7b069581ee3..8988bab2ce8de 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py @@ -28,6 +28,7 @@ from sqlalchemy import delete, select from sqlalchemy.orm import Session +from airflow._shared.serialization import CLASSNAME, SCHEMA_ID from airflow._shared.timezones.timezone import utc, utcnow from airflow.models.hitl import HITLDetail from airflow.models.log import Log @@ -408,6 +409,75 @@ def test_should_respond_400_for_invalid_option( assert response.status_code == 400 assert "Invalid options" in response.json()["detail"] + @pytest.mark.usefixtures("sample_hitl_detail") + @pytest.mark.parametrize("reserved_key", [CLASSNAME, SCHEMA_ID]) + @pytest.mark.parametrize( + "make_params_input", + [ + pytest.param(lambda key: {key: "x"}, id="top-level"), + pytest.param(lambda key: {"nested": {key: "x"}}, id="nested-dict"), + pytest.param(lambda key: {"items": [{key: "x"}]}, id="inside-list"), + ], + ) + def test_should_respond_400_for_serde_reserved_key_in_params_input( + self, + test_client: TestClient, + sample_ti_url_identifier: str, + reserved_key: str, + make_params_input: Callable[[str], dict[str, Any]], + ) -> None: + """A params_input carrying a serde-reserved key at any depth is rejected (400) at submission time.""" + response = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": make_params_input(reserved_key)}, + ) + assert response.status_code == 400 + assert reserved_key in response.json()["detail"] + + @time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False) + @pytest.mark.usefixtures("sample_hitl_detail") + def test_rejected_reserved_key_leaves_task_resumable( + self, + test_client: TestClient, + sample_ti_url_identifier: str, + sample_ti: TaskInstance, + session: Session, + ) -> None: + """A rejected reserved-key response records nothing, so the parked task is still resumable by a corrected resubmission.""" + ti = session.get(TIModel, sample_ti.id) + assert ti is not None + ti.state = TaskInstanceState.AWAITING_INPUT + ti.next_method = "execute_complete" + ti.next_kwargs = {} + ti.trigger_id = None + session.commit() + + rejected = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": {CLASSNAME: "x"}}, + ) + assert rejected.status_code == 400 + + session.expire_all() + parked = session.get(TIModel, sample_ti.id) + assert parked is not None + assert parked.state == TaskInstanceState.AWAITING_INPUT + detail = session.scalar(select(HITLDetail).where(HITLDetail.ti_id == sample_ti.id)) + assert detail is not None + assert detail.response_received is False + + accepted = test_client.patch( + f"{sample_ti_url_identifier}/hitlDetails", + json={"chosen_options": ["Approve"], "params_input": {"input_1": 2}}, + ) + assert accepted.status_code == 200 + + session.expire_all() + resumed = session.get(TIModel, sample_ti.id) + assert resumed is not None + assert resumed.state == TaskInstanceState.SCHEDULED + assert "event" in (resumed.next_kwargs or {}) + @time_machine.travel(datetime(2025, 7, 3, 0, 0, 0), tick=False) @pytest.mark.usefixtures("sample_hitl_detail_respondent") def test_should_respond_200_to_assigned_users( From e2e3c04cf65c291ab18c870fa2487759d70aa526 Mon Sep 17 00:00:00 2001 From: haseebmalik18 Date: Tue, 4 Aug 2026 14:21:16 -0400 Subject: [PATCH 2/2] Share HITL reserved-key constant and document the params_input restriction --- airflow-core/docs/tutorial/hitl.rst | 6 ++++++ .../api_fastapi/core_api/routes/public/hitl.py | 16 ++++++++-------- .../core_api/routes/public/test_hitl.py | 6 +++--- 3 files changed, 17 insertions(+), 11 deletions(-) diff --git a/airflow-core/docs/tutorial/hitl.rst b/airflow-core/docs/tutorial/hitl.rst index aea843486f404..a1dc91da68e95 100644 --- a/airflow-core/docs/tutorial/hitl.rst +++ b/airflow-core/docs/tutorial/hitl.rst @@ -221,6 +221,12 @@ calls involved (``~`` works as a wildcard for ``dag_id`` and ``dag_run_id``): PATCH /api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails {"chosen_options": ["Approve"], "params_input": {}} +.. note:: + + Keys in ``params_input`` may not be Airflow's reserved serialization keys (``__classname__`` or + ``__id__``). A response containing one is rejected with ``400`` at submission time, because it + could not be serialized when the task resumes. + .. note:: ``response_timeout`` and timeout defaults are enforced by the scheduler, which does not run diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py index 413c66f69f90c..204db40180a92 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py @@ -82,19 +82,19 @@ log = structlog.get_logger(__name__) -def _find_serde_reserved_key(value: object) -> str | None: - """ - Return the first serde-reserved key found at any depth in ``value``, else ``None``. +# Keys that ``serde.serialize`` refuses at any depth. A ``params_input`` carrying one could not be +# serialized into the resume event, so the response is rejected before it is stored. Extend this +# tuple if serde gains another reserved key; the write-side check and its tests read from here. +_SERDE_RESERVED_KEYS = (CLASSNAME, SCHEMA_ID) + - ``serialize`` refuses a dict holding these keys, so a ``params_input`` carrying one cannot be - packed into the resume event and the parked task could never restart. Reject it on submission. - """ - reserved = (CLASSNAME, SCHEMA_ID) +def _find_serde_reserved_key(value: object) -> str | None: + """Return the first ``_SERDE_RESERVED_KEYS`` entry found at any depth in ``value``, else ``None``.""" stack: list[object] = [value] while stack: current = stack.pop() if isinstance(current, Mapping): - for key in reserved: + for key in _SERDE_RESERVED_KEYS: if key in current: return key stack.extend(current.values()) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py index 8988bab2ce8de..171c167225bb1 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py @@ -28,8 +28,8 @@ from sqlalchemy import delete, select from sqlalchemy.orm import Session -from airflow._shared.serialization import CLASSNAME, SCHEMA_ID from airflow._shared.timezones.timezone import utc, utcnow +from airflow.api_fastapi.core_api.routes.public.hitl import _SERDE_RESERVED_KEYS from airflow.models.hitl import HITLDetail from airflow.models.log import Log from airflow.models.taskinstance import TaskInstance as TIModel @@ -410,7 +410,7 @@ def test_should_respond_400_for_invalid_option( assert "Invalid options" in response.json()["detail"] @pytest.mark.usefixtures("sample_hitl_detail") - @pytest.mark.parametrize("reserved_key", [CLASSNAME, SCHEMA_ID]) + @pytest.mark.parametrize("reserved_key", _SERDE_RESERVED_KEYS) @pytest.mark.parametrize( "make_params_input", [ @@ -454,7 +454,7 @@ def test_rejected_reserved_key_leaves_task_resumable( rejected = test_client.patch( f"{sample_ti_url_identifier}/hitlDetails", - json={"chosen_options": ["Approve"], "params_input": {CLASSNAME: "x"}}, + json={"chosen_options": ["Approve"], "params_input": {_SERDE_RESERVED_KEYS[0]: "x"}}, ) assert rejected.status_code == 400