chore(vpc): add scan unused services logic to VPC checks (#4137)

2026-07-24 13:01:56 +00:00 · 2024-05-30 10:59:48 -04:00
parent dfa940440c
commit 2725d476a4
8 changed files with 206 additions and 58 deletions
@@ -69,3 +69,15 @@ You should enable Public Access Block at the account level to prevent the exposu
 VPC Flow Logs provide visibility into network traffic that traverses the VPC and can be used to detect anomalous traffic or insight during security workflows. Nevertheless, Prowler will only check if the Flow Logs are enabled for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).

  - `vpc_flow_logs_enabled`
+
+VPC subnets must not have public IP addresses by default to prevent the exposure of your resources to the internet. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
+
+  - `vpc_subnet_no_public_ip_by_default`
+
+VPCs should have separate private and public subnets to prevent the exposure of your resources to the internet. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
+
+  - `vpc_subnet_separate_private_public`
+
+VPCs should have subnets in different availability zones to prevent a single point of failure. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
+
+  - `vpc_subnet_different_az`
@@ -28,9 +28,9 @@ class VPC(AWSService):
        self.__describe_flow_logs__()
        self.__describe_peering_route_tables__()
        self.__describe_vpc_endpoint_service_permissions__()
-        self.__describe_network_interfaces__()
        self.vpc_subnets = {}
        self.__threading_call__(self.__describe_vpc_subnets__)
+        self.__describe_network_interfaces__()

    def __describe_vpcs__(self, regional_client):
        logger.info("VPC - Describing VPCs...")
@@ -192,6 +192,19 @@ class VPC(AWSService):
                    )["NetworkInterfaces"]
                    if enis:
                        vpc.in_use = True
+                    for subnet in vpc.subnets:
+                        enis = regional_client.describe_network_interfaces(
+                            Filters=[
+                                {
+                                    "Name": "subnet-id",
+                                    "Values": [
+                                        subnet.id,
+                                    ],
+                                },
+                            ]
+                        )["NetworkInterfaces"]
+                        if enis:
+                            subnet.in_use = True
                except Exception as error:
                    logger.error(
                        f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -395,6 +408,7 @@ class VpcSubnet(BaseModel):
    cidr_block: Optional[str]
    availability_zone: str
    public: bool
+    in_use: bool = False
    nat_gateway: bool
    region: str
    mapPublicIpOnLaunch: bool
@@ -6,28 +6,29 @@ class vpc_subnet_different_az(Check):
    def execute(self):
        findings = []
        for vpc in vpc_client.vpcs.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = vpc.region
-            report.resource_tags = vpc.tags
-            report.status = "FAIL"
-            report.status_extended = (
-                f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
-            )
-            report.resource_id = vpc.id
-            report.resource_arn = vpc.arn
-            if vpc.subnets:
-                availability_zone = None
-                for subnet in vpc.subnets:
-                    if (
-                        availability_zone
-                        and subnet.availability_zone != availability_zone
-                    ):
-                        report.status = "PASS"
-                        report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has subnets in more than one availability zone."
-                        break
-                    availability_zone = subnet.availability_zone
-                    report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only subnets in {availability_zone}."
+            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.name if vpc.name else vpc.id} has no subnets."
+                )
+                report.resource_id = vpc.id
+                report.resource_arn = vpc.arn
+                if vpc.subnets:
+                    availability_zone = None
+                    for subnet in vpc.subnets:
+                        if (
+                            availability_zone
+                            and subnet.availability_zone != availability_zone
+                        ):
+                            report.status = "PASS"
+                            report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has subnets in more than one availability zone."
+                            break
+                        availability_zone = subnet.availability_zone
+                        report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only subnets in {availability_zone}."

-            findings.append(report)
+                findings.append(report)

        return findings
@@ -7,17 +7,19 @@ class vpc_subnet_no_public_ip_by_default(Check):
        findings = []
        for vpc in vpc_client.vpcs.values():
            for subnet in vpc.subnets:
-                report = Check_Report_AWS(self.metadata())
-                report.region = subnet.region
-                report.resource_tags = subnet.tags
-                report.resource_id = subnet.id
-                report.resource_arn = subnet.arn
-                if subnet.mapPublicIpOnLaunch:
-                    report.status = "FAIL"
-                    report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."
-                else:
-                    report.status = "PASS"
-                    report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} does NOT assign public IP by default."
-                findings.append(report)
+                # Check if ignoring flag is set and if the VPC Subnet is in use
+                if vpc_client.provider.scan_unused_services or subnet.in_use:
+                    report = Check_Report_AWS(self.metadata())
+                    report.region = subnet.region
+                    report.resource_tags = subnet.tags
+                    report.resource_id = subnet.id
+                    report.resource_arn = subnet.arn
+                    if subnet.mapPublicIpOnLaunch:
+                        report.status = "FAIL"
+                        report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."
+                    else:
+                        report.status = "PASS"
+                        report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} does NOT assign public IP by default."
+                    findings.append(report)

        return findings
@@ -6,28 +6,29 @@ class vpc_subnet_separate_private_public(Check):
    def execute(self):
        findings = []
        for vpc in vpc_client.vpcs.values():
