Simplify lakeformation shares - single shared_db and single resource link table (#1016)

### Feature or Bugfix
- Feature

### Detail
Implements design specified in #746. Multiple share requests share a
single shared Glue database instead of a shared database per share
request. Same for resource link tables, multiple shares in the same
target account get permissions to the same resource link table.

In addition, this PR includes:
- Unified LakeFormation processor for same account and cross account
shares (remove duplicated code)
- Code clean-up of Lake Formation manager and client: remove duplicates,
added more error handling
- Ensure "no-failures" if a resource already exists or is already
deleted or if a permission is already granted or revoked
- Redesigned tests and increased coverage

#### Upgrade LakeFormation version
LakeFormation v3
([docs](https://docs.aws.amazon.com/lake-formation/latest/dg/optimize-ram.html))
simplifies cross-account sharing requests. We want to see if there are
advantages of implementing the capability of LakeFormation to define the
external IAM role directly, instead of using first an AWS account and
then the role. In addition, v4 introduces LF hybrid mode. We are not
using it at the moment, but it can be a great addition in future
features. Because we already distinguish between "old" and "new" shares,
it is relatively easy to upgrade to a new LF version without affecting
"old" shares.

**Does it make sense to upgrade the sharing mechanism?**
- Is it possible technically for all principals? --> No, which is a
blocker
```Could not grant principal arn:aws:quicksight:us-east-1:11111111111:group/default/dataall permissions ['DESCRIBE', 'SELECT'] to {'Table': {'DatabaseName': 'dataall_imported_c3_tvnews81', 'Name': 'children_books_raw', 'CatalogId': '2222222222222'}}  due to: An error occurred (InvalidInputException) when calling the GrantPermissions operation: Cross account requests are only allowed for AWS Accounts, Organizations, IAM Principals and All IAMPrincipals```

- What actions are simplified? In this [commit ](f5b9bee) we can see how the final implementation would look like. 

**How does the migration look like?**

1. Upgrade LF version: It is possible to change the LF cross account settings programmatically using [put_data_lake_settings](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lakeformation/client/put_data_lake_settings.html) API. We already use this API in a custom resource of the environment stack. We could introduce the LF version as part of that call and make sure that it is executed on update (not only on create). 
2. Modify code in sharing task: for old_shares use the AWS account and for new shares use the IAM roles as principals of the grant. for old_shares execute additional grant for principals in target, for new shares skip. See [commit ](f5b9bee).
3. Backwards compatibility for shares done with V2:
  - At the moment the code is meant to work for versions >= v2. v4 is very recent and most AWS accounts use v3
  - Grant: If a RAM request is v2 and the LF version is upgraded to v3, new tables added to the RAM invitation can use v3 underlying code (I tested one table shared with old v2 procedure and added one table to the request with new code changes and it succeeds. In source the principals are the target IAM roles not the target account)
  - Revoke: not tested

**Conclusion**
- We cannot implement LF v3 direct sharing with principals because it does not support Quicksight groups.
- I think that setting the LF version in the environment stack is a good idea regardless of it being explicitly used It gives us control to set the latest version in the accounts we use. This is a feature that should be part of a separate PR.
- v2 RAM invitations still work with v3 direct-IAM role code. Therefore we could keep using v2 shares without the need to inspect shares and upgrade them. This means that the upgrade is pretty seamless, and we can take it at a later time when the Quicksight blocker is resolved.


Issues resolved:
- [X] Issue: deletion of new "shared" database does not happen when other old shares are still present in environment account.
- [X] ~[Pre-existing issue]:  In cross-account shares revoke ALL does not revoke permissions to QS group (by-design)~
- [X]  [Pre-existing issue]: in same account requesters have access to original table. It is needed! If we remove access to the original table the resource link access to data does not work.
- [X] ~Issue: when revoking, permissions to the shared db are not revoked for principal~
- [X] ~Issue: shared db is not deleted when all shares are revoked~
- [X] ~Issue: duplication of LF permissions. Pivot role and QS group were granted permissions on the original db and on the shared db with every new share~
- [ ]  [Pre-existing issue]: orphan LF resources on failed shares. If a share fails it can be deleted from data.all UI but there are resources that are created in LF that require manual clean-up (ugly operation assuming pivot role) 
- [X]  ~Changes in UI: data consumption details~


### Next steps and remarks
- Because there are shared resources between shares, it is possible that parallel shares on the same tables have conflicts: e.g. 2 shares trying to create a resource link simultaneously. I have tried testing with shares approved in a matter of seconds and did not run into issues (at the end a look-up is way faster than updating a policy). But we should keep an eye on this.
- Upgrade Lake Formation version. Not possible at this moment, because it would not work for Quicksight groups. To be re-evaluated in the future.
- I think the orphan LF resources on failed shares can be done in a separate PR to limit the scope of this PR. 
- I tried to revoke the requester access to the original table in same account sharing, but it is not possible. For the requester to be able to access the underlying data of the resource link, it needs to have access to the table. 
- The issue "deletion of new "shared" database does not happen when other old shares are still present in environment account." has not a big impact, but I implemented a solution nevertheless.

### Relates
- #746 

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

Author: dlpzx

backend/dataall/base/utils/alarm_service.py

@@ -74,4 +74,3 @@ def publish_message_to_alarms_topic(self, subject, message):
                 return response
             except ClientError as e:
                 logger.error(f'Failed to deliver message due to: {e} ')
-                raise e

backend/dataall/base/utils/naming_convention.py

@@ -7,7 +7,7 @@ class NamingConventionPattern(Enum):
 
     S3 = {'regex': '[^a-zA-Z0-9-]', 'separator': '-', 'max_length': 63}
     IAM = {'regex': '[^a-zA-Z0-9-_]', 'separator': '-', 'max_length': 63}
-    GLUE = {'regex': '[^a-zA-Z0-9_]', 'separator': '_', 'max_length': 63}
+    GLUE = {'regex': '[^a-zA-Z0-9_]', 'separator': '_', 'max_length': 240}  # Limit 255 - 15 extra chars buffer
     GLUE_ETL = {'regex': '[^a-zA-Z0-9-]', 'separator': '-', 'max_length': 52}
     NOTEBOOK = {'regex': '[^a-zA-Z0-9-]', 'separator': '-', 'max_length': 63}
     MLSTUDIO_DOMAIN = {'regex': '[^a-zA-Z0-9-]', 'separator': '-', 'max_length': 63}

backend/dataall/modules/dataset_sharing/api/resolvers.py

@@ -182,18 +182,12 @@ def resolve_group(context: Context, source: ShareObject, **kwargs):
 def resolve_consumption_data(context: Context, source: ShareObject, **kwargs):
     if not source:
         return None
-    with context.engine.scoped_session() as session:
-        ds: Dataset = DatasetRepository.get_dataset_by_uri(session, source.datasetUri)
-        if ds:
-            S3AccessPointName = utils.slugify(
-                source.datasetUri + '-' + source.principalId,
-                max_length=50, lowercase=True, regex_pattern='[^a-zA-Z0-9-]', separator='-'
-            )
-            return {
-                's3AccessPointName': S3AccessPointName,
-                'sharedGlueDatabase': (ds.GlueDatabaseName + '_shared_' + source.shareUri)[:254] if ds else 'Not created',
-                's3bucketName': ds.S3BucketName,
-            }
+    return ShareObjectService.resolve_share_object_consumption_data(
+        uri=source.shareUri,
+        datasetUri=source.datasetUri,
+        principalId=source.principalId,
+        environmentUri=source.environmentUri
+    )
 
 
 def resolve_share_object_statistics(context: Context, source: ShareObject, **kwargs):

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

@@ -16,16 +16,26 @@ def __init__(self, account_id, region, database):
 
     def create_database(self, location):
         try:
-            existing_database = self.database_exists()
+            log.info(
+                f'Creating database {self._database} '
+                f'in account {self._account_id}...'
+            )
+            existing_database = self.get_glue_database()
             if existing_database:
                 glue_database_created = True
             else:
                 self._create_glue_database(location)
                 glue_database_created = True
+            log.info(
+                f'Successfully created database {self._database}'
+                f'in account {self._account_id}'
+            )
             return glue_database_created
         except ClientError as e:
             log.error(
-                f'Failed to create database {self._database} on account {self._account_id} due to {e}'
+                f'Failed to create database {self._database} '
+                f'in account {self._account_id} '
+                f'due to {e}'
             )
             raise e
 
@@ -39,89 +49,141 @@ def _create_glue_database(self, location):
             }
             if location:
                 db_input['LocationUri'] = location
-            log.info(f'Creating Glue database with input: {db_input}')
             response = self._client.create_database(CatalogId=self._account_id, DatabaseInput=db_input)
-            log.info(f'response Create Database: {response}')
             return response
         except ClientError as e:
-            log.debug(f'Failed to create database {database}', e)
             raise e
 
-    def database_exists(self):
+    def get_glue_database(self):
+        try:
+            log.info(
+                f'Getting database {self._database} '
+                f'in account {self._account_id}...'
+            )
+            database = self._client.get_database(CatalogId=self._account_id, Name=self._database)
+            return database
+        except ClientError:
+            log.info(f'Database {self._database} not found in account {self._account_id}')
+            return False
+
+    def database_exists(self, database_name):
         try:
-            self._client.get_database(CatalogId=self._account_id, Name=self._database)
+            log.info(
+                f'Check database exists {self._database} '
+                f'in account {self._account_id}...'
+            )
+            self._client.get_database(CatalogId=self._account_id, Name=database_name)
             return True
         except ClientError:
-            log.info(f'Database {self._database} does not exist on account {self._account_id}...')
+            log.info(f'Database {database_name} not found in account {self._account_id}')
             return False
 
     def table_exists(self, table_name):
         try:
+            log.info(
+                f'Check table exists {table_name} '
+                f'in database {self._database} '
+                f'in account {self._account_id}...'
+            )
             table = (
                 self._client.get_table(
                     CatalogId=self._account_id, DatabaseName=self._database, Name=table_name
                 )
             )
-            log.info(f'Glue table found: {table_name}')
+            log.info(f'Glue table {table_name} not found in account {self._account_id} in database {self._database}')
             return table
         except ClientError:
             log.info(f'Glue table not found: {table_name}')
             return None
 
     def delete_table(self, table_name):
         database = self._database
-        log.info(
-            'Deleting table {} in database {}'.format(
-                table_name, database
-            )
-        )
-        response = self._client.delete_table(
-            CatalogId=self._account_id,
-            DatabaseName=database,
-            Name=table_name
-        )
-
-        return response
+        try:
+            log.info(
+                f'Deleting table {table_name} '
+                f'in database {self._database} '
+                f'in catalog {self._account_id}...'
+            )
+            response = self._client.delete_table(
+                CatalogId=self._account_id,
+                DatabaseName=database,
+                Name=table_name
+            )
+            log.info(
+                f'Successfully deleted table {table_name} '
+                f'in database {database} '
+                f'in catalog {self._account_id} '
+                f'response: {response}'
+            )
+            return response
+        except ClientError as e:
+            log.error(
+                f'Could not delete table {table_name} '
+                f'in database {database} '
+                f'in catalog {self._account_id} '
+                f'due to: {e}'
+            )
+            raise e
 
-    def create_resource_link(self, resource_link_name, resource_link_input):
+    def create_resource_link(self, resource_link_name, table, catalog_id):
         account_id = self._account_id
-        database = self._database
+        shared_database = self._database
+        resource_link_input = {
+            'Name': table.GlueTableName,
+            'TargetTable': {
+                'CatalogId': catalog_id,
+                'DatabaseName': table.GlueDatabaseName,
+                'Name': table.GlueTableName,
+            },
+        }
 
-        log.info(
-            f'Creating ResourceLink {resource_link_name} in database {account_id}://{database}'
-        )
         try:
+            log.info(
+                f'Creating ResourceLink {resource_link_name}  '
+                f'in database {shared_database}...'
+            )
             resource_link = self.table_exists(resource_link_name)
             if resource_link:
                 log.info(
-                    f'ResourceLink {resource_link_name} already exists in database {account_id}://{database}'
+                    f'ResourceLink {resource_link_name} '
+                    f'in database {account_id}://{shared_database}'
+                    f'already exists: {resource_link}'
                 )
             else:
                 resource_link = self._client.create_table(
                     CatalogId=account_id,
-                    DatabaseName=database,
+                    DatabaseName=shared_database,
                     TableInput=resource_link_input,
                 )
                 log.info(
-                    f'Successfully created ResourceLink {resource_link_name} in database {account_id}://{database}'
+                    f'Successfully created ResourceLink {resource_link_name} '
+                    f'in database {account_id}://{shared_database} '
+                    f'response: {resource_link}'
                 )
             return resource_link
         except ClientError as e:
             log.error(
                 f'Could not create ResourceLink {resource_link_name} '
-                f'in database {account_id}://{database} '
+                f'in database {account_id}://{shared_database} '
                 f'due to: {e}'
             )
             raise e
 
     def delete_database(self):
         account_id = self._account_id
         database = self._database
-
-        log.info(f'Deleting database {account_id}://{database} ...')
         try:
-            if self.database_exists():
+            log.info(
+                f'Deleting database {self._database} '
+                f'in account {self._account_id}...'
+            )
+            existing_database = self.get_glue_database()
+            if existing_database:
                 self._client.delete_database(CatalogId=account_id, Name=database)
+            log.info(
+                f'Successfully deleted database {database} '
+                f'in account {account_id}'
+            )
             return True
         except ClientError as e:
             log.error(