Multiple permission roots (data-dot-all#1259)

### Feature or Bugfix
- Bugfix


### Detail
- GET_DATASET_TABLE (FOLDER) permissions are granted to the group only
if they are not granted already
- these permissions are removed if group is not admin|steward and there
are no other shares of this item.

### Relates
- data-dot-all#1174

### Security
Please answer the questions below briefly where applicable, or write
`N/A`. Based on
[OWASP 10](https://owasp.org/Top10/en/).

- Does this PR introduce or modify any input fields or queries - this
includes
fetching data from storage outside the application (e.g. a database, an
S3 bucket)?
  - Is the input sanitized?
- What precautions are you taking before deserializing the data you
consume?
  - Is injection prevented by parametrizing queries?
  - Have you ensured no `eval` or similar functions are used?
- Does this PR introduce any functionality or component that requires
authorization?
- How have you ensured it respects the existing AuthN/AuthZ mechanisms?
  - Are you logging failed auth attempts?
- Are you using or adding any cryptographic features?
  - Do you use a standard proven implementations?
  - Are the used keys controlled by the customer? Where are they stored?
- Are you introducing any new policies/roles/users?
  - Have you used the least-privilege principle? How?


By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license.

---------

Co-authored-by: Sofia Sazonova <[email protected]>

backend/dataall/core/permissions/db/resource_policy/resource_policy_repositories.py

@@ -70,7 +70,9 @@ def has_group_resource_permission(
         else:
             return policy
 
-    def query_all_resource_policies(session, group_uri: str, resource_uri: str, resource_type: str = None):
+    def query_all_resource_policies(
+        session, group_uri: str, resource_uri: str, resource_type: str = None, permissions: List[str] = None
+    ):
         resource_policy = session.query(ResourcePolicy).filter(
             ResourcePolicy.resourceUri == resource_uri,
         )
@@ -84,6 +86,9 @@ def query_all_resource_policies(session, group_uri: str, resource_uri: str, reso
                 ResourcePolicy.resourceType == resource_type,
             )
 
+        if permissions is not None:
+            resource_policy = resource_policy.filter(ResourcePolicy.permissions.in_(permissions))
+
         return resource_policy
 
     @staticmethod
@@ -95,9 +100,9 @@ def find_resource_policy(session, group_uri: str, resource_uri: str, resource_ty
 
     @staticmethod
     def find_all_resource_policies(
-        session, group_uri: str, resource_uri: str, resource_type: str = None
+        session, group_uri: str, resource_uri: str, resource_type: str = None, permissions: List[str] = None
     ) -> List[ResourcePolicy]:
         resource_policy = ResourcePolicyRepository.query_all_resource_policies(
-            session, group_uri, resource_uri, resource_type
+            session, group_uri, resource_uri, resource_type, permissions
         )
         return resource_policy.all()

backend/dataall/core/permissions/services/resource_policy_service.py

@@ -77,7 +77,7 @@ def check_user_resource_permission(session, username: str, groups: [str], resour
             return resource_policy
 
     @staticmethod
-    def find_resource_policies(session, group, resource_uri, resource_type):
+    def find_resource_policies(session, group, resource_uri, resource_type, permissions: List[str] = None):
         """
 
         :param session:
@@ -90,7 +90,7 @@ def find_resource_policies(session, group, resource_uri, resource_type):
             group, resource_uri, resource_type
         )
         policies = ResourcePolicyRepository.find_all_resource_policies(
-            session, group_uri=group, resource_uri=resource_uri, resource_type=resource_type
+            session, group_uri=group, resource_uri=resource_uri, resource_type=resource_type, permissions=permissions
         )
         return policies
 

backend/dataall/modules/dataset_sharing/db/share_object_repositories.py

@@ -899,6 +899,32 @@ def find_all_share_items(session, share_uri, share_type, status=None):
             query = query.filter(ShareObjectItem.status.in_(status))
         return query.all()
 
+    @staticmethod
+    def find_all_other_share_items(
+        session, not_this_share_uri, item_uri, share_type, principal_type, principal_uri, item_status=None
+    ) -> List[ShareObjectItem]:
+        """
+        Find all shares from principal (principal_uri) to item (item_uri), that are not from specified share (not_this_share_uri)
+        """
+        query = (
+            session.query(ShareObjectItem)
+            .join(ShareObject, ShareObjectItem.shareUri == ShareObject.shareUri)
+            .filter(
+                (
+                    and_(
+                        ShareObjectItem.itemUri == item_uri,
+                        ShareObjectItem.itemType == share_type,
+                        ShareObject.principalType == principal_type,
+                        ShareObject.principalId == principal_uri,
+                        ShareObject.shareUri != not_this_share_uri,
+                    )
+                )
+            )
+        )
+        if item_status:
+            query = query.filter(ShareObjectItem.status.in_(item_status))
+        return query.all()
+
     @staticmethod
     def other_approved_share_item_table_exists(session, environment_uri, item_uri, share_item_uri):
         share_item_shared_states = ShareItemSM.get_share_item_shared_states()

backend/dataall/modules/dataset_sharing/services/data_sharing_service.py

@@ -73,10 +73,6 @@ def approve_share(cls, engine: Engine, share_uri: str) -> bool:
                 new_share_state = share_sm.run_transition(ShareObjectActions.Start.value)
                 share_sm.update_state(session, share, new_share_state)
 
-                need_grant_permissions = (
-                    share.groupUri != dataset.SamlAdminGroupName and share.principalType == PrincipalType.Group.value
-                )
-
                 (shared_tables, shared_folders, shared_buckets) = ShareObjectRepository.get_share_data_items(
                     session, share_uri, ShareItemStatus.Share_Approved.value
                 )
@@ -290,7 +286,7 @@ def revoke_share(cls, engine: Engine, share_uri: str):
                     env_group,
                 )
                 log.info(f'revoking folders succeeded = {revoked_folders_succeed}')
-                if share.groupUri != dataset.SamlAdminGroupName and share.principalType == PrincipalType.Group.value:
+                if share.groupUri != dataset.SamlAdminGroupName and share.groupUri != dataset.stewards:
                     log.info('Deleting FOLDER READ permissions...')
                     ShareItemService.delete_dataset_folder_read_permission(session, share)
                 log.info('Revoking permissions to S3 buckets')
@@ -313,7 +309,7 @@ def revoke_share(cls, engine: Engine, share_uri: str):
                 ).process_revoked_shares()
 
                 log.info(f'revoking tables succeeded = {revoked_tables_succeed}')
-                if share.groupUri != dataset.SamlAdminGroupName and share.principalType == PrincipalType.Group.value:
+                if share.groupUri != dataset.SamlAdminGroupName and share.groupUri != dataset.stewards:
                     log.info('Deleting TABLE READ permissions...')
                     ShareItemService.delete_dataset_table_read_permission(session, share)
 

backend/dataall/modules/dataset_sharing/services/share_item_service.py

@@ -245,9 +245,23 @@ def delete_dataset_table_read_permission(session, share):
             session, share.shareUri, ShareableType.Table.value, [ShareItemStatus.Revoke_Succeeded.value]
         )
         for table in share_table_items:
-            ResourcePolicyService.delete_resource_policy(
-                session=session, group=share.groupUri, resource_uri=table.itemUri
+            other_shares = ShareObjectRepository.find_all_other_share_items(
+                session,
+                not_this_share_uri=share.shareUri,
+                item_uri=table.itemUri,
+                share_type=ShareableType.Table,
+                principal_type='GROUP',
+                principal_uri=share.groupUri,
+                item_status=[ShareItemStatus.Share_Succeeded],
             )
+            log.info(
+                f'Table {table.itemUri} has been shared with group {share.groupUri} in {len(other_shares)} more shares'
+            )
+            if len(other_shares) == 0:
+                log.info('Delete permissions...')
+                ResourcePolicyService.delete_resource_policy(
+                    session=session, group=share.groupUri, resource_uri=table.itemUri
+                )
 
     @staticmethod
     def delete_dataset_folder_read_permission(session, share):
@@ -258,8 +272,22 @@ def delete_dataset_folder_read_permission(session, share):
             session, share.shareUri, ShareableType.StorageLocation.value, [ShareItemStatus.Revoke_Succeeded.value]
         )
         for location in share_folder_items:
-            ResourcePolicyService.delete_resource_policy(
-                session=session,
-                group=share.groupUri,
-                resource_uri=location.itemUri,
+            other_shares = ShareObjectRepository.find_all_other_share_items(
+                session,
+                not_this_share_uri=share.shareUri,
+                item_uri=location.itemUri,
+                share_type=ShareableType.StorageLocation,
+                principal_type='GROUP',
+                principal_uri=share.groupUri,
+                item_status=[ShareItemStatus.Share_Succeeded],
+            )
+            log.info(
+                f'Location {location.itemUri} has been shared with group {share.groupUri} in {len(other_shares)} more shares'
             )
+            if len(other_shares) == 0:
+                log.info('Delete permissions...')
+                ResourcePolicyService.delete_resource_policy(
+                    session=session,
+                    group=share.groupUri,
+                    resource_uri=location.itemUri,
+                )

backend/dataall/modules/dataset_sharing/services/share_object_service.py

@@ -533,13 +533,29 @@ def attach_dataset_table_read_permission(session, share):
             session, share.shareUri, ShareableType.Table.value, [ShareItemStatus.Share_Succeeded.value]
         )
         for table in share_table_items:
-            ResourcePolicyService.attach_resource_policy(
-                session=session,
+            existing_policy = ResourcePolicyService.find_resource_policies(
+                session,
                 group=share.groupUri,
-                permissions=DATASET_TABLE_READ,
                 resource_uri=table.itemUri,
                 resource_type=DatasetTable.__name__,
+                permissions=DATASET_TABLE_READ,
             )
+            # toDo: separate policies from list DATASET_TABLE_READ, because in future only one of them can be granted (Now they are always granted together)
+            if len(existing_policy) == 0:
+                log.info(
+                    f'Attaching new resource permission policy {DATASET_TABLE_READ} to table {table.itemUri} for group {share.groupUri}'
+                )
+                ResourcePolicyService.attach_resource_policy(
+                    session=session,
+                    group=share.groupUri,
+                    permissions=DATASET_TABLE_READ,
+                    resource_uri=table.itemUri,
+                    resource_type=DatasetTable.__name__,
+                )
+            else:
+                log.info(
+                    f'Resource permission policy {DATASET_TABLE_READ} to table {table.itemUri} for group {share.groupUri} already exists. Skip... '
+                )
 
     @staticmethod
     def attach_dataset_folder_read_permission(session, share):
@@ -550,10 +566,27 @@ def attach_dataset_folder_read_permission(session, share):
             session, share.shareUri, ShareableType.StorageLocation.value, [ShareItemStatus.Share_Succeeded.value]
         )
         for location in share_folder_items:
-            ResourcePolicyService.attach_resource_policy(
-                session=session,
+            existing_policy = ResourcePolicyService.find_resource_policies(
+                session,
                 group=share.groupUri,
-                permissions=DATASET_FOLDER_READ,
                 resource_uri=location.itemUri,
                 resource_type=DatasetStorageLocation.__name__,
+                permissions=DATASET_FOLDER_READ,
             )
+            # toDo: separate policies from list DATASET_TABLE_READ, because in future only one of them can be granted (Now they are always granted together)
+            if len(existing_policy) == 0:
+                log.info(
+                    f'Attaching new resource permission policy {DATASET_FOLDER_READ} to folder {location.itemUri} for group {share.groupUri}'
+                )
+
+                ResourcePolicyService.attach_resource_policy(
+                    session=session,
+                    group=share.groupUri,
+                    permissions=DATASET_FOLDER_READ,
+                    resource_uri=location.itemUri,
+                    resource_type=DatasetStorageLocation.__name__,
+                )
+            else:
+                log.info(
+                    f'Resource permission policy {DATASET_FOLDER_READ} to table {location.itemUri} for group {share.groupUri} already exists. Skip... '
+                )