Fix Check other share exists before clean up (#769)

### Feature or Bugfix
<!-- please choose -->
- Bugfix

### Detail
- Fix method to detect if other share objects exist on the environment
before cleaning up environment-level shared resources (i.e. RAM
invitation and PivotRole permissions)
- Originally, if TeamA in EnvA had 2 shares approved and succeeded on
DatasetB and TeamA rejects 1 of the pre-existing shares, the method
`other_approved_share_object_exists` was returning `False`and deleting
necessary permissions for the other existing Share
- Also disables the other existing shares ability to Revoke the still
existing share since pivotRole no longer has permissions

- Also fixes the removal of dataall QS Group permissions if there are
still existing shares to EnvA

### Security
NA
```
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: dlpzx <[email protected]>

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

@@ -774,17 +774,23 @@ def find_all_share_items(session, share_uri, share_type):
         )
 
     @staticmethod
-    def other_approved_share_object_exists(session, environment_uri, dataset_uri):
+    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()
         return (
             session.query(ShareObject)
+            .join(
+                ShareObjectItem,
+                ShareObject.shareUri == ShareObjectItem.shareUri,
+            )
             .filter(
                 and_(
-                    Environment.environmentUri == environment_uri,
-                    ShareObject.status == ShareObjectStatus.Approved.value,
-                    ShareObject.datasetUri == dataset_uri,
+                    ShareObject.environmentUri == environment_uri,
+                    ShareObjectItem.itemUri == item_uri,
+                    ShareObjectItem.shareItemUri != share_item_uri,
+                    ShareObjectItem.status.in_(share_item_shared_states),
                 )
             )
-            .all()
+            .first()
         )
 
     @staticmethod

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

@@ -207,7 +207,7 @@ def revoke_share(cls, engine: Engine, share_uri: str):
             log.info(f'Still remaining LF resources shared = {existing_shared_items}')
             if not existing_shared_items and revoked_tables:
                 log.info("Clean up LF remaining resources...")
-                clean_up_tables = processor.clean_up_share()
+                clean_up_tables = processor.delete_shared_database()
                 log.info(f"Clean up LF successful = {clean_up_tables}")
 
             existing_pending_items = ShareObjectRepository.check_pending_share_items(session, share_uri)

backend/dataall/modules/dataset_sharing/services/share_managers/lf_share_manager.py

@@ -51,10 +51,6 @@ def process_approved_shares(self) -> [str]:
     def process_revoked_shares(self) -> [str]:
         return NotImplementedError
 
-    @abc.abstractmethod
-    def clean_up_share(self):
-        return NotImplementedError
-
     def get_share_principals(self) -> [str]:
         """
         Builds list of principals of the share request
@@ -421,7 +417,7 @@ def share_table_with_target_account(cls, **data):
             )
             raise e
 
-    def revoke_external_account_access_on_source_account(self) -> [dict]:
+    def revoke_external_account_access_on_source_account(self, db_name, table_name) -> [dict]:
         """
         1) Revokes access to external account
         if dataset is not shared with any other team from the same workspace
@@ -440,29 +436,28 @@ def revoke_external_account_access_on_source_account(self) -> [dict]:
         client = aws_session.client(
             'lakeformation', region_name=self.source_environment.region
         )
-        revoke_entries = []
-        for table in self.revoked_tables:
-            revoke_entries.append(
-                {
-                    'Id': str(uuid.uuid4()),
-                    'Principal': {
-                        'DataLakePrincipalIdentifier': self.target_environment.AwsAccountId
-                    },
-                    'Resource': {
-                        'TableWithColumns': {
-                            'DatabaseName': table.GlueDatabaseName,
-                            'Name': table.GlueTableName,
-                            'ColumnWildcard': {},
-                            'CatalogId': self.source_environment.AwsAccountId,
-                        }
-                    },
-                    'Permissions': ['DESCRIBE', 'SELECT'],
-                    'PermissionsWithGrantOption': ['DESCRIBE', 'SELECT'],
-                }
-            )
-            LakeFormationClient.batch_revoke_permissions(
-                client, self.source_environment.AwsAccountId, revoke_entries
-            )
+        revoke_entries = [
+            {
+                'Id': str(uuid.uuid4()),
+                'Principal': {
+                    'DataLakePrincipalIdentifier': self.target_environment.AwsAccountId
+                },
+                'Resource': {
+                    'TableWithColumns': {
+                        'DatabaseName': db_name,
+                        'Name': table_name,
+                        'ColumnWildcard': {},
+                        'CatalogId': self.source_environment.AwsAccountId,
+                    }
+                },
+                'Permissions': ['DESCRIBE', 'SELECT'],
+                'PermissionsWithGrantOption': ['DESCRIBE', 'SELECT'],
+            }
+        ]
+
+        LakeFormationClient.batch_revoke_permissions(
+            client, self.source_environment.AwsAccountId, revoke_entries
+        )
         return revoke_entries
 
     def handle_share_failure(

backend/dataall/modules/dataset_sharing/services/share_processors/lf_process_cross_account_share.py

@@ -124,7 +124,7 @@ def process_revoked_shares(self) -> bool:
             a) update its status to REVOKE_IN_PROGRESS with Action Start
             b) check if item exists on glue catalog raise error if not and flag item status to failed
             c) revoke table resource link: undo grant permission to resource link table for team role in target account
-            d) revoke source table access: undo grant permission to table for team role in source account
+            d) revoke source table access: undo grant permission to table for team role in source account (and for QS Group if no other shares present for table)
             e) delete resource link table
             h) update share item status to REVOKE_SUCCESSFUL with Action Success
 
@@ -157,10 +157,23 @@ def process_revoked_shares(self) -> bool:
 
                 self.revoke_table_resource_link_access(table, principals)
 
+                other_table_shares_in_env = False
+                if ShareObjectRepository.other_approved_share_item_table_exists(
+                    self.session,
+                    self.target_environment.environmentUri,
+                    share_item.itemUri,
+                    share_item.shareItemUri
+                ):
+                    other_table_shares_in_env = True
+                    principals = [p for p in principals if "arn:aws:quicksight" not in p]
+
                 self.revoke_source_table_access(table, principals)
 
                 self.delete_resource_link_table(table)
 
+                if not other_table_shares_in_env:
+                    self.revoke_external_account_access_on_source_account(table.GlueDatabaseName, table.GlueTableName)
+
                 new_state = revoked_item_SM.run_transition(ShareItemActions.Success.value)
                 revoked_item_SM.update_state_single_item(self.session, share_item, new_state)
 
@@ -171,24 +184,3 @@ def process_revoked_shares(self) -> bool:
                 success = False
 
         return success
-
-    def clean_up_share(self) -> bool:
-        """"
-        1) deletes deprecated shared db in target account
-        2) checks if there are other share objects from this source account to this target account.
-        If not, it revokes external account access of the target account to the source account.
-        Returns
-        -------
-        True if clean-up succeeds
-        """
-
-        self.delete_shared_database()
-
-        if not ShareObjectRepository.other_approved_share_object_exists(
-                self.session,
-                self.target_environment.environmentUri,
-                self.dataset.datasetUri,
-        ):
-            self.revoke_external_account_access_on_source_account()
-
-        return True

backend/dataall/modules/dataset_sharing/services/share_processors/lf_process_same_account_share.py

@@ -157,13 +157,3 @@ def process_revoked_shares(self) -> bool:
                 success = False
 
         return success
-
-    def clean_up_share(self) -> bool:
-        """"
-        1) deletes deprecated shared db in target account
-        Returns
-        -------
-        True if clean-up succeeds
-        """
-        self.delete_shared_database()
-        return True