From 38e024216cd51e2f24b3210ec9008df7f1e54cac Mon Sep 17 00:00:00 2001 From: Sergio Garcia <38561120+sergargar@users.noreply.github.com> Date: Mon, 30 Sep 2024 14:53:04 -0400 Subject: [PATCH] chore(ec2): enhance security group with any open port check (#5215) --- ...ss_from_internet_to_any_port.metadata.json | 4 +- ...allow_ingress_from_internet_to_any_port.py | 100 ++++++++---------- .../providers/aws/services/ec2/ec2_service.py | 24 ++++- .../ec2_launch_template_no_public_ip_test.py | 5 +- ..._ingress_from_internet_to_any_port_test.py | 31 ++---- 5 files changed, 77 insertions(+), 87 deletions(-) diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.metadata.json b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.metadata.json index cc0eae560f..2cf5f6cb3e 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.metadata.json +++ b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.metadata.json @@ -10,8 +10,8 @@ "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id", "Severity": "high", "ResourceType": "AwsEc2SecurityGroup", - "Description": "Ensure no security groups allow ingress from 0.0.0.0/0 or ::/0 to any port.", - "Risk": "If Security groups are not properly configured the attack surface is increased.", + "Description": "Ensure no security groups allow ingress from 0.0.0.0/0 or ::/0 to any port and not attached to a network interface with not allowed network interface types or instance owners. By default, the allowed network interface types are 'api_gateway_managed' and 'vpc_endpoint', and the allowed instance owners are 'amazon-elb', you can customize these values by setting the 'ec2_allowed_interface_types' and 'ec2_allowed_instance_owners' variables.", + "Risk": "The security group allows all traffic from the internet to any port. This could allow an attacker to access the instance.", "RelatedUrl": "", "Remediation": { "Code": { diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py index 4a340eccc7..b9b504e6c4 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py +++ b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py @@ -12,26 +12,25 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check): def execute(self): findings = [] for security_group_arn, security_group in ec2_client.security_groups.items(): - # Check if ignoring flag is set and if the VPC and the SG is in use - if ec2_client.provider.scan_unused_services or ( - security_group.vpc_id in vpc_client.vpcs - and vpc_client.vpcs[security_group.vpc_id].in_use - and len(security_group.network_interfaces) > 0 + # Only execute the check if the check ec2_securitygroup_allow_ingress_from_internet_to_all_ports has not failed + if not ec2_client.is_failed_check( + ec2_securitygroup_allow_ingress_from_internet_to_all_ports.__name__, + security_group_arn, ): - report = Check_Report_AWS(self.metadata()) - report.region = security_group.region - report.status = "PASS" - report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have any port open to the Internet." - report.resource_details = security_group.name - report.resource_id = security_group.id - report.resource_arn = security_group_arn - report.resource_tags = security_group.tags - # only proceed if check "..._to_all_ports" did not run or did not FAIL to avoid to report open ports twice - if not ec2_client.is_failed_check( - ec2_securitygroup_allow_ingress_from_internet_to_all_ports.__name__, - security_group_arn, + # Check if ignoring flag is set and if the VPC and the SG is in use + if ec2_client.provider.scan_unused_services or ( + security_group.vpc_id in vpc_client.vpcs + and vpc_client.vpcs[security_group.vpc_id].in_use + and len(security_group.network_interfaces) > 0 ): - # Loop through every security group's ingress rule and check it + report = Check_Report_AWS(self.metadata()) + report.region = security_group.region + report.status = "PASS" + report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have any port open to the Internet." + report.resource_details = security_group.name + report.resource_id = security_group.id + report.resource_arn = security_group_arn + report.resource_tags = security_group.tags for ingress_rule in security_group.ingress_rules: if check_security_group( ingress_rule, "-1", ports=None, any_address=True @@ -44,11 +43,8 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check): ) if report.status == "FAIL": - break # no need to check other ingress rules because at least one failed already - else: - report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet and therefore was not checked against a specific port." - - findings.append(report) + break + findings.append(report) return findings @@ -57,46 +53,34 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check): report, security_group_name: str, security_group_id: str, - enis: [NetworkInterface], + enis: list[NetworkInterface], ): - report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but is exclusively not attached to any network interface." + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but it is not attached to any network interface." for eni in enis: - if self.is_allowed_eni_type(eni_type=eni.type): + if eni.type in ec2_client.audit_config.get( + "ec2_allowed_interface_types", [] + ): report.status = "PASS" - report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but is exclusively attached to an allowed network interface type ({eni.type})." + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet and it is attached to an allowed network interface type ({eni.type})." continue - - eni_owner = self.get_eni_owner(eni=eni) - if self.is_allowed_eni_owner(eni_owner=eni_owner): + if eni.attachment.instance_owner_id in ec2_client.audit_config.get( + "ec2_allowed_instance_owners", [] + ): report.status = "PASS" - report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but is exclusively attached to an allowed network interface instance owner ({eni_owner})." + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet and it is attached to an allowed network interface instance owner ({eni.attachment.instance_owner_id})." continue + if eni.type not in ec2_client.audit_config.get( + "ec2_allowed_interface_types", [] + ): + report.status = "FAIL" + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but its network interface type ({eni.type}) is not allowed." + elif eni.attachment.instance_owner_id not in ec2_client.audit_config.get( + "ec2_allowed_instance_owners", [] + ): + report.status = "FAIL" + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but its network interface instance owner ({eni.attachment.instance_owner_id}) is not allowed." else: report.status = "FAIL" - report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet and neither its network interface type ({eni.type}) nor its network interface instance owner ({eni_owner}) are part of the allowed network interfaces." - - break # no need to check other network interfaces because at least one failed already - - @staticmethod - def is_allowed_eni_type(eni_type: str) -> bool: - return eni_type in ec2_client.audit_config.get( - "ec2_allowed_interface_types", [] - ) - - @staticmethod - def get_eni_owner(eni) -> str: - eni_owner = "" - if ( - hasattr(eni, "attachment") - and isinstance(eni.attachment, dict) - and "InstanceOwnerId" in eni.attachment - ): - eni_owner = eni.attachment["InstanceOwnerId"] - - return eni_owner - - @staticmethod - def is_allowed_eni_owner(eni_owner: str) -> bool: - return eni_owner in ec2_client.audit_config.get( - "ec2_allowed_instance_owners", [] - ) + report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but neither its network interface type ({eni.type}) nor its network interface instance owner ({eni.attachment.instance_owner_id}) are allowed." + if report.status == "FAIL": + break diff --git a/prowler/providers/aws/services/ec2/ec2_service.py b/prowler/providers/aws/services/ec2/ec2_service.py index 5d3896ce3b..7ca7b7719f 100644 --- a/prowler/providers/aws/services/ec2/ec2_service.py +++ b/prowler/providers/aws/services/ec2/ec2_service.py @@ -275,11 +275,22 @@ class EC2(AWSService): ipv6_address = ip_address(ipv6_address_str) if ipv6_address.is_global: public_ip_addresses.append(ipv6_address) - + attachment = Attachment( + attachment_id=interface.get("Attachment", {}).get( + "AttachmentId", "" + ), + instance_id=interface.get("Attachment", {}).get( + "InstanceId", "" + ), + instance_owner_id=interface.get("Attachment", {}).get( + "InstanceOwnerId", "" + ), + status=interface.get("Attachment", {}).get("Status", ""), + ) self.network_interfaces[id] = NetworkInterface( id=id, association=interface.get("Association", {}), - attachment=interface.get("Attachment", {}), + attachment=attachment, private_ip=interface.get("PrivateIpAddress"), type=interface["InterfaceType"], subnet_id=interface["SubnetId"], @@ -669,10 +680,17 @@ class Volume(BaseModel): tags: Optional[list] = [] +class Attachment(BaseModel): + attachment_id: str = "" + instance_id: str = "" + instance_owner_id: str = "" + status: str = "" + + class NetworkInterface(BaseModel): id: str association: dict - attachment: dict + attachment: Attachment private_ip: Optional[str] public_ip_addresses: list[Union[IPv4Address, IPv6Address]] type: str diff --git a/tests/providers/aws/services/ec2/ec2_launch_template_no_public_ip/ec2_launch_template_no_public_ip_test.py b/tests/providers/aws/services/ec2/ec2_launch_template_no_public_ip/ec2_launch_template_no_public_ip_test.py index a7275aa3db..d582b73698 100644 --- a/tests/providers/aws/services/ec2/ec2_launch_template_no_public_ip/ec2_launch_template_no_public_ip_test.py +++ b/tests/providers/aws/services/ec2/ec2_launch_template_no_public_ip/ec2_launch_template_no_public_ip_test.py @@ -10,6 +10,7 @@ from moto import mock_aws from prowler.config.config import encoding_format_utf_8 from prowler.providers.aws.services.ec2.ec2_service import ( + Attachment, LaunchTemplate, LaunchTemplateVersion, NetworkInterface, @@ -190,7 +191,7 @@ class Test_ec2_launch_template_no_public_ip: network_interface = NetworkInterface( id="eni-1234567890", association={}, - attachment={}, + attachment=Attachment(), private_ip="", public_ip_addresses=[IPv4Address("192.175.48.10")], type="interface", @@ -287,7 +288,7 @@ class Test_ec2_launch_template_no_public_ip: network_interface = NetworkInterface( id="eni-1234567890", association={}, - attachment={}, + attachment=Attachment(), private_ip="", public_ip_addresses=[IPv6Address("::1234:5678")], type="interface", diff --git a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port_test.py b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port_test.py index 4e3ea6b636..2c83564bc9 100644 --- a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port_test.py +++ b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port_test.py @@ -3,6 +3,7 @@ from unittest import mock from boto3 import client, resource from moto import mock_aws +from prowler.providers.aws.services.ec2.ec2_service import Attachment from tests.providers.aws.utils import ( AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1, @@ -97,9 +98,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: self, default_sg_id, default_sg_name, network_interface_response ): eni = network_interface_response.get("NetworkInterface", {}) - att = eni.get("Attachment", {}) eni_type = eni.get("InterfaceType", "") - eni_owner = att.get("InstanceOwnerId", "") from prowler.providers.aws.services.ec2.ec2_service import EC2 from prowler.providers.aws.services.vpc.vpc_service import VPC @@ -133,7 +132,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet and neither its network interface type ({eni_type}) nor its network interface instance owner ({eni_owner}) are part of the allowed network interfaces." + == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet but its network interface type ({eni_type}) is not allowed." ) assert ( sg.resource_arn @@ -175,7 +174,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: "eni_instance_owner": "NOT_ALLOWED", "report": { "status": "PASS", - "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet but is exclusively attached to an allowed network interface type (vpc_endpoint).", + "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet and it is attached to an allowed network interface type (vpc_endpoint).", }, }, { @@ -183,7 +182,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: "eni_instance_owner": "amazon-elb", "report": { "status": "PASS", - "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet but is exclusively attached to an allowed network interface instance owner (amazon-elb).", + "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet and it is attached to an allowed network interface instance owner (amazon-elb).", }, }, { @@ -191,7 +190,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: "eni_instance_owner": "NOT_ALLOWED_INSTANCE_OWNER", "report": { "status": "FAIL", - "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet and neither its network interface type (NOT_ALLOWED_ENI_TYPE) nor its network interface instance owner (NOT_ALLOWED_INSTANCE_OWNER) are part of the allowed network interfaces.", + "status_extended": "Security group SG_name (SG_id) has at least one port open to the Internet but its network interface type (NOT_ALLOWED_ENI_TYPE) is not allowed.", }, }, ] @@ -202,7 +201,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: eni = NetworkInterface( id="1", association={}, - attachment={"InstanceOwnerId": test["eni_instance_owner"]}, + attachment=Attachment(instance_owner_id=test["eni_instance_owner"]), private_ip="1", public_ip_addresses=[], type=test["eni_interface_type"], @@ -303,7 +302,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet but is exclusively attached to an allowed network interface type ({eni_type})." + == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet and it is attached to an allowed network interface type ({eni_type})." ) assert ( sg.resource_arn @@ -396,7 +395,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet but is exclusively attached to an allowed network interface instance owner ({eni_owner})." + == f"Security group {default_sg_name} ({default_sg_id}) has at least one port open to the Internet and it is attached to an allowed network interface instance owner ({eni_owner})." ) assert ( sg.resource_arn @@ -650,7 +649,6 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: "SecurityGroups" ][0] default_sg_id = default_sg["GroupId"] - default_sg_name = default_sg["GroupName"] ec2_client.authorize_security_group_ingress( GroupId=default_sg_id, IpPermissions=[ @@ -721,15 +719,4 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port: result_specific_port = check_specific_port.execute() # One default sg per region - assert len(result_specific_port) == 3 - # Search changed sg - for sg in result_specific_port: - if sg.resource_id == default_sg_id: - assert sg.status == "PASS" - assert sg.region == AWS_REGION_US_EAST_1 - assert ( - sg.status_extended - == f"Security group {sg.resource_details} ({sg.resource_id}) has all ports open to the Internet and therefore was not checked against a specific port." - ) - assert sg.resource_tags == [] - assert sg.resource_details == default_sg_name + assert len(result_specific_port) == 2