fix: create virtual dataset validation

superset/commands/dataset/create.py

@@ -25,13 +25,15 @@
 from superset.commands.dataset.exceptions import (
     DatabaseNotFoundValidationError,
     DatasetCreateFailedError,
+    DatasetDataAccessIsNotAllowed,
     DatasetExistsValidationError,
     DatasetInvalidError,
     TableNotFoundValidationError,
 )
 from superset.daos.dataset import DatasetDAO
 from superset.daos.exceptions import DAOCreateFailedError
-from superset.extensions import db
+from superset.exceptions import SupersetSecurityException
+from superset.extensions import db, security_manager
 
 logger = logging.getLogger(__name__)
 
@@ -82,6 +84,15 @@ def validate(self) -> None:
         ):
             exceptions.append(TableNotFoundValidationError(table_name))
 
+        if sql:
+            try:
+                security_manager.raise_for_access(
+                    database=database,
+                    sql=sql,
+                    schema=schema,
+                )
+            except SupersetSecurityException as ex:
+                exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
         try:
             owners = self.populate_owners(owner_ids)
             self._properties["owners"] = owners

superset/commands/dataset/exceptions.py

@@ -144,6 +144,13 @@ def __init__(self) -> None:
         super().__init__([_("Owners are invalid")], field_name="owners")
 
 
+class DatasetDataAccessIsNotAllowed(ValidationError):
+    status = 422
+
+    def __init__(self, message: str) -> None:
+        super().__init__([_(message)], field_name="sql")
+
+
 class DatasetNotFoundError(CommandException):
     status = 404
     message = _("Dataset does not exist")

superset/security/manager.py

@@ -1828,6 +1828,8 @@ def raise_for_access(
         query_context: Optional["QueryContext"] = None,
         table: Optional["Table"] = None,
         viz: Optional["BaseViz"] = None,
+        sql: Optional[str] = None,
+        schema: Optional[str] = None,
     ) -> None:
         """
         Raise an exception if the user cannot access the resource.
@@ -1838,6 +1840,8 @@ def raise_for_access(
         :param query_context: The query context
         :param table: The Superset table (requires database)
         :param viz: The visualization
+        :param sql: The SQL string (requires database)
+        :param schema: Optional schema name
         :raises SupersetSecurityException: If the user cannot access the resource
         """
 
@@ -1846,7 +1850,19 @@ def raise_for_access(
         from superset.connectors.sqla.models import SqlaTable
         from superset.models.dashboard import Dashboard
         from superset.models.slice import Slice
+        from superset.models.sql_lab import Query
         from superset.sql_parse import Table
+        from superset.utils.core import shortid
+
+        if sql and database:
+            query = Query(
+                database=database,
+                sql=sql,
+                schema=schema,
+                client_id=shortid()[:10],
+                user_id=get_user_id(),
+            )
+            self.get_session.expunge(query)
 
         if database and table or query:
             if query:

tests/integration_tests/datasets/commands_tests.py

@@ -586,6 +586,21 @@ def test_create_dataset_command(self, mock_g, mock_g2):
             engine.execute("DROP TABLE test_create_dataset_command")
         db.session.commit()
 
+    @patch("superset.security.manager.g")
+    @patch("superset.commands.utils.g")
+    def test_create_dataset_command_not_allowed(self, mock_g, mock_g2):
+        mock_g.user = security_manager.find_user("gamma")
+        mock_g2.user = mock_g.user
+        examples_db = get_example_database()
+        with self.assertRaises(DatasetInvalidError):
+            _ = CreateDatasetCommand(
+                {
+                    "sql": "select * from ab_user",
+                    "database": examples_db.id,
+                    "table_name": "exp1",
+                }
+            ).run()
+
 
 class TestDatasetWarmUpCacheCommand(SupersetTestCase):
     def test_warm_up_cache_command_table_not_found(self):

tests/integration_tests/security_tests.py

@@ -1642,6 +1642,30 @@ def test_raise_for_access_query(self, mock_can_access, mock_is_owner):
         with self.assertRaises(SupersetSecurityException):
             security_manager.raise_for_access(query=query)
 
+    @patch("superset.utils.core.g")
+    @patch("superset.security.manager.g")
+    def test_raise_for_access_sql_fails(self, mock_sm_g, mock_g):
+        mock_g.user = mock_sm_g.user = security_manager.find_user("gamma")
+        with self.assertRaises(SupersetSecurityException):
+            security_manager.raise_for_access(
+                database=get_example_database(), schema="bar", sql="SELECT * FROM foo"
+            )
+
+    @patch("superset.security.SupersetSecurityManager.is_owner")
+    @patch("superset.security.SupersetSecurityManager.can_access")
+    @patch("superset.utils.core.g")
+    @patch("superset.security.manager.g")
+    def test_raise_for_access_sql(
+        self, mock_sm_g, mock_g, mock_can_access, mock_is_owner
+    ):
+        mock_g.user = mock_sm_g.user = security_manager.find_user("gamma")
+        mock_can_access.return_value = True
+        mock_is_owner.return_value = True
+
+        security_manager.raise_for_access(
+            database=get_example_database(), schema="bar", sql="SELECT * FROM foo"
+        )
+
     @patch("superset.security.manager.g")
     @patch("superset.security.SupersetSecurityManager.is_owner")
     @patch("superset.security.SupersetSecurityManager.can_access")