This is an automated email from the ASF dual-hosted git repository.
o-nikolas 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 3b9d2a22211 Scope XCom Execution API to teams in multi-team mode
(#68850)
3b9d2a22211 is described below
commit 3b9d2a222113b47faf07fcebfe3adef034b96e8b
Author: Niko Oliveira <[email protected]>
AuthorDate: Mon Jun 22 16:53:47 2026 -0700
Scope XCom Execution API to teams in multi-team mode (#68850)
XCom was the only shared resource without team isolation at the
task api level. Any task could read, overwrite, or delete another
team's XCom. Enforce team ownership so reads are allowed for the
task's own team or global dags, and writes/deletes only for its own
team, matching team-scoped variables/connections.
No cross-team sharing at this time. Gated on the multi_team setting
(no-op when disabled).
---
.../api_fastapi/execution_api/routes/xcoms.py | 59 ++++++--
.../airflow/api_fastapi/execution_api/security.py | 15 ++
.../execution_api/versions/head/test_xcoms.py | 163 ++++++++++++++++++++-
3 files changed, 224 insertions(+), 13 deletions(-)
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 a7592e4f0f2..9ab8eb20071 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
@@ -38,31 +38,70 @@ from airflow.models.xcom import XComModel
from airflow.utils.db import get_query_count
-async def has_xcom_access(
+def has_xcom_access(
dag_id: str,
run_id: str,
task_id: str,
xcom_key: Annotated[str, Path(alias="key", min_length=1)],
request: Request,
+ session: SessionDep,
token=CurrentTIToken,
) -> bool:
- """Check if the task has access to the XCom."""
+ """
+ Check whether the requesting task may access the XCom for ``dag_id``.
+
+ In multi-team mode, XCom access is scoped by team ownership (resolved via
the
+ ``dag -> bundle -> team`` chain). There is no cross-team XCom sharing:
+
+ * reads (``GET``/``HEAD``) are allowed for the requester's own team or for
+ global (teamless) dags;
+ * writes and deletes are allowed only for the requester's own team; a team
+ task may not mutate a global dag's XCom, mirroring how team-scoped
+ Variables and Connections behave.
+
+ When multi-team mode is disabled this is a no-op and all access is allowed,
+ consistent with Airflow's single-team security model where workers within a
+ deployment trust each other. Note this enforces the boundary at the
Execution
+ API only; it does not constrain code paths with direct database access
(e.g.
+ the Dag File Processor or Triggerer).
+ """
+ from airflow.configuration import conf
+
write = request.method not in {"GET", "HEAD", "OPTIONS"}
log.debug(
- "Checking %s XCom access for xcom from TaskInstance with key '%s' to
XCom '%s'",
+ "Checking %s XCom access for task instance '%s' to XCom '%s' on dag
'%s'",
"write" if write else "read",
token.id,
xcom_key,
+ dag_id,
+ )
+
+ if not conf.getboolean("core", "multi_team"):
+ return True
+
+ from airflow.api_fastapi.execution_api.security import (
+ _team_name_for_dag_stmt,
+ _team_name_for_ti_stmt,
)
- # The current version of Airflow does not support true
- # multi-tenancy yet (this is well-documented at
- #
https://airflow.apache.org/docs/apache-airflow/stable/security/security_model.html#limiting-dag-author-access-to-subset-of-dags),
- # so for now we always return 'True' here.
- # When we introduce true multi-tenancy in the future
- # this would be the place to do add a check.
- return True
+ requester_team = session.scalar(_team_name_for_ti_stmt(token.id))
+ target_team = session.scalar(_team_name_for_dag_stmt(dag_id))
+
+ # Same team (including a teamless task accessing a global, teamless dag)
is always allowed.
+ if target_team == requester_team:
+ return True
+ # Reads may additionally reach global (teamless) dags; writes and deletes
may not.
+ if not write and target_team is None:
+ return True
+
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail={
+ "reason": "access_denied",
+ "message": "Task does not have access to this XCom in multi-team
mode",
+ },
+ )
router = APIRouter(
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
index dabad06dbaf..9de3493061f 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/security.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/security.py
@@ -274,3 +274,18 @@ def _team_name_for_ti_stmt(ti_id):
.join(DagBundleModel.teams)
.where(TaskInstance.id == ti_id)
)
+
+
+def _team_name_for_dag_stmt(dag_id):
+ """Build the select statement resolving ``DagModel.dag_id ->
Team.name``."""
+ from airflow.models import DagModel
+ from airflow.models.dagbundle import DagBundleModel
+ from airflow.models.team import Team
+
+ return (
+ select(Team.name)
+ .select_from(DagModel)
+ .join(DagBundleModel, DagBundleModel.name == DagModel.bundle_name)
+ .join(DagBundleModel.teams)
+ .where(DagModel.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 899435f717a..277e2a78fd8 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
@@ -19,19 +19,25 @@ from __future__ import annotations
import logging
import urllib.parse
+from uuid import uuid4
import pytest
from fastapi import FastAPI, HTTPException, Path, Request, status
-from sqlalchemy import delete, select
+from sqlalchemy import delete, select, update
from airflow._shared.timezones import timezone
+from airflow.api_fastapi.execution_api.datamodels.token import TIClaims,
TIToken
from airflow.api_fastapi.execution_api.datamodels.xcom import XComResponse
+from airflow.api_fastapi.execution_api.security import require_auth
from airflow.models.dagrun import DagRun
from airflow.models.taskmap import TaskMap
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.serialization.serde import deserialize, serialize
from airflow.utils.session import create_session
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils.config import conf_vars
pytestmark = pytest.mark.db_test
@@ -53,7 +59,7 @@ def access_denied(client):
assert isinstance(last_route.app, FastAPI)
exec_app = last_route.app
- async def _(
+ def _(
request: Request,
dag_id: str = Path(),
run_id: str = Path(),
@@ -61,7 +67,8 @@ def access_denied(client):
xcom_key: str = Path(alias="key"),
token=CurrentTIToken,
):
- await has_xcom_access(dag_id, run_id, task_id, xcom_key, request,
token)
+ with create_session() as session:
+ has_xcom_access(dag_id, run_id, task_id, xcom_key, request,
session, token)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
@@ -609,3 +616,153 @@ class TestXComsDeleteEndpoint:
)
).first()
assert xcom_ti is not None
+
+
+class TestXComTeamAccess:
+ """Multi-team isolation for the Execution API XCom routes (no cross-team
sharing)."""
+
+ @staticmethod
+ def _make_dag(session, dag_maker, dag_id, team_name):
+ """
+ Create a dag (with a running dag run and a ``task`` TI) owned by
``team_name``.
+
+ ``team_name=None`` leaves the dag global (its bundle has no team).
Always
+ assigns a fresh bundle so the dag's team ownership is independent of
any
+ shared default bundle state. Returns ``(dag_run, task_instance)``.
+ """
+ from airflow.models import DagModel
+ from airflow.models.dagbundle import DagBundleModel
+ from airflow.models.team import Team
+
+ with dag_maker(dag_id=dag_id, session=session):
+ EmptyOperator(task_id="task")
+ dr = dag_maker.create_dagrun(run_id="run1", state=DagRunState.RUNNING)
+ ti = dr.get_task_instance("task")
+
+ bundle_name = f"bundle-{dag_id}"
+ bundle = DagBundleModel(name=bundle_name)
+ if team_name is not None:
+ team = session.scalar(select(Team).where(Team.name == team_name))
or Team(name=team_name)
+ bundle.teams.append(team)
+ session.add(bundle)
+ session.flush()
+ session.execute(update(DagModel).where(DagModel.dag_id ==
dag_id).values(bundle_name=bundle_name))
+ session.commit()
+ return dr, ti
+
+ @staticmethod
+ def _insert_xcom(session, dag_run, dag_id, key="k", value="v"):
+ session.add(
+ XComModel(
+ key=key,
+ value=value,
+ dag_run_id=dag_run.id,
+ run_id=dag_run.run_id,
+ task_id="task",
+ dag_id=dag_id,
+ )
+ )
+ session.commit()
+
+ @staticmethod
+ def _authenticate_as(exec_app, ti_id):
+ async def _auth(request: Request) -> TIToken:
+ return TIToken(id=ti_id, claims=TIClaims(scope="execution"))
+
+ exec_app.dependency_overrides[require_auth] = _auth
+
+ @staticmethod
+ def _url(dag_id, key="k"):
+ return f"/execution/xcoms/{dag_id}/run1/task/{key}"
+
+ def test_multi_team_disabled_allows_cross_team(self, client, exec_app,
session, dag_maker):
+ """With multi-team disabled, the check is a no-op even across teams."""
+ _, requester_ti = self._make_dag(session, dag_maker,
f"req_{uuid4().hex}", "team_a")
+ target_dag = f"tgt_{uuid4().hex}"
+ target_dr, _ = self._make_dag(session, dag_maker, target_dag, "team_b")
+ self._insert_xcom(session, target_dr, target_dag)
+ self._authenticate_as(exec_app, requester_ti.id)
+
+ with conf_vars({("core", "multi_team"): "False"}):
+ response = client.get(self._url(target_dag))
+
+ assert response.status_code == 200, response.json()
+ assert response.json() == {"key": "k", "value": "v"}
+
+ def test_same_team_read_write_delete_allowed(self, client, exec_app,
session, dag_maker):
+ """A task may read, write, and delete XCom within its own team."""
+ dag_id = f"dag_{uuid4().hex}"
+ dag_run, ti = self._make_dag(session, dag_maker, dag_id, "team_a")
+ self._insert_xcom(session, dag_run, dag_id, key="existing", value="v")
+ self._authenticate_as(exec_app, ti.id)
+
+ with conf_vars({("core", "multi_team"): "True"}):
+ read = client.get(self._url(dag_id, key="existing"))
+ write = client.post(self._url(dag_id, key="newkey"), json="w")
+ delete_ = client.delete(self._url(dag_id, key="existing"))
+
+ assert read.status_code == 200, read.json()
+ assert read.json() == {"key": "existing", "value": "v"}
+ assert write.status_code == 201, write.json()
+ assert delete_.status_code == 200, delete_.json()
+
+ def test_same_team_cross_dag_read_allowed(self, client, exec_app, session,
dag_maker):
+ """A task may read another dag's XCom when both dags belong to the
same team."""
+ _, requester_ti = self._make_dag(session, dag_maker,
f"req_{uuid4().hex}", "team_a")
+ target_dag = f"tgt_{uuid4().hex}"
+ target_dr, _ = self._make_dag(session, dag_maker, target_dag, "team_a")
+ self._insert_xcom(session, target_dr, target_dag)
+ self._authenticate_as(exec_app, requester_ti.id)
+
+ with conf_vars({("core", "multi_team"): "True"}):
+ response = client.get(self._url(target_dag))
+
+ assert response.status_code == 200, response.json()
+ assert response.json() == {"key": "k", "value": "v"}
+
+ @pytest.mark.parametrize("method", ["get", "post", "delete"])
+ def test_cross_team_access_forbidden(self, client, exec_app, session,
dag_maker, method):
+ """A task cannot read, write, or delete another team's XCom."""
+ _, requester_ti = self._make_dag(session, dag_maker,
f"req_{uuid4().hex}", "team_a")
+ target_dag = f"tgt_{uuid4().hex}"
+ self._make_dag(session, dag_maker, target_dag, "team_b")
+ self._authenticate_as(exec_app, requester_ti.id)
+
+ kwargs = {"json": "v"} if method == "post" else {}
+ with conf_vars({("core", "multi_team"): "True"}):
+ response = getattr(client, method)(self._url(target_dag), **kwargs)
+
+ assert response.status_code == 403, response.json()
+ assert response.json()["detail"]["reason"] == "access_denied"
+
+ def test_global_dag_read_allowed_but_write_forbidden(self, client,
exec_app, session, dag_maker):
+ """A team task may read a global (teamless) dag's XCom but not mutate
it."""
+ _, requester_ti = self._make_dag(session, dag_maker,
f"req_{uuid4().hex}", "team_a")
+ global_dag = f"global_{uuid4().hex}"
+ global_dr, _ = self._make_dag(session, dag_maker, global_dag, None)
+ self._insert_xcom(session, global_dr, global_dag, key="k", value="v")
+ self._authenticate_as(exec_app, requester_ti.id)
+
+ with conf_vars({("core", "multi_team"): "True"}):
+ read = client.get(self._url(global_dag, key="k"))
+ write = client.post(self._url(global_dag, key="k2"), json="w")
+ delete_ = client.delete(self._url(global_dag, key="k"))
+
+ assert read.status_code == 200, read.json()
+ assert write.status_code == 403, write.json()
+ assert delete_.status_code == 403, delete_.json()
+
+ def test_teamless_requester_scoping(self, client, session, dag_maker):
+ """A teamless requester (default token) reaches global dags but not
team dags."""
+ team_dag = f"team_{uuid4().hex}"
+ self._make_dag(session, dag_maker, team_dag, "team_b")
+ global_dag = f"global_{uuid4().hex}"
+ global_dr, _ = self._make_dag(session, dag_maker, global_dag, None)
+ self._insert_xcom(session, global_dr, global_dag, key="k", value="v")
+
+ with conf_vars({("core", "multi_team"): "True"}):
+ forbidden = client.get(self._url(team_dag, key="k"))
+ allowed = client.get(self._url(global_dag, key="k"))
+
+ assert forbidden.status_code == 403, forbidden.json()
+ assert allowed.status_code == 200, allowed.json()
