feat(vpc): Ensure Amazon EC2 Is Configured to Use VPC Endpoints Created for the Amazon EC2 Service (#4872)

prowler/providers/aws/services/vpc/vpc_endpoint_for_ec2_enabled/vpc_endpoint_for_ec2_enabled.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "vpc_endpoint_for_ec2_enabled",
+  "CheckTitle": "Amazon EC2 should be configured to use VPC endpoints that are created for the Amazon EC2 service.",
+  "CheckType": [],
+  "ServiceName": "ec2",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsEc2VPC",
+  "Description": "Ensure that a service endpoint for Amazon EC2 is created for each VPC. The check fails if a VPC does not have a VPC endpoint created for the Amazon EC2 service.",
+  "Risk": "Without VPC endpoints, network traffic between your VPC and Amazon EC2 may traverse the public internet, increasing the risk of unintended access or data exposure.",
+  "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/service-vpc-endpoint-enabled.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ec2-controls.html#ec2-10",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To improve the security posture of your VPC, configure Amazon EC2 to use an interface VPC endpoint powered by AWS PrivateLink.",
+      "Url": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/interface-vpc-endpoints.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/vpc/vpc_endpoint_for_ec2_enabled/vpc_endpoint_for_ec2_enabled.py

@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.vpc.vpc_client import vpc_client
+
+
+class vpc_endpoint_for_ec2_enabled(Check):
+    def execute(self):
+        findings = []
+        for vpc_id, vpc in vpc_client.vpcs.items():
+            if vpc_client.provider.scan_unused_services or vpc.in_use:
+                report = Check_Report_AWS(self.metadata())
+                report.region = vpc.region
+                report.resource_tags = vpc.tags
+                report.status = "FAIL"
+                report.status_extended = f"VPC {vpc.id} has no EC2 endpoint."
+                report.resource_id = vpc.id
+                report.resource_arn = vpc.arn
+                for endpoint in vpc_client.vpc_endpoints:
+                    if endpoint.vpc_id == vpc_id and "ec2" in endpoint.service_name:
+                        report.status = "PASS"
+                        report.status_extended = (
+                            f"VPC {vpc.id} has an EC2 {endpoint.type} endpoint."
+                        )
+                        break
+
+                findings.append(report)
+
+        return findings

prowler/providers/aws/services/vpc/vpc_service.py

@@ -239,6 +239,7 @@ class VPC(AWSService):
                                     state=endpoint["State"],
                                     policy_document=endpoint_policy,
                                     owner_id=endpoint["OwnerId"],
+                                    type=endpoint["VpcEndpointType"],
                                     region=regional_client.region,
                                     tags=endpoint.get("Tags"),
                                 )
@@ -453,6 +454,7 @@ class VpcEndpoint(BaseModel):
     state: str
     policy_document: Optional[dict]
     owner_id: str
+    type: str
     region: str
     tags: Optional[list] = []
 

tests/providers/aws/services/vpc/vpc_endpoint_for_ec2_enabled/vpc_endpoint_for_ec2_enabled_test.py

@@ -0,0 +1,156 @@
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+
+class Test_vpc_endpoint_for_ec2_enabled:
+    @mock_aws
+    def test_vpc_no_endpoints(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc_id = ec2_client.describe_vpcs()["Vpcs"][0]["VpcId"]
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled import (
+                    vpc_endpoint_for_ec2_enabled,
+                )
+
+                check = vpc_endpoint_for_ec2_enabled()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].resource_id == vpc_id
+                assert result[0].region == AWS_REGION_US_EAST_1
+                assert result[0].status == "FAIL"
+                assert result[0].status_extended == f"VPC {vpc_id} has no EC2 endpoint."
+
+    @mock_aws
+    def test_vpc_no_ec2_endpoint(self):
+        # Create VPC Mocked Resources
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+
+        vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
+        vpc_default_id = ec2_client.describe_vpcs()["Vpcs"][0]["VpcId"]
+
+        route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
+        ec2_client.create_vpc_endpoint(
+            VpcId=vpc["VpcId"],
+            ServiceName="com.amazonaws.vpce.us-east-1.s3",
+            RouteTableIds=[route_table["RouteTableId"]],
+            VpcEndpointType="Interface",
+        )
+
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled import (
+                    vpc_endpoint_for_ec2_enabled,
+                )
+
+                check = vpc_endpoint_for_ec2_enabled()
+                result = check.execute()
+
+                # One VPC is created by default, so we should have 2 results
+                assert len(result) == 2
+                for result_vpc in result:
+                    if result_vpc.resource_id == vpc["VpcId"]:
+                        assert result_vpc.region == AWS_REGION_US_EAST_1
+                        assert result_vpc.status == "FAIL"
+                        assert (
+                            result_vpc.status_extended
+                            == f"VPC {vpc['VpcId']} has no EC2 endpoint."
+                        )
+                        assert (
+                            result_vpc.resource_arn
+                            == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:123456789012:vpc/{vpc['VpcId']}"
+                        )
+                    else:
+                        assert result_vpc.region == AWS_REGION_US_EAST_1
+                        assert result_vpc.status == "FAIL"
+                        assert (
+                            result_vpc.status_extended
+                            == f"VPC {vpc_default_id} has no EC2 endpoint."
+                        )
+                        assert result_vpc.resource_id == vpc_default_id
+
+    @mock_aws
+    def test_vpc_ec2_endpoint_enabled(self):
+        # Create VPC Mocked Resources
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+
+        vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
+        vpc_default_id = ec2_client.describe_vpcs()["Vpcs"][0]["VpcId"]
+
+        route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
+        ec2_client.create_vpc_endpoint(
+            VpcId=vpc["VpcId"],
+            ServiceName="com.amazonaws.vpce.us-east-1.ec2",
+            RouteTableIds=[route_table["RouteTableId"]],
+            VpcEndpointType="Gateway",
+        )
+
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.vpc.vpc_endpoint_for_ec2_enabled.vpc_endpoint_for_ec2_enabled import (
+                    vpc_endpoint_for_ec2_enabled,
+                )
+
+                check = vpc_endpoint_for_ec2_enabled()
+                result = check.execute()
+
+                # One VPC is created by default, so we should have 2 results
+                assert len(result) == 2
+                for result_vpc in result:
+                    if result_vpc.resource_id == vpc["VpcId"]:
+                        assert result_vpc.region == AWS_REGION_US_EAST_1
+                        assert result_vpc.status == "PASS"
+                        assert (
+                            result_vpc.status_extended
+                            == f"VPC {vpc['VpcId']} has an EC2 Gateway endpoint."
+                        )
+                        assert (
+                            result_vpc.resource_arn
+                            == f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:123456789012:vpc/{vpc['VpcId']}"
+                        )
+                    else:
+                        assert result_vpc.region == AWS_REGION_US_EAST_1
+                        assert result_vpc.status == "FAIL"
+                        assert (
+                            result_vpc.status_extended
+                            == f"VPC {vpc_default_id} has no EC2 endpoint."
+                        )
+                        assert result_vpc.resource_id == vpc_default_id

tests/providers/aws/services/vpc/vpc_service_test.py

@@ -256,6 +256,7 @@ class Test_VPC_Service:
         assert vpc.vpc_endpoints[0].tags == [
             {"Key": "test", "Value": "test"},
         ]
+        assert vpc.vpc_endpoints[0].type == "Gateway"
 
     # Test VPC Describe VPC Endpoint Services
     @mock_aws