Refactor: uncouple datasets and dataset_sharing modules - part 3 (#1186)

### Feature or Bugfix
- Refactoring

### Detail
Remove and move logic from dataset to datasets_sharing module. This is
needed as explained in full PR [AFTER 2.4] Refactor: uncouple datasets
and dataset_sharing modules
#1179
- [X] Split the KMS client and create one dedicated KMS client for
datasets

### Relates
#1179

### Security
Please answer the questions below briefly where applicable, or write
`N/A`. Based on
[OWASP 10](https://owasp.org/Top10/en/). `N/A`

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

backend/dataall/modules/datasets/aws/kms_dataset_client.py

@@ -0,0 +1,52 @@
+import logging
+
+from dataall.base.aws.sts import SessionHelper
+from botocore.exceptions import ClientError
+
+
+log = logging.getLogger(__name__)
+
+
+class KmsClient:
+    _DEFAULT_POLICY_NAME = 'default'
+
+    def __init__(self, account_id: str, region: str):
+        session = SessionHelper.remote_session(accountid=account_id, region=region)
+        self._client = session.client('kms', region_name=region)
+        self._account_id = account_id
+        self.region = region
+
+    def get_key_id(self, key_alias: str):
+        # The same client function is defined in the data_sharing module. Duplication is allowed to avoid coupling.
+        try:
+            response = self._client.describe_key(
+                KeyId=key_alias,
+            )
+        except ClientError as e:
+            if e.response['Error']['Code'] == 'AccessDenied':
+                raise Exception(
+                    f'Data.all Environment Pivot Role does not have kms:DescribeKey Permission for key {key_alias}: {e}'
+                )
+            log.error(f'Failed to get kms key id of {key_alias}: {e}')
+            return None
+        else:
+            return response['KeyMetadata']['KeyId']
+
+    def check_key_exists(self, key_alias: str):
+        try:
+            key_exist = False
+            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:
+                    key_exist = True
+                    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 in account {self._account_id}: {e}'
+                )
+            log.error(f'Failed to list KMS key aliases in account {self._account_id}: {e}')
+            return None
+        else:
+            return key_exist

backend/dataall/modules/datasets/services/dataset_service.py

@@ -10,7 +10,7 @@
 from dataall.core.stacks.services.stack_service import StackService
 from dataall.core.tasks.service_handlers import Worker
 from dataall.base.aws.sts import SessionHelper
-from dataall.modules.dataset_sharing.aws.kms_client import KmsClient
+from dataall.modules.datasets.aws.kms_dataset_client import KmsClient
 from dataall.base.context import get_context
 from dataall.core.permissions.services.group_policy_service import GroupPolicyService
 from dataall.core.environment.services.environment_service import EnvironmentService