Fix permission check on menus (#35781)

airflow/www/security_manager.py

@@ -137,15 +137,7 @@ def has_access(
             user = g.user
 
         is_authorized_method = self._get_auth_manager_is_authorized_method(resource_name)
-        if is_authorized_method:
-            return is_authorized_method(action_name, resource_pk, user)
-        else:
-            # This means the page the user is trying to access is specific to the auth manager used
-            # Example: the user list view in FabAuthManager
-            action_name = ACTION_CAN_READ if action_name == ACTION_CAN_ACCESS_MENU else action_name
-            return get_auth_manager().is_authorized_custom_view(
-                fab_action_name=action_name, fab_resource_name=resource_name, user=user
-            )
+        return is_authorized_method(action_name, resource_pk, user)
 
     def create_admin_standalone(self) -> tuple[str | None, str | None]:
         """Perform the required steps when initializing airflow for standalone mode.
@@ -331,7 +323,7 @@ def get_variable_key(resource_pk):
             ),
         }
 
-    def _get_auth_manager_is_authorized_method(self, fab_resource_name: str) -> Callable | None:
+    def _get_auth_manager_is_authorized_method(self, fab_resource_name: str) -> Callable:
         is_authorized_method = self._auth_manager_is_authorized_map.get(fab_resource_name)
         if is_authorized_method:
             return is_authorized_method
@@ -340,12 +332,19 @@ def _get_auth_manager_is_authorized_method(self, fab_resource_name: str) -> Call
             # least one dropdown child
             return self._is_authorized_category_menu(fab_resource_name)
         else:
-            return None
+            # This means the page the user is trying to access is specific to the auth manager used
+            # Example: the user list view in FabAuthManager
+            return lambda action, resource_pk, user: get_auth_manager().is_authorized_custom_view(
+                fab_action_name=ACTION_CAN_READ if action == ACTION_CAN_ACCESS_MENU else action,
+                fab_resource_name=fab_resource_name,
+                user=user,
+            )
 
     def _is_authorized_category_menu(self, category: str) -> Callable:
         items = {item.name for item in self.appbuilder.menu.find(category).childs}
         return lambda action, resource_pk, user: any(
-            self._get_auth_manager_is_authorized_method(fab_resource_name=item) for item in items
+            self._get_auth_manager_is_authorized_method(fab_resource_name=item)(action, resource_pk, user)
+            for item in items
         )
 
     """

tests/www/test_security_manager.py

@@ -0,0 +1,115 @@
+#
+# 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 import mock
+from unittest.mock import Mock
+
+import pytest
+
+from airflow.security.permissions import (
+    ACTION_CAN_READ,
+    RESOURCE_ADMIN_MENU,
+    RESOURCE_BROWSE_MENU,
+    RESOURCE_DOCS_MENU,
+    RESOURCE_VARIABLE,
+)
+from airflow.www import app as application
+
+
+@pytest.fixture()
+def app():
+    return application.create_app(testing=True)
+
+
+@pytest.fixture()
+def app_builder(app):
+    return app.appbuilder
+
+
+@pytest.fixture()
+def security_manager(app_builder):
+    return app_builder.sm
+
+
+@pytest.mark.db_test
+class TestAirflowSecurityManagerV2:
+    @pytest.mark.parametrize(
+        "resource_name, auth_manager_methods, expected",
+        [
+            (RESOURCE_VARIABLE, {"is_authorized_variable": True}, True),
+            (RESOURCE_VARIABLE, {"is_authorized_variable": False}, False),
+            (RESOURCE_DOCS_MENU, {"is_authorized_view": True}, True),
+            (RESOURCE_DOCS_MENU, {"is_authorized_view": False}, False),
+            (
+                RESOURCE_ADMIN_MENU,
+                {
+                    "is_authorized_view": False,
+                    "is_authorized_variable": False,
+                    "is_authorized_connection": True,
+                    "is_authorized_dag": False,
+                    "is_authorized_configuration": False,
+                    "is_authorized_pool": False,
+                },
+                True,
+            ),
+            (
+                RESOURCE_ADMIN_MENU,
+                {
+                    "is_authorized_view": False,
+                    "is_authorized_variable": False,
+                    "is_authorized_connection": False,
+                    "is_authorized_dag": False,
+                    "is_authorized_configuration": False,
+                    "is_authorized_pool": False,
+                },
+                False,
+            ),
+            (
+                RESOURCE_BROWSE_MENU,
+                {
+                    "is_authorized_dag": False,
+                    "is_authorized_view": False,
+                },
+                False,
+            ),
+            (
+                RESOURCE_BROWSE_MENU,
+                {
+                    "is_authorized_dag": False,
+                    "is_authorized_view": True,
+                },
+                True,
+            ),
+        ],
+    )
+    @mock.patch("airflow.www.security_manager.get_auth_manager")
+    def test_has_access(
+        self, mock_get_auth_manager, security_manager, resource_name, auth_manager_methods, expected
+    ):
+        user = Mock()
+        auth_manager = Mock()
+        for method_name, method_return in auth_manager_methods.items():
+            method = Mock(return_value=method_return)
+            getattr(auth_manager, method_name).side_effect = method
+            mock_get_auth_manager.return_value = auth_manager
+        result = security_manager.has_access(ACTION_CAN_READ, resource_name, user=user)
+        assert result == expected
+        if len(auth_manager_methods) > 1 and not expected:
+            for method_name in auth_manager_methods:
+                getattr(auth_manager, method_name).assert_called()