fix: Users being able to update datasets across DBs (#17348)

* add database id back

* add condition to verify dataset is being changed

* Update superset/datasets/dao.py

Co-authored-by: Beto Dealmeida <roberto@dealmeida.net>

Co-authored-by: Beto Dealmeida <roberto@dealmeida.net>

superset-frontend/src/SqlLab/components/ResultSet/index.tsx

@@ -147,6 +147,7 @@ const ResultSetErrorMessage = styled.div`
 `;
 
 const updateDataset = async (
+  dbId: number,
   datasetId: number,
   sql: string,
   columns: Array<Record<string, any>>,
@@ -159,6 +160,7 @@ const updateDataset = async (
     sql,
     columns,
     owners,
+    database_id: dbId,
   });
 
   const data: JsonResponse = await SupersetClient.put({
@@ -272,10 +274,11 @@ export default class ResultSet extends React.PureComponent<
   };
 
   handleOverwriteDataset = async () => {
-    const { sql, results } = this.props.query;
+    const { sql, results, dbId } = this.props.query;
     const { datasetToOverwrite } = this.state;
 
     await updateDataset(
+      dbId,
       datasetToOverwrite.datasetId,
       sql,
       results.selected_columns.map(d => ({ column_name: d.name })),

superset/connectors/sqla/models.py

@@ -1734,14 +1734,13 @@ def before_update(
 
         for attr in ["database_id", "schema", "table_name"]:
             history = state.get_history(attr, True)
-
             if history.has_changes():
                 break
         else:
             return None
 
         if not DatasetDAO.validate_uniqueness(
-            target.database_id, target.schema, target.table_name
+            target.database_id, target.schema, target.table_name, target.id
         ):
             raise Exception(get_dataset_exist_error_msg(target.full_name))
 

superset/datasets/dao.py

@@ -85,12 +85,22 @@ def validate_table_exists(
             return False
 
     @staticmethod
-    def validate_uniqueness(database_id: int, schema: Optional[str], name: str) -> bool:
+    def validate_uniqueness(
+        database_id: int,
+        schema: Optional[str],
+        name: str,
+        dataset_id: Optional[int] = None,
+    ) -> bool:
         dataset_query = db.session.query(SqlaTable).filter(
             SqlaTable.table_name == name,
             SqlaTable.schema == schema,
             SqlaTable.database_id == database_id,
         )
+
+        if dataset_id:
+            # make sure the dataset found is different from the target (if any)
+            dataset_query = dataset_query.filter(SqlaTable.id != dataset_id)
+
         return not db.session.query(dataset_query.exists()).scalar()
 
     @staticmethod