Add IMPORT_ERRORS_ALL permission for import errors of files with no registered Dag

airflow-core/docs/core-concepts/auth-manager/index.rst

@@ -140,13 +140,48 @@ These authorization methods are:
 * ``is_authorized_asset_alias``: Return whether the user is authorized to access Airflow asset aliases. Some details about the asset alias can be provided (e.g. the asset alias ID).
 * ``is_authorized_pool``: Return whether the user is authorized to access Airflow pools. Some details about the pool can be provided (e.g. the pool name).
 * ``is_authorized_variable``: Return whether the user is authorized to access Airflow variables. Some details about the variable can be provided (e.g. the variable key).
-* ``is_authorized_view``: Return whether the user is authorized to access a specific view in Airflow. The view is specified through ``access_view`` (e.g. ``AccessView.CLUSTER_ACTIVITY``).
+* ``is_authorized_view``: Return whether the user is authorized to access a specific view in Airflow. The view is specified through ``access_view`` (e.g. ``AccessView.CLUSTER_ACTIVITY``). An optional ``team_name`` scopes the check to a team -- see :ref:`team-scoped-view-authorization` below.
 * ``is_authorized_custom_view``: Return whether the user is authorized to access a specific view not defined in Airflow. This view can be provided by the auth manager itself or a plugin defined by the user.
 * ``filter_authorized_menu_items``: Given the list of menu items in the UI, return the list of menu items the user has access to.
 
 It should be noted that the ``method`` parameter listed above may only have relevance for a specific subset of the auth manager's authorization methods.
 For example, the ``configuration`` resource is by definition read-only, so only the ``GET`` parameter is relevant in the context of ``is_authorized_configuration``.
 
+.. _team-scoped-view-authorization:
+
+Team-scoped view authorization
+''''''''''''''''''''''''''''''
+
+.. versionadded:: 3.4.0
+   ``is_authorized_view`` accepts an optional ``team_name`` argument.
+
+In a multi-team deployment, access to a read-only view can be restricted to the users of a
+specific team. ``is_authorized_view`` accepts an optional ``team_name`` for that purpose. An
+auth manager that implements multi-team isolation honors it and only authorizes users who
+belong to ``team_name``; an auth manager without multi-team support accepts the argument and
+ignores it, which authorizes the view across all teams (the same behaviour it had before the
+argument existed).
+
+Core never calls ``is_authorized_view`` with ``team_name`` directly. It goes through
+``BaseAuthManager.authorize_view``, which first checks whether the auth manager's
+``is_authorized_view`` accepts ``team_name``. This keeps auth managers that predate the
+argument working: a custom or out-of-tree auth manager whose ``is_authorized_view`` still has
+the old ``(access_view, user)`` signature is called *without* ``team_name`` -- it does not
+raise -- and a ``RemovedInAirflow4Warning`` is emitted, warning that some views that should be
+restricted to a single team are instead authorized across all teams until the auth manager is
+upgraded.
+
+To make an auth manager team-aware, add ``team_name`` to the override (managers without
+multi-team support may accept and ignore it):
+
+.. code-block:: python
+
+    def is_authorized_view(
+        self, *, access_view: AccessView, user: MyUser, team_name: str | None = None
+    ) -> bool: ...
+
+The fallback for the older signature is removed in Airflow 4, which requires ``team_name``.
+
 JWT token management by auth managers
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 The auth manager is responsible for creating the JWT token needed to interact with Airflow public API.

airflow-core/newsfragments/69790.feature.rst

@@ -0,0 +1 @@
+Auth managers can scope read-only view authorization to a team via an optional ``team_name`` argument on ``is_authorized_view``, added to the auth manager interface in Airflow 3.4.0. It is supported by the Simple auth manager (shipped in Airflow) from 3.4.0, and by the Amazon, FAB, and Keycloak providers from the releases after 9.32.0, 3.7.2, and 0.8.1 respectively; older provider releases and auth managers that have not adopted the argument keep working but cannot restrict views per team. See :doc:`the auth manager documentation </core-concepts/auth-manager/index>` for details.

airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py

@@ -17,12 +17,13 @@
 # under the License.
 from __future__ import annotations
 
+import inspect
 import logging
 import warnings
 from abc import ABCMeta, abstractmethod
 from collections import defaultdict
 from enum import Enum
-from functools import cache
+from functools import cache, cached_property
 from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar
 
 from jwt import InvalidTokenError
