You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ashb (via GitHub)" <gi...@apache.org> on 2026/10/14 22:25:12 UTC
[PR] Resolve a trigger's XCom requests as its deferred task instance [airflow]
ashb opened a new pull request, #74772:
URL: https://github.com/apache/airflow/pull/74772
Note: _None_ of this is a security boundary, and should not be treated as
such.
The Execution API resolves XCom reads and writes for a task inside a loop
to the caller's own iteration, using the task identity in the caller's JWT.
Triggers have no such identity: the triggerer talks to an in-process API
that treats every request as the all-zero task instance. A trigger's XCom
requests therefore resolve as an anonymous caller rather than as the
deferred task it runs for, so it can neither read (a 400) nor write the
XComs of a task inside a loop.
Triggers can also share one runner coroutine and one comms channel (the
"shared streams" idea), so only the asyncio task running a trigger's code
knows which trigger it is; a context variable carries that trigger's task
instance id to each request frame. Shared-stream groups are started from
whichever trigger subscribes first but serve several triggers, so they start
with it cleared rather than inheriting that trigger's id. This isn't ideal,
but will do "for now".
The trigger runner process (the one running async code) sets the task
instance id, and since that process runs user code a trigger can claim to be
any task instance. The (in-process) API therefore checks the claim. A write
acts as the task instance only when it is deferred and its trigger is
already assigned to the calling triggerer. A read acts as it whenever that
try is still live, because after a user marks a deferred task failed or
successful the trigger row is gone and the triggerer still runs the
trigger's on_kill(). The most a misbehaving trigger can do is write as
another task that is deferred on the same triggerer, or read as any live
task, and it can already read and write every task's XComs by coordinates
through the in-process API. As stated, this is _not_ a security boundary we
can rely on
Requests that fail the check are rejected rather than falling back to the
anonymous identity, deliberately: a trigger whose task instance is no
longer deferred, or whose trigger was reassigned, should not still be
writing its task's XComs, and falling back would only work for tasks
outside loops. Clearing a deferred task archives its try before on_kill()
runs, so on_kill() after a clear cannot read its task's XComs; a trigger
that needs a job id there should keep it on the trigger, as the in-tree
triggers already do.
Only the in-process API that the triggerer creates knows which triggerer
is calling, so every other API rejects the header rather than trusting it.
A triggerer that eventually talks to a real API server will need its own
credential before this can work there.
Only the XCom routes honour the trigger's task instance id for now.
get_previous_task_instance also resolves relative to the caller, so a
trigger asking for its previous task instance still gets the latest pass
rather than the one at its own loop position. Every other request from the
triggerer keeps its current identity, so access rules and the live-attempt
check on those routes are unchanged.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Resolve a trigger's XCom requests as its deferred task instance [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74772:
URL: https://github.com/apache/airflow/pull/74772#discussion_r4282576227
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py:
##########
@@ -522,6 +528,121 @@ def test_loop_task_reads_from_inside_the_loop_stay_on_one_pass(client, outside_l
assert client.get(f"{fanout_url}/item/-1").json() == "1-1"
+@pytest.fixture
+def deferred_loop_pass(session, outside_loop_run):
+ dr, _, _, steps, _ = outside_loop_run
+ trigger = Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={})
+ trigger.triggerer_id = TRIGGERER_ID
+ session.add(trigger)
+ session.flush()
+ steps[1].state = TaskInstanceState.DEFERRED
+ steps[1].trigger_id = trigger.id
+ session.commit()
+ yield dr, steps, trigger
+ clear_db_triggers()
+
+
+def test_trigger_in_a_loop_reads_and_writes_as_its_deferred_pass(session, deferred_loop_pass):
+ dr, steps, _ = deferred_loop_pass
+ headers = {TRIGGER_TI_ID_HEADER: str(steps[1].id)}
+ step_url = f"/xcoms/{dr.dag_id}/{dr.run_id}/body.step/key"
+
+ sessions_begun = []
+
+ def record(api_session, *_):
+ sessions_begun.append(api_session)
+
+ with TestClient(InProcessExecutionAPI(triggerer_id=TRIGGERER_ID).app) as api:
+ event.listen(Session, "after_begin", record)
+ try:
+ with capture_orm_selects("task_instance") as task_instance_selects:
+ assert api.get(step_url, headers=headers).json() == {"key": "key", "value": "step-1"}
+ assert api.head(step_url, headers=headers).headers["Content-Range"] == "map_indexes 1"
+ assert api.get(f"{step_url}/slice", headers=headers).json() == ["step-1"]
+ assert api.get(f"{step_url}/item/0", headers=headers).json() == "step-1"
+ assert api.post(f"{step_url}-written", json="from-trigger", headers=headers).status_code == 201
+ assert api.delete(step_url, headers=headers).status_code == 200
+ finally:
+ event.remove(Session, "after_begin", record)
+
+ liveness_checks = [
+ statement
+ for statement in task_instance_selects
+ if statement.startswith("SELECT task_instance.working_set FROM")
+ ]
+ assert len(liveness_checks) == 1
+ assert len(set(map(id, sessions_begun))) == 6
+ session.expire_all()
+ assert XComModelV2.get_for_attempt(steps[1].id, "key-written", session=session).value == "from-trigger"
+ assert XComModelV2.get_for_attempt(steps[1].id, "key", session=session) is None
+ assert XComModelV2.get_for_attempt(steps[0].id, "key", session=session).value == "step-0"
+
+
+@pytest.mark.parametrize(
+ ("api_triggerer_id", "header", "expected_status"),
+ [
+ pytest.param(None, None, 403, id="api-without-triggerer-id"),
+ pytest.param(TRIGGERER_ID, "not-a-uuid", 400, id="malformed-header"),
+ ],
+)
+def test_trigger_ti_id_header_rejected(deferred_loop_pass, api_triggerer_id, header, expected_status):
+ dr, steps, _ = deferred_loop_pass
+
+ with TestClient(InProcessExecutionAPI(triggerer_id=api_triggerer_id).app) as api:
+ response = api.get(
+ f"/xcoms/{dr.dag_id}/{dr.run_id}/body.step/key",
+ headers={TRIGGER_TI_ID_HEADER: header or str(steps[1].id)},
+ )
+
+ assert response.status_code == expected_status
+
+
+@pytest.mark.parametrize(
+ ("method", "change", "expected_status", "expected_value"),
+ [
+ pytest.param(
+ "post",
+ {"triggerer_id": TRIGGERER_ID + 1},
+ 403,
+ None,
+ id="write-trigger-assigned-to-another-triggerer",
+ ),
+ pytest.param("post", {"state": TaskInstanceState.RUNNING}, 403, None, id="write-task-not-deferred"),
+ pytest.param(
+ "get",
+ {"state": TaskInstanceState.FAILED, "trigger_removed": True},
+ 200,
+ "step-1",
+ id="read-live-task-no-longer-deferred",
+ ),
+ pytest.param("get", {"archived": True}, 403, None, id="read-archived-try"),
Review Comment:
These cases only use `post` and `get`, so nothing pins which rule HEAD and DELETE get. HEAD and DELETE only run in the happy-path test, where the task is deferred on this triggerer and passes both checks. So changing `request.method in ("GET", "HEAD")` in `_resolve_trigger_token` to `== "GET"`, or to `!= "POST"` (which puts DELETE on the looser read rule), still passes everything. A `head` case with the trigger removed (expect 200) and a `delete` case with `RUNNING` (expect 403) would cover it. Asserting `detail["message"]` on the 403s would also show which check denied.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -252,29 +256,86 @@ def _build_xcom_read(
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(error)) from error
-def xcom_query(
- dag_id: str,
- run_id: str,
- task_id: str,
- key: str,
+def _resolve_trigger_token(
+ request: Request,
session: SessionDep,
- dag_bag: DagBagDep,
- map_index: Annotated[int | None, Query()] = None,
- previous_iteration: bool = False,
token: ExecutionToken = CurrentExecutionToken,
-) -> Select:
- return _build_xcom_read(
- dag_id=dag_id,
- run_id=run_id,
- task_id=task_id,
- key=key,
- session=session,
- dag_bag=dag_bag,
- token=token,
- map_index=map_index,
- previous_iteration=previous_iteration,
- sequence=True,
- ).statement
+ dependency_solver=Depends(current_dependency_solver),
+) -> ExecutionToken:
+ """
+ Return the identity an XCom request acts as.
+
+ A request made by a trigger's code carries its task instance's id in ``TRIGGER_TI_ID_HEADER``. The
+ triggerer subprocess can forge the header, so it is only honoured on the triggerer's in-process API.
+
+ A write acts as the claimed task instance only while it is deferred on a trigger assigned to the calling
+ triggerer. A read acts as it while that try is live, in any state, because a trigger's ``on_kill`` runs
+ after the task instance leaves ``deferred`` and the trigger row is gone; reads are not a security
+ boundary on this API, which already serves every XCom. Anything else gets a 403.
+ """
+ header = request.headers.get(TRIGGER_TI_ID_HEADER)
+ # The endpoint runs with Cadwyn's re-solved dependencies, so checking there reuses the endpoint's session.
+ if header is None or dependency_solver != "cadwyn":
+ return token
+
+ triggerer_id = request.scope.get(_IN_PROCESS_TRIGGERER_ID)
+ if triggerer_id is None:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail={
+ "reason": "access_denied",
+ "message": f"{TRIGGER_TI_ID_HEADER} is only accepted from the triggerer",
+ },
+ )
+
+ try:
+ claimed = UUID(header)
+ except ValueError:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail={"reason": "invalid_request", "message": f"{TRIGGER_TI_ID_HEADER} must be a UUID"},
+ )
+
+ if request.method in ("GET", "HEAD"):
+ permitted = (
+ session.scalar(
+ select(TaskInstance.working_set)
+ .where(TaskInstance.id == claimed)
+ .execution_options(include_all_attempts=True)
+ )
+ is True
+ )
+ denial = "Task instance is not a live attempt"
+ else:
+ permitted = (
+ session.scalar(
+ select(TaskInstance.id)
+ .join(Trigger, Trigger.id == TaskInstance.trigger_id)
+ .where(
+ TaskInstance.id == claimed,
+ Trigger.triggerer_id == triggerer_id,
Review Comment:
What happens to the task when a trigger gets reassigned mid-run? If triggerer A misses its heartbeat and the trigger moves to B, A's copy keeps running until A cancels it. An XCom write from A in that window now gets a 403 here. The trigger raises before yielding an event, `cleanup_finished_triggers` queues it as failed, and `Trigger.submit_failure` (which selects on `trigger_id` and `DEFERRED` only, not on the owning triggerer) sends the task to `__fail__` while B is still running it. Before this, a write from a trigger outside a loop resolved by coordinates and went through. It's narrow, since no in-tree trigger writes XComs from `run()`, but should `handle_failed_triggers` skip triggers this triggerer no longer owns?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Resolve a trigger's XCom requests as its deferred task instance [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #74772:
URL: https://github.com/apache/airflow/pull/74772#issuecomment-6196861842
Hi maintainer, this PR was merged without a milestone set.
We've automatically set the milestone to **[Airflow 3.4.1](https://github.com/apache/airflow/milestone/154)** based on: backport label targeting v3-4-test
If this milestone is not correct, please update it to the appropriate milestone.
> This comment was generated by [Milestone Tag Assistant](https://github.com/apache/airflow/blob/main/.github/workflows/milestone-tag-assistant.yml).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Resolve a trigger's XCom requests as its deferred task instance [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #74772:
URL: https://github.com/apache/airflow/pull/74772#issuecomment-6196871117
### Backport failed to create: v3-4-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/38767996333'> Run details </a>
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-4-test</td>
<td><a href="https://github.com/apache/airflow/commit/61f1d4b85e1553bc973e15b88656d3b46b44ff60"><img src='https://img.shields.io/badge/Commit-61f1d4b-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker 61f1d4b v3-4-test
```
This should apply the commit to the v3-4-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
If you don't have cherry-picker installed, see the [installation guide](https://github.com/apache/airflow/blob/main/dev/README.md#how-to-backport-pr-with-cherry-picker-cli).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Resolve a trigger's XCom requests as its deferred task instance [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb merged PR #74772:
URL: https://github.com/apache/airflow/pull/74772
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org