-            report = Check_Report_AWS(self.metadata())
-            report.region = vpc.region
-            report.resource_tags = vpc.tags
-            report.status = "FAIL"
-            report.status_extended = (
-                f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
-            )
-            report.resource_id = vpc.id
-            report.resource_arn = vpc.arn
-            if vpc.subnets:
-                public = False
-                private = False
-                for subnet in vpc.subnets:
-                    if subnet.public:
-                        public = True
-                        report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only public subnets."
-                    if not subnet.public:
-                        private = True
-                        report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only private subnets."
-                    if public and private:
-                        report.status = "PASS"
-                        report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has private and public subnets."
-            findings.append(report)
+            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.name if vpc.name else vpc.id} has no subnets."
+                )
+                report.resource_id = vpc.id
+                report.resource_arn = vpc.arn
+                if vpc.subnets:
+                    public = False
+                    private = False
+                    for subnet in vpc.subnets:
+                        if subnet.public:
+                            public = True
+                            report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only public subnets."
+                        if not subnet.public:
+                            private = True
+                            report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only private subnets."
+                        if public and private:
+                            report.status = "PASS"
+                            report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has private and public subnets."
+                findings.append(report)

        return findings
@@ -166,3 +166,31 @@ class Test_vpc_subnet_different_az:
                        assert result.region == AWS_REGION_US_EAST_1
                if not found:
                    assert False
+
+    @mock_aws
+    def test_vpc_no_subnets_but_unused(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        ec2_client.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
+
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider(
+            audited_regions=[AWS_REGION_US_EAST_1], scan_unused_services=False
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_subnet_different_az.vpc_subnet_different_az.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                from prowler.providers.aws.services.vpc.vpc_subnet_different_az.vpc_subnet_different_az import (
+                    vpc_subnet_different_az,
+                )
+
+                check = vpc_subnet_different_az()
+                results = check.execute()
+
+                assert len(results) == 0
@@ -102,3 +102,51 @@ class Test_vpc_subnet_no_public_ip_by_default:
                            result.status_extended
                            == f"VPC subnet {subnet_private['Subnet']['SubnetId']} does NOT assign public IP by default."
                        )
+
+    @mock_aws
+    def test_vpc_with_map_ip_on_launch_but_unused(self):
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc = ec2_client.create_vpc(
+            CidrBlock="172.28.7.0/24", InstanceTenancy="default"
+        )
+        subnet_private = ec2_client.create_subnet(
+            VpcId=vpc["Vpc"]["VpcId"],
+            CidrBlock="172.28.7.192/26",
+            AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
+            TagSpecifications=[
+                {
+                    "ResourceType": "subnet",
+                    "Tags": [
+                        {"Key": "Name", "Value": "subnet_name"},
+                    ],
+                },
+            ],
+        )
+
+        ec2_client.modify_subnet_attribute(
+            SubnetId=subnet_private["Subnet"]["SubnetId"],
+            MapPublicIpOnLaunch={"Value": True},
+        )
+
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider(
+            audited_regions=[AWS_REGION_US_EAST_1], scan_unused_services=False
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_subnet_no_public_ip_by_default.vpc_subnet_no_public_ip_by_default.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                from prowler.providers.aws.services.vpc.vpc_subnet_no_public_ip_by_default.vpc_subnet_no_public_ip_by_default import (
+                    vpc_subnet_no_public_ip_by_default,
+                )
+
+                check = vpc_subnet_no_public_ip_by_default()
+                results = check.execute()
+
+                assert len(results) == 0
@@ -129,6 +129,48 @@ class Test_vpc_subnet_separate_private_public:
                if not found:
                    assert False

+    @mock_aws
+    def test_vpc_subnet_only_public_but_unused(self):
+        # Create EC2 Mocked Resources
+        ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
+        vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+        subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
+        # Create IGW and attach to VPC
+        igw = ec2.create_internet_gateway()
+        vpc.attach_internet_gateway(InternetGatewayId=igw.id)
+        # Set IGW as default route for public subnet
+        route_table = ec2.create_route_table(VpcId=vpc.id)
+        route_table.associate_with_subnet(SubnetId=subnet.id)
+        ec2_client.create_route(
+            RouteTableId=route_table.id,
+            DestinationCidrBlock="0.0.0.0/0",
+            GatewayId=igw.id,
+        )
+
+        from prowler.providers.aws.services.vpc.vpc_service import VPC
+
+        aws_provider = set_mocked_aws_provider(
+            audited_regions=[AWS_REGION_US_EAST_1], scan_unused_services=False
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.vpc.vpc_subnet_separate_private_public.vpc_subnet_separate_private_public.vpc_client",
+                new=VPC(aws_provider),
+            ):
+                from prowler.providers.aws.services.vpc.vpc_subnet_separate_private_public.vpc_subnet_separate_private_public import (
+                    vpc_subnet_separate_private_public,
+                )
+
+                check = vpc_subnet_separate_private_public()
+                results = check.execute()
+
+                assert len(results) == 0
+
    @mock_aws
    def test_vpc_subnet_private_and_public(self):
        ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)