Fix permissions on share workflows (#914)

### Feature or Bugfix
- Bugfix

### Detail
There has been a mismatch on the permissions-granting in the shares.
There are 2 types of permissions: on the share_object and preview
permissions for requesters in the shared_tabled

In this PR:
- delete share object permissions from RDS when the share object is
deleted
- attach table permissions to requesters for approved tables in
`approve_share_object`
- delete table permissions to requesters for revoked tables in
`revoke_items_share_object`
- delete table permissions to requesters for revoked tables in
`remove_shared_item` if the removed item is a table and was in
Share_Failed
- remove the deletion of permissions on the dataset level (which should
not have been there in the first place)

### Relates

### 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.

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

@@ -849,17 +849,20 @@ def _find_all_share_item(session, share, status, share_type_model, share_type_ur
         )
 
     @staticmethod
-    def find_all_share_items(session, share_uri, share_type):
-        return (
+    def find_all_share_items(session, share_uri, share_type, status=None):
+        query = (
             session.query(ShareObjectItem).filter(
                 (
                     and_(
                         ShareObjectItem.shareUri == share_uri,
                         ShareObjectItem.itemType == share_type
                     )
                 )
-            ).all()
+            )
         )
+        if status :
+            query = query.filter(ShareObjectItem.status.in_(status))
+        return query.all()
 
     @staticmethod
     def other_approved_share_item_table_exists(session, environment_uri, item_uri, share_item_uri):

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

@@ -57,11 +57,16 @@ def revoke_items_share_object(uri, revoked_uris):
 
             share_sm.update_state(session, share, new_share_state)
 
-            ResourcePolicy.delete_resource_policy(
-                session=session,
-                group=share.groupUri,
-                resource_uri=dataset.datasetUri,
-            )
+            if share.groupUri != dataset.SamlAdminGroupName:
+                revoke_table_items = ShareObjectRepository.find_all_share_items(
+                    session, uri, ShareableType.Table.value, [ShareItemStatus.Revoke_Approved.value]
+                )
+                for item in revoke_table_items:
+                    ResourcePolicy.delete_resource_policy(
+                        session=session,
+                        group=share.groupUri,
+                        resource_uri=item.itemUri,
+                    )
 
             ShareNotificationService(
                 session=session,
@@ -141,6 +146,13 @@ def add_shared_item(uri: str, data: dict = None):
     def remove_shared_item(uri: str):
         with get_context().db_engine.scoped_session() as session:
             share_item = ShareObjectRepository.get_share_item_by_uri(session, uri)
+            if share_item.itemType == ShareableType.Table.value and share_item.status == ShareItemStatus.Share_Failed.value:
+                share = ShareObjectRepository.get_share_by_uri(session, share_item.shareUri)
+                ResourcePolicy.delete_resource_policy(
+                    session=session,
+                    group=share.groupUri,
+                    resource_uri=share_item.itemUri,
+                )
 
             item_sm = ShareItemSM(share_item.status)
             item_sm.run_transition(ShareItemActions.RemoveItem.value)

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

@@ -206,15 +206,18 @@ def approve_share_object(cls, uri: str):
             cls._run_transitions(session, share, states, ShareObjectActions.Approve)
 
             # GET TABLES SHARED AND APPROVE SHARE FOR EACH TABLE
-            share_table_items = ShareObjectRepository.find_all_share_items(session, uri, ShareableType.Table.value)
-            for table in share_table_items:
-                ResourcePolicy.attach_resource_policy(
-                    session=session,
-                    group=share.principalId,
-                    permissions=DATASET_TABLE_READ,
-                    resource_uri=table.itemUri,
-                    resource_type=DatasetTable.__name__,
+            if share.groupUri != dataset.SamlAdminGroupName:
+                share_table_items = ShareObjectRepository.find_all_share_items(
+                    session, uri, ShareableType.Table.value, [ShareItemStatus.Share_Approved.value]
                 )
+                for table in share_table_items:
+                    ResourcePolicy.attach_resource_policy(
+                        session=session,
+                        group=share.groupUri,
+                        permissions=DATASET_TABLE_READ,
+                        resource_uri=table.itemUri,
+                        resource_type=DatasetTable.__name__,
+                    )
 
             share.rejectPurpose = ""
             session.commit()
@@ -261,11 +264,6 @@ def reject_share_object(cls, uri: str, reject_purpose: str):
         with context.db_engine.scoped_session() as session:
             share, dataset, states = cls._get_share_data(session, uri)
             cls._run_transitions(session, share, states, ShareObjectActions.Reject)
-            ResourcePolicy.delete_resource_policy(
-                session=session,
-                group=share.groupUri,
-                resource_uri=dataset.datasetUri,
-            )
 
             # Update Reject Purpose
             share.rejectPurpose = reject_purpose
@@ -295,6 +293,28 @@ def delete_share_object(cls, uri: str):
                 )
 
             if new_state == ShareObjectStatus.Deleted.value:
+                # Delete share resource policy permissions
+                # Deleting REQUESTER permissions
+                ResourcePolicy.delete_resource_policy(
+                    session=session,
+                    group=share.groupUri,
+                    resource_uri=share.shareUri,
+                )
+
+                # Deleting APPROVER permissions
+                ResourcePolicy.delete_resource_policy(
+                    session=session,
+                    group=dataset.SamlAdminGroupName,
+                    resource_uri=share.shareUri,
+                )
+                if dataset.stewards != dataset.SamlAdminGroupName:
+                    ResourcePolicy.delete_resource_policy(
+                        session=session,
+                        group=dataset.stewards,
+                        resource_uri=share.shareUri,
+                    )
+
+                # Delete share
                 session.delete(share)
 
             return True