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 5aeba034521 Authorize team-scoped plugin API endpoints in multi-team
mode (#71919)
5aeba034521 is described below
commit 5aeba034521864f2ea27ef1d5f3f15c6abf296b2
Author: Niko Oliveira <[email protected]>
AuthorDate: Tue Aug 25 11:38:15 2026 -0700
Authorize team-scoped plugin API endpoints in multi-team mode (#71919)
A plugin's team_name has so far only affected what the API returns about
the plugin, not what its own endpoints allow. A team-scoped plugin's
endpoints were reachable by every authenticated user.
When multi-team mode is enabled, a team-scoped plugin's app is now mounted
behind a middleware that rejects users who are not authorized for that team,
using the auth manager's existing team check. Requests are authenticated the
same way as the core API (bearer token or session cookie). Global plugins
and
deployments with multi-team disabled are unaffected.
Root middlewares from a team-scoped plugin are skipped with a warning: they
wrap every request to the API server, including other teams' and core
routes, so they cannot be limited to one team. A team plugin that needs
middleware should apply it inside its own app.
---
airflow-core/src/airflow/api_fastapi/app.py | 35 ++++
.../auth/middlewares/team_authorization.py | 110 +++++++++++
airflow-core/src/airflow/plugins_manager.py | 15 +-
.../auth/middlewares/test_team_authorization.py | 207 +++++++++++++++++++++
airflow-core/tests/unit/api_fastapi/test_app.py | 75 ++++++++
.../tests/unit/plugins/test_plugins_manager.py | 52 ++++++
6 files changed, 491 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/app.py
b/airflow-core/src/airflow/api_fastapi/app.py
index 233965bd56e..55ec201b9ee 100644
--- a/airflow-core/src/airflow/api_fastapi/app.py
+++ b/airflow-core/src/airflow/api_fastapi/app.py
@@ -25,6 +25,7 @@ from urllib.parse import urlsplit
from fastapi import FastAPI
from fastapi.routing import Mount
+from starlette.middleware import Middleware
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.common.exceptions import init_error_handlers
@@ -221,8 +222,10 @@ def get_auth_manager() -> BaseAuthManager:
def init_plugins(app: FastAPI) -> None:
"""Integrate FastAPI app, middlewares and UI plugins."""
from airflow import plugins_manager
+ from airflow.api_fastapi.auth.middlewares.team_authorization import
TeamAuthorizationMiddleware
apps, root_middlewares = plugins_manager.get_fastapi_plugins()
+ multi_team = conf.getboolean("core", "multi_team")
for subapp_dict in apps:
name = subapp_dict.get("name")
@@ -241,6 +244,25 @@ def init_plugins(app: FastAPI) -> None:
log.error("Plugin %s attempted to use reserved url_prefix '%s'",
name, url_prefix)
continue
+ if multi_team and (team_name := subapp_dict.get("team_name")) is not
None:
+ # Airflow applies no authorization to mounted plugin apps, so a
team-scoped
+ # plugin's endpoints would otherwise be reachable by any
authenticated user.
+ # `app.mount()` cannot attach middleware, so build the Mount
directly.
+ log.debug(
+ "Adding subapplication %s under prefix %s, restricted to team
%s",
+ name,
+ url_prefix,
+ team_name,
+ )
+ app.router.routes.append(
+ Mount(
+ url_prefix,
+ app=subapp,
+ middleware=[Middleware(TeamAuthorizationMiddleware,
team_name=team_name)],
+ )
+ )
+ continue
+
log.debug("Adding subapplication %s under prefix %s", name, url_prefix)
app.mount(url_prefix, subapp)
@@ -258,5 +280,18 @@ def init_plugins(app: FastAPI) -> None:
log.error("'middleware' value for %s is should be callable: %s",
name, middleware)
continue
+ if multi_team and (team_name := middleware_dict.get("team_name")) is
not None:
+ # Root middlewares wrap every request to the API server, including
other
+ # teams' and core routes, so they cannot be scoped to one team. A
team plugin
+ # that needs middleware should apply it inside its own FastAPI app.
+ log.warning(
+ "Skipping root middleware %s from team-scoped plugin (team
%s): root middlewares "
+ "apply to the entire API server and cannot be restricted to a
single team. "
+ "Apply it within the plugin's own fastapi_apps instead.",
+ name,
+ team_name,
+ )
+ continue
+
log.debug("Adding root middleware %s", name)
app.add_middleware(middleware, *args, **kwargs)
diff --git
a/airflow-core/src/airflow/api_fastapi/auth/middlewares/team_authorization.py
b/airflow-core/src/airflow/api_fastapi/auth/middlewares/team_authorization.py
new file mode 100644
index 00000000000..4ffde678bea
--- /dev/null
+++
b/airflow-core/src/airflow/api_fastapi/auth/middlewares/team_authorization.py
@@ -0,0 +1,110 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi import HTTPException, Request, status
+from fastapi.responses import JSONResponse
+from starlette.concurrency import run_in_threadpool
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from airflow.api_fastapi.app import get_auth_manager
+from airflow.api_fastapi.auth.managers.base_auth_manager import
COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.auth.managers.models.resource_details import
TeamDetails
+from airflow.api_fastapi.core_api.security import (
+ USER_INJECTED_BY_TRUSTED_MIDDLEWARE,
+ resolve_user_from_token,
+)
+
+if TYPE_CHECKING:
+ from airflow.api_fastapi.auth.managers.base_auth_manager import
ResourceMethod
+ from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+
+# Airflow authorizes against a small set of methods, so map the request's HTTP
method onto
+# one of them. This lets an auth manager grant a team read-only access to its
plugin if it
+# distinguishes methods; managers that only check membership can ignore it.
+_HTTP_METHOD_TO_RESOURCE_METHOD: dict[str, ResourceMethod] = {
+ "GET": "GET",
+ "HEAD": "GET",
+ "OPTIONS": "GET",
+ "POST": "POST",
+ "PUT": "PUT",
+ "PATCH": "PUT",
+ "DELETE": "DELETE",
+}
+
+
+class TeamAuthorizationMiddleware(BaseHTTPMiddleware):
+ """
+ Restrict a team-scoped plugin's FastAPI app to users authorized for that
team.
+
+ Plugin apps are mounted on the shared API server, which applies no
authorization of
+ its own to them, so without this every authenticated user can reach a
team's plugin
+ endpoints. This wraps only the owning team's sub-app, leaving global
plugins and core
+ routes untouched.
+ """
+
+ def __init__(self, app, team_name: str) -> None:
+ super().__init__(app)
+ self.team_name = team_name
+
+ async def dispatch(self, request: Request, call_next):
+ try:
+ user = await self._resolve_user(request)
+ except HTTPException as exception:
+ # Unauthenticated or invalid token. Mirror the core API's error
shape rather
+ # than letting the exception escape, since a plugin sub-app does
not
+ # necessarily install Airflow's exception handlers.
+ return JSONResponse(status_code=exception.status_code,
content={"detail": exception.detail})
+
+ authorized = await run_in_threadpool(
+ get_auth_manager().is_authorized_team,
+ method=_HTTP_METHOD_TO_RESOURCE_METHOD.get(request.method.upper(),
"GET"),
+ user=user,
+ details=TeamDetails(name=self.team_name),
+ )
+ if not authorized:
+ return JSONResponse(
+ status_code=status.HTTP_403_FORBIDDEN,
+ content={"detail": f"You are not authorized to access team
{self.team_name!r}."},
+ )
+
+ return await call_next(request)
+
+ async def _resolve_user(self, request: Request) -> BaseUser:
+ """
+ Build the requesting user, mirroring the core API's ``get_user``
dependency.
+
+ FastAPI dependencies are not available inside middleware, so the token
is read
+ from the request directly: a bearer header for API clients, otherwise
the session
+ cookie used by the UI.
+ """
+ user: BaseUser | None = getattr(request.state, "user", None)
+ if user and getattr(request.state, "user_authenticated_via", None) is (
+ USER_INJECTED_BY_TRUSTED_MIDDLEWARE
+ ):
+ return user
+
+ authorization = request.headers.get("Authorization")
+ if authorization and authorization.lower().startswith("bearer "):
+ token_str: str | None = authorization[len("bearer ") :].strip()
+ else:
+ token_str = request.cookies.get(COOKIE_NAME_JWT_TOKEN)
+
+ return await resolve_user_from_token(token_str)
diff --git a/airflow-core/src/airflow/plugins_manager.py
b/airflow-core/src/airflow/plugins_manager.py
index df1ade2e7a1..ff3c620b934 100644
--- a/airflow-core/src/airflow/plugins_manager.py
+++ b/airflow-core/src/airflow/plugins_manager.py
@@ -315,7 +315,14 @@ def get_flask_plugins() -> tuple[list[Any], list[Any],
list[Any]]:
@cache
def get_fastapi_plugins() -> tuple[list[Any], list[Any]]:
- """Collect extension points for the API."""
+ """
+ Collect extension points for the API.
+
+ Each returned dict is a shallow copy of the plugin's own dict with the
owning
+ plugin's ``team_name`` added, so the API server can authorize a team-scoped
+ plugin's app without re-deriving which plugin it came from. The plugin's
dicts are
+ left untouched, so this does not alter what ``get_plugin_info`` reports.
+ """
log.debug("Initialize FastAPI plugins")
# Validate here (the API-server, DB-available path) so callers cannot mount
@@ -325,8 +332,10 @@ def get_fastapi_plugins() -> tuple[list[Any], list[Any]]:
fastapi_apps: list[Any] = []
fastapi_root_middlewares: list[Any] = []
for plugin in _get_plugins()[0]:
- fastapi_apps.extend(plugin.fastapi_apps)
- fastapi_root_middlewares.extend(plugin.fastapi_root_middlewares)
+ fastapi_apps.extend({**app, "team_name": plugin.team_name} for app in
plugin.fastapi_apps)
+ fastapi_root_middlewares.extend(
+ {**middleware, "team_name": plugin.team_name} for middleware in
plugin.fastapi_root_middlewares
+ )
return fastapi_apps, fastapi_root_middlewares
diff --git
a/airflow-core/tests/unit/api_fastapi/auth/middlewares/test_team_authorization.py
b/airflow-core/tests/unit/api_fastapi/auth/middlewares/test_team_authorization.py
new file mode 100644
index 00000000000..b9754f52b8e
--- /dev/null
+++
b/airflow-core/tests/unit/api_fastapi/auth/middlewares/test_team_authorization.py
@@ -0,0 +1,207 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException, Request, Response, status
+
+from airflow.api_fastapi.auth.managers.base_auth_manager import
COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
+from airflow.api_fastapi.core_api.security import
USER_INJECTED_BY_TRUSTED_MIDDLEWARE
+
+MIDDLEWARE_MODULE = "airflow.api_fastapi.auth.middlewares.team_authorization"
+
+
[email protected]
+def middleware():
+ from airflow.api_fastapi.auth.middlewares.team_authorization import
TeamAuthorizationMiddleware
+
+ return TeamAuthorizationMiddleware(app=MagicMock(), team_name="team_a")
+
+
[email protected]
+def mock_request():
+ request = MagicMock(spec=Request)
+ request.cookies = {}
+ request.headers = {}
+ request.method = "GET"
+ request.state = MagicMock()
+ request.state.user = None
+ request.state.user_authenticated_via = None
+ return request
+
+
[email protected]
+def mock_user():
+ return MagicMock(spec=BaseUser)
+
+
+class TestTeamAuthorizationMiddleware:
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_authorized_team_member_passes_through(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "a_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+ expected = Response(status_code=200)
+ call_next = AsyncMock(return_value=expected)
+
+ response = await middleware.dispatch(mock_request, call_next)
+
+ assert response is expected
+ call_next.assert_awaited_once_with(mock_request)
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_user_outside_team_is_forbidden(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "a_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
False
+ call_next = AsyncMock()
+
+ response = await middleware.dispatch(mock_request, call_next)
+
+ assert response.status_code == status.HTTP_403_FORBIDDEN
+ call_next.assert_not_awaited()
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(
+ f"{MIDDLEWARE_MODULE}.resolve_user_from_token",
+ side_effect=HTTPException(status_code=401, detail="Not authenticated"),
+ )
+ @pytest.mark.asyncio
+ async def test_unauthenticated_request_is_rejected(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request
+ ):
+ call_next = AsyncMock()
+
+ response = await middleware.dispatch(mock_request, call_next)
+
+ assert response.status_code == status.HTTP_401_UNAUTHORIZED
+ call_next.assert_not_awaited()
+ # The team check is unreachable without a user.
+
mock_get_auth_manager.return_value.is_authorized_team.assert_not_called()
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_authorizes_against_the_mounted_team(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "a_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+
+ await middleware.dispatch(mock_request,
AsyncMock(return_value=Response()))
+
+ kwargs =
mock_get_auth_manager.return_value.is_authorized_team.call_args.kwargs
+ assert kwargs["user"] is mock_user
+ assert kwargs["details"].name == "team_a"
+
+ @pytest.mark.parametrize(
+ ("http_method", "expected_resource_method"),
+ [
+ ("GET", "GET"),
+ ("HEAD", "GET"),
+ ("OPTIONS", "GET"),
+ ("POST", "POST"),
+ ("PUT", "PUT"),
+ ("PATCH", "PUT"),
+ ("DELETE", "DELETE"),
+ ("SOMETHING_ELSE", "GET"),
+ ],
+ )
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_maps_http_method_to_resource_method(
+ self,
+ mock_resolve_user,
+ mock_get_auth_manager,
+ middleware,
+ mock_request,
+ mock_user,
+ http_method,
+ expected_resource_method,
+ ):
+ mock_request.method = http_method
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "a_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+
+ await middleware.dispatch(mock_request,
AsyncMock(return_value=Response()))
+
+ kwargs =
mock_get_auth_manager.return_value.is_authorized_team.call_args.kwargs
+ assert kwargs["method"] == expected_resource_method
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_prefers_bearer_token_over_cookie(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ mock_request.headers = {"Authorization": "Bearer header_token"}
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "cookie_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+
+ await middleware.dispatch(mock_request,
AsyncMock(return_value=Response()))
+
+ mock_resolve_user.assert_awaited_once_with("header_token")
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_reuses_user_from_trusted_middleware(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ """A cookie request already authenticated by JWTRefreshMiddleware is
not re-resolved."""
+ mock_request.state.user = mock_user
+ mock_request.state.user_authenticated_via =
USER_INJECTED_BY_TRUSTED_MIDDLEWARE
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+
+ await middleware.dispatch(mock_request,
AsyncMock(return_value=Response()))
+
+ mock_resolve_user.assert_not_awaited()
+ assert
mock_get_auth_manager.return_value.is_authorized_team.call_args.kwargs["user"]
is mock_user
+
+ @patch(f"{MIDDLEWARE_MODULE}.get_auth_manager")
+ @patch(f"{MIDDLEWARE_MODULE}.resolve_user_from_token")
+ @pytest.mark.asyncio
+ async def test_ignores_untrusted_state_user(
+ self, mock_resolve_user, mock_get_auth_manager, middleware,
mock_request, mock_user
+ ):
+ """A ``state.user`` without the trust sentinel must not bypass token
validation."""
+ mock_request.state.user = MagicMock(spec=BaseUser)
+ mock_request.state.user_authenticated_via = None
+ mock_request.cookies = {COOKIE_NAME_JWT_TOKEN: "a_token"}
+ mock_resolve_user.return_value = mock_user
+ mock_get_auth_manager.return_value.is_authorized_team.return_value =
True
+
+ await middleware.dispatch(mock_request,
AsyncMock(return_value=Response()))
+
+ mock_resolve_user.assert_awaited_once_with("a_token")
+ assert
mock_get_auth_manager.return_value.is_authorized_team.call_args.kwargs["user"]
is mock_user
diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py
b/airflow-core/tests/unit/api_fastapi/test_app.py
index 9fd9a3edad1..26e505021da 100644
--- a/airflow-core/tests/unit/api_fastapi/test_app.py
+++ b/airflow-core/tests/unit/api_fastapi/test_app.py
@@ -25,6 +25,8 @@ from fastapi import FastAPI
import airflow.api_fastapi.app as app_module
import airflow.plugins_manager as plugins_manager
+from tests_common.test_utils.config import conf_vars
+
pytestmark = pytest.mark.db_test
@@ -114,6 +116,79 @@ def test_plugin_with_invalid_url_prefix(caplog,
invalid_prefix, expected_message
assert not any(r.path == invalid_prefix for r in app.routes)
+class TestInitPluginsTeamAuthorization:
+ """A team-scoped plugin's app must be mounted behind the team authorization
+ middleware, since Airflow applies no authorization to plugin apps
itself."""
+
+ @staticmethod
+ def _mount_for(app, url_prefix):
+ return next(route for route in app.routes if getattr(route, "path",
None) == url_prefix)
+
+ @staticmethod
+ def _has_team_middleware(mount, team_name):
+ from airflow.api_fastapi.auth.middlewares.team_authorization import
TeamAuthorizationMiddleware
+
+ # Starlette applies a Mount's middleware by wrapping the sub-app, so
the mounted
+ # app *is* the middleware instance when the plugin is team-scoped.
+ return isinstance(mount.app, TeamAuthorizationMiddleware) and
mount.app.team_name == team_name
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_team_plugin_app_is_wrapped(self):
+ fastapi_apps = [
+ {"name": "team_a_app", "app": FastAPI(), "url_prefix": "/team_a",
"team_name": "team_a"}
+ ]
+ app = FastAPI()
+ with mock.patch.object(plugins_manager, "get_fastapi_plugins",
return_value=(fastapi_apps, [])):
+ app_module.init_plugins(app)
+
+ assert self._has_team_middleware(self._mount_for(app, "/team_a"),
"team_a")
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_global_plugin_app_is_not_wrapped(self):
+ fastapi_apps = [{"name": "global_app", "app": FastAPI(), "url_prefix":
"/global", "team_name": None}]
+ app = FastAPI()
+ with mock.patch.object(plugins_manager, "get_fastapi_plugins",
return_value=(fastapi_apps, [])):
+ app_module.init_plugins(app)
+
+ mount = self._mount_for(app, "/global")
+ assert not self._has_team_middleware(mount, None)
+
+ @conf_vars({("core", "multi_team"): "False"})
+ def test_team_plugin_app_is_not_wrapped_when_multi_team_disabled(self):
+ fastapi_apps = [
+ {"name": "team_a_app", "app": FastAPI(), "url_prefix": "/team_a",
"team_name": "team_a"}
+ ]
+ app = FastAPI()
+ with mock.patch.object(plugins_manager, "get_fastapi_plugins",
return_value=(fastapi_apps, [])):
+ app_module.init_plugins(app)
+
+ assert not self._has_team_middleware(self._mount_for(app, "/team_a"),
"team_a")
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_team_plugin_root_middleware_is_skipped(self, caplog):
+ root_middlewares = [
+ {"name": "team_a_middleware", "middleware": mock.MagicMock(),
"team_name": "team_a"}
+ ]
+ app = FastAPI()
+ with mock.patch.object(plugins_manager, "get_fastapi_plugins",
return_value=([], root_middlewares)):
+ with mock.patch.object(app, "add_middleware") as
mock_add_middleware:
+ app_module.init_plugins(app)
+
+ mock_add_middleware.assert_not_called()
+ assert any("Skipping root middleware team_a_middleware" in rec.message
for rec in caplog.records)
+
+ @conf_vars({("core", "multi_team"): "True"})
+ def test_global_root_middleware_is_still_added(self):
+ middleware = mock.MagicMock()
+ root_middlewares = [{"name": "global_middleware", "middleware":
middleware, "team_name": None}]
+ app = FastAPI()
+ with mock.patch.object(plugins_manager, "get_fastapi_plugins",
return_value=([], root_middlewares)):
+ with mock.patch.object(app, "add_middleware") as
mock_add_middleware:
+ app_module.init_plugins(app)
+
+ mock_add_middleware.assert_called_once_with(middleware)
+
+
class TestGetCookiePath:
def test_default_returns_slash(self):
"""When no base_url is configured, get_cookie_path() should return
'/'."""
diff --git a/airflow-core/tests/unit/plugins/test_plugins_manager.py
b/airflow-core/tests/unit/plugins/test_plugins_manager.py
index ad23415912b..7f458ef4d90 100644
--- a/airflow-core/tests/unit/plugins/test_plugins_manager.py
+++ b/airflow-core/tests/unit/plugins/test_plugins_manager.py
@@ -640,3 +640,55 @@ class TestValidatePluginTeams:
assert "unknown_team" in recorded["team_plugin"]
warnings = [msg for _, level, msg in caplog.record_tuples if level ==
logging.WARNING]
assert any("team_plugin" in msg and "unknown_team" in msg for msg in
warnings)
+
+
+class TestGetFastapiPluginsTeamName:
+ """``get_fastapi_plugins`` must tell the API server which team each app
belongs to,
+ since that is what lets ``init_plugins`` authorize a team-scoped plugin's
app."""
+
+ @staticmethod
+ def _plugins():
+ class GlobalPlugin(AirflowPlugin):
+ name = "global_plugin"
+
+ class TeamPlugin(AirflowPlugin):
+ name = "team_plugin"
+ team_name = "team_a"
+
+ global_plugin = GlobalPlugin()
+ team_plugin = TeamPlugin()
+ # Per-instance dicts so a mutation would be visible to the assertions
below.
+ global_plugin.fastapi_apps = [{"name": "global_app", "app": object(),
"url_prefix": "/global"}]
+ global_plugin.fastapi_root_middlewares = [{"name": "global_mw",
"middleware": object()}]
+ team_plugin.fastapi_apps = [{"name": "team_app", "app": object(),
"url_prefix": "/team"}]
+ team_plugin.fastapi_root_middlewares = [{"name": "team_mw",
"middleware": object()}]
+ return global_plugin, team_plugin
+
+ def test_team_name_is_added_to_apps_and_middlewares(self):
+ from airflow import plugins_manager
+
+ global_plugin, team_plugin = self._plugins()
+ with mock_plugin_manager(plugins=[global_plugin, team_plugin]):
+ apps, middlewares = plugins_manager.get_fastapi_plugins()
+
+ assert {app["name"]: app["team_name"] for app in apps} == {
+ "global_app": None,
+ "team_app": "team_a",
+ }
+ assert {mw["name"]: mw["team_name"] for mw in middlewares} == {
+ "global_mw": None,
+ "team_mw": "team_a",
+ }
+
+ def test_plugin_dicts_are_not_mutated(self):
+ """The plugin's own dicts must stay clean so ``get_plugin_info`` (and
therefore
+ the public API response) does not gain an unexpected ``team_name``
key."""
+ from airflow import plugins_manager
+
+ global_plugin, team_plugin = self._plugins()
+ with mock_plugin_manager(plugins=[global_plugin, team_plugin]):
+ plugins_manager.get_fastapi_plugins()
+
+ for plugin in (global_plugin, team_plugin):
+ assert "team_name" not in plugin.fastapi_apps[0]
+ assert "team_name" not in plugin.fastapi_root_middlewares[0]
