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()