ghndrx/prowler
1
0
Fork 0
mirror of https://github.com/ghndrx/prowler.git synced 2026-02-10 14:55:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(vpc services): list to dicts in vpc and subnets (#2310)

Browse Source
This commit is contained in:
Nacho Rivera
2023-05-04 15:35:02 +02:00
committed by GitHub
parent 9b551ef0ba
commit 05e4911d6f
9 changed files with 59 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
prowler/providers/aws/services/networkfirewall/networkfirewall_in_all_vpc/networkfirewall_in_all_vpc.py
View File

@@ -8,7 +8,7 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class networkfirewall_in_all_vpc(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs:
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_id = vpc.id

27
prowler/providers/aws/services/vpc/vpc_different_regions/vpc_different_regions.py
View File

@@ -5,21 +5,20 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class vpc_different_regions(Check):
def execute(self):
findings = []
region = None
for vpc in vpc_client.vpcs:
vpc_regions = set()
for vpc in vpc_client.vpcs.values():
if not vpc.default:
report = Check_Report_AWS(self.metadata())
# This is a global check under the vpc service: region, resource_id and tags are not relevant here but we keep them for consistency
report.region = vpc.region
report.resource_id = vpc.id
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = f"VPCs found only in one region {vpc.region}."
if region and vpc.region != region:
report.status = "PASS"
report.status_extended = "VPCs found in more than one region."
break
region = vpc.region
vpc_regions.add(vpc.region)
report = Check_Report_AWS(self.metadata())
# This is a global check under the vpc service: region, resource_id and tags are not relevant here but we keep them for consistency
report.region = vpc_client.region
report.resource_id = vpc_client.audited_account
report.status = "FAIL"
report.status_extended = "VPCs found only in one region"
if len(vpc_regions) > 1:
report.status = "PASS"
report.status_extended = "VPCs found in more than one region."
findings.append(report)

2
prowler/providers/aws/services/vpc/vpc_flow_logs_enabled/vpc_flow_logs_enabled.py
View File

@@ -5,7 +5,7 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class vpc_flow_logs_enabled(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs:
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags

29
prowler/providers/aws/services/vpc/vpc_service.py
View File

@@ -17,7 +17,7 @@ class VPC:
self.audited_account = audit_info.audited_account
self.audit_resources = audit_info.audit_resources
self.regional_clients = generate_regional_clients(self.service, audit_info)
self.vpcs = []
self.vpcs = {}
self.vpc_peering_connections = []
self.vpc_endpoints = []
self.vpc_endpoint_services = []
@@ -28,8 +28,13 @@ class VPC:
self.__describe_flow_logs__()
self.__describe_peering_route_tables__()
self.__describe_vpc_endpoint_service_permissions__()
self.vpc_subnets = []
self.vpc_subnets = {}
self.__threading_call__(self.__describe_vpc_subnets__)
self.region = (
audit_info.profile_region
if audit_info.profile_region
else list(self.regional_clients.keys())[0]
)
def __get_session__(self):
return self.session
@@ -52,14 +57,12 @@ class VPC:
if not self.audit_resources or (
is_resource_filtered(vpc["VpcId"], self.audit_resources)
):
self.vpcs.append(
VPCs(
id=vpc["VpcId"],
default=vpc["IsDefault"],
cidr_block=vpc["CidrBlock"],
region=regional_client.region,
tags=vpc.get("Tags"),
)
self.vpcs[vpc["VpcId"]] = VPCs(
id=vpc["VpcId"],
default=vpc["IsDefault"],
cidr_block=vpc["CidrBlock"],
region=regional_client.region,
tags=vpc.get("Tags"),
)
except Exception as error:
logger.error(
@@ -135,7 +138,7 @@ class VPC:
def __describe_flow_logs__(self):
logger.info("VPC - Describing flow logs...")
try:
for vpc in self.vpcs:
for vpc in self.vpcs.values():
regional_client = self.regional_clients[vpc.region]
flow_logs = regional_client.describe_flow_logs(
Filters=[
@@ -289,9 +292,9 @@ class VPC:
public=public,
tags=subnet.get("Tags"),
)
self.vpc_subnets.append(object)
self.vpc_subnets[subnet["SubnetId"]] = object
# Add it to the VPC object
for vpc in self.vpcs:
for vpc in self.vpcs.values():
if vpc.id == subnet["VpcId"]:
vpc.subnets.append(object)
except Exception as error:

2
prowler/providers/aws/services/vpc/vpc_subnet_different_az/vpc_subnet_different_az.py
View File

@@ -5,7 +5,7 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class vpc_subnet_different_az(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs:
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags

2
prowler/providers/aws/services/vpc/vpc_subnet_separate_private_public/vpc_subnet_separate_private_public.py
View File

@@ -5,7 +5,7 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class vpc_subnet_separate_private_public(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs:
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags

22
tests/providers/aws/services/networkfirewall/networkfirewall_in_all_vpc/networkfirewall_in_all_vpc_test.py
View File

@@ -20,7 +20,7 @@ class Test_networkfirewall_in_all_vpc:
networkfirewall_client.network_firewalls = []
vpc_client = mock.MagicMock
vpc_client.region = AWS_REGION
vpc_client.vpcs = []
vpc_client.vpcs = {}
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_service.NetworkFirewall",
new=networkfirewall_client,
@@ -55,8 +55,8 @@ class Test_networkfirewall_in_all_vpc:
]
vpc_client = mock.MagicMock
vpc_client.region = AWS_REGION
vpc_client.vpcs = [
VPCs(
vpc_client.vpcs = {
VPC_ID_PROTECTED: VPCs(
id=VPC_ID_PROTECTED,
default=False,
cidr_block="192.168.0.0/16",
@@ -76,7 +76,7 @@ class Test_networkfirewall_in_all_vpc:
],
tags=[],
)
]
}
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_service.NetworkFirewall",
new=networkfirewall_client,
@@ -110,8 +110,8 @@ class Test_networkfirewall_in_all_vpc:
networkfirewall_client.network_firewalls = []
vpc_client = mock.MagicMock
vpc_client.region = AWS_REGION
vpc_client.vpcs = [
VPCs(
vpc_client.vpcs = {
VPC_ID_UNPROTECTED: VPCs(
id=VPC_ID_UNPROTECTED,
default=False,
cidr_block="192.168.0.0/16",
@@ -131,7 +131,7 @@ class Test_networkfirewall_in_all_vpc:
],
tags=[],
)
]
}
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_service.NetworkFirewall",
new=networkfirewall_client,
@@ -175,8 +175,8 @@ class Test_networkfirewall_in_all_vpc:
]
vpc_client = mock.MagicMock
vpc_client.region = AWS_REGION
vpc_client.vpcs = [
VPCs(
vpc_client.vpcs = {
VPC_ID_UNPROTECTED: VPCs(
id=VPC_ID_UNPROTECTED,
default=False,
cidr_block="192.168.0.0/16",
@@ -196,7 +196,7 @@ class Test_networkfirewall_in_all_vpc:
],
tags=[],
),
VPCs(
VPC_ID_PROTECTED: VPCs(
id=VPC_ID_PROTECTED,
default=False,
cidr_block="192.168.0.0/16",
@@ -216,7 +216,7 @@ class Test_networkfirewall_in_all_vpc:
],
tags=[],
),
]
}
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_service.NetworkFirewall",
new=networkfirewall_client,

32
tests/providers/aws/services/vpc/vpc_different_regions/vpc_different_regions_test.py
View File

@@ -23,7 +23,7 @@ class Test_vpc_different_regions:
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=["us-east-1", "eu-west-1"],
profile_region="us-east-1",
credentials=None,
assumed_role_info=None,
audited_regions=["us-east-1", "eu-west-1"],
@@ -37,14 +37,10 @@ class Test_vpc_different_regions:
def test_vpc_different_regions(self):
# VPC Region 1
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
ec2_client.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
# VPC Region 2
ec2_client_eu = client("ec2", region_name="eu-west-1")
vpc_eu = ec2_client_eu.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
ec2_client_eu.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
from prowler.providers.aws.services.vpc.vpc_service import VPC
@@ -67,24 +63,19 @@ class Test_vpc_different_regions:
result = check.execute()
assert len(result) == 1
assert result[0].region in ["us-east-1", "eu-west-1"]
assert result[0].status == "PASS"
assert result[0].region == "us-east-1"
assert (
result[0].status_extended == "VPCs found in more than one region."
)
assert result[0].resource_id in [
vpc["Vpc"]["VpcId"],
vpc_eu["Vpc"]["VpcId"],
]
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_tags == []
assert result[0].status == "PASS"
@mock_ec2
def test_vpc_only_one_regions(self):
ec2_client = client("ec2", region_name=AWS_REGION)
# VPC Region 1
vpc = ec2_client.create_vpc(
CidrBlock="172.28.6.0/24", InstanceTenancy="default"
)
ec2_client.create_vpc(CidrBlock="172.28.6.0/24", InstanceTenancy="default")
from prowler.providers.aws.services.vpc.vpc_service import VPC
@@ -108,10 +99,7 @@ class Test_vpc_different_regions:
assert len(result) == 1
assert result[0].status == "FAIL"
assert result[0].region == AWS_REGION
assert (
result[0].status_extended
== f"VPCs found only in one region {AWS_REGION}."
)
assert result[0].resource_id == vpc["Vpc"]["VpcId"]
assert result[0].region == "us-east-1"
assert result[0].status_extended == "VPCs found only in one region"
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_tags == []

10
tests/providers/aws/services/vpc/vpc_service_test.py
View File

@@ -90,7 +90,7 @@ class Test_VPC_Service:
assert (
len(vpc.vpcs) == 3
) # Number of AWS regions + created VPC, one default VPC per region
for vpc in vpc.vpcs:
for vpc in vpc.vpcs.values():
if vpc.cidr_block == "10.0.0.0/16":
assert vpc.tags == [
{"Key": "test", "Value": "test"},
@@ -117,9 +117,9 @@ class Test_VPC_Service:
audit_info = self.set_mocked_audit_info()
vpc = VPC(audit_info)
# Search created VPC among default ones
for vpc in vpc.vpcs:
if vpc.id == new_vpc["VpcId"]:
assert vpc.flow_log is True
for vpc_iter in vpc.vpcs.values():
if vpc_iter.id == new_vpc["VpcId"]:
assert vpc_iter.flow_log is True
# Test VPC Describe VPC Peering connections
@mock_ec2
@@ -303,7 +303,7 @@ class Test_VPC_Service:
assert (
len(vpc.vpcs) == 3
) # Number of AWS regions + created VPC, one default VPC per region
for vpc in vpc.vpcs:
for vpc in vpc.vpcs.values():
if vpc.cidr_block == "172.28.7.0/24":
assert vpc.subnets[0].id == subnet["Subnet"]["SubnetId"]
assert vpc.subnets[0].default is False