feat(rds): add new check rds_instance_inside_vpc

prowler/providers/aws/services/rds/rds_instance_inside_vpc/rds_instance_inside_vpc.metadata.json

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "rds_instance_inside_vpc",
+  "CheckTitle": "Check if RDS instances are deployed within a VPC.",
+  "CheckType": [
+    "Software and Configuration Checks, AWS Security Best Practices"
+  ],
+  "ServiceName": "rds",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:rds:region:account-id:db-instance",
+  "Severity": "high",
+  "ResourceType": "AwsRdsDbInstance",
+  "Description": "Check if RDS instances are deployed within a VPC.",
+  "Risk": "If your RDS instances are not deployed within a VPC, they are not isolated from the public internet and are exposed to potential security threats. Deploying RDS instances within a VPC allows you to control inbound and outbound traffic to and from the instances, and provides an additional layer of security to your database instances.",
+  "RelatedUrl": "https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_VPC.WorkingWithRDSInstanceinaVPC.html#USER_VPC.Subnets",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws rds modify-db-instance --db-instance-identifier <instance-identifier> --vpc-security-group-ids <vpc-security-group-ids>",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/rds-controls.html#rds-18",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that your RDS instances are deployed within a VPC to provide an additional layer of security to your database instances.",
+      "Url": "https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Overview.DBInstance.Modifying.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/rds/rds_instance_inside_vpc/rds_instance_inside_vpc.py

@@ -0,0 +1,25 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.rds.rds_client import rds_client
+
+
+class rds_instance_inside_vpc(Check):
+    def execute(self):
+        findings = []
+        for db_instance_arn, db_instance in rds_client.db_instances.items():
+            report = Check_Report_AWS(self.metadata())
+            report.region = db_instance.region
+            report.resource_id = db_instance.id
+            report.resource_arn = db_instance_arn
+            report.resource_tags = db_instance.tags
+            if db_instance.vpc_id:
+                report.status = "PASS"
+                report.status_extended = f"RDS Instance {db_instance.id} is deployed in a VPC {db_instance.vpc_id}."
+            else:
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"RDS Instance {db_instance.id} is not deployed in a VPC."
+                )
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/rds/rds_service.py

@@ -53,6 +53,7 @@
                         is_resource_filtered(arn, self.audit_resources)
                     ):
                         if instance["Engine"] != "docdb":
