chore(ec2): enhance security group with any open port check (#5215)

This commit is contained in:
Sergio Garcia
2024-09-30 14:53:04 -04:00
committed by GitHub
parent 8e4847ec89
commit 38e024216c
5 changed files with 77 additions and 87 deletions
@@ -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": {
@@ -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
@@ -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
@@ -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",
@@ -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