Skip to content

Commit

Permalink
fix(ec2 tests): add region and delete search sg checks (#2788)
Browse files Browse the repository at this point in the history
  • Loading branch information
@n4ch04
n4ch04 authored Aug 31, 2023
1 parent 38a7dc1 commit dbd29c0
Show file tree
Hide file tree
Showing 19 changed files with 162 additions and 127 deletions.
22 changes: 12 additions & 10 deletions ...om_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -122,9 +121,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has all ports open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has all ports open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -178,9 +178,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have all ports open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have all ports open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -239,9 +240,10 @@ def test_ec2_compliant_default_sg_only_open_to_one_port(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have all ports open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have all ports open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...7_27018/ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has MongoDB ports 27017 and 27018 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has MongoDB ports 27017 and 27018 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have MongoDB ports 27017 and 27018 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have MongoDB ports 27017 and 27018 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...tp_port_20_21/ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has FTP ports 20 and 21 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has FTP ports 20 and 21 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have FTP ports 20 and 21 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have FTP ports 20 and 21 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
12 changes: 9 additions & 3 deletions ...ernet_to_tcp_port_22/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,11 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has SSH port 22 open to the Internet."
)
assert search(
"has SSH port 22 open to the Internet",
sg.status_extended,
Expand Down Expand Up @@ -180,9 +185,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have SSH port 22 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have SSH port 22 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...t_to_tcp_port_3389/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -122,9 +121,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Microsoft RDP port 3389 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Microsoft RDP port 3389 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -180,9 +180,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Microsoft RDP port 3389 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Microsoft RDP port 3389 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...c2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Casandra ports 7199, 8888 and 9160 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Casandra ports 7199, 8888 and 9160 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Casandra ports 7199, 8888 and 9160 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Casandra ports 7199, 8888 and 9160 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...group_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Elasticsearch/Kibana ports 9200, 9300 and 5601 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Elasticsearch/Kibana ports 9200, 9300 and 5601 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Elasticsearch/Kibana ports 9200, 9300 and 5601 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Elasticsearch/Kibana ports 9200, 9300 and 5601 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
14 changes: 8 additions & 6 deletions ...t_kafka_9092/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,8 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Kafka port 9092 open to the Internet", sg.status_extended
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Kafka port 9092 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -185,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Kafka port 9092 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Kafka port 9092 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...d_11211/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Memcached port 11211 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Memcached port 11211 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Memcached port 11211 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Memcached port 11211 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...t_mysql_3306/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has MySQL port 3306 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has MySQL port 3306 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have MySQL port 3306 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have MySQL port 3306 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
15 changes: 8 additions & 7 deletions ...1_2483/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from re import search
from unittest import mock

from boto3 import client, session
Expand Down Expand Up @@ -126,9 +125,10 @@ def test_ec2_non_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "FAIL"
assert search(
"has Oracle ports 1521 and 2483 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) has Oracle ports 1521 and 2483 open to the Internet."
)
assert (
sg.resource_arn
Expand Down Expand Up @@ -186,9 +186,10 @@ def test_ec2_compliant_default_sg(self):
for sg in result:
if sg.resource_id == default_sg_id:
assert sg.status == "PASS"
assert search(
"does not have Oracle ports 1521 and 2483 open to the Internet",
sg.status_extended,
assert sg.region == AWS_REGION
assert (
sg.status_extended
== f"Security group {default_sg_name} ({default_sg_id}) does not have Oracle ports 1521 and 2483 open to the Internet."
)
assert (
sg.resource_arn
Expand Down
Loading

0 comments on commit dbd29c0

Please sign in to comment.