Introduce filter_authorized_menu_items to filter menu items based on permissions

airflow/api_fastapi/auth/managers/base_auth_manager.py

@@ -26,7 +26,7 @@
 
 from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
 from airflow.api_fastapi.auth.managers.models.resource_details import BackfillDetails, DagDetails
-from airflow.api_fastapi.common.types import MenuItem
+from airflow.api_fastapi.common.types import ExtraMenuItem, MenuItem
 from airflow.configuration import conf
 from airflow.models import DagModel
 from airflow.typing_compat import Literal
@@ -281,6 +281,15 @@ def is_authorized_custom_view(self, *, method: ResourceMethod | str, resource_na
         :param user: the user to performing the action
         """
 
+    @abstractmethod
+    def filter_authorized_menu_items(self, menu_items: list[MenuItem], *, user: T) -> list[MenuItem]:
+        """
+        Filter menu items based on user permissions.
+
+        :param menu_items: list of all menu items
+        :param user: the user
+        """
+
     def batch_is_authorized_connection(
         self,
         requests: Sequence[IsAuthorizedConnectionRequest],
@@ -429,7 +438,11 @@ def get_fastapi_app(self) -> FastAPI | None:
         """
         return None
 
-    def get_menu_items(self, *, user: T) -> list[MenuItem]:
+    def get_authorized_menu_items(self, *, user: T) -> list[MenuItem]:
+        """Get all menu items the user has access to."""
+        return self.filter_authorized_menu_items(list(MenuItem), user=user)
+
+    def get_extra_menu_items(self, *, user: T) -> list[ExtraMenuItem]:
         """
         Provide additional links to be added to the menu.
 

airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py

@@ -36,6 +36,7 @@
 from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
 from airflow.api_fastapi.auth.managers.models.resource_details import BackfillDetails
 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
 
 if TYPE_CHECKING:
@@ -249,6 +250,11 @@ def is_authorized_custom_view(
     ):
         return self._is_authorized(method="GET", allow_role=SimpleAuthManagerRole.VIEWER, user=user)
 
+    def filter_authorized_menu_items(
+        self, menu_items: list[MenuItem], *, user: SimpleAuthManagerUser
+    ) -> list[MenuItem]:
+        return menu_items
+
     def get_fastapi_app(self) -> FastAPI | None:
         """
         Specify a sub FastAPI application specific to the auth manager.

airflow/api_fastapi/common/types.py

@@ -76,8 +76,23 @@ class Mimetype(str, Enum):
 
 
 @dataclass
-class MenuItem:
-    """Define a menu item."""
+class ExtraMenuItem:
+    """Define a menu item that can be added to the menu by auth managers or plugins."""
 
     text: str
     href: str
+
+
+class MenuItem(Enum):
+    """Define all menu items defined in the menu."""
+
+    ASSETS = "Assets"
+    ASSET_EVENTS = "Asset Events"
+    CONNECTIONS = "Connections"
+    DAGS = "Dags"
+    DOCS = "Docs"
+    PLUGINS = "Plugins"
+    POOLS = "Pools"
+    PROVIDERS = "Providers"
+    VARIABLES = "Variables"
+    XCOMS = "XComs"

airflow/api_fastapi/core_api/routes/ui/auth.py

@@ -31,7 +31,7 @@
 def get_auth_links(
     user: GetUserDep,
 ) -> MenuItemCollectionResponse:
-    menu_items = get_auth_manager().get_menu_items(user=user)
+    menu_items = get_auth_manager().get_extra_menu_items(user=user)
 
     return MenuItemCollectionResponse(
         menu_items=cast(list[MenuItem], menu_items),

newsfragments/aip-79.significant.rst

@@ -10,6 +10,8 @@ As part of this change the following breaking changes have occurred:
 
   - A new abstract method ``serialize_user`` needs to be implemented
 
+  - A new abstract method ``filter_authorized_menu_items`` needs to be implemented
+
   - The property ``security_manager`` has been removed from the interface
 
   - The method ``get_url_logout`` is now optional

providers/amazon/src/airflow/providers/amazon/aws/auth_manager/aws_auth_manager.py

@@ -63,6 +63,7 @@
         AssetDetails,
         ConfigurationDetails,
     )
+    from airflow.api_fastapi.common.types import MenuItem
 
 
 class AwsAuthManager(BaseAuthManager[AwsAuthManagerUser]):
@@ -226,6 +227,25 @@ def is_authorized_custom_view(
             entity_id=resource_name,
         )
 
+    def filter_authorized_menu_items(
+        self, menu_items: list[MenuItem], *, user: AwsAuthManagerUser
+    ) -> list[MenuItem]:
+        requests: dict[str, IsAuthorizedRequest] = {}
+        for menu_item in menu_items:
+            requests[menu_item.value] = self._get_menu_item_request(menu_item.value)
+
+        batch_is_authorized_results = self.avp_facade.get_batch_is_authorized_results(
+            requests=list(requests.values()), user=user
+        )
+
+        def _has_access_to_menu_item(request: IsAuthorizedRequest):
+            result = self.avp_facade.get_batch_is_authorized_single_result(
+                batch_is_authorized_results=batch_is_authorized_results, request=request, user=user
+            )
+            return result["decision"] == "ALLOW"
+
+        return [menu_item for menu_item in menu_items if _has_access_to_menu_item(requests[menu_item.value])]
+
     def batch_is_authorized_connection(
         self,
         requests: Sequence[IsAuthorizedConnectionRequest],
@@ -360,6 +380,14 @@ def get_fastapi_app(self) -> FastAPI | None:
 
         return app
 
+    @staticmethod
+    def _get_menu_item_request(menu_item_text: str) -> IsAuthorizedRequest:
+        return {
+            "method": "MENU",
+            "entity_type": AvpEntities.MENU,
+            "entity_id": menu_item_text,
+        }
+
     def _check_avp_schema_version(self):
         if not self.avp_facade.is_policy_store_schema_up_to_date():
             self.log.warning(

providers/amazon/tests/unit/amazon/aws/auth_manager/test_aws_auth_manager.py

@@ -37,6 +37,7 @@
     PoolDetails,
     VariableDetails,
 )
+from airflow.api_fastapi.common.types import MenuItem
 from airflow.providers.amazon.aws.auth_manager.avp.entities import AvpEntities
 from airflow.providers.amazon.aws.auth_manager.aws_auth_manager import AwsAuthManager
 from airflow.providers.amazon.aws.auth_manager.user import AwsAuthManagerUser
@@ -399,6 +400,77 @@ def test_batch_is_authorized_connection(
         )
         assert result
 
+    def test_filter_authorized_menu_items(self, auth_manager):
+        batch_is_authorized_output = [
+            {
+                "request": {
+                    "principal": {"entityType": "Airflow::User", "entityId": "test_user_id"},
+                    "action": {"actionType": "Airflow::Action", "actionId": "Menu.MENU"},
+                    "resource": {"entityType": "Airflow::Menu", "entityId": MenuItem.CONNECTIONS.value},
+                },
+                "decision": "DENY",
+            },
+            {
+                "request": {
+                    "principal": {"entityType": "Airflow::User", "entityId": "test_user_id"},
+                    "action": {"actionType": "Airflow::Action", "actionId": "Menu.MENU"},
+                    "resource": {"entityType": "Airflow::Menu", "entityId": MenuItem.VARIABLES.value},
+                },
+                "decision": "ALLOW",
+            },
+            {
+                "request": {
+                    "principal": {"entityType": "Airflow::User", "entityId": "test_user_id"},
+                    "action": {"actionType": "Airflow::Action", "actionId": "Menu.MENU"},
+                    "resource": {"entityType": "Airflow::Menu", "entityId": MenuItem.ASSETS.value},
+                },
+                "decision": "DENY",
+            },
+            {
+                "request": {
+                    "principal": {"entityType": "Airflow::User", "entityId": "test_user_id"},
+                    "action": {"actionType": "Airflow::Action", "actionId": "Menu.MENU"},
+                    "resource": {"entityType": "Airflow::Menu", "entityId": MenuItem.DAGS.value},
+                },
+                "decision": "ALLOW",
+            },
+        ]
+        auth_manager.avp_facade.get_batch_is_authorized_results = Mock(
+            return_value=batch_is_authorized_output
+        )
+
+        result = auth_manager.filter_authorized_menu_items(
+            [MenuItem.CONNECTIONS, MenuItem.VARIABLES, MenuItem.ASSETS, MenuItem.DAGS],
+            user=AwsAuthManagerUser(user_id="test_user_id", groups=[]),
+        )
+
+        auth_manager.avp_facade.get_batch_is_authorized_results.assert_called_once_with(
+            requests=[
+                {
+                    "method": "MENU",
+                    "entity_type": AvpEntities.MENU,
+                    "entity_id": MenuItem.CONNECTIONS.value,
+                },
+                {
+                    "method": "MENU",
+                    "entity_type": AvpEntities.MENU,
+                    "entity_id": MenuItem.VARIABLES.value,
+                },
+                {
+                    "method": "MENU",
+                    "entity_type": AvpEntities.MENU,
+                    "entity_id": MenuItem.ASSETS.value,
+                },
+                {
+                    "method": "MENU",
+                    "entity_type": AvpEntities.MENU,
+                    "entity_id": MenuItem.DAGS.value,
+                },
+            ],
+            user=ANY,
+        )
+        assert result == [MenuItem.VARIABLES, MenuItem.DAGS]
+
     @patch.object(AwsAuthManager, "avp_facade")
     def test_batch_is_authorized_dag(
         self,

providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py

@@ -44,7 +44,7 @@
     PoolDetails,
     VariableDetails,
 )
-from airflow.api_fastapi.common.types import MenuItem
+from airflow.api_fastapi.common.types import ExtraMenuItem, MenuItem
 from airflow.cli.cli_config import (
     DefaultHelpParser,
     GroupCommand,
@@ -143,6 +143,19 @@
     AccessView.WEBSITE: RESOURCE_WEBSITE,
 }
 
+_MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE = {
+    MenuItem.ASSETS: RESOURCE_ASSET,
+    MenuItem.ASSET_EVENTS: RESOURCE_ASSET,
+    MenuItem.CONNECTIONS: RESOURCE_CONNECTION,
+    MenuItem.DAGS: RESOURCE_DAG,
+    MenuItem.DOCS: RESOURCE_DOCS,
+    MenuItem.PLUGINS: RESOURCE_PLUGIN,
+    MenuItem.POOLS: RESOURCE_POOL,
+    MenuItem.PROVIDERS: RESOURCE_PROVIDER,
+    MenuItem.VARIABLES: RESOURCE_VARIABLE,
+    MenuItem.XCOMS: RESOURCE_XCOM,
+}
+
 
 class FabAuthManager(BaseAuthManager[User]):
     """
@@ -360,6 +373,17 @@ def is_authorized_custom_view(
         fab_action_name = get_fab_action_from_method_map().get(method, method)
         return (fab_action_name, resource_name) in self._get_user_permissions(user)
 
+    def filter_authorized_menu_items(self, menu_items: list[MenuItem], user: User) -> list[MenuItem]:
+        return [
+            menu_item
+            for menu_item in menu_items
+            if self._is_authorized(
+                method="MENU",
+                resource_type=_MAP_MENU_ITEM_TO_FAB_RESOURCE_TYPE.get(menu_item, menu_item.value),
+                user=user,
+            )
+        ]
+
     @provide_session
     def get_permitted_dag_ids(
         self,
@@ -433,7 +457,7 @@ def get_url_logout(self) -> str | None:
     def register_views(self) -> None:
         self.security_manager.register_views()
 
-    def get_menu_items(self, *, user: User) -> list[MenuItem]:
+    def get_extra_menu_items(self, *, user: User) -> list[ExtraMenuItem]:
         # Contains the list of menu items. ``resource_type`` is the name of the resource in FAB
         # permission model to check whether the user is allowed to see this menu item
         items = [
@@ -465,7 +489,7 @@ def get_menu_items(self, *, user: User) -> list[MenuItem]:
         ]
 
         return [
-            MenuItem(text=item["text"], href=item["href"])
+            ExtraMenuItem(text=item["text"], href=item["href"])
             for item in items
             if self._is_authorized(method="MENU", resource_type=item["resource_type"], user=user)
         ]

providers/fab/tests/unit/fab/auth_manager/test_fab_auth_manager.py

@@ -26,6 +26,7 @@
 from flask import Flask, g
 
 from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
+from airflow.api_fastapi.common.types import MenuItem
 from airflow.exceptions import AirflowConfigException
 from airflow.providers.fab.www.extensions.init_appbuilder import init_appbuilder
 from airflow.providers.standard.operators.empty import EmptyOperator
@@ -463,6 +464,43 @@ def test_is_authorized_custom_view(
         result = auth_manager.is_authorized_custom_view(method=method, resource_name=resource_name, user=user)
         assert result == expected_result
 
+    @pytest.mark.parametrize(
+        "menu_items, user_permissions, expected_result",
+        [
+            (
+                [MenuItem.ASSETS, MenuItem.DAGS],
+                [(ACTION_CAN_ACCESS_MENU, RESOURCE_ASSET), (ACTION_CAN_ACCESS_MENU, RESOURCE_DAG)],
+                [MenuItem.ASSETS, MenuItem.DAGS],
+            ),
+            (
+                [MenuItem.ASSETS, MenuItem.DAGS],
+                [(ACTION_CAN_READ, RESOURCE_ASSET), (ACTION_CAN_READ, RESOURCE_DAG)],
+                [],
+            ),
+            (
+                [MenuItem.ASSET_EVENTS, MenuItem.VARIABLES],
+                [(ACTION_CAN_ACCESS_MENU, RESOURCE_ASSET), (ACTION_CAN_READ, RESOURCE_VARIABLE)],
+                [MenuItem.ASSET_EVENTS],
+            ),
+            (
+                [],
+                [],
+                [],
+            ),
+        ],
+    )
+    def test_filter_authorized_menu_items(
+        self,
+        menu_items: list[MenuItem],
+        user_permissions,
+        expected_result,
+        auth_manager,
+    ):
+        user = Mock()
+        user.perms = user_permissions
+        result = auth_manager.filter_authorized_menu_items(menu_items, user=user)
+        assert result == expected_result
+
     @pytest.mark.parametrize(
         "method, user_permissions, expected_results",
         [
@@ -576,8 +614,8 @@ def test_get_url_logout(self, auth_manager):
         assert result == f"http://localhost:8080{AUTH_MANAGER_FASTAPI_APP_PREFIX}/logout/"
 
     @mock.patch.object(FabAuthManager, "_is_authorized", return_value=True)
-    def test_get_menu_items(self, _, auth_manager_with_appbuilder, flask_app):
+    def test_get_extra_menu_items(self, _, auth_manager_with_appbuilder, flask_app):
         auth_manager_with_appbuilder.register_views()
-        result = auth_manager_with_appbuilder.get_menu_items(user=Mock())
+        result = auth_manager_with_appbuilder.get_extra_menu_items(user=Mock())
         assert len(result) == 5
         assert all(item.href.startswith(AUTH_MANAGER_FASTAPI_APP_PREFIX) for item in result)

tests/api_fastapi/auth/managers/simple/test_simple_auth_manager.py

@@ -24,6 +24,7 @@
 from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
 from airflow.api_fastapi.auth.managers.models.resource_details import AccessView
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
+from airflow.api_fastapi.common.types import MenuItem
 
 from tests_common.test_utils.config import conf_vars
 
@@ -214,3 +215,10 @@ def test_is_authorized_methods_viewer_role_required_for_get(
             getattr(auth_manager, api)(method=method, user=SimpleAuthManagerUser(username="test", role=role))
             is result
         )
+
+    def test_filter_authorized_menu_items(self, auth_manager):
+        items = [MenuItem.ASSETS]
+        results = auth_manager.filter_authorized_menu_items(
+            items, user=SimpleAuthManagerUser(username="test", role=None)
+        )
+        assert results == items