@@ -44,6 +45,7 @@
 )
 from airflow.api_fastapi.common.types import ExtraMenuItem, MenuItem
 from airflow.configuration import conf
+from airflow.exceptions import RemovedInAirflow4Warning
 from airflow.models import Connection, DagModel, Pool, Variable
 from airflow.models.dagbundle import DagBundleModel
 from airflow.models.revoked_token import RevokedToken
@@ -357,13 +359,42 @@ def is_authorized_view(
         *,
         access_view: AccessView,
         user: T,
+        team_name: str | None = None,
     ) -> bool:
         """
         Return whether the user is authorized to access a read-only state of the installation.
 
         :param access_view: the specific read-only view/state the authorization request is about.
         :param user: the user to performing the action
+        :param team_name: team the view is scoped to, if any. Managers without multi-team
+            support may accept and ignore it, which authorizes the view globally.
+        """
+
+    @cached_property
+    def _is_authorized_view_team_aware(self) -> bool:
+        """Whether this manager's ``is_authorized_view`` override accepts ``team_name``."""
+        params = inspect.signature(self.is_authorized_view).parameters
+        return "team_name" in params or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values())
+
+    def authorize_view(self, *, access_view: AccessView, user: T, team_name: str | None = None) -> bool:
         """
+        Authorize a read-only view, tolerating auth managers that predate ``team_name``.
+
+        Core calls this instead of :meth:`is_authorized_view` on team-scoped paths: an
+        override still on the old ``(access_view, user)`` signature would otherwise raise
+        ``TypeError``. Removed in Airflow 4.
+        """
+        if self._is_authorized_view_team_aware:
+            return self.is_authorized_view(access_view=access_view, user=user, team_name=team_name)
+        warnings.warn(
+            f"The '{type(self).__name__}' auth manager is not team-aware, so team-scoped views are "
+            "authorized across all teams and may be visible to users of other teams. Add the "
+            "'team_name' argument to its is_authorized_view (or upgrade the provider). Airflow 4 "
+            "will require team-aware auth managers.",
+            RemovedInAirflow4Warning,
+            stacklevel=2,
+        )
+        return self.is_authorized_view(access_view=access_view, user=user)
 
     @abstractmethod
     def is_authorized_custom_view(self, *, method: ResourceMethod, resource_name: str, user: T) -> bool:

airflow-core/src/airflow/api_fastapi/auth/managers/models/resource_details.py

@@ -101,6 +101,9 @@ class AccessView(Enum):
     CLUSTER_ACTIVITY = "CLUSTER_ACTIVITY"
     DOCS = "DOCS"
     IMPORT_ERRORS = "IMPORT_ERRORS"
+    # Import errors for files with no registered Dag: there is no per-Dag key to
+    # authorize on, so they get their own admin-by-default view.
+    IMPORT_ERRORS_ALL = "IMPORT_ERRORS_ALL"
     JOBS = "JOBS"
     PLUGINS = "PLUGINS"
     PROVIDERS = "PROVIDERS"

airflow-core/src/airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py

@@ -37,7 +37,7 @@
 
 from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
 from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
-from airflow.api_fastapi.auth.managers.models.resource_details import TeamDetails
+from airflow.api_fastapi.auth.managers.models.resource_details import AccessView, TeamDetails
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
 from airflow.api_fastapi.common.types import MenuItem
 from airflow.configuration import AIRFLOW_HOME, conf
@@ -47,7 +47,6 @@
 
     from airflow.api_fastapi.auth.managers.base_auth_manager import ResourceMethod
     from airflow.api_fastapi.auth.managers.models.resource_details import (
-        AccessView,
         AssetAliasDetails,
         AssetDetails,
         ConfigurationDetails,
@@ -352,8 +351,17 @@ def is_authorized_variable(
             team_name=details.team_name if details else None,
         )
 
-    def is_authorized_view(self, *, access_view: AccessView, user: SimpleAuthManagerUser) -> bool:
-        return self._is_authorized(method="GET", allow_role=SimpleAuthManagerRole.VIEWER, user=user)
+    def is_authorized_view(
+        self, *, access_view: AccessView, user: SimpleAuthManagerUser, team_name: str | None = None
+    ) -> bool:
+        # Import errors for files with no registered Dag are admin-only; every other
+        # view stays readable by viewers.
+        allow_role = (
+            SimpleAuthManagerRole.ADMIN
+            if access_view == AccessView.IMPORT_ERRORS_ALL
+            else SimpleAuthManagerRole.VIEWER
+        )
+        return self._is_authorized(method="GET", allow_role=allow_role, user=user, team_name=team_name)
 
     def is_authorized_custom_view(
         self, *, method: ResourceMethod, resource_name: str, user: SimpleAuthManagerUser

airflow-core/src/airflow/api_fastapi/core_api/routes/public/import_error.py

@@ -55,6 +55,7 @@
     requires_access_view,
 )
 from airflow.models import DagModel
+from airflow.models.dagbundle import DagBundleModel
 from airflow.models.errors import ParseImportError
 
 REDACTED_STACKTRACE = "REDACTED - you do not have read permission on all Dags in the file"
@@ -99,12 +100,21 @@ def get_import_error(
         ).all()
     )
 
