feat(ec2): Client VPN Endpoints Should Have Client Connection Logging Enabled (#4804)

Co-authored-by: Sergio <sergio@prowler.com>

prowler/providers/aws/services/ec2/ec2_client_vpn_endpoint_connection_logging_enabled/ec2_client_vpn_endpoint_connection_logging_enabled.metadata.json

@@ -0,0 +1,30 @@
+{
+  "Provider": "aws",
+  "CheckID": "ec2_client_vpn_endpoint_connection_logging_enabled",
+  "CheckTitle": "EC2 Client VPN endpoints should have client connection logging enabled.",
+  "CheckType": [],
+  "ServiceName": "ec2",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "low",
+  "ResourceType": "AwsEc2ClientVpnEndpoint",
+  "Description": "This control checks whether an AWS Client VPN endpoint has client connection logging enabled. The control fails if the endpoint doesn't have client connection logging enabled.",
+  "Risk": "Client VPN endpoints allow remote clients to securely connect to resources in a Virtual Private Cloud (VPC) in AWS. Connection logs allow you to track user activity on the VPN endpoint and provides visibility.",
+  "RelatedUrl": "https://docs.aws.amazon.com/vpn/latest/clientvpn-admin/what-is.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ec2-controls.html#ec2-51",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To enable connection logging, see Enable connection logging for an existing Client VPN endpoint in the AWS Client VPN Administrator Guide.",
+      "Url": "https://docs.aws.amazon.com/config/latest/developerguide/ec2-client-vpn-connection-log-enabled.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/ec2/ec2_client_vpn_endpoint_connection_logging_enabled/ec2_client_vpn_endpoint_connection_logging_enabled.py

@@ -0,0 +1,24 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.ec2.ec2_client import ec2_client
+
+
+class ec2_client_vpn_endpoint_connection_logging_enabled(Check):
+    def execute(self):
+        findings = []
+        for vpn_arn, vpn_endpoint in ec2_client.vpn_endpoints.items():
+            report = Check_Report_AWS(self.metadata())
+            report.region = vpn_endpoint.region
+            report.resource_id = vpn_endpoint.id
+            report.resource_arn = vpn_arn
+            report.resource_tags = vpn_endpoint.tags
+
+            if vpn_endpoint.connection_logging:
+                report.status = "PASS"
+                report.status_extended = f"Client VPN endpoint {vpn_endpoint.id} in region {vpn_endpoint.region} has client connection logging enabled."
+            else:
+                report.status = "FAIL"
+                report.status_extended = f"Client VPN endpoint {vpn_endpoint.id} in region {vpn_endpoint.region} does not have client connection logging enabled."
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/ec2/ec2_service.py

@@ -49,6 +49,8 @@ class EC2(AWSService):
         self.__threading_call__(
             self.__get_launch_template_versions__, self.launch_templates
         )
+        self.vpn_endpoints = {}
+        self.__threading_call__(self._describe_vpn_endpoints)
         self.transit_gateways = {}
         self.__threading_call__(self._describe_transit_gateways)
 
@@ -507,6 +509,33 @@ class EC2(AWSService):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def _describe_vpn_endpoints(self, regional_client):
+        try:
+            describe_client_vpn_endpoints_paginator = regional_client.get_paginator(
+                "describe_client_vpn_endpoints"
+            )
+
+            for page in describe_client_vpn_endpoints_paginator.paginate():
+                for vpn_endpoint in page["ClientVpnEndpoints"]:
+                    vpn_endpoint_arn = f"arn:aws:ec2:{regional_client.region}:{self.audited_account}:client-vpn-endpoint/{vpn_endpoint['ClientVpnEndpointId']}"
+                    if not self.audit_resources or (
+                        is_resource_filtered(vpn_endpoint_arn, self.audit_resources)
+                    ):
+                        self.vpn_endpoints[vpn_endpoint_arn] = VpnEndpoint(
+                            id=vpn_endpoint["ClientVpnEndpointId"],
+                            arn=vpn_endpoint_arn,
+                            connection_logging=vpn_endpoint["ConnectionLogOptions"][
+                                "Enabled"
+                            ],
+                            region=regional_client.region,
+                            tags=vpn_endpoint.get("Tags"),
+                        )
+
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def _describe_transit_gateways(self, regional_client):
         try:
             describe_transit_gateways_paginator = regional_client.get_paginator(
@@ -662,6 +691,13 @@ class LaunchTemplate(BaseModel):
     versions: list[LaunchTemplateVersion] = []
 
 
+class VpnEndpoint(BaseModel):
+    id: str
+    connection_logging: bool
+    region: str
+    tags: Optional[list] = []
+
+
 class TransitGateway(BaseModel):
     id: str
     auto_accept_shared_attachments: bool

tests/providers/aws/services/ec2/ec2_client_vpn_endpoint_connection_logging_enabled/ec2_client_vpn_endpoint_connection_logging_enabled_test.py

@@ -0,0 +1,135 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from prowler.providers.aws.services.ec2.ec2_service import VpnEndpoint
+from tests.providers.aws.utils import (
+    AWS_REGION_EU_WEST_1,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_ec2_client_vpn_endpoint_connection_logging_enabled:
+    @mock_aws
+    def test_ec2_no_vpn_endpoints(self):
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client",
+            new=EC2(aws_provider),
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled import (
+                ec2_client_vpn_endpoint_connection_logging_enabled,
+            )
+
+            check = ec2_client_vpn_endpoint_connection_logging_enabled()
+            result = check.execute()
+
+            assert len(result) == 0
+
+    @mock_aws
+    def test_ec2_vpn_endpoint_without_connection_logging(self):
+        # Create EC2 Mocked Resources
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        ec2 = EC2(aws_provider)
+
+        # Assuming that you mock the creation of a VPN endpoint without connection logging
+        ec2.vpn_endpoints = {
+            "arn:aws:ec2:us-east-1:123456789012:client-vpn-endpoint/cvpn-endpoint-1234567890abcdef0": VpnEndpoint(
+                id="cvpn-endpoint-1234567890abcdef0",
+                connection_logging=False,
+                region=AWS_REGION_US_EAST_1,
+                tags=None,
+            )
+        }
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client",
+            new=ec2,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled import (
+                ec2_client_vpn_endpoint_connection_logging_enabled,
+            )
+
+            check = ec2_client_vpn_endpoint_connection_logging_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags is None
+            assert (
+                result[0].status_extended
+                == f"Client VPN endpoint cvpn-endpoint-1234567890abcdef0 in region {AWS_REGION_US_EAST_1} does not have client connection logging enabled."
+            )
+            assert (
+                result[0].resource_arn
+                == "arn:aws:ec2:us-east-1:123456789012:client-vpn-endpoint/cvpn-endpoint-1234567890abcdef0"
+            )
+
+    @mock_aws
+    def test_ec2_vpn_endpoint_with_connection_logging(self):
+        # Create EC2 Mocked Resources
+        from prowler.providers.aws.services.ec2.ec2_service import EC2
+
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
+        )
+
+        ec2 = EC2(aws_provider)
+
+        # Assuming that you mock the creation of a VPN endpoint with connection logging
+        ec2.vpn_endpoints = {
+            "arn:aws:ec2:us-east-1:123456789012:client-vpn-endpoint/cvpn-endpoint-1234567890abcdef0": VpnEndpoint(
+                id="cvpn-endpoint-1234567890abcdef0",
+                connection_logging=True,
+                region=AWS_REGION_US_EAST_1,
+                tags=None,
+            )
+        }
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ), mock.patch(
+            "prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client",
+            new=ec2,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.ec2.ec2_client_vpn_endpoint_connection_logging_enabled.ec2_client_vpn_endpoint_connection_logging_enabled import (
+                ec2_client_vpn_endpoint_connection_logging_enabled,
+            )
+
+            check = ec2_client_vpn_endpoint_connection_logging_enabled()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert result[0].resource_tags is None
+            assert (
+                result[0].status_extended
+                == f"Client VPN endpoint cvpn-endpoint-1234567890abcdef0 in region {AWS_REGION_US_EAST_1} has client connection logging enabled."
+            )
+            assert (
+                result[0].resource_arn
+                == "arn:aws:ec2:us-east-1:123456789012:client-vpn-endpoint/cvpn-endpoint-1234567890abcdef0"
+            )

tests/providers/aws/services/ec2/ec2_service_test.py

@@ -22,7 +22,26 @@ 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
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "DescribeClientVpnEndpoints":
+        return {
+            "ClientVpnEndpoints": [
+                {
+                    "ClientVpnEndpointId": "cvpn-endpoint-1234567890abcdef0",
+                    "ConnectionLogOptions": {
+                        "Enabled": True,
+                        "CloudwatchLogGroup": "string",
+                        "CloudwatchLogStream": "string",
+                    },
+                    "Tags": [{"Key": "vpnendpoint", "Value": "test"}],
+                }
+            ]
+        }
+    # Si no es la operación que queremos interceptar, llamamos al método original
+    return make_api_call(self, operation_name, kwarg)
 
 
 class Test_EC2_Service:
@@ -708,6 +727,21 @@ class Test_EC2_Service:
             == KNOWN_SECRET_USER_DATA
         )
 
+    # Test EC2 Describe VPN Endpoints
+    @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+    def test_describe_vpn_endpoints(self):
+        # EC2 client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        ec2 = EC2(aws_provider)
+
+        assert len(ec2.vpn_endpoints) == 1
+        vpn_arn = f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:client-vpn-endpoint/cvpn-endpoint-1234567890abcdef0"
+        assert vpn_arn in ec2.vpn_endpoints
+        vpn_endpoint = ec2.vpn_endpoints[vpn_arn]
+        assert vpn_endpoint.id == "cvpn-endpoint-1234567890abcdef0"
+        assert vpn_endpoint.connection_logging
+        assert vpn_endpoint.region == AWS_REGION_US_EAST_1
+
     # Test EC2 Describe Launch Templates
     @mock_aws
     def test_describe_transit_gateways(self):