From e479d187d2e5f95dc203295bf9e0eec79d5d0799 Mon Sep 17 00:00:00 2001 From: Niko Oliveira Date: Fri, 19 Jun 2026 17:14:10 -0700 Subject: [PATCH] Scope XCom Execution API to teams in multi-team mode 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 +++++-- .../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 a7592e4f0f2ec..9ab8eb200712a 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.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 dabad06dbaf52..9de3493061f32 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 899435f717a9f..277e2a78fd8dc 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 @@ 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 @@ async def _( 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 @@ def test_xcom_delete_endpoint(self, client, create_task_instance, session): ) ).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()