-    # No Dags matched for this file -- either the file genuinely contains
-    # no Dags (parse failed before any Dag was defined), or the name keys
-    # did not resolve. Return the raw error in this case; a proper
-    # admin-only path for unregistered files is tracked in a follow-up
-    # issue (see https://github.com/apache/airflow/issues/67461).
+    # No Dags matched for this file, so there is no per-Dag key to authorize on:
+    # gate on ``IMPORT_ERRORS_ALL``, scoped to the file's team via its bundle.
+    # Denying rather than failing open -- returning the row would leak the file's
+    # existence, and for team-scoped bundles the cross-team repository structure.
     if not file_dag_ids:
+        team_name = (
+            DagBundleModel.get_team_name(error.bundle_name, session=session) if error.bundle_name else None
+        )
+        if not auth_manager.authorize_view(
+            access_view=AccessView.IMPORT_ERRORS_ALL, user=user, team_name=team_name
+        ):
+            raise HTTPException(
+                status.HTTP_403_FORBIDDEN,
+                "You do not have permission to view import errors for files with no registered Dag",
+            )
         return error
 
     # Can the user read any Dags in the file?
@@ -157,7 +167,33 @@ def get_import_errors(
     readable_dag_ids = auth_manager.get_authorized_dag_ids(method="GET", user=user)
 
     # Subquery for files that have any Dags
-    files_with_any_dags = select(DagModel.relative_fileloc).distinct().subquery()
+    files_with_any_dags = select(DagModel.relative_fileloc, DagModel.bundle_name).distinct().subquery()
+
+    # Errors for files with no registered Dag are authorized on ``IMPORT_ERRORS_ALL``,
+    # scoped to the file's team via its bundle. Filtering here rather than redacting in
+    # the loop keeps unauthorized rows out of the count and pagination too, so their
+    # existence -- and for team-scoped bundles the cross-team file names -- does not leak.
+    unregistered_errors = session.execute(
+        select(ParseImportError.id, ParseImportError.bundle_name)
+        .outerjoin(
+            files_with_any_dags,
+            (ParseImportError.filename == files_with_any_dags.c.relative_fileloc)
+            & (ParseImportError.bundle_name == files_with_any_dags.c.bundle_name),
+        )
+        .where(files_with_any_dags.c.relative_fileloc.is_(None))
+    ).all()
+    team_name_by_bundle = DagBundleModel.get_team_names(
+        {bundle_name for _, bundle_name in unregistered_errors if bundle_name}, session=session
+    )
+    authorized_unregistered_ids = {
+        error_id
+        for error_id, bundle_name in unregistered_errors
+        if auth_manager.authorize_view(
+            access_view=AccessView.IMPORT_ERRORS_ALL,
+            user=user,
+            team_name=team_name_by_bundle.get(bundle_name) if bundle_name else None,
+        )
+    }
 
     # Files (identified by ``(relative_fileloc, bundle_name)``) where the
     # user can read at least one Dag. Used to decide which import errors
@@ -195,7 +231,10 @@ def get_import_errors(
         select(ParseImportError, file_dags_cte.c.dag_id)
         .outerjoin(
             files_with_any_dags,
-            ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+            and_(
+                ParseImportError.filename == files_with_any_dags.c.relative_fileloc,
+                ParseImportError.bundle_name == files_with_any_dags.c.bundle_name,
+            ),
         )
         .outerjoin(
             file_dags_cte,
@@ -206,7 +245,9 @@ def get_import_errors(
         )
         .where(
             or_(
-                files_with_any_dags.c.relative_fileloc.is_(None),
+                # Unregistered-file errors the caller is authorized to see.
+                ParseImportError.id.in_(authorized_unregistered_ids),
+                # Files where the caller can read at least one Dag.
                 file_dags_cte.c.dag_id.isnot(None),
             )
         )
@@ -259,12 +300,8 @@ def get_import_errors(
     for import_error, file_dag_ids_iter in import_errors_result:
         dag_ids = [dag_id for _, dag_id in file_dag_ids_iter if dag_id is not None]
 
-        # No Dags matched for this file -- either the file genuinely has
-        # no Dags yet (parse failed before any Dag was defined), or the
-        # name keys did not resolve. Append the raw error in this case;
-        # a proper admin-only path for unregistered files is tracked in
-        # a follow-up issue
-        # (see https://github.com/apache/airflow/issues/67461).
+        # Unauthorized unregistered-file errors were already filtered out above,
+        # so anything reaching here may show its raw stacktrace.
         if not dag_ids:
             import_errors.append(import_error)
             continue

airflow-core/tests/unit/api_fastapi/auth/managers/test_base_auth_manager.py

@@ -26,6 +26,7 @@
 from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager, T
 from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
 from airflow.api_fastapi.auth.managers.models.resource_details import (
+    AccessView,
     ConnectionDetails,
     DagDetails,
     PoolDetails,
@@ -34,14 +35,14 @@
 )
 from airflow.api_fastapi.auth.tokens import JWTGenerator, JWTValidator
 from airflow.api_fastapi.common.types import MenuItem
+from airflow.exceptions import RemovedInAirflow4Warning
 from airflow.models.team import Team
 
 from tests_common.test_utils.config import conf_vars
 
 if TYPE_CHECKING:
     from airflow.api_fastapi.auth.managers.base_auth_manager import ResourceMethod
     from airflow.api_fastapi.auth.managers.models.resource_details import (
-        AccessView,
         AssetAliasDetails,
         AssetDetails,
         ConfigurationDetails,
@@ -132,7 +133,11 @@ def is_authorized_variable(
         raise NotImplementedError()
 
     def is_authorized_view(
-        self, *, access_view: AccessView, user: BaseAuthManagerUserTest | None = None
+        self,
+        *,
+        access_view: AccessView,
+        user: BaseAuthManagerUserTest | None = None,
+        team_name: str | None = None,
     ) -> bool:
         raise NotImplementedError()
 
@@ -157,6 +162,47 @@ class TestBaseAuthManager:
     def test_init_non_multi_team_mode(self, auth_manager):
         assert auth_manager.init() is None
 
+    @patch.object(EmptyAuthManager, "is_authorized_view")
+    def test_authorize_view_passes_team_name_to_team_aware_manager(
+        self, mock_is_authorized_view, auth_manager
+    ):
+        mock_is_authorized_view.return_value = True
+
+        result = auth_manager.authorize_view(access_view=AccessView.DOCS, user=None, team_name="team_a")
+
+        assert result is True
+        mock_is_authorized_view.assert_called_once_with(
+            access_view=AccessView.DOCS, user=None, team_name="team_a"
+        )
+
+    def test_authorize_view_drops_team_name_and_warns_for_legacy_manager(self):
+        # A manager whose is_authorized_view predates team_name (out-of-tree, or a provider
+        # released before the argument existed) must keep working: authorize_view falls back
+        # to a global check and warns that some views are not team-restricted.
+        class LegacyAuthManager(EmptyAuthManager):
+            def is_authorized_view(self, *, access_view, user=None):  # old signature, no team_name
+                return True
+
+        manager = LegacyAuthManager()
+
+        with pytest.warns(RemovedInAirflow4Warning, match="not team-aware"):
+            result = manager.authorize_view(access_view=AccessView.DOCS, user=None, team_name="team_a")
+
+        assert result is True
+
+    def test_authorize_view_treats_kwargs_override_as_team_aware(self):
+        class KwargsAuthManager(EmptyAuthManager):
+            def is_authorized_view(self, *, access_view, user=None, **kwargs):
+                return kwargs.get("team_name") == "team_a"
+
+        manager = KwargsAuthManager()
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")  # a warning here would fail the test
+            result = manager.authorize_view(access_view=AccessView.DOCS, user=None, team_name="team_a")
+
+        assert result is True
+
     @conf_vars({("core", "multi_team"): "True"})
     @pytest.mark.parametrize(
         ("auth_manager_teams", "db_teams", "expected"),