feat(ec2): Add 2 new checks + fixers related with EC2 service (#3827)

Co-authored-by: Sergio <sergio@prowler.com>
2026-07-24 21:11:53 +00:00 · 2024-04-24 11:43:19 +02:00
parent 022b7ef756
commit acac3fc693
17 changed files with 805 additions and 15 deletions
@@ -0,0 +1,67 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from prowler.providers.aws.services.ec2.ec2_service import (
+    EbsSnapshotBlockPublicAccess,
+    InstanceMetadataDefaults,
+)
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+
+# Since moto does not support the ec2 metadata service, we need to mock the response for some functions
+def mock_get_instance_metadata_defaults(http_tokens, instances, region):
+    return InstanceMetadataDefaults(
+        http_tokens=http_tokens, instances=instances, region=region
+    )
+
+
+def mock_get_snapshot_block_public_access_state(status, snapshots, region):
+    return EbsSnapshotBlockPublicAccess(
+        status=status, snapshots=snapshots, region=region
+    )
+
+
+def mock_enable_snapshot_block_public_access(State):
+    return {"State": State}
+
+
+class Test_ec2_ebs_snapshot_account_block_public_access_fixer:
+    @mock_aws
+    def test_ec2_ebs_snapshot_account_block_public_access_fixer(self):
+        ec2_service = mock.MagicMock()
+        ec2_client = mock.MagicMock()
+        ec2_service.regional_clients = {AWS_REGION_US_EAST_1: ec2_client}
+
+        ec2_client.instance_metadata_defaults = [
+            mock_get_instance_metadata_defaults(
+                http_tokens="required", instances=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            mock_get_snapshot_block_public_access_state(
+                status="block-all-sharing", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+
+        ec2_client.enable_snapshot_block_public_access = (
+            mock_enable_snapshot_block_public_access
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access_fixer.ec2_client",
+            ec2_service,
+        ):
+
+            from prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access_fixer import (
+                fixer,
+            )
+
+            # By default, the account has not public access blocked
+            assert fixer(region=AWS_REGION_US_EAST_1)
@@ -0,0 +1,146 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_ec2_ebs_snapshot_account_block_public_access:
+    @mock_aws
+    def test_ec2_ebs_block_public_access_state_unblocked(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            EbsSnapshotBlockPublicAccess,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            EbsSnapshotBlockPublicAccess(
+                status="unblocked", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+        ec2_client.account_arn_template = (
+            f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+        )
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access.ec2_client",
+            new=ec2_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access import (
+                ec2_ebs_snapshot_account_block_public_access,
+            )
+
+            check = ec2_ebs_snapshot_account_block_public_access()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Public access is not blocked for EBS Snapshots."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+
+    @mock_aws
+    def test_ec2_ebs_block_public_access_state_block_new_sharing(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            EbsSnapshotBlockPublicAccess,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            EbsSnapshotBlockPublicAccess(
+                status="block-new-sharing", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+        ec2_client.account_arn_template = (
+            f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+        )
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access.ec2_client",
+            new=ec2_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access import (
+                ec2_ebs_snapshot_account_block_public_access,
+            )
+
+            check = ec2_ebs_snapshot_account_block_public_access()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Public access is blocked only for new EBS Snapshots."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+
+    @mock_aws
+    def test_ec2_ebs_block_public_access_state_block_all_sharing(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            EbsSnapshotBlockPublicAccess,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            EbsSnapshotBlockPublicAccess(
+                status="block-all-sharing", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+        ec2_client.account_arn_template = (
+            f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+        )
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access.ec2_client",
+            new=ec2_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access import (
+                ec2_ebs_snapshot_account_block_public_access,
+            )
+
+            check = ec2_ebs_snapshot_account_block_public_access()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "Public access is blocked for all EBS Snapshots."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
@@ -0,0 +1,68 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from prowler.providers.aws.services.ec2.ec2_service import (
+    EbsSnapshotBlockPublicAccess,
+    InstanceMetadataDefaults,
+)
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+
+# Since moto does not support the ec2 metadata service, we need to mock the response for some functions
+def mock_get_instance_metadata_defaults(http_tokens, instances, region):
+    return InstanceMetadataDefaults(
+        http_tokens=http_tokens, instances=instances, region=region
+    )
+
+
+def mock_get_snapshot_block_public_access_state(status, snapshots, region):
+    return EbsSnapshotBlockPublicAccess(
+        status=status, snapshots=snapshots, region=region
+    )
+
+
+def mock_modify_instance_metadata_defaults(HttpTokens):
+    if HttpTokens == "required":
+        return {"Return": True}
+
+
+class Test_ec2_instance_account_imdsv2_enabled_fixer:
+    @mock_aws
+    def test_ec2_instance_account_imdsv2_enabled_fixer(self):
+        ec2_service = mock.MagicMock()
+        ec2_client = mock.MagicMock()
+        ec2_service.regional_clients = {AWS_REGION_US_EAST_1: ec2_client}
+
+        ec2_client.instance_metadata_defaults = [
+            mock_get_instance_metadata_defaults(
+                http_tokens="required", instances=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            mock_get_snapshot_block_public_access_state(
+                status="block-all-sharing", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+
+        ec2_client.modify_instance_metadata_defaults = (
+            mock_modify_instance_metadata_defaults
+        )
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled_fixer.ec2_client",
+            ec2_service,
+        ):
+
+            from prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled_fixer import (
+                fixer,
+            )
+
+            # By default, the account has not public access blocked
+            assert fixer(region=AWS_REGION_US_EAST_1)
@@ -0,0 +1,102 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_ec2_instance_account_imdsv2_enabled:
+    @mock_aws
+    def test_ec2_imdsv2_required(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            InstanceMetadataDefaults,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.instance_metadata_defaults = [
+            InstanceMetadataDefaults(
+                http_tokens="required", instances=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+        ec2_client.account_arn_template = (
+            f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+        )
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled.ec2_client",
+            new=ec2_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled import (
+                ec2_instance_account_imdsv2_enabled,
+            )
+
+            check = ec2_instance_account_imdsv2_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "IMDSv2 is enabled by default for EC2 instances."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+
+    @mock_aws
+    def test_ec2_imdsv2_none(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            InstanceMetadataDefaults,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.instance_metadata_defaults = [
+            InstanceMetadataDefaults(
+                http_tokens=None, instances=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+
+        ec2_client.account_arn_template = (
+            f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+        )
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled.ec2_client",
+            new=ec2_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_instance_account_imdsv2_enabled.ec2_instance_account_imdsv2_enabled import (
+                ec2_instance_account_imdsv2_enabled,
+            )
+
+            check = ec2_instance_account_imdsv2_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "IMDSv2 is not enabled by default for EC2 instances."
+            )
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
@@ -3,6 +3,8 @@ import re
 from base64 import b64decode
 from datetime import datetime

+import botocore
+import mock
 from boto3 import client, resource
 from dateutil.tz import tzutc
 from freezegun import freeze_time
@@ -19,6 +21,8 @@ from tests.providers.aws.utils import (
 EXAMPLE_AMI_ID = "ami-12c6146b"
 MOCK_DATETIME = datetime(2023, 1, 4, 7, 27, 30, tzinfo=tzutc())

+orig = botocore.client.BaseClient._make_api_call
+

 class Test_EC2_Service:
    # Test EC2 Service
@@ -336,6 +340,106 @@ class Test_EC2_Service:
            if result.region == AWS_REGION_US_EAST_1:
                assert result.status

+    # Test EC2 get_snapshot_block_public_access_state
+    def test__get_snapshot_block_public_access_state__(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            EbsSnapshotBlockPublicAccess,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.ebs_block_public_access_snapshots_states = [
+            EbsSnapshotBlockPublicAccess(
+                status="block-all-sharing", snapshots=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_client.ec2_client",
+            new=ec2_client,
+        ):
+            assert (
+                ec2_client.ebs_block_public_access_snapshots_states[0].status
+                == "block-all-sharing"
+            )
+
+    # Test EC2 __get_resources_for_regions__
+    @mock_aws
+    def test__get_resources_for_regions__(self):
+        # Generate EC2 Client
+        ec2_resource = resource("ec2", region_name=AWS_REGION_US_EAST_1)
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        # Get AMI image
+        image_response = ec2_client.describe_images()
+        image_id = image_response["Images"][0]["ImageId"]
+        # Create EC2 Instances running
+        ec2_resource.create_instances(
+            MinCount=1,
+            MaxCount=1,
+            ImageId=image_id,
+        )
+        # Create Volume
+        volume_id = ec2_client.create_volume(
+            AvailabilityZone=AWS_REGION_US_EAST_1,
+            Encrypted=False,
+            Size=40,
+            TagSpecifications=[
+                {
+                    "ResourceType": "volume",
+                    "Tags": [
+                        {"Key": "test", "Value": "test"},
+                    ],
+                },
+            ],
+        )["VolumeId"]
+        ec2_client.create_snapshot(
+            VolumeId=volume_id,
+            TagSpecifications=[
+                {
+                    "ResourceType": "snapshot",
+                    "Tags": [
+                        {"Key": "test", "Value": "test"},
+                    ],
+                },
+            ],
+        )
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+        ec2 = EC2(aws_provider)
+        assert ec2.attributes_for_regions[AWS_REGION_US_EAST_1]["has_snapshots"]
+        assert ec2.attributes_for_regions[AWS_REGION_US_EAST_1]["has_instances"]
+        assert ec2.attributes_for_regions[AWS_REGION_US_EAST_1]["has_volumes"]
+
+    # Test __get_instance_metadata_defaults__
+    @mock_aws
+    def test__get_instance_metadata_defaults__(self):
+        from prowler.providers.aws.services.ec2.ec2_service import (
+            InstanceMetadataDefaults,
+        )
+
+        ec2_client = mock.MagicMock()
+        ec2_client.instance_metadata_defaults = [
+            InstanceMetadataDefaults(
+                http_tokens="required", instances=True, region=AWS_REGION_US_EAST_1
+            )
+        ]
+        ec2_client.audited_account = AWS_ACCOUNT_NUMBER
+        ec2_client.region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_client.ec2_client",
+            new=ec2_client,
+        ):
+            assert ec2_client.instance_metadata_defaults[0].http_tokens == "required"
+
    # Test EC2 Describe Addresses
    @mock_aws
    def test__describe_addresses__(self):