From ff244138d9b4e76da495121ca050029ebe1c8a90 Mon Sep 17 00:00:00 2001 From: Mario Rodriguez Lopez <101330800+MarioRgzLpz@users.noreply.github.com> Date: Thu, 22 Aug 2024 14:26:01 +0200 Subject: [PATCH] feat(ec2): Ensure automatic acceptance of VPC attachment requests is disabled (#4765) --- .../providers/aws/services/ec2/ec2_service.py | 41 ++++ .../__init__.py | 0 ..._auto_accept_vpc_attachments.metadata.json | 32 +++ ...nsitgateway_auto_accept_vpc_attachments.py | 24 ++ .../aws/services/ec2/ec2_service_test.py | 37 +++ ...ateway_auto_accept_vpc_attachments_test.py | 213 ++++++++++++++++++ 6 files changed, 347 insertions(+) create mode 100644 prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/__init__.py create mode 100644 prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.metadata.json create mode 100644 prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.py create mode 100644 tests/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments_test.py diff --git a/prowler/providers/aws/services/ec2/ec2_service.py b/prowler/providers/aws/services/ec2/ec2_service.py index 8876ef1dbf..82126d6483 100644 --- a/prowler/providers/aws/services/ec2/ec2_service.py +++ b/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.transit_gateways = {} + self.__threading_call__(self._describe_transit_gateways) def __get_volume_arn_template__(self, region): return ( @@ -505,6 +507,38 @@ class EC2(AWSService): 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( + "describe_transit_gateways" + ) + + for page in describe_transit_gateways_paginator.paginate(): + for transit_gateway in page["TransitGateways"]: + if not self.audit_resources or ( + is_resource_filtered( + transit_gateway["TransitGatewayArn"], self.audit_resources + ) + ): + self.transit_gateways[transit_gateway["TransitGatewayArn"]] = ( + TransitGateway( + id=transit_gateway["TransitGatewayId"], + auto_accept_shared_attachments=( + transit_gateway["Options"][ + "AutoAcceptSharedAttachments" + ] + == "enable" + ), + region=regional_client.region, + tags=transit_gateway.get("Tags"), + ) + ) + + except Exception as error: + logger.error( + f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + class Instance(BaseModel): id: str @@ -626,3 +660,10 @@ class LaunchTemplate(BaseModel): arn: str region: str versions: list[LaunchTemplateVersion] = [] + + +class TransitGateway(BaseModel): + id: str + auto_accept_shared_attachments: bool + region: str + tags: Optional[list] = [] diff --git a/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/__init__.py b/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.metadata.json b/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.metadata.json new file mode 100644 index 0000000000..2da3d2da95 --- /dev/null +++ b/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.metadata.json @@ -0,0 +1,32 @@ +{ + "Provider": "aws", + "CheckID": "ec2_transitgateway_auto_accept_vpc_attachments", + "CheckTitle": "Amazon EC2 Transit Gateways should not automatically accept VPC attachment requests", + "CheckType": [ + "Infrastructure Security" + ], + "ServiceName": "ec2", + "SubServiceName": "transit-gateway", + "ResourceIdTemplate": "arn:aws:ec2:region:account-id:transit-gateway/tgw-id", + "Severity": "high", + "ResourceType": "AwsEc2TransitGateway", + "Description": "Ensure EC2 transit gateways are not automatically accepting shared VPC attachments. We get a fail if a transit gateway is configured to automatically accept shared VPC attachment requests.", + "Risk": "Turning on AutoAcceptSharedAttachments allows a transit gateway to automatically accept any cross-account VPC attachment requests without verification. This increases the risk of unauthorized VPC attachments, compromising network security.", + "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/ec2-transit-gateway-auto-vpc-attach-disabled.html", + "Remediation": { + "Code": { + "CLI": "", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/ec2-controls.html#ec2-23", + "Terraform": "" + }, + "Recommendation": { + "Text": "Turn off AutoAcceptSharedAttachments to ensure that only authorized VPC attachment requests are accepted", + "Url": "https://docs.aws.amazon.com/vpc/latest/tgw/tgw-transit-gateways.html#tgw-modifying" + } + }, + "Categories": [], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.py b/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.py new file mode 100644 index 0000000000..8be9641773 --- /dev/null +++ b/prowler/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments.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_transitgateway_auto_accept_vpc_attachments(Check): + def execute(self): + findings = [] + for tgw_arn, tgw in ec2_client.transit_gateways.items(): + report = Check_Report_AWS(self.metadata()) + report.region = tgw.region + report.resource_id = tgw.id + report.resource_arn = tgw_arn + report.resource_tags = tgw.tags + + if tgw.auto_accept_shared_attachments: + report.status = "FAIL" + report.status_extended = f"Transit Gateway {tgw.id} in region {tgw.region} is configured to automatically accept shared VPC attachments." + else: + report.status = "PASS" + report.status_extended = f"Transit Gateway {tgw.id} in region {tgw.region} does not automatically accept shared VPC attachments." + + findings.append(report) + + return findings diff --git a/tests/providers/aws/services/ec2/ec2_service_test.py b/tests/providers/aws/services/ec2/ec2_service_test.py index b9fa8bcccd..c53456fc46 100644 --- a/tests/providers/aws/services/ec2/ec2_service_test.py +++ b/tests/providers/aws/services/ec2/ec2_service_test.py @@ -707,3 +707,40 @@ class Test_EC2_Service: b64decode(version2.template_data["UserData"]).decode(encoding_format_utf_8) == KNOWN_SECRET_USER_DATA ) + + # Test EC2 Describe Launch Templates + @mock_aws + def test_describe_transit_gateways(self): + # Generate EC2 Client + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + + # Create EC2 Transit Gateway API + response = ec2_client.create_transit_gateway( + Description="Test Transit Gateway", + Options={ + "AmazonSideAsn": 64512, + "AutoAcceptSharedAttachments": "enable", + }, + TagSpecifications=[ + { + "ResourceType": "transit-gateway", + "Tags": [{"Key": "Name", "Value": "test-tgw"}], + } + ], + ) + + # EC2 client for this test class + aws_provider = set_mocked_aws_provider( + [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1] + ) + ec2 = EC2(aws_provider) + + transit_arn = response["TransitGateway"]["TransitGatewayArn"] + + assert len(ec2.transit_gateways) == 1 + assert ( + ec2.transit_gateways[transit_arn].id + == response["TransitGateway"]["TransitGatewayId"] + ) + assert ec2.transit_gateways[transit_arn].auto_accept_shared_attachments + assert ec2.transit_gateways[transit_arn].region == AWS_REGION_US_EAST_1 diff --git a/tests/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments_test.py b/tests/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments_test.py new file mode 100644 index 0000000000..fa31325833 --- /dev/null +++ b/tests/providers/aws/services/ec2/ec2_transitgateway_auto_accept_vpc_attachments/ec2_transitgateway_auto_accept_vpc_attachments_test.py @@ -0,0 +1,213 @@ +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_ec2_transitgateway_auto_accept_vpc_attachments: + @mock_aws + def test_no_transit_gateways(self): + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + ec2_client.transit_gateways = [] + + from prowler.providers.aws.services.ec2.ec2_service import EC2 + + 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, + ), mock.patch( + "prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments.ec2_client", + new=EC2(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments import ( + ec2_transitgateway_auto_accept_vpc_attachments, + ) + + check = ec2_transitgateway_auto_accept_vpc_attachments() + result = check.execute() + + assert len(result) == 0 + + @mock_aws + def test_transit_gateway_default_options(self): + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + tgw = ec2_client.create_transit_gateway( + Description="Test TGW with auto-accept enabled", + ) + tgw_id = tgw["TransitGateway"]["TransitGatewayId"] + + from prowler.providers.aws.services.ec2.ec2_service import EC2 + + 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, + ), mock.patch( + "prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments.ec2_client", + new=EC2(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments import ( + ec2_transitgateway_auto_accept_vpc_attachments, + ) + + check = ec2_transitgateway_auto_accept_vpc_attachments() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Transit Gateway {tgw_id} in region {AWS_REGION_US_EAST_1} does not automatically accept shared VPC attachments." + ) + assert result[0].resource_id == tgw_id + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_transit_gateway_autoaccept_enabled(self): + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + tgw = ec2_client.create_transit_gateway( + Description="Test TGW with auto-accept enabled", + Options={ + "AutoAcceptSharedAttachments": "enable", + }, + ) + tgw_id = tgw["TransitGateway"]["TransitGatewayId"] + + from prowler.providers.aws.services.ec2.ec2_service import EC2 + + 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, + ), mock.patch( + "prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments.ec2_client", + new=EC2(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments import ( + ec2_transitgateway_auto_accept_vpc_attachments, + ) + + check = ec2_transitgateway_auto_accept_vpc_attachments() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Transit Gateway {tgw_id} in region {AWS_REGION_US_EAST_1} is configured to automatically accept shared VPC attachments." + ) + assert result[0].resource_id == tgw_id + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_transit_gateway_autoaccept_disabled(self): + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + tgw = ec2_client.create_transit_gateway( + Description="Test TGW with auto-accept disabled", + Options={ + "AutoAcceptSharedAttachments": "disable", + }, + ) + tgw_id = tgw["TransitGateway"]["TransitGatewayId"] + + from prowler.providers.aws.services.ec2.ec2_service import EC2 + + 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, + ), mock.patch( + "prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments.ec2_client", + new=EC2(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments import ( + ec2_transitgateway_auto_accept_vpc_attachments, + ) + + check = ec2_transitgateway_auto_accept_vpc_attachments() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"Transit Gateway {tgw_id} in region {AWS_REGION_US_EAST_1} does not automatically accept shared VPC attachments." + ) + assert result[0].resource_id == tgw_id + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_multiple_transit_gateways(self): + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + + # Create TGW with auto-accept enabled + tgw_with_auto_accept = ec2_client.create_transit_gateway( + Description="TGW with auto-accept enabled", + Options={ + "AutoAcceptSharedAttachments": "enable", + }, + ) + tgw_with_auto_accept_id = tgw_with_auto_accept["TransitGateway"][ + "TransitGatewayId" + ] + + # Create TGW with auto-accept disabled + tgw_without_auto_accept = ec2_client.create_transit_gateway( + Description="TGW with auto-accept disabled", + Options={ + "AutoAcceptSharedAttachments": "disable", + }, + ) + tgw_without_auto_accept_id = tgw_without_auto_accept["TransitGateway"][ + "TransitGatewayId" + ] + + from prowler.providers.aws.services.ec2.ec2_service import EC2 + + 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, + ), mock.patch( + "prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments.ec2_client", + new=EC2(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.ec2.ec2_transitgateway_auto_accept_vpc_attachments.ec2_transitgateway_auto_accept_vpc_attachments import ( + ec2_transitgateway_auto_accept_vpc_attachments, + ) + + check = ec2_transitgateway_auto_accept_vpc_attachments() + result = check.execute() + + assert len(result) == 2 + + # Check the TGW with auto-accept enabled + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"Transit Gateway {tgw_with_auto_accept_id} in region {AWS_REGION_US_EAST_1} is configured to automatically accept shared VPC attachments." + ) + assert result[0].resource_id == tgw_with_auto_accept_id + assert result[0].region == AWS_REGION_US_EAST_1 + + # Check the TGW with auto-accept disabled + assert result[1].status == "PASS" + assert ( + result[1].status_extended + == f"Transit Gateway {tgw_without_auto_accept_id} in region {AWS_REGION_US_EAST_1} does not automatically accept shared VPC attachments." + ) + assert result[1].resource_id == tgw_without_auto_accept_id + assert result[1].region == AWS_REGION_US_EAST_1