+                            print(instance.get("DBSubnetGroup", {}))
                             self.db_instances[arn] = DBInstance(
                                 id=instance["DBInstanceIdentifier"],
                                 arn=arn,
@@ -100,6 +101,7 @@
                                 copy_tags_to_snapshot=instance.get(
                                     "CopyTagsToSnapshot"
                                 ),
+                                vpc_id=instance.get("DBSubnetGroup", {}).get("VpcId"),
                             )
         except Exception as error:
             logger.error(
@@ -109,10 +111,7 @@
     def _describe_db_parameters(self, regional_client):
         logger.info("RDS - Describe DB Parameters...")
         try:
-            for (
-                instance_arn,
-                instance,
-            ) in self.db_instances.items():
+            for instance in self.db_instances.values():
                 if instance.region == regional_client.region:
                     for parameter_group in instance.parameter_groups:
                         describe_db_parameters_paginator = (
@@ -137,23 +136,26 @@
                     describe_db_certificates_paginator = regional_client.get_paginator(
                         "describe_certificates"
                     )
-                    for page in describe_db_certificates_paginator.paginate(
-                        CertificateIdentifier=instance.ca_cert
-                    ):
-                        for certificate in page["Certificates"]:
-                            instance.cert.append(
-                                Certificate(
-                                    id=certificate["CertificateIdentifier"],
-                                    arn=certificate["CertificateArn"],
-                                    type=certificate["CertificateType"],
-                                    valid_from=certificate["ValidFrom"],
-                                    valid_till=certificate["ValidTill"],
-                                    customer_override=certificate["CustomerOverride"],
-                                    customer_override_valid_till=certificate.get(
-                                        "CustomerOverrideValidTill"
-                                    ),
+                    if instance.ca_cert:
+                        for page in describe_db_certificates_paginator.paginate(
+                            CertificateIdentifier=instance.ca_cert
+                        ):
+                            for certificate in page["Certificates"]:
+                                instance.cert.append(
+                                    Certificate(
+                                        id=certificate["CertificateIdentifier"],
+                                        arn=certificate["CertificateArn"],
+                                        type=certificate["CertificateType"],
+                                        valid_from=certificate["ValidFrom"],
+                                        valid_till=certificate["ValidTill"],
+                                        customer_override=certificate[
+                                            "CustomerOverride"
+                                        ],
+                                        customer_override_valid_till=certificate.get(
+                                            "CustomerOverrideValidTill"
+                                        ),
+                                    )
                                 )
-                            )
 
         except Exception as error:
             logger.error(
@@ -507,6 +509,7 @@
     ca_cert: Optional[str]
     cert: list[Certificate] = []
     copy_tags_to_snapshot: Optional[bool]
+    vpc_id: Optional[str]
 
 
 class DBCluster(BaseModel):

tests/providers/aws/services/rds/rds_instance_inside_vpc/rds_instance_inside_vpc_test.py

@@ -0,0 +1,169 @@
+from unittest import mock
+
+from boto3 import client
+from moto import mock_aws
+
+from prowler.providers.aws.services.rds.rds_service import DBInstance
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_rds_instance_inside_vpc:
+    @mock_aws
+    def test_rds_no_instances(self):
+        from prowler.providers.aws.services.rds.rds_service import RDS
+
+        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.rds.rds_instance_inside_vpc.rds_instance_inside_vpc.rds_client",
+                new=RDS(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.rds.rds_instance_inside_vpc.rds_instance_inside_vpc import (
+                    rds_instance_inside_vpc,
+                )
+
+                check = rds_instance_inside_vpc()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @mock_aws
+    def test_rds_instance_inside_vpc(self):
+        rds_conn = client("rds", region_name=AWS_REGION_US_EAST_1)
+        ec2_conn = client("ec2")
+
+        # Step 1: Create the VPC
+        vpc_id = ec2_conn.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]
+        subnet_1_id = ec2_conn.create_subnet(CidrBlock="10.0.1.0/24", VpcId=vpc_id)[
+            "Subnet"
+        ]["SubnetId"]
+        subnet_2_id = ec2_conn.create_subnet(CidrBlock="10.0.2.0/24", VpcId=vpc_id)[
+            "Subnet"
+        ]["SubnetId"]
+        subnet_group_name = "my-rds-subnet-group"
+        rds_conn.create_db_subnet_group(
+            DBSubnetGroupName=subnet_group_name,
+            DBSubnetGroupDescription="Subnet group for RDS instance in VPC",
+            SubnetIds=[
+                subnet_1_id,
+                subnet_2_id,
+            ],
+        )
+        rds_conn.create_db_instance(
+            DBInstanceIdentifier="db-master-1",
+            AllocatedStorage=10,
+            Engine="postgres",
+            DBName="staging-postgres",
+            DBInstanceClass="db.m1.small",
+            StorageEncrypted=True,
+            DeletionProtection=True,
+            PubliclyAccessible=True,
+            AutoMinorVersionUpgrade=True,
+            BackupRetentionPeriod=10,
+            Port=5432,
+            DBSubnetGroupName=subnet_group_name,
+            Tags=[{"Key": "test", "Value": "test"}],
+        )
+
+        from prowler.providers.aws.services.rds.rds_service import RDS
+
+        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.rds.rds_instance_inside_vpc.rds_instance_inside_vpc.rds_client",
+                new=RDS(aws_provider),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.rds.rds_instance_inside_vpc.rds_instance_inside_vpc import (
+                    rds_instance_inside_vpc,
+                )
+
+                check = rds_instance_inside_vpc()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"RDS Instance db-master-1 is deployed in a VPC {vpc_id}."
+                )
+                assert result[0].resource_id == "db-master-1"
+                assert result[0].region == AWS_REGION_US_EAST_1
+                assert (
+                    result[0].resource_arn
+                    == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
+                )
+                assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]
+
+    def test_rds_instance_not_in_vpc(self):
+        rds_client = mock.MagicMock
+        instance_arn = (
+            f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
+        )
+        rds_client.db_instances = {
+            instance_arn: DBInstance(
+                id="db-master-1",
+                arn=instance_arn,
+                engine="postgres",
+                cloudwatch_logs=None,
+                deletion_protection=True,
+                auto_minor_version_upgrade=True,
+                enhanced_monitoring_arn=None,
+                endpoint={
+                    "Address": "db-master-1.us-east-1.rds.amazonaws.com",
+                    "Port": 5432,
+                },
+                engine_version="12.3",
+                status="available",
+                public=False,
+                encrypted=False,
+                iam_auth=False,
+                region=AWS_REGION_US_EAST_1,
+                multi_az=False,
+                username="admin",
+                tags=[{"Key": "test", "Value": "test"}],
+                copy_tags_to_snapshot=None,
+            )
+        }
+
+        with mock.patch(
+            "prowler.providers.aws.services.rds.rds_service.RDS",
+            new=rds_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.rds.rds_instance_inside_vpc.rds_instance_inside_vpc.rds_client",
+            new=rds_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.rds.rds_instance_inside_vpc.rds_instance_inside_vpc import (
+                rds_instance_inside_vpc,
+            )
+
+            check = rds_instance_inside_vpc()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "RDS Instance db-master-1 is not deployed in a VPC."
+            )
+            assert result[0].resource_id == "db-master-1"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
+            )
+            assert result[0].resource_tags == [{"Key": "test", "Value": "test"}]