feat(ec2): Ensure automatic acceptance of VPC attachment requests is disabled (#4765)

This commit is contained in:
Mario Rodriguez Lopez
2024-08-22 14:26:01 +02:00
committed by GitHub
parent 903f9c576f
commit ff244138d9
6 changed files with 347 additions and 0 deletions
@@ -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] = []
@@ -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": ""
}
@@ -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
@@ -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
@@ -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