Introduce check for IAM actions in share_verify bucket and access poi…

…nts + reapply with list of allowed actions (#1407)

### Feature or Bugfix
<!-- please choose -->
- Bugfix

### Detail
- Share verifier now detects "s3:*" permission as an error 
- Share reapplied with enforced valid actions

### Relates
- <URL or Ticket>

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

---------

Co-authored-by: Sofia Sazonova <sazonova@amazon.co.uk>

backend/dataall/modules/s3_datasets_shares/services/s3_share_managed_policy_service.py

@@ -17,6 +17,8 @@
 IAM_S3_BUCKETS_STATEMENT_SID = 'BucketStatement'
 EMPTY_STATEMENT_SID = 'EmptyStatement'
 
+S3_ALLOWED_ACTIONS = ['s3:List*', 's3:Describe*', 's3:GetObject']
+
 
 class S3SharePolicyService(ManagedPolicy):
     def __init__(self, role_name, account, region, environmentUri, resource_prefix):
@@ -63,14 +65,10 @@ def add_missing_resources_to_policy_statement(
         :return
         """
         policy_name = self.generate_policy_name()
+        policy_actions = S3_ALLOWED_ACTIONS if resource_type == 's3' else [f'{resource_type}:*']
         index = self._get_statement_by_sid(policy_document, statement_sid)
         if index is None:
             log.info(f'{statement_sid} does NOT exists for Managed policy {policy_name} ' f'creating statement...')
-            policy_actions = (
-                [f'{resource_type}:List*', f'{resource_type}:Describe*', f'{resource_type}:GetObject']
-                if resource_type == 's3'
-                else [f'{resource_type}:*']
-            )
             additional_policy = {
                 'Sid': statement_sid,
                 'Effect': 'Allow',
@@ -79,6 +77,8 @@ def add_missing_resources_to_policy_statement(
             }
             policy_document['Statement'].append(additional_policy)
         else:
+            # Enforce, that actions are valid
+            policy_document['Statement'][index]['Action'] = policy_actions
             for target_resource in target_resources:
                 if target_resource not in policy_document['Statement'][index]['Resource']:
                     log.info(
@@ -128,6 +128,23 @@ def check_resource_in_policy_statement(target_resources: list, existing_policy_s
                 return False
         return True
 
+    @staticmethod
+    def check_s3_actions_in_policy_statement(existing_policy_statement: dict) -> (bool, str, str):
+        """
+        Checks if all required s3 actions are allowed in the existing policy and there is no unallowed actions
+        :param existing_policy_statement:
+        :return: bool, allowed missing actions string, not allowed actions string
+        """
+        statement_actions = set(existing_policy_statement['Action'])
+        allowed_actions = set(S3_ALLOWED_ACTIONS)
+        missing_actions = allowed_actions - statement_actions
+        extra_actions = statement_actions - allowed_actions
+        return (
+            not (missing_actions or extra_actions),
+            ','.join(missing_actions),
+            ','.join(extra_actions),
+        )
+
     @staticmethod
     def _get_statement_by_sid(policy, sid):
         for index, statement in enumerate(policy['Statement']):

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

@@ -226,6 +226,34 @@ def check_target_role_access_policy(self) -> None:
                     f'{self.bucket_name}',
                 )
             )
+        else:
+            policy_check, missing_permissions, extra_permissions = (
+                share_policy_service.check_s3_actions_in_policy_statement(
+                    existing_policy_statement=policy_document['Statement'][s3_statement_index]
+                )
+            )
+            if not policy_check:
+                logger.info(f'IAM Policy Statement {IAM_S3_ACCESS_POINTS_STATEMENT_SID}S3 has invalid actions')
+                if missing_permissions:
+                    self.folder_errors.append(
+                        ShareErrorFormatter.missing_permission_error_msg(
+                            self.target_requester_IAMRoleName,
+                            'IAM Policy Action',
+                            missing_permissions,
+                            'S3 Bucket',
+                            f'{self.bucket_name}',
+                        )
+                    )
+                if extra_permissions:
+                    self.folder_errors.append(
+                        ShareErrorFormatter.not_allowed_permission_error_msg(
+                            self.target_requester_IAMRoleName,
+                            'IAM Policy Action',
+                            extra_permissions,
+                            'S3 Bucket',
+                            f'{self.bucket_name}',
+                        )
+                    )
 
         if kms_key_id:
             kms_statement_index = S3SharePolicyService._get_statement_by_sid(

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

@@ -118,7 +118,7 @@ def check_s3_iam_access(self) -> None:
             existing_policy_statement=policy_document['Statement'][s3_statement_index],
         ):
             logger.info(
-                f'IAM Policy Statement {IAM_S3_BUCKETS_STATEMENT_SID}KMS does not contain resources {s3_target_resources}'
+                f'IAM Policy Statement {IAM_S3_BUCKETS_STATEMENT_SID}S3 does not contain resources {s3_target_resources}'
             )
             self.bucket_errors.append(
                 ShareErrorFormatter.missing_permission_error_msg(
@@ -129,14 +129,42 @@ def check_s3_iam_access(self) -> None:
                     f'{self.bucket_name}',
                 )
             )
+        else:
+            policy_check, missing_permissions, extra_permissions = (
+                share_policy_service.check_s3_actions_in_policy_statement(
+                    existing_policy_statement=policy_document['Statement'][s3_statement_index]
+                )
+            )
+            if not policy_check:
+                logger.info(f'IAM Policy Statement {IAM_S3_BUCKETS_STATEMENT_SID}S3 has invalid actions')
+                if missing_permissions:
+                    self.bucket_errors.append(
+                        ShareErrorFormatter.missing_permission_error_msg(
+                            self.target_requester_IAMRoleName,
+                            'IAM Policy Action',
+                            missing_permissions,
+                            'S3 Bucket',
+                            f'{self.bucket_name}',
+                        )
+                    )
+                if extra_permissions:
+                    self.bucket_errors.append(
+                        ShareErrorFormatter.not_allowed_permission_error_msg(
+                            self.target_requester_IAMRoleName,
+                            'IAM Policy Action',
+                            extra_permissions,
+                            'S3 Bucket',
+                            f'{self.bucket_name}',
+                        )
+                    )
 
         if kms_key_id:
             kms_statement_index = S3SharePolicyService._get_statement_by_sid(
                 policy_document, f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS'
             )
             kms_target_resources = [f'arn:aws:kms:{self.bucket_region}:{self.source_account_id}:key/{kms_key_id}']
             if kms_statement_index is None:
-                logger.info(f'IAM Policy Statement {IAM_S3_BUCKETS_STATEMENT_SID}S3 does not exist')
+                logger.info(f'IAM Policy Statement {IAM_S3_BUCKETS_STATEMENT_SID}KMS does not exist')
                 self.bucket_errors.append(
                     ShareErrorFormatter.missing_permission_error_msg(
                         self.target_requester_IAMRoleName,

backend/dataall/modules/s3_datasets_shares/services/share_managers/share_manager_utils.py

@@ -19,3 +19,9 @@ def missing_permission_error_msg(requestor, permission_type, permissions, resour
         requestor = ShareErrorFormatter._stringify(requestor)
         permissions = ShareErrorFormatter._stringify(permissions)
         return f'Requestor {requestor} missing {permission_type} permissions: {permissions} for {resource_type} Target: {target_resource}'
+
+    @staticmethod
+    def not_allowed_permission_error_msg(requestor, permission_type, permissions, resource_type, target_resource):
+        requestor = ShareErrorFormatter._stringify(requestor)
+        permissions = ShareErrorFormatter._stringify(permissions)
+        return f'Requestor {requestor} has not allowed {permission_type} permissions: {permissions} for {resource_type} Target: {target_resource}'

tests/modules/s3_datasets_shares/tasks/test_s3_access_point_share_manager.py

@@ -10,7 +10,10 @@
 from dataall.core.organizations.db.organization_models import Organization
 from dataall.modules.s3_datasets_shares.aws.s3_client import S3ControlClient
 from dataall.modules.shares_base.db.share_object_models import ShareObject, ShareObjectItem
-from dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service import S3SharePolicyService
+from dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service import (
+    S3SharePolicyService,
+    S3_ALLOWED_ACTIONS,
+)
 from dataall.modules.s3_datasets_shares.services.share_managers import S3AccessPointShareManager
 from dataall.modules.s3_datasets.db.dataset_models import DatasetStorageLocation, S3Dataset
 from dataall.modules.shares_base.services.sharing_service import ShareData
@@ -237,7 +240,7 @@ def _create_target_dataset_access_control_policy(bucket_name, access_point_name)
             {
                 'Sid': f'{IAM_S3_ACCESS_POINTS_STATEMENT_SID}S3',
                 'Effect': 'Allow',
-                'Action': ['s3:*'],
+                'Action': S3_ALLOWED_ACTIONS,
                 'Resource': [
                     f'arn:aws:s3:::{bucket_name}',
                     f'arn:aws:s3:::{bucket_name}/*',
@@ -350,7 +353,7 @@ def test_grant_target_role_access_policy_test_empty_policy(
             {
                 'Sid': f'{IAM_S3_ACCESS_POINTS_STATEMENT_SID}S3',
                 'Effect': 'Allow',
-                'Action': ['s3:List*', 's3:Describe*', 's3:GetObject'],
+                'Action': S3_ALLOWED_ACTIONS,
                 'Resource': [
                     f'arn:aws:s3:::{location1.S3BucketName}',
                     f'arn:aws:s3:::{location1.S3BucketName}/*',
@@ -1232,6 +1235,53 @@ def test_check_target_role_access_policy(mocker, share_manager):
     assert len(share_manager.folder_errors) == 0
 
 
+def test_check_target_role_access_policy_wrong_permissions(mocker, share_manager):
+    # Given
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=True,
+    )
+
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_attached',
+        return_value=True,
+    )
+
+    bad_policy = _create_target_dataset_access_control_policy(
+        share_manager.bucket_name, share_manager.access_point_name
+    )
+    bad_policy['Statement'][0]['Action'] = ['s3:*']
+
+    iam_get_policy_mock = mocker.patch(
+        'dataall.base.aws.iam.IAM.get_managed_policy_default_version',
+        return_value=(
+            'v1',
+            bad_policy,
+        ),
+    )
+
+    mocker.patch(
+        'dataall.base.aws.iam.IAM.get_role_arn_by_name',
+        side_effect=lambda account_id, region, role_name: f'arn:aws:iam::{account_id}:role/{role_name}',
+    )
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'some-key-2112'
+
+    # When
+    share_manager.check_target_role_access_policy()
+    # Then
+    iam_get_policy_mock.assert_called()
+    kms_client().get_key_id.assert_called()
+    assert len(share_manager.folder_errors) == 2
+    message_missing = 'missing IAM Policy Action permissions:'
+    message_extra = 'has not allowed IAM Policy Action permissions: s3:*'
+    assert message_missing in share_manager.folder_errors[0]
+    for action in S3_ALLOWED_ACTIONS:
+        assert action in share_manager.folder_errors[0]
+    assert message_extra in share_manager.folder_errors[1]
+
+
 def test_check_target_role_access_policy_existing_policy_bucket_and_key_not_included(mocker, share_manager):
     # Given
     mocker.patch(

tests/modules/s3_datasets_shares/tasks/test_s3_bucket_share_manager.py

@@ -9,7 +9,10 @@
 from dataall.core.organizations.db.organization_models import Organization
 from dataall.modules.shares_base.db.share_object_models import ShareObject
 from dataall.modules.s3_datasets_shares.services.share_managers import S3BucketShareManager
-from dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service import S3SharePolicyService
+from dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service import (
+    S3SharePolicyService,
+    S3_ALLOWED_ACTIONS,
+)
 from dataall.modules.s3_datasets.db.dataset_models import S3Dataset, DatasetBucket
 from dataall.modules.shares_base.services.sharing_service import ShareData
 
@@ -1385,7 +1388,7 @@ def test_check_s3_iam_access(mocker, dataset2, share2_manager):
             {
                 'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S3',
                 'Effect': 'Allow',
-                'Action': ['s3:*'],
+                'Action': ['s3:List*', 's3:Describe*', 's3:GetObject'],
                 'Resource': [f'arn:aws:s3:::{dataset2.S3BucketName}', f'arn:aws:s3:::{dataset2.S3BucketName}/*'],
             },
             {
@@ -1418,6 +1421,55 @@ def test_check_s3_iam_access(mocker, dataset2, share2_manager):
     assert (len(share2_manager.bucket_errors)) == 0
 
 
+def test_check_s3_iam_access_wrong_actions(mocker, dataset2, share2_manager):
+    # Given policy with some other bucket as resource
+    # Check if the correct resource is attached/appended
+
+    policy = {
+        'Version': '2012-10-17',
+        'Statement': [
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}S3',
+                'Effect': 'Allow',
+                'Action': ['s3:*'],
+                'Resource': [f'arn:aws:s3:::{dataset2.S3BucketName}', f'arn:aws:s3:::{dataset2.S3BucketName}/*'],
+            },
+            {
+                'Sid': f'{IAM_S3_BUCKETS_STATEMENT_SID}KMS',
+                'Effect': 'Allow',
+                'Action': ['kms:*'],
+                'Resource': [f'arn:aws:kms:{dataset2.region}:{dataset2.AwsAccountId}:key/kms-key'],
+            },
+        ],
+    }
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_exists',
+        return_value=True,
+    )
+    mocker.patch(
+        'dataall.modules.s3_datasets_shares.services.s3_share_managed_policy_service.S3SharePolicyService.check_if_policy_attached',
+        return_value=True,
+    )
+    # Gets policy with S3 and KMS
+    iam_update_role_policy_mock_1 = mocker.patch(
+        'dataall.base.aws.iam.IAM.get_managed_policy_default_version', return_value=('v1', policy)
+    )
+
+    kms_client = mock_kms_client(mocker)
+    kms_client().get_key_id.return_value = 'kms-key'
+
+    share2_manager.check_s3_iam_access()
+    # Then
+    iam_update_role_policy_mock_1.assert_called_once()
+    assert (len(share2_manager.bucket_errors)) == 2
+    message_missing = 'missing IAM Policy Action permissions:'
+    message_extra = 'has not allowed IAM Policy Action permissions: s3:*'
+    assert message_missing in share2_manager.bucket_errors[0]
+    for action in S3_ALLOWED_ACTIONS:
+        assert action in share2_manager.bucket_errors[0]
+    assert message_extra in share2_manager.bucket_errors[1]
+
+
 def test_check_s3_iam_access_no_policy(mocker, dataset2, share2_manager):
     # Given
     # There is not existing IAM policy in the requesters account for the dataset's S3bucket