sec(datasets): BI-5998 require connection read permission on any data…

…set source action (#755)

* sec(datasets): BI-5998 require connection read permission on any dataset source action

* remove redundant permission check

* fix typo

* bring back the second permissions check

lib/dl_api_lib/dl_api_lib/dataset/utils.py

@@ -22,6 +22,7 @@
     DataSourceCollectionBase,
     DataSourceCollectionFactory,
 )
+import dl_core.exc as exc
 from dl_core.us_dataset import Dataset
 from dl_core.us_manager.local_cache import USEntryBuffer
 from dl_core.us_manager.us_manager import USManagerBase
@@ -50,7 +51,7 @@ def _iter_data_source_collections(
 
 def check_permissions_for_origin_sources(
     dataset: Dataset,
-    source_ids: Iterable[str],
+    source_ids: Optional[Iterable[str]],
     permission_kind: USPermissionKind,
     us_entry_buffer: USEntryBuffer,
 ) -> None:
@@ -62,7 +63,10 @@ def check_permissions_for_origin_sources(
     ):
         data_source = dsrc_coll.get_opt(role=DataSourceRole.origin)
         if data_source is not None:
-            bi_utils.need_permission_on_entry(data_source.connection, permission_kind)
+            try:
+                bi_utils.need_permission_on_entry(data_source.connection, permission_kind)
+            except exc.ReferencedUSEntryNotFound:
+                LOGGER.info(f"Connection for source {data_source.id} not found => skipping permission check")
 
 
 def log_dataset_field_stats(dataset: Dataset) -> None:

lib/dl_api_lib/dl_api_lib/dataset/validator.py

@@ -1147,6 +1147,14 @@ def apply_source_action(
     ) -> None:
         """Apply update to the data source configuration"""
 
+        self._sync_us_manager.load_dependencies(self._ds)
+        check_permissions_for_origin_sources(  # any source update requires sufficient permissions on the connection
+            dataset=self._ds,
+            source_ids=None,
+            permission_kind=USPermissionKind.read,
+            us_entry_buffer=self._us_manager.get_entry_buffer(),
+        )
+
         source_data = source_data.copy()
         source_id = source_data.pop("id") or str(uuid.uuid4())
         component_ref = DatasetComponentRef(component_type=ComponentType.data_source, component_id=source_id)
@@ -1178,12 +1186,17 @@ def add_source(title: str) -> None:
                 parameters=source_data["parameters"],
             )
 
-            check_permissions_for_origin_sources(
-                dataset=self._ds,
-                source_ids=[source_id],
-                permission_kind=USPermissionKind.read,
-                us_entry_buffer=self._us_manager.get_entry_buffer(),
-            )
+            # need to check permissions again in case added source refers to an unchecked connection
+            # this can only happen when the first source is being added,
+            # because we don't support more than one connection in a single ds (see `source_can_be_added`)
+            existing_source_id = self._ds.get_single_data_source_id(ignore_source_ids=[source_id])
+            if existing_source_id is None:  # dataset is empty
+                check_permissions_for_origin_sources(
+                    dataset=self._ds,
+                    source_ids=[source_id],
+                    permission_kind=USPermissionKind.read,
+                    us_entry_buffer=self._us_manager.get_entry_buffer(),
+                )
 
         if action in (DatasetAction.update_source, DatasetAction.delete_source):
             dsrc_coll = self._get_data_source_coll_strict(source_id=source_id)