Authorize team-scoped plugin API endpoints in multi-team mode

airflow-core/src/airflow/api_fastapi/app.py

@@ -25,6 +25,7 @@
 
 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)

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)

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