mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
feat(EC2): Add new check for security group port restrictions (#4594)
This commit is contained in:
Mario Rodriguez Lopez
committed by
GitHub
GitHub
parent
e7d0d49809
commit
49ff901195
@@ -18,6 +18,7 @@ The following list includes all the AWS checks with configurable variables that
|
||||
| `ec2_elastic_ip_shodan` | `shodan_api_key` | String |
|
||||
| `ec2_securitygroup_with_many_ingress_egress_rules` | `max_security_group_rules` | Integer |
|
||||
| `ec2_instance_older_than_specific_days` | `max_ec2_instance_age_in_days` | Integer |
|
||||
| `ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports`| `ec2_sg_high_risk_ports` | List of Integer |
|
||||
| `vpc_endpoint_connections_trust_boundaries` | `trusted_account_ids` | List of Strings |
|
||||
| `vpc_endpoint_services_allowed_principals_trust_boundaries` | `trusted_account_ids` | List of Strings |
|
||||
| `cloudwatch_log_group_retention_policy_specific_days_enabled` | `log_group_retention_days` | Integer |
|
||||
@@ -126,6 +127,21 @@ aws:
|
||||
[
|
||||
"amazon-elb"
|
||||
]
|
||||
# aws.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports
|
||||
ec2_sg_high_risk_ports:
|
||||
[
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
|
||||
# AWS VPC Configuration (vpc_endpoint_connections_trust_boundaries, vpc_endpoint_services_allowed_principals_trust_boundaries)
|
||||
# Single account environment: No action required. The AWS account number will be automatically added by the checks.
|
||||
|
||||
@@ -41,6 +41,21 @@ aws:
|
||||
[
|
||||
"amazon-elb"
|
||||
]
|
||||
# aws.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports
|
||||
ec2_sg_high_risk_ports:
|
||||
[
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
|
||||
# AWS VPC Configuration (vpc_endpoint_connections_trust_boundaries, vpc_endpoint_services_allowed_principals_trust_boundaries)
|
||||
# Single account environment: No action required. The AWS account number will be automatically added by the checks.
|
||||
|
||||
+34
@@ -0,0 +1,34 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports",
|
||||
"CheckTitle": "Ensure no security groups allow ingress from 0.0.0.0/0 or ::/0 to high risk ports.",
|
||||
"CheckType": [
|
||||
"Infrastructure Security"
|
||||
],
|
||||
"ServiceName": "ec2",
|
||||
"SubServiceName": "securitygroup",
|
||||
"ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
|
||||
"Severity": "critical",
|
||||
"ResourceType": "AwsEc2SecurityGroup",
|
||||
"Description": "Ensure no security groups allow ingress from 0.0.0.0/0 or ::/0 to ports 25(SMTP), 110(POP3), 135(RCP), 143(IMAP), 445(CIFS), 3000(Go, Node.js, and Ruby web developemnt frameworks), 4333(ahsp), 5000(Python web development frameworks), 5500(fcp-addr-srvr1), 8080(proxy), 8088(legacy HTTP port).",
|
||||
"Risk": "If Security groups are not properly configured the attack surface is increased.",
|
||||
"RelatedUrl": "",
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Use a Zero Trust approach. Narrow ingress traffic as much as possible. Consider north-south as well as east-west traffic.",
|
||||
"Url": "https://docs.aws.amazon.com/vpc/latest/userguide/VPC_SecurityGroups.html"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
|
||||
)
|
||||
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
|
||||
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
|
||||
|
||||
|
||||
class ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports(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
|
||||
):
|
||||
check_ports = ec2_client.audit_config.get(
|
||||
"ec2_high_risk_ports",
|
||||
[25, 110, 135, 143, 445, 3000, 4333, 5000, 5500, 8080, 8088],
|
||||
)
|
||||
for port in check_ports:
|
||||
report = Check_Report_AWS(self.metadata())
|
||||
report.region = security_group.region
|
||||
report.resource_details = security_group.name
|
||||
report.resource_id = security_group.id
|
||||
report.resource_arn = security_group_arn
|
||||
report.resource_tags = security_group.tags
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have port {port} open to the Internet."
|
||||
# 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,
|
||||
):
|
||||
# Loop through every security group's ingress rule and check it
|
||||
for ingress_rule in security_group.ingress_rules:
|
||||
if check_security_group(
|
||||
ingress_rule, "tcp", [port], any_address=True
|
||||
):
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has port {port} (high risk port) open to the Internet."
|
||||
break
|
||||
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 port {port}."
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -79,6 +79,19 @@ config_aws = {
|
||||
"max_ec2_instance_age_in_days": 180,
|
||||
"ec2_allowed_interface_types": ["api_gateway_managed", "vpc_endpoint"],
|
||||
"ec2_allowed_instance_owners": ["amazon-elb"],
|
||||
"ec2_sg_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
],
|
||||
"trusted_account_ids": [],
|
||||
"log_group_retention_days": 365,
|
||||
"max_idle_disconnect_timeout_in_seconds": 600,
|
||||
|
||||
@@ -41,6 +41,21 @@ aws:
|
||||
[
|
||||
"amazon-elb"
|
||||
]
|
||||
# aws.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports
|
||||
ec2_sg_high_risk_ports:
|
||||
[
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
|
||||
# AWS VPC Configuration (vpc_endpoint_connections_trust_boundaries, vpc_endpoint_services_allowed_principals_trust_boundaries)
|
||||
# Single account environment: No action required. The AWS account number will be automatically added by the checks.
|
||||
|
||||
@@ -92,4 +92,3 @@ eks_required_log_types:
|
||||
"controllerManager",
|
||||
"scheduler",
|
||||
]
|
||||
|
||||
|
||||
+475
@@ -0,0 +1,475 @@
|
||||
from unittest import mock
|
||||
|
||||
from boto3 import client, resource
|
||||
from moto import mock_aws
|
||||
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_REGION_EU_WEST_1,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports:
|
||||
@mock_aws
|
||||
def test_ec2_default_sgs(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
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_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result = check.execute()
|
||||
|
||||
# One default sg per region, each port should be checked
|
||||
expected_findings_count = (
|
||||
len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3
|
||||
) # 3 security groups
|
||||
assert len(result) == expected_findings_count
|
||||
# Search changed sg
|
||||
for sg in result:
|
||||
# All are compliant by default
|
||||
assert sg.status == "PASS"
|
||||
|
||||
@mock_aws
|
||||
def test_ec2_non_compliant_default_sg(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
security_groups = ec2_client.describe_security_groups(GroupNames=["default"])[
|
||||
"SecurityGroups"
|
||||
]
|
||||
default_sg_id = security_groups[0]["GroupId"]
|
||||
default_sg_name = security_groups[0]["GroupName"]
|
||||
ec2_client.authorize_security_group_ingress(
|
||||
GroupId=default_sg_id,
|
||||
IpPermissions=[
|
||||
{
|
||||
"IpProtocol": "tcp",
|
||||
"FromPort": 20,
|
||||
"ToPort": 9000,
|
||||
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
|
||||
}
|
||||
],
|
||||
)
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
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_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result = check.execute()
|
||||
|
||||
# One default sg per region, each port should be checked
|
||||
expected_findings_count = (
|
||||
len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3
|
||||
) # 3 security groups
|
||||
assert len(result) == expected_findings_count
|
||||
# Search changed sg
|
||||
iterator = 0
|
||||
for sg in result:
|
||||
port = ec2_client.audit_config["ec2_high_risk_ports"][iterator]
|
||||
if sg.resource_id == default_sg_id:
|
||||
assert sg.status == "FAIL"
|
||||
assert sg.region == AWS_REGION_US_EAST_1
|
||||
assert (
|
||||
sg.status_extended
|
||||
== f"Security group {default_sg_name} ({default_sg_id}) has port {port} (high risk port) open to the Internet."
|
||||
)
|
||||
assert (
|
||||
sg.resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:security-group/{default_sg_id}"
|
||||
)
|
||||
assert sg.resource_details == default_sg_name
|
||||
assert sg.resource_tags == []
|
||||
iterator = (iterator + 1) % len(
|
||||
ec2_client.audit_config["ec2_high_risk_ports"]
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_ec2_compliant_default_sg(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
|
||||
"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=[
|
||||
{
|
||||
"IpProtocol": "tcp",
|
||||
"FromPort": 20,
|
||||
"ToPort": 9000,
|
||||
"IpRanges": [{"CidrIp": "123.123.123.123/32"}],
|
||||
}
|
||||
],
|
||||
)
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
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_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result = check.execute()
|
||||
|
||||
# One default sg per region, each port should be checked
|
||||
expected_findings_count = (
|
||||
len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3
|
||||
) # 3 security groups
|
||||
assert len(result) == expected_findings_count
|
||||
# Search changed sg
|
||||
iterator = 0
|
||||
for sg in result:
|
||||
port = ec2_client.audit_config["ec2_high_risk_ports"][iterator]
|
||||
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 {default_sg_name} ({default_sg_id}) does not have port {port} open to the Internet."
|
||||
)
|
||||
assert (
|
||||
sg.resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:security-group/{default_sg_id}"
|
||||
)
|
||||
assert sg.resource_details == default_sg_name
|
||||
assert sg.resource_tags == []
|
||||
iterator = (iterator + 1) % len(
|
||||
ec2_client.audit_config["ec2_high_risk_ports"]
|
||||
)
|
||||
|
||||
@mock_aws
|
||||
def test_ec2_default_sgs_ignoring(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
aws_provider = set_mocked_aws_provider(
|
||||
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
|
||||
scan_unused_services=False,
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_ec2_default_sgs_ignoring_vpc_in_use(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
|
||||
ec2.create_network_interface(SubnetId=subnet.id)
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
aws_provider = set_mocked_aws_provider(
|
||||
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
|
||||
scan_unused_services=False,
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result = check.execute()
|
||||
|
||||
expected_findings_count = len(
|
||||
ec2_client.audit_config["ec2_high_risk_ports"]
|
||||
) # 1 security group in use
|
||||
assert len(result) == expected_findings_count
|
||||
for sg in result:
|
||||
assert sg.status == "PASS"
|
||||
assert sg.region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
|
||||
# Create EC2 Mocked Resources
|
||||
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
|
||||
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
|
||||
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
|
||||
"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=[
|
||||
{
|
||||
"IpProtocol": "-1",
|
||||
"FromPort": -1,
|
||||
"ToPort": -1,
|
||||
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
|
||||
},
|
||||
{
|
||||
"IpProtocol": "tcp",
|
||||
"FromPort": 22,
|
||||
"ToPort": 22,
|
||||
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
|
||||
},
|
||||
],
|
||||
)
|
||||
ec2_client.audit_config = {
|
||||
"ec2_high_risk_ports": [
|
||||
25,
|
||||
110,
|
||||
135,
|
||||
143,
|
||||
445,
|
||||
3000,
|
||||
4333,
|
||||
5000,
|
||||
5500,
|
||||
8080,
|
||||
8088,
|
||||
]
|
||||
}
|
||||
|
||||
from prowler.providers.aws.services.ec2.ec2_service import EC2
|
||||
from prowler.providers.aws.services.vpc.vpc_service import VPC
|
||||
|
||||
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_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
|
||||
new=EC2(aws_provider),
|
||||
) as ec2_client_instance, mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
|
||||
new=VPC(aws_provider),
|
||||
) as vpc_client_instance:
|
||||
# Run check for all ports
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
|
||||
)
|
||||
|
||||
check_all_ports = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
|
||||
)
|
||||
result_all_ports = check_all_ports.execute()
|
||||
|
||||
# Verify that the all ports check has detected the issue
|
||||
assert any(
|
||||
sg.status == "FAIL" and sg.resource_id == default_sg_id
|
||||
for sg in result_all_ports
|
||||
)
|
||||
|
||||
# use the same mock objects for the specific port check
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_client",
|
||||
new=ec2_client_instance,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.vpc_client",
|
||||
new=vpc_client_instance,
|
||||
):
|
||||
# Now run the specific port check
|
||||
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports import (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports,
|
||||
)
|
||||
|
||||
check_specific_port = (
|
||||
ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports()
|
||||
)
|
||||
result_specific_port = check_specific_port.execute()
|
||||
|
||||
# One default sg per region, each port should be checked
|
||||
expected_findings_count = (
|
||||
len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3
|
||||
) # 2 default security groups + 1 added
|
||||
assert len(result_specific_port) == expected_findings_count
|
||||
# Search changed sg
|
||||
iterator = 0
|
||||
for sg in result_specific_port:
|
||||
port = ec2_client.audit_config["ec2_high_risk_ports"][iterator]
|
||||
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 port {port}."
|
||||
)
|
||||
assert sg.resource_tags == []
|
||||
assert sg.resource_details == default_sg_name
|
||||
iterator = (iterator + 1) % len(
|
||||
ec2_client.audit_config["ec2_high_risk_ports"]
|
||||
)
|
||||
Reference in New Issue
Block a user