Version 1.0.15

Making RAM share acceptance more explicit
Fixing producer IAM policy
Improved documentation for producer rights after creating data product

Author: IanMeyers

README.md

@@ -204,7 +204,6 @@ You can also use [examples/0\_5\_setup\_account\_as.py](examples/0_5_setup_accou
 Accounts can be both producers and consumers, so you may wish to run this step against the account used above. You may also have Accounts that are Consumer only, and cannot create data shares. This step is only run once per AWS Account and must be run using credentials that have AdministratorAccess as well as being Lake Formation Data Lake Admin:
 
 ```python
-import logging
 from data_mesh_util.lib.constants import *
 from data_mesh_util import DataMeshMacros as data_mesh_macros
 
@@ -244,7 +243,6 @@ The above Steps 1.1 and 1.2 can be run for any number of accounts that you requi
 Creating a data product replicates Glue Catalog metadata from the Producer's account into the Data Mesh account, while leaving the source storage at rest within the Producer. The data mesh objects are shared back to the Producer account to enable local control without accessing the data mesh. Data Products can be created from Glue Catalog Databases or one-or-more Tables, but all permissions are managed at Table level. Producers can run this as many times as they require. To create a data product:
 
 ```python
-import logging
 from data_mesh_util import DataMeshProducer as dmp
 
 data_mesh_account = 'insert data mesh account number here'
@@ -285,12 +283,13 @@ data_mesh_producer.create_data_products(
 
 You can also use [examples/1\_create\_data\_product.py](examples/1_create_data_product.py) as an example to build your own application.
 
+Please note that upon creation of a data product, you will see a new Database and Table created in the Data Mesh Account, and this Database and Table have been shared back to the producer AWS Account using Resource Access Manager (RAM). Your producer Account may now be able to query data both from within the data mesh and from their own account, but the security Principal used for Data Mesh Utils may require additional permissions to use Athena or other query services.
+
 ### Step 3: Request access to a Data Product Table
 
 As a consumer, you can gain view public metadata by assuming the `DataMeshReadOnly` role in the mesh account. You can then create an access request for data products using:
 
 ```python
-import logging
 from data_mesh_util import DataMeshConsumer as dmc
 
 data_mesh_account = 'insert data mesh account number here'
@@ -303,7 +302,6 @@ consumer_credentials = {
 }
 data_mesh_consumer = dmp.DataMeshConsumer(
     data_mesh_account_id=data_mesh_account,
-    log_level=logging.DEBUG,
     region_name=aws_region,
     use_credentials=consumer_credentials
 )
@@ -329,7 +327,6 @@ You can also use [examples/2\_consumer\_request\_access.py](examples/2_consumer_
 In this step, you will grant permissions to the Consumer who has requested access:
 
 ```python
-import logging
 from data_mesh_util import DataMeshProducer as dmp
 
 data_mesh_account = 'insert data mesh account number here'
@@ -342,7 +339,6 @@ producer_credentials = {
 }
 data_mesh_producer = dmp.DataMeshProducer(
    data_mesh_account_id=data_mesh_account,
-   log_level=logging.DEBUG,
    region_name=aws_region,
 	use_credentials=producer_credentials
 )
@@ -381,7 +377,6 @@ You can also use [examples/3\_grant\_data\_product\_access.py](examples/3_grant_
 Permissions have been granted, but the Consumer must allow those grants to be imported into their account:
 
 ```python
-import logging
 from data_mesh_util import DataMeshConsumer as dmc
 
 data_mesh_account = 'insert data mesh account number here'
@@ -394,7 +389,6 @@ consumer_credentials = {
 }
 data_mesh_consumer = dmp.DataMeshConsumer(
     data_mesh_account_id=data_mesh_account,
-    log_level=logging.DEBUG,
     region_name=aws_region,
     use_credentials=consumer_credentials
 )

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = aws-data-mesh-utils
-version = 1.0.14
+version = 1.0.15
 author = Ian Meyers
 author_email = [email protected]
 license = Apache 2.0

src/data_mesh_util/DataMeshConsumer.py

@@ -136,13 +136,30 @@ def finalize_subscription(self, subscription_id: str) -> None:
             source_account=self._data_mesh_account_id
         )
 
-        self._consumer_automator.accept_pending_lf_resource_shares(
-            sender_account=self._data_mesh_account_id
-        )
+        shares = []
+        for k, v in subscription.get(RAM_SHARES).items():
+            shares.append(v.get('arn'))
+
+        # accept the RAM shares attached to the subscription
+        accepted, active, not_found = self._consumer_automator.accept_lf_resource_shares(
+            share_list=shares)
+
+        if len(accepted) > 0:
+            self._logger.info(f"Accepted {len(accepted)} RAM Shares: {str(accepted)}")
+
+        if len(active) > 0:
+            self._logger.info(
+                f"{len(active)} RAM Shares already in Active state")
+
+        if len(not_found) > 0:
+            self._logger.warning(
+                f"Unable to resolve {len(not_found)} RAM Shares: {str(not_found)}")
 
+        # mark the subscription as finalized
         self._subscription_tracker.mark_subscription_as_imported(
             subscription_id=subscription_id
         )
+        self._logger.info(f"Subscription Import Complete")
 
     def get_subscription(self, request_id: str) -> dict:
         return self._subscription_tracker.get_subscription(subscription_id=request_id)

src/data_mesh_util/lib/ApiAutomator.py

@@ -219,6 +219,13 @@ def configure_iam(self, policy_name: str, policy_desc: str, policy_template: str
 
         self._logger.debug("Waiting for User to be ready for inclusion in AssumeRolePolicy")
 
+        # attach the data access policy to the group
+        iam_client.attach_group_policy(
+            GroupName=group_name,
+            PolicyArn=policy_arn
+        )
+        self._logger.info(f"Attached Policy {policy_arn} to Group {group_name}")
+
         role_created = False
         retries = 0
         while role_created is False and retries < 5:
@@ -265,7 +272,7 @@ def configure_iam(self, policy_name: str, policy_desc: str, policy_template: str
                     PolicyArn=policy_arn
                 )
                 policy_attached = True
-                self._logger.info(f"Attached Policy {policy_arn} to {role_name}")
+                self._logger.info(f"Attached Policy {policy_arn} to Role {role_name}")
             except iam_client.exceptions.MalformedPolicyDocumentException as mpde:
                 if "Invalid principal" in str(mpde):
                     # this is raised when something within IAM hasn't yet propagated correctly.
@@ -1147,14 +1154,54 @@ def add_bucket_policy_entry(self, principal_account: str, access_path: str):
         # put the policy back into the bucket store
         s3_client.put_bucket_policy(Bucket=bucket_name, Policy=json.dumps(new_policy))
 
+    def accept_lf_resource_shares(self, share_list: list) -> tuple:
+        '''
+        Causes a list of RAM shares to be accepted by the caller
+        :param share_list:
+        :return:
+        '''
+        ram_client = self._get_client('ram')
+
+        get_response = ram_client.get_resource_share_invitations()
+
+        # only accept peding lakeformation shares from the source account
+        shares_accepted = []
+        shares_active = []
+        shares_not_found = []
+        for r in get_response.get('resourceShareInvitations'):
+            share_arn = r.get('resourceShareArn')
+            if share_arn in share_list:
+                if r.get('status') == 'PENDING':
+                    ram_client.accept_resource_share_invitation(
+                        resourceShareInvitationArn=r.get('resourceShareInvitationArn')
+                    )
+                    shares_accepted.append(share_arn)
+                    self._logger.info(f"Accepted RAM Share {share_arn}")
+                elif r.get('status') == 'ACCEPTED':
+                    shares_active.append(share_arn)
+                else:
+                    shares_not_found.append(share_arn)
+            else:
+                shares_not_found.append(share_arn)
+
+        return shares_accepted, shares_active, shares_not_found
+
     def accept_pending_lf_resource_shares(self, sender_account: str, filter_resource_arn: str = None):
+        '''
+        Accepts all pending resource shares of the specified type from the sending account. Used to automatically accept
+        shares from the data mesh back to the producer
+
+        :param sender_account:
+        :param filter_resource_arn:
+        :return:
+        '''
         ram_client = self._get_client('ram')
 
         get_response = ram_client.get_resource_share_invitations()
 
         accepted_share = False
         for r in get_response.get('resourceShareInvitations'):
-            # only accept peding lakeformation shares from the source account
+            # only accept pending lakeformation shares from the source account
             if r.get('senderAccountId') == sender_account and 'LakeFormation' in r.get('resourceShareName') and r.get(
                     'status') == 'PENDING':
                 if filter_resource_arn is None or r.get('resourceShareArn') == filter_resource_arn:

src/data_mesh_util/resource/producer_account_policy.pystache

@@ -21,10 +21,10 @@
                 "glue:GetDatabase",
                 "glue:CreateDatabase",
                 "glue:CreateTable",
-                "glue:CreateTables",
                 "glue:CreateCrawler",
                 "glue:Update*",
-                "glue:TagResource"
+                "glue:TagResource",
+                "glue:SearchTables"
             ],
             "Resource": "*"
         },