Add back missing permissions to UserModelView controls.

airflow/www/views.py

@@ -57,6 +57,7 @@
 from flask_appbuilder.actions import action
 from flask_appbuilder.fieldwidgets import Select2Widget
 from flask_appbuilder.models.sqla.filters import BaseFilter
+from flask_appbuilder.security.decorators import has_access
 from flask_appbuilder.security.views import (
     PermissionModelView,
     PermissionViewModelView,
@@ -4245,7 +4246,72 @@ class CustomViewMenuModelView(ViewMenuModelView):
     ]
 
 
-class CustomUserDBModelView(UserDBModelView):
+class CustomUserInfoEditView(UserInfoEditView):
+    """Customize permission names for FAB's builtin UserInfoEditView."""
+
+    class_permission_name = permissions.RESOURCE_MY_PROFILE
+    route_base = "/userinfoeditview"
+    method_permission_name = {
+        'this_form_get': 'edit',
+        'this_form_post': 'edit',
+    }
+    base_permissions = [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
+
+
+class CustomUserStatsChartView(UserStatsChartView):
+    """Customize permission names for FAB's builtin UserStatsChartView."""
+
+    class_permission_name = permissions.RESOURCE_USER_STATS_CHART
+    route_base = "/userstatschartview"
+    method_permission_name = {
+        'chart': 'read',
+        'list': 'read',
+    }
+    base_permissions = [permissions.ACTION_CAN_READ]
+
+
+class MultiResourceUserMixin:
+    """Remaps UserModelView permissions to new resources and actions."""
+
+    _class_permission_name = permissions.RESOURCE_USER
+
+    class_permission_name_mapping = {
+        'userinfoedit': permissions.RESOURCE_MY_PROFILE,
+        'userinfo': permissions.RESOURCE_MY_PROFILE,
+    }
+
+    method_permission_name = {
+        'userinfo': 'read',
+        'download': 'read',
+        'show': 'read',
+        'list': 'read',
+        'edit': 'edit',
+        'userinfoedit': 'edit',
+        'delete': 'delete',
+    }
+
+    base_permissions = [
+        permissions.ACTION_CAN_READ,
+        permissions.ACTION_CAN_EDIT,
+        permissions.ACTION_CAN_DELETE,
+    ]
+
+    @expose("/show/<pk>", methods=["GET"])
+    @has_access
+    def show(self, pk):
+        pk = self._deserialize_pk_if_composite(pk)
+        widgets = self._show(pk)
+        widgets['show'].template_args['actions'].pop('userinfoedit')
+        return self.render_template(
+            self.show_template,
+            pk=pk,
+            title=self.show_title,
+            widgets=widgets,
+            related_views=self._related_views,
+        )
+
+
+class CustomUserDBModelView(MultiResourceUserMixin, UserDBModelView):
     """Customize permission names for FAB's builtin UserDBModelView."""
 
     _class_permission_name = permissions.RESOURCE_USER
@@ -4259,15 +4325,15 @@ class CustomUserDBModelView(UserDBModelView):
 
     method_permission_name = {
         'add': 'create',
-        'userinfo': 'read',
         'download': 'read',
         'show': 'read',
         'list': 'read',
         'edit': 'edit',
+        'delete': 'delete',
         'resetmypassword': 'read',
         'resetpasswords': 'read',
-        'userinfoedit': 'edit',
-        'delete': 'delete',
+        'userinfo': 'read',
+        'userinfoedit': 'read',
     }
 
     base_permissions = [
@@ -4287,85 +4353,32 @@ def class_permission_name(self):
                 return self.class_permission_name_mapping.get(action_name, self._class_permission_name)
             if method_name:
                 return self.class_permission_name_mapping.get(method_name, self._class_permission_name)
-
         return self._class_permission_name
 
     @class_permission_name.setter
     def class_permission_name(self, name):
         self._class_permission_name = name
 
 
-class CustomUserInfoEditView(UserInfoEditView):
-    """Customize permission names for FAB's builtin UserInfoEditView."""
-
-    class_permission_name = permissions.RESOURCE_MY_PROFILE
-    route_base = "/userinfoeditview"
-    method_permission_name = {
-        'this_form_get': 'read',
-        'this_form_post': 'edit',
-    }
-    base_permissions = [permissions.ACTION_CAN_EDIT, permissions.ACTION_CAN_READ]
-
-
-class CustomUserStatsChartView(UserStatsChartView):
-    """Customize permission names for FAB's builtin UserStatsChartView."""
-
-    class_permission_name = permissions.RESOURCE_USER_STATS_CHART
-    route_base = "/userstatschartview"
-    method_permission_name = {
-        'chart': 'read',
-        'list': 'read',
-    }
-    base_permissions = [permissions.ACTION_CAN_READ]
-
-
-class CustomUserLDAPModelView(UserLDAPModelView):
+class CustomUserLDAPModelView(MultiResourceUserMixin, UserLDAPModelView):
     """Customize permission names for FAB's builtin UserLDAPModelView."""
 
-    class_permission_name = permissions.RESOURCE_MY_PROFILE
-    method_permission_name = {
-        'userinfo': 'read',
-        'list': 'read',
-    }
-    base_permissions = [
-        permissions.ACTION_CAN_READ,
-    ]
+    pass
 
 
-class CustomUserOAuthModelView(UserOAuthModelView):
+class CustomUserOAuthModelView(MultiResourceUserMixin, UserOAuthModelView):
     """Customize permission names for FAB's builtin UserOAuthModelView."""
 
-    class_permission_name = permissions.RESOURCE_MY_PROFILE
-    method_permission_name = {
-        'userinfo': 'read',
-        'list': 'read',
-    }
-    base_permissions = [
-        permissions.ACTION_CAN_READ,
-    ]
+    pass
 
 
-class CustomUserOIDModelView(UserOIDModelView):
+class CustomUserOIDModelView(MultiResourceUserMixin, UserOIDModelView):
     """Customize permission names for FAB's builtin UserOIDModelView."""
 
-    class_permission_name = permissions.RESOURCE_MY_PROFILE
-    method_permission_name = {
-        'userinfo': 'read',
-        'list': 'read',
-    }
-    base_permissions = [
-        permissions.ACTION_CAN_READ,
-    ]
+    pass
 
 
-class CustomUserRemoteUserModelView(UserRemoteUserModelView):
+class CustomUserRemoteUserModelView(MultiResourceUserMixin, UserRemoteUserModelView):
     """Customize permission names for FAB's builtin UserRemoteUserModelView."""
 
-    class_permission_name = permissions.RESOURCE_MY_PROFILE
-    method_permission_name = {
-        'userinfo': 'read',
-        'list': 'read',
-    }
-    base_permissions = [
-        permissions.ACTION_CAN_READ,
-    ]
+    pass

tests/www/views/test_views_custom_user_views.py

@@ -0,0 +1,199 @@
+#
+# 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.
+
+import unittest
+
+from flask_appbuilder import SQLA
+from parameterized import parameterized
+
+from airflow import settings
+from airflow.security import permissions
+from airflow.www import app as application
+from airflow.www.views import CustomUserDBModelView
+from tests.test_utils.api_connexion_utils import create_user, delete_role
+from tests.test_utils.www import check_content_in_response, check_content_not_in_response, client_with_login
+
+
+class TestSecurity(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        settings.configure_orm()
+        cls.session = settings.Session
+        cls.app = application.create_app(testing=True)
+        cls.appbuilder = cls.app.appbuilder
+        cls.app.config['WTF_CSRF_ENABLED'] = False
+        cls.security_manager = cls.appbuilder.sm
+
+        cls.delete_roles()
+
+    def setUp(self):
+        self.db = SQLA(self.app)
+        self.appbuilder.add_view(CustomUserDBModelView, "CustomUserDBModelView", category="ModelViews")
+        self.client = self.app.test_client()  # type:ignore
+
+    @classmethod
+    def delete_roles(cls):
+        for role_name in ['role_edit_one_dag']:
+            delete_role(cls.app, role_name)
+
+    @parameterized.expand(
+        [
+            (
+                "/resetpassword/form?pk={user.id}",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_PASSWORD),
+                'Reset Password Form',
+            ),
+            (
+                "/resetmypassword/form",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PASSWORD),
+                'Reset Password Form',
+            ),
+            (
+                "/users/userinfo/",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PROFILE),
+                'Your user information',
+            ),
+            (
+                "/userinfoeditview/form",
+                (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_MY_PROFILE),
+                'Edit User',
+            ),
+            ("/users/add", (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_USER), 'Add User'),
+            ("/users/list/", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'List Users'),
+            ("/users/show/{user.id}", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'Show User'),
+            ("/users/edit/{user.id}", (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_USER), 'Edit User'),
+        ]
+    )
+    def test_user_model_view_with_access(self, url, permission, expected_text):
+        user_without_access = create_user(
+            self.app,
+            username="no_access",
+            role_name="role_no_access",
+            permissions=[
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+            ],
+        )
+        client = client_with_login(
+            self.app,
+            username="no_access",
+            password="no_access",
+        )
+        response = client.get(url.replace("{user.id}", str(user_without_access.id)), follow_redirects=True)
+        check_content_not_in_response(expected_text, response)
+
+    @parameterized.expand(
+        [
+            (
+                "/resetpassword/form?pk={user.id}",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_PASSWORD),
+                'Reset Password Form',
+            ),
+            (
+                "/resetmypassword/form",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PASSWORD),
+                'Reset Password Form',
+            ),
+            (
+                "/users/userinfo/",
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PROFILE),
+                'Your user information',
+            ),
+            (
+                "/userinfoeditview/form",
+                (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_MY_PROFILE),
+                'Edit User',
+            ),
+            ("/users/add", (permissions.ACTION_CAN_CREATE, permissions.RESOURCE_USER), 'Add User'),
+            ("/users/list/", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'List Users'),
+            ("/users/show/{user.id}", (permissions.ACTION_CAN_READ, permissions.RESOURCE_USER), 'Show User'),
+            ("/users/edit/{user.id}", (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_USER), 'Edit User'),
+        ]
+    )
+    def test_user_model_view_without_access(self, url, permission, expected_text):
+
+        user_with_access = create_user(
+            self.app,
+            username="has_access",
+            role_name="role_has_access",
+            permissions=[(permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE), permission],
+        )
+
+        client = client_with_login(
+            self.app,
+            username="has_access",
+            password="has_access",
+        )
+        response = client.get(url.replace("{user.id}", str(user_with_access.id)), follow_redirects=True)
+        check_content_in_response(expected_text, response)
+
+    def test_user_model_view_without_delete_access(self):
+
+        user_to_delete = create_user(
+            self.app,
+            username="user_to_delete",
+            role_name="user_to_delete",
+        )
+
+        create_user(
+            self.app,
+            username="no_access",
+            role_name="role_no_access",
+            permissions=[
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+            ],
+        )
+
+        client = client_with_login(
+            self.app,
+            username="no_access",
+            password="no_access",
+        )
+
+        response = client.post(f"/users/delete/{user_to_delete.id}", follow_redirects=True)
+
+        check_content_not_in_response("Deleted Row", response)
+        assert bool(self.security_manager.get_user_by_id(user_to_delete.id)) is True
+
+    def test_user_model_view_with_delete_access(self):
+
+        user_to_delete = create_user(
+            self.app,
+            username="user_to_delete",
+            role_name="user_to_delete",
+        )
+
+        create_user(
+            self.app,
+            username="has_access",
+            role_name="role_has_access",
+            permissions=[
+                (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+                (permissions.ACTION_CAN_DELETE, permissions.RESOURCE_USER),
+            ],
+        )
+
+        client = client_with_login(
+            self.app,
+            username="has_access",
+            password="has_access",
+        )
+
+        response = client.post(f"/users/delete/{user_to_delete.id}", follow_redirects=True)
+        check_content_in_response("Deleted Row", response)
+        check_content_not_in_response(user_to_delete.username, response)
+        assert bool(self.security_manager.get_user_by_id(user_to_delete.id)) is False