fix: catalog permission check (apache#29581)

superset/security/manager.py

@@ -860,7 +860,7 @@ def get_catalogs_accessible_by_user(
             if len(parts) == 2 and default_catalog:
                 accessible_catalogs.add(default_catalog)
             elif len(parts) == 3:
-                accessible_catalogs.add(parts[2])
+                accessible_catalogs.add(parts[1])
 
         # datasource_access
         if perms := self.user_view_menu_names("datasource_access"):
@@ -911,7 +911,12 @@ def get_datasources_accessible_by_user(  # pylint: disable=invalid-name
                 return datasource_names
 
         if schema:
-            schema_perm = self.get_schema_perm(database, catalog, schema)
+            default_catalog = database.get_default_catalog()
+            schema_perm = self.get_schema_perm(
+                database.database_name,
+                catalog or default_catalog,
+                schema,
+            )
             if schema_perm and self.can_access("schema_access", schema_perm):
                 return datasource_names
 

tests/unit_tests/security/manager_test.py

@@ -32,7 +32,7 @@
 )
 from superset.sql_parse import Table
 from superset.superset_typing import AdhocColumn, AdhocMetric
-from superset.utils.core import override_user
+from superset.utils.core import DatasourceName, override_user
 
 
 def test_security_manager(app_context: None) -> None:
@@ -760,3 +760,67 @@ def test_raise_for_access_catalog(
         == """You need access to the following tables: `db2.public.ab_user`,
             `all_database_access` or `all_datasource_access` permission"""
     )
+
+
+def test_get_datasources_accessible_by_user_schema_access(
+    mocker: MockerFixture,
+    app_context: None,
+) -> None:
+    """
+    Test that `get_datasources_accessible_by_user` works with schema permissions.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "can_access_database", return_value=False)
+
+    database = mocker.MagicMock()
+    database.database_name = "db1"
+    database.get_default_catalog.return_value = "catalog2"
+
+    can_access = mocker.patch.object(sm, "can_access", return_value=True)
+
+    datasource_names = [
+        DatasourceName("table1", "schema1", "catalog2"),
+        DatasourceName("table2", "schema1", "catalog2"),
+    ]
+
+    assert sm.get_datasources_accessible_by_user(
+        database,
+        datasource_names,
+        catalog=None,
+        schema="schema1",
+    ) == [
+        DatasourceName("table1", "schema1", "catalog2"),
+        DatasourceName("table2", "schema1", "catalog2"),
+    ]
+
+    # Even though we passed `catalog=None,` the schema check uses the default catalog
+    # when building the schema permission, since the DB supports catalog.
+    can_access.assert_called_with("schema_access", "[db1].[catalog2].[schema1]")
+
+
+def test_get_catalogs_accessible_by_user_schema_access(
+    mocker: MockerFixture,
+    app_context: None,
+) -> None:
+    """
+    Test that `get_catalogs_accessible_by_user` works with schema permissions.
+    """
+    sm = SupersetSecurityManager(appbuilder)
+    mocker.patch.object(sm, "can_access_database", return_value=False)
+    mocker.patch.object(
+        sm,
+        "user_view_menu_names",
+        side_effect=[
+            set(),  # catalog_access
+            {"[db1].[catalog2].[schema1]"},  # schema_access
+            set(),  # datasource_access
+        ],
+    )
+
+    database = mocker.MagicMock()
+    database.database_name = "db1"
+    database.get_default_catalog.return_value = "catalog2"
+
+    catalogs = {"catalog1", "catalog2"}
+
+    assert sm.get_catalogs_accessible_by_user(database, catalogs) == {"catalog2"}