[Gh 904] Central Catalog Support (#1021)

### Feature or Bugfix
- Feature

### Detail

PR containing all the code raised in PR -
#905 + Unit Tests +
Addressing comments raised on that PR. Copy pasting details from PR -

Detect if the source database is a resource link
If it is a resource link, check that the catalog account has been
onboarded to data.all
Check for the presence of owner_account_id tag on the database
The tag needs to exist and the value has to match the account id of the
share approver

Credits - @blitzmohit 

## Testing 

Running Unit tests - ✅ 
Testing on AWS Deployed data.all instance with the Original PR -  ✅ 
Sanity testing after addressing comments - **[EDIT]** ✅ ( Testing done )

### Relates
- #904

### 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)? No
  - 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? No
- 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? No
  - 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? Yes
  - Have you used the least-privilege principle? How? Yes


By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license.

---------

Co-authored-by: trajopadhye <[email protected]>

backend/dataall/base/aws/sts.py

@@ -359,3 +359,15 @@ def generate_console_url(credentials, session_duration=None, region='eu-west-1',
 
         # Send final URL to stdout
         return request_url
+
+    @staticmethod
+    def is_assumable_pivot_role(accountid):
+        try:
+            SessionHelper.remote_session(accountid=accountid)
+        except ClientError as e:
+            log.error(f'Failed to assume dataall pivot role session in environment with account id {accountid} due to {e}')
+            return False
+        except Exception as e:
+            log.error(f'Unexpected error while assuming data.all pivot role in environment with account id {accountid} due to {e}')
+            return False
+        return True

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

@@ -13,6 +13,7 @@ def __init__(self, account_id, region, database):
         self._client = aws_session.client('glue', region_name=region)
         self._database = database
         self._account_id = account_id
+        self._region = region
 
     def create_database(self, location):
         try:
@@ -125,14 +126,14 @@ def delete_table(self, table_name):
             )
             raise e
 
-    def create_resource_link(self, resource_link_name, table, catalog_id):
+    def create_resource_link(self, resource_link_name, table, catalog_id, database):
         account_id = self._account_id
         shared_database = self._database
         resource_link_input = {
             'Name': table.GlueTableName,
             'TargetTable': {
                 'CatalogId': catalog_id,
-                'DatabaseName': table.GlueDatabaseName,
+                'DatabaseName': database,
                 'Name': table.GlueTableName,
             },
         }
@@ -192,3 +193,43 @@ def delete_database(self):
                 f'due to: {e}'
             )
             raise e
+
+    def get_source_catalog(self):
+        """ Get the source catalog account details """
+        try:
+            log.info(f'Fetching source catalog details for database {self._database}...')
+            response = self._client.get_database(CatalogId=self._account_id, Name=self._database)
+            linked_database = response.get('Database', {}).get('TargetDatabase', {})
+            log.info(f'Fetched source catalog details for database {self._database} are: {linked_database}...')
+            if linked_database:
+                return {'account_id' : linked_database.get('CatalogId'), 'database_name' : linked_database.get('DatabaseName'), 'region' : linked_database.get('Region', self._region)}
+
+        except self._client.exceptions.EntityNotFoundException as enoFnd:
+            log.exception(f'Could not fetch source catalog details for database {self._database} due to {enoFnd}')
+            raise enoFnd
+        except Exception as e:
+            log.exception(f'Error fetching source catalog details for database {self._database} due to {e}')
+            raise e
+        return None
+
+    def get_database_tags(self):
+        # Get tags from the glue database
+        account_id = self._account_id
+        database = self._database
+        region = self._region
+
+        try:
+            log.info(f'Getting tags for database {database}')
+            resource_arn = f'arn:aws:glue:{region}:{account_id}:database/{database}'
+            response = self._client.get_tags(ResourceArn=resource_arn)
+            tags = response['Tags']
+
+            log.info(f'Successfully retrieved tags: {tags}')
+
+            return tags
+        except self._client.exceptions.EntityNotFoundException as entNotFound:
+            log.exception(f'Could not get tags for database {database} due to {entNotFound}')
+            raise entNotFound
+        except Exception as e:
+            log.exception(f'Error fetching tags for {database} due to {e}')
+            raise e

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

@@ -64,7 +64,8 @@ def _accept_resource_share_invitation(self, resource_share_invitation_arn):
                 raise e
 
     @staticmethod
-    def accept_ram_invitation(source_account_id, source_region, target_account_id, target_region, source_database, source_table):
+    def accept_ram_invitation(source_account_id, source_region, source_database, source_table, target_account_id,
+                              target_region):
         """
         Accepts RAM invitations on the target account
         """