[query] deprecate can_only_access_owned_queries

UPDATING.md

@@ -22,6 +22,11 @@ This file documents any backwards-incompatible changes in Superset and
 assists people when migrating to a new version.
 
 ## Next
+* [9046](https://github.com/apache/incubator-superset/pull/9046): Replaces `can_only_access_owned_queries` by
+`all_query_access` favoring a white list approach. Since a new permission is introduced use `superset init` 
+to create and associate it by default to the `Admin` role. Note that, by default, all non `Admin` users will
+not be able to access queries they do not own.
+
 * [8901](https://github.com/apache/incubator-superset/pull/8901): The datasource's update
 timestamp has been added to the query object's cache key to ensure updates to
 datasources are always reflected in associated query results. As a consequence all

superset/security/manager.py

@@ -117,6 +117,7 @@ class SupersetSecurityManager(SecurityManager):
         "can_override_role_permissions",
         "can_approve",
         "can_update_role",
+        "all_query_access",
     }
 
     READ_ONLY_PERMISSION = {"can_show", "can_list"}
@@ -132,7 +133,6 @@ class SupersetSecurityManager(SecurityManager):
         "schema_access",
         "datasource_access",
         "metric_access",
-        "can_only_access_owned_queries",
     }
 
     ACCESSIBLE_PERMS = {"can_userinfo"}
@@ -177,15 +177,13 @@ def can_access(self, permission_name: str, view_name: str) -> bool:
             return self.is_item_public(permission_name, view_name)
         return self._has_view_access(user, permission_name, view_name)
 
-    def can_only_access_owned_queries(self) -> bool:
+    def can_access_all_queries(self) -> bool:
         """
-        Return True if the user can only access owned queries, False otherwise.
+        Return True if the user can access all queries, False otherwise.
 
-        :returns: Whether the use can only access owned queries
+        :returns: Whether the user can access all queries
         """
-        return self.can_access(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
-        )
+        return self.can_access("all_query_access", "all_query_access")
 
     def all_datasource_access(self) -> bool:
         """
@@ -538,12 +536,9 @@ def create_custom_permissions(self) -> None:
         """
         Create custom FAB permissions.
         """
-
         self.add_permission_view_menu("all_datasource_access", "all_datasource_access")
         self.add_permission_view_menu("all_database_access", "all_database_access")
-        self.add_permission_view_menu(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
-        )
+        self.add_permission_view_menu("all_query_access", "all_query_access")
 
     def create_missing_perms(self) -> None:
         """

superset/views/core.py

@@ -2597,10 +2597,15 @@ def search_queries(self) -> Response:
         :returns: Response with list of sql query dicts
         """
         query = db.session.query(Query)
-        if security_manager.can_only_access_owned_queries():
-            search_user_id = g.user.get_user_id()
-        else:
+        if security_manager.can_access_all_queries():
             search_user_id = request.args.get("user_id")
+        elif (
+            request.args.get("user_id") is not None
+            and request.args.get("user_id") != g.user.get_user_id()
+        ):
+            return Response(status=403, mimetype="application/json")
+        else:
+            search_user_id = g.user.get_user_id()
         database_id = request.args.get("database_id")
         search_text = request.args.get("search_text")
         status = request.args.get("status")

superset/views/sql_lab.py

@@ -40,12 +40,12 @@
 class QueryFilter(BaseFilter):  # pylint: disable=too-few-public-methods
     def apply(self, query: BaseQuery, value: Callable) -> BaseQuery:
         """
-        Filter queries to only those owned by current user if
-        can_only_access_owned_queries permission is set.
+        Filter queries to only those owned by current user. If
+        can_access_all_queries permission is set a user can list all queries
 
         :returns: query
         """
-        if security_manager.can_only_access_owned_queries():
+        if not security_manager.can_access_all_queries():
             query = query.filter(Query.user_id == g.user.get_user_id())
         return query
 

tests/sqllab_tests.py

@@ -223,42 +223,21 @@ def test_search_query_on_time(self):
         data = json.loads(resp)
         self.assertEqual(2, len(data))
 
-    def test_search_query_with_owner_only_perms(self) -> None:
+    def test_search_query_only_owned(self) -> None:
         """
-        Test a search query with can_only_access_owned_queries perm added to
-        Admin and make sure only Admin queries show up.
+        Test a search query with a user that does not have can_access_all_queries.
         """
-        session = db.session
-
-        # Add can_only_access_owned_queries perm to Admin user
-        owned_queries_view = security_manager.find_permission_view_menu(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
-        )
-        security_manager.add_permission_role(
-            security_manager.find_role("Admin"), owned_queries_view
-        )
-        session.commit()
-
-        # Test search_queries for Admin user
+        # Test search_queries for Alpha user
         self.run_some_queries()
-        self.login("admin")
+        self.login("gamma_sqllab")
 
-        user_id = security_manager.find_user("admin").id
+        user_id = security_manager.find_user("gamma_sqllab").id
         data = self.get_json_resp("/superset/search_queries")
-        self.assertEqual(2, len(data))
+
+        self.assertEqual(1, len(data))
         user_ids = {k["userId"] for k in data}
         self.assertEqual(set([user_id]), user_ids)
 
-        # Remove can_only_access_owned_queries from Admin
-        owned_queries_view = security_manager.find_permission_view_menu(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
-        )
-        security_manager.del_permission_role(
-            security_manager.find_role("Admin"), owned_queries_view
-        )
-
-        session.commit()
-
     def test_alias_duplicate(self):
         self.run_sql(
             "SELECT name as col, gender as col FROM birth_names LIMIT 10",
@@ -356,45 +335,54 @@ def test_queryview_filter(self) -> None:
         assert admin.username in user_queries
         assert gamma_sqllab.username in user_queries
 
-    def test_queryview_filter_owner_only(self) -> None:
+    def test_queryview_can_access_all_queries(self) -> None:
         """
-        Test queryview api with can_only_access_owned_queries perm added to
-        Admin and make sure only Admin queries show up.
+        Test queryview api with can_access_all_queries perm added to
+        gamma and make sure all queries show up.
         """
         session = db.session
 
-        # Add can_only_access_owned_queries perm to Admin user
-        owned_queries_view = security_manager.find_permission_view_menu(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
+        # Add all_query_access perm to Gamma user
+        all_queries_view = security_manager.find_permission_view_menu(
+            "all_query_access", "all_query_access"
         )
+
         security_manager.add_permission_role(
-            security_manager.find_role("Admin"), owned_queries_view
+            security_manager.find_role("gamma_sqllab"), all_queries_view
         )
         session.commit()
 
         # Test search_queries for Admin user
         self.run_some_queries()
-        self.login("admin")
-
+        self.login("gamma_sqllab")
         url = "/queryview/api/read"
         data = self.get_json_resp(url)
-        admin = security_manager.find_user("admin")
-        self.assertEqual(2, len(data["result"]))
-        all_admin_user_queries = all(
-            [result.get("username") == admin.username for result in data["result"]]
-        )
-        assert all_admin_user_queries is True
+        self.assertEqual(3, len(data["result"]))
 
-        # Remove can_only_access_owned_queries from Admin
-        owned_queries_view = security_manager.find_permission_view_menu(
-            "can_only_access_owned_queries", "can_only_access_owned_queries"
+        # Remove all_query_access from gamma sqllab
+        all_queries_view = security_manager.find_permission_view_menu(
+            "all_query_access", "all_query_access"
         )
         security_manager.del_permission_role(
-            security_manager.find_role("Admin"), owned_queries_view
+            security_manager.find_role("gamma_sqllab"), all_queries_view
         )
 
         session.commit()
 
+    def test_queryview_admin_can_access_all_queries(self) -> None:
+        """
+        Test queryview api with all_query_access perm added to
+        Admin and make sure only Admin queries show up. This is the default
+        """
+        # Test search_queries for Admin user
+        self.run_some_queries()
+        self.login("admin")
+
+        url = "/queryview/api/read"
+        data = self.get_json_resp(url)
+        admin = security_manager.find_user("admin")
+        self.assertEqual(3, len(data["result"]))
+
     def test_api_database(self):
         self.login("admin")
         self.create_fake_db()
@@ -407,7 +395,7 @@ def test_api_database(self):
             "page": 0,
             "page_size": -1,
         }
-        url = "api/v1/database/?{}={}".format("q", prison.dumps(arguments))
+        url = f"api/v1/database/?q={prison.dumps(arguments)}"
         self.assertEqual(
             {"examples", "fake_db_100"},
             {r.get("database_name") for r in self.get_json_resp(url)["result"]},