Bugfix 956

backend/dataall/core/stacks/db/enums.py

@@ -6,7 +6,7 @@ class StackStatus(Enum):
     CREATE_COMPLETE = 'CREATE_COMPLETE'
     CREATE_FAILED = 'CREATE_FAILED'
     DELETE_IN_PROGRESS = 'DELETE_IN_PROGRESS'
-    DELETE_COMPLETE = 'DELETE_FAILED'
+    DELETE_COMPLETE = 'DELETE_COMPLETE'
     DELETE_FAILED = 'DELETE_FAILED'
     ROLLBACK_IN_PROGRESS = 'ROLLBACK_IN_PROGRESS'
     ROLLBACK_COMPLETE = 'ROLLBACK_COMPLETE'

backend/dataall/modules/dataset_sharing/aws/kms_client.py

@@ -55,6 +55,24 @@ def get_key_id(self, key_alias: str):
         else:
             return response['KeyMetadata']['KeyId']
 
+    def get_key_id_using_list_aliases(self, key_alias: str):
+        try:
+            key_id = None
+            paginator = self._client.get_paginator('list_aliases')
+            for page in paginator.paginate():
+                key_aliases = [alias["AliasName"] for alias in page['Aliases']]
+                if key_alias in key_aliases:
+                    # Retrieve the key_id corresponding to the matching key_alias
+                    key_id = [alias["TargetKeyId"] for alias in page['Aliases'] if alias["AliasName"] == key_alias][0]
+                    break
+        except ClientError as e:
+            if e.response['Error']['Code'] == 'AccessDenied':
+                raise Exception(f'Data.all Environment Pivot Role does not have kms:ListAliases Permission for key {key_alias}: {e}')
+            log.error(f'Failed to get kms key id of {key_alias}: {e}')
+            return None
+        else:
+            return key_id
+
     def check_key_exists(self, key_alias: str):
         try:
             key_exist = False

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

@@ -19,11 +19,9 @@ def trigger_table_sharing_failure_alarm(
             target_environment: Environment,
     ):
         log.info('Triggering share failure alarm...')
-        subject = (
-            f'ALARM: DATAALL Table {table.GlueTableName} Sharing Failure Notification'
-        )
+        subject = f'Data.all Share Failure for Table {table.GlueTableName}'[:100]
         message = f"""
-    You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the table {table.GlueTableName} with Lake Formation.
+    You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the table {table.GlueTableName} with Lake Formation.
 
     Alarm Details:
         - State Change:               	OK -> ALARM
@@ -51,9 +49,9 @@ def trigger_revoke_table_sharing_failure_alarm(
             target_environment: Environment,
     ):
         log.info('Triggering share failure alarm...')
-        subject = f'ALARM: DATAALL Table {table.GlueTableName} Revoking LF permissions Failure Notification'
+        subject = f'Data.all Revoke LF Permissions Failure for Table {table.GlueTableName}'[:100]
         message = f"""
-    You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to revoke Lake Formation permissions for table {table.GlueTableName} with Lake Formation.
+    You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to revoke Lake Formation permissions for table {table.GlueTableName} with Lake Formation.
 
     Alarm Details:
         - State Change:               	OK -> ALARM
@@ -76,11 +74,9 @@ def trigger_revoke_table_sharing_failure_alarm(
 
     def trigger_dataset_sync_failure_alarm(self, dataset: Dataset, error: str):
         log.info(f'Triggering dataset {dataset.name} tables sync failure alarm...')
-        subject = (
-            f'ALARM: DATAALL Dataset {dataset.name} Tables Sync Failure Notification'
-        )
+        subject = f'Data.all Dataset Tables Sync Failure for {dataset.name}'[:100]
         message = f"""
-You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to synchronize Dataset {dataset.name} tables from AWS Glue to the Search Catalog.
+You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to synchronize Dataset {dataset.name} tables from AWS Glue to the Search Catalog.
 
 Alarm Details:
     - State Change:               	OK -> ALARM
@@ -101,11 +97,9 @@ def trigger_folder_sharing_failure_alarm(
         target_environment: Environment,
     ):
         log.info('Triggering share failure alarm...')
-        subject = (
-            f'ALARM: DATAALL Folder {folder.S3Prefix} Sharing Failure Notification'
-        )
+        subject = f'Data.all Folder Share Failure for {folder.S3Prefix}'[:100]
         message = f"""
-You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the folder {folder.S3Prefix} with S3 Access Point.
+You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the folder {folder.S3Prefix} with S3 Access Point.
 Alarm Details:
     - State Change:               	OK -> ALARM
     - Reason for State Change:      S3 Folder sharing failure
@@ -129,11 +123,9 @@ def trigger_revoke_folder_sharing_failure_alarm(
         target_environment: Environment,
     ):
         log.info('Triggering share failure alarm...')
-        subject = (
-            f'ALARM: DATAALL Folder {folder.S3Prefix} Sharing Revoke Failure Notification'
-        )
+        subject = f'Data.all Folder Share Revoke Failure for {folder.S3Prefix}'[:100]
         message = f"""
-You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the folder {folder.S3Prefix} with S3 Access Point.
+You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to share the folder {folder.S3Prefix} with S3 Access Point.
 Alarm Details:
     - State Change:               	OK -> ALARM
     - Reason for State Change:      S3 Folder sharing Revoke failure
@@ -173,11 +165,9 @@ def handle_bucket_sharing_failure(self, bucket: DatasetBucket,
                                       target_environment: Environment,
                                       alarm_type: str):
         log.info(f'Triggering {alarm_type} failure alarm...')
-        subject = (
-            f'ALARM: DATAALL S3 Bucket {bucket.S3BucketName} {alarm_type} Failure Notification'
-        )
+        subject = f'Data.all S3 Bucket Failure for {bucket.S3BucketName} {alarm_type}'[:100]
         message = f"""
-You are receiving this email because your DATAALL {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to {alarm_type} the S3 Bucket {bucket.S3BucketName}.
+You are receiving this email because your Data.all {self.envname} environment in the {self.region} region has entered the ALARM state, because it failed to {alarm_type} the S3 Bucket {bucket.S3BucketName}.
 Alarm Details:
     - State Change:               	OK -> ALARM
     - Reason for State Change:      S3 Bucket {alarm_type} failure

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

@@ -23,6 +23,7 @@
 IAM_ACCESS_POINT_ROLE_POLICY = "targetDatasetAccessControlPolicy"
 DATAALL_ALLOW_OWNER_SID = "AllowAllToAdmin"
 DATAALL_ACCESS_POINT_KMS_DECRYPT_SID = "DataAll-Access-Point-KMS-Decrypt"
+DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID = "KMSPivotRolePermissions"
 
 
 class S3AccessPointShareManager:
@@ -331,14 +332,25 @@ def update_dataset_bucket_key_policy(self):
         )
         key_alias = f"alias/{self.dataset.KmsAlias}"
         kms_client = KmsClient(self.source_account_id, self.source_environment.region)
-        kms_key_id = kms_client.get_key_id(key_alias)
+        kms_key_id = kms_client.get_key_id_using_list_aliases(key_alias)
         existing_policy = kms_client.get_key_policy(kms_key_id)
         target_requester_arn = IAM.get_role_arn_by_name(self.target_account_id, self.target_requester_IAMRoleName)
+        pivot_role_name = SessionHelper.get_delegation_role_name()
 
         if existing_policy:
             existing_policy = json.loads(existing_policy)
             counter = count()
             statements = {item.get("Sid", next(counter)): item for item in existing_policy.get("Statement", {})}
+
+            if DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID in statements.keys():
+                logger.info(
+                    f'KMS key policy already contains share statement {DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}')
+            else:
+                logger.info(
+                    f'KMS key policy does not contain statement {DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}, generating a new one')
+                statements[DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID] \
+                    = self.generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, self.dataset_account_id)
+
             if DATAALL_ACCESS_POINT_KMS_DECRYPT_SID in statements.keys():
                 logger.info(
                     f'KMS key policy contains share statement {DATAALL_ACCESS_POINT_KMS_DECRYPT_SID}, '
@@ -353,12 +365,14 @@ def update_dataset_bucket_key_policy(self):
                 statements[DATAALL_ACCESS_POINT_KMS_DECRYPT_SID] = (self.generate_default_kms_decrypt_policy_statement
                                                                     (target_requester_arn))
             existing_policy["Statement"] = list(statements.values())
+
         else:
             logger.info('KMS key policy does not contain any statements, generating a new one')
             existing_policy = {
                 "Version": "2012-10-17",
                 "Statement": [
-                    self.generate_default_kms_decrypt_policy_statement(target_requester_arn)
+                    self.generate_default_kms_decrypt_policy_statement(target_requester_arn),
+                    self.generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, self.dataset_account_id)
                 ]
             }
         kms_client.put_key_policy(
@@ -474,7 +488,7 @@ def delete_dataset_bucket_key_policy(
         )
         key_alias = f"alias/{dataset.KmsAlias}"
         kms_client = KmsClient(dataset.AwsAccountId, dataset.region)
-        kms_key_id = kms_client.get_key_id(key_alias)
+        kms_key_id = kms_client.get_key_id_using_list_aliases(key_alias)
         existing_policy = json.loads(kms_client.get_key_policy(kms_key_id))
         target_requester_arn = IAM.get_role_arn_by_name(self.target_account_id, self.target_requester_IAMRoleName)
         counter = count()
@@ -510,7 +524,7 @@ def handle_share_failure(self, error: Exception) -> None:
             self.target_folder, self.share, self.target_environment
         )
 
-    def handle_revoke_failure(self, error: Exception) -> None:
+    def handle_revoke_failure(self, error: Exception) -> bool:
         """
         Handles share failure by raising an alarm to alarmsTopic
         Returns
@@ -526,6 +540,7 @@ def handle_revoke_failure(self, error: Exception) -> None:
         DatasetAlarmService().trigger_revoke_folder_sharing_failure_alarm(
             self.target_folder, self.share, self.target_environment
         )
+        return True
 
     @staticmethod
     def generate_default_kms_decrypt_policy_statement(target_requester_arn):
@@ -541,6 +556,29 @@ def generate_default_kms_decrypt_policy_statement(target_requester_arn):
             "Resource": "*"
         }
 
+    @staticmethod
+    def generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, dataset_account_id):
+        return {
+            "Sid": f"{DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}",
+            "Effect": "Allow",
+            "Principal": {
+                "AWS": [
+                    f"arn:aws:iam::{dataset_account_id}:role/{pivot_role_name}"
+                ]
+            },
+            "Action": [
+                "kms:Decrypt",
+                "kms:Encrypt",
+                "kms:GenerateDataKey*",
+                "kms:PutKeyPolicy",
+                "kms:GetKeyPolicy",
+                "kms:ReEncrypt*",
+                "kms:TagResource",
+                "kms:UntagResource"
+            ],
+            "Resource": "*"
+        }
+
     def add_target_arn_to_statement_principal(self, statement, target_requester_arn):
         principal_list = self.get_principal_list(statement)
         if f"{target_requester_arn}" not in principal_list:

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

@@ -20,6 +20,7 @@
 DATAALL_ALLOW_OWNER_SID = "AllowAllToAdmin"
 IAM_S3BUCKET_ROLE_POLICY = "dataall-targetDatasetS3Bucket-AccessControlPolicy"
 DATAALL_BUCKET_KMS_DECRYPT_SID = "DataAll-Bucket-KMS-Decrypt"
+DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID = "KMSPivotRolePermissions"
 
 
 class S3BucketShareManager:
@@ -271,13 +272,25 @@ def grant_dataset_bucket_key_policy(self):
             )
             key_alias = f"alias/{self.target_bucket.KmsAlias}"
             kms_client = KmsClient(self.source_account_id, self.source_environment.region)
-            kms_key_id = kms_client.get_key_id(key_alias)
+            kms_key_id = kms_client.get_key_id_using_list_aliases(key_alias)
             existing_policy = kms_client.get_key_policy(kms_key_id)
             target_requester_arn = IAM.get_role_arn_by_name(self.target_account_id, self.target_requester_IAMRoleName)
+            pivot_role_name = SessionHelper.get_delegation_role_name()
+
             if existing_policy:
                 existing_policy = json.loads(existing_policy)
                 counter = count()
                 statements = {item.get("Sid", next(counter)): item for item in existing_policy.get("Statement", {})}
+
+                if DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID in statements.keys():
+                    logger.info(
+                        f'KMS key policy already contains share statement {DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}')
+                else:
+                    logger.info(
+                        f'KMS key policy does not contain statement {DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}, generating a new one')
+                    statements[DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID] \
+                        = self.generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, self.source_account_id)
+
                 if DATAALL_BUCKET_KMS_DECRYPT_SID in statements.keys():
                     logger.info(
                         f'KMS key policy contains share statement {DATAALL_BUCKET_KMS_DECRYPT_SID}, updating the current one')
@@ -289,12 +302,14 @@ def grant_dataset_bucket_key_policy(self):
                     statements[DATAALL_BUCKET_KMS_DECRYPT_SID] = self.generate_default_kms_decrypt_policy_statement(
                         target_requester_arn)
                 existing_policy["Statement"] = list(statements.values())
+
             else:
                 logger.info('KMS key policy does not contain any statements, generating a new one')
                 existing_policy = {
                     "Version": "2012-10-17",
                     "Statement": [
-                        self.generate_default_kms_decrypt_policy_statement(target_requester_arn)
+                        self.generate_default_kms_decrypt_policy_statement(target_requester_arn),
+                        self.generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, self.source_account_id)
                     ]
                 }
             kms_client.put_key_policy(
@@ -394,7 +409,7 @@ def delete_target_role_bucket_key_policy(
             )
             key_alias = f"alias/{target_bucket.KmsAlias}"
             kms_client = KmsClient(target_bucket.AwsAccountId, target_bucket.region)
-            kms_key_id = kms_client.get_key_id(key_alias)
+            kms_key_id = kms_client.get_key_id_using_list_aliases(key_alias)
             existing_policy = json.loads(kms_client.get_key_policy(kms_key_id))
             target_requester_arn = IAM.get_role_arn_by_name(self.target_account_id, self.target_requester_IAMRoleName)
             counter = count()
@@ -444,7 +459,7 @@ def handle_revoke_failure(self, error: Exception) -> bool:
             f'with target account {self.target_environment.AwsAccountId}/{self.target_environment.region} '
             f'due to: {error}'
         )
-        DatasetAlarmService().trigger_revoke_folder_sharing_failure_alarm(
+        DatasetAlarmService().trigger_revoke_s3_bucket_sharing_failure_alarm(
             self.target_bucket, self.share, self.target_environment
         )
         return True
@@ -482,3 +497,26 @@ def generate_default_kms_decrypt_policy_statement(target_requester_arn):
             "Action": "kms:Decrypt",
             "Resource": "*"
         }
+
+    @staticmethod
+    def generate_enable_pivot_role_permissions_policy_statement(pivot_role_name, source_account_id):
+        return {
+            "Sid": f"{DATAALL_KMS_PIVOT_ROLE_PERMISSIONS_SID}",
+            "Effect": "Allow",
+            "Principal": {
+                "AWS": [
+                    f"arn:aws:iam::{source_account_id}:role/{pivot_role_name}"
+                ]
+            },
+            "Action": [
+                "kms:Decrypt",
+                "kms:Encrypt",
+                "kms:GenerateDataKey*",
+                "kms:PutKeyPolicy",
+                "kms:GetKeyPolicy",
+                "kms:ReEncrypt*",
+                "kms:TagResource",
+                "kms:UntagResource",
+            ],
+            "Resource": "*"
+        }

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

@@ -111,11 +111,14 @@ def process_approved_shares(self) -> bool:
                     shared_item_SM.update_state_single_item(self.session, share_item, new_state)
 
                 except Exception as e:
-                    self.handle_share_failure(table=table, share_item=share_item, error=e)
+                    # must run first to ensure state transitions to failed
                     new_state = shared_item_SM.run_transition(ShareItemActions.Failure.value)
                     shared_item_SM.update_state_single_item(self.session, share_item, new_state)
                     success = False
 
+                    # statements which can throw exceptions but are not critical
+                    self.handle_share_failure(table=table, share_item=share_item, error=e)
+
         return success
 
     def process_revoked_shares(self) -> bool:
@@ -178,9 +181,12 @@ def process_revoked_shares(self) -> bool:
                 revoked_item_SM.update_state_single_item(self.session, share_item, new_state)
 
             except Exception as e:
-                self.handle_revoke_failure(share_item=share_item, table=table, error=e)
+                # must run first to ensure state transitions to failed
                 new_state = revoked_item_SM.run_transition(ShareItemActions.Failure.value)
                 revoked_item_SM.update_state_single_item(self.session, share_item, new_state)
                 success = False
 
+                # statements which can throw exceptions but are not critical
+                self.handle_revoke_failure(share_item=share_item, table=table, error=e)
+
         return success