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

feat(vpc): add vpc, nacl or subnet names in findings (#2928)

Browse Source
This commit is contained in:
Sergio Garcia
2023-10-18 16:07:53 +02:00
committed by GitHub
parent 236f57ab0e
commit eb3cb97115
15 changed files with 170 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
prowler/providers/aws/services/ec2/ec2_networkacl_allow_ingress_any_port/ec2_networkacl_allow_ingress_any_port.py
View File

@@ -15,13 +15,11 @@ class ec2_networkacl_allow_ingress_any_port(Check):
report.resource_arn = network_acl.arn
report.resource_tags = network_acl.tags
report.status = "PASS"
report.status_extended = f"Network ACL {network_acl.id} does not have every port open to the Internet."
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} does not have every port open to the Internet."
# If some entry allows it, that ACL is not securely configured
if check_network_acl(network_acl.entries, tcp_protocol, check_port):
report.status = "FAIL"
report.status_extended = (
f"Network ACL {network_acl.id} has every port open to the Internet."
)
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} has every port open to the Internet."
findings.append(report)
return findings

4
prowler/providers/aws/services/ec2/ec2_networkacl_allow_ingress_tcp_port_22/ec2_networkacl_allow_ingress_tcp_port_22.py
View File

@@ -14,12 +14,12 @@ class ec2_networkacl_allow_ingress_tcp_port_22(Check):
report.resource_arn = network_acl.arn
report.resource_tags = network_acl.tags
report.status = "PASS"
report.status_extended = f"Network ACL {network_acl.id} does not have SSH port 22 open to the Internet."
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} does not have SSH port 22 open to the Internet."
report.resource_id = network_acl.id
# If some entry allows it, that ACL is not securely configured
if check_network_acl(network_acl.entries, tcp_protocol, check_port):
report.status = "FAIL"
report.status_extended = f"Network ACL {network_acl.id} has SSH port 22 open to the Internet."
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} has SSH port 22 open to the Internet."
findings.append(report)
return findings

4
prowler/providers/aws/services/ec2/ec2_networkacl_allow_ingress_tcp_port_3389/ec2_networkacl_allow_ingress_tcp_port_3389.py
View File

@@ -14,12 +14,12 @@ class ec2_networkacl_allow_ingress_tcp_port_3389(Check):
report.resource_arn = network_acl.arn
report.resource_tags = network_acl.tags
report.status = "PASS"
report.status_extended = f"Network ACL {network_acl.id} does not have Microsoft RDP port 3389 open to the Internet."
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} does not have Microsoft RDP port 3389 open to the Internet."
report.resource_id = network_acl.id
# If some entry allows it, that ACL is not securely configured
if check_network_acl(network_acl.entries, tcp_protocol, check_port):
report.status = "FAIL"
report.status_extended = f"Network ACL {network_acl.id} has Microsoft RDP port 3389 open to the Internet."
report.status_extended = f"Network ACL {network_acl.name if network_acl.name else network_acl.id} has Microsoft RDP port 3389 open to the Internet."
findings.append(report)
return findings

6
prowler/providers/aws/services/ec2/ec2_service.py
View File

@@ -157,10 +157,15 @@ class EC2(AWSService):
if not self.audit_resources or (
is_resource_filtered(arn, self.audit_resources)
):
nacl_name = ""
for tag in nacl.get("Tags", []):
if tag["Key"] == "Name":
nacl_name = tag["Value"]
self.network_acls.append(
NetworkACL(
id=nacl["NetworkAclId"],
arn=arn,
name=nacl_name,
region=regional_client.region,
entries=nacl["Entries"],
tags=nacl.get("Tags"),
@@ -458,6 +463,7 @@ class SecurityGroup(BaseModel):
class NetworkACL(BaseModel):
id: str
arn: str
name: str
region: str
entries: list[dict]
tags: Optional[list] = []

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

@@ -15,15 +15,11 @@ class networkfirewall_in_all_vpc(Check):
report.resource_arn = vpc.arn
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = (
f"VPC {vpc.id} does not have Network Firewall enabled."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} does not have Network Firewall enabled."
for firewall in networkfirewall_client.network_firewalls:
if firewall.vpc_id == vpc.id:
report.status = "PASS"
report.status_extended = (
f"VPC {vpc.id} has Network Firewall enabled."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has Network Firewall enabled."
break
findings.append(report)

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

@@ -12,10 +12,14 @@ class vpc_flow_logs_enabled(Check):
report.resource_id = vpc.id
report.resource_arn = vpc.arn
report.status = "FAIL"
report.status_extended = f"VPC {vpc.id} Flow logs are disabled."
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} Flow logs are disabled."
)
if vpc.flow_log:
report.status = "PASS"
report.status_extended = f"VPC {vpc.id} Flow logs are enabled."
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} Flow logs are enabled."
)
findings.append(report)

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

@@ -38,9 +38,14 @@ class VPC(AWSService):
if not self.audit_resources or (
is_resource_filtered(arn, self.audit_resources)
):
vpc_name = ""
for tag in vpc.get("Tags", []):
if tag["Key"] == "Name":
vpc_name = tag["Value"]
self.vpcs[vpc["VpcId"]] = VPCs(
arn=arn,
id=vpc["VpcId"],
name=vpc_name,
default=vpc["IsDefault"],
cidr_block=vpc["CidrBlock"],
region=regional_client.region,
@@ -278,10 +283,15 @@ class VPC(AWSService):
public = True
if "NatGatewayId" in route:
nat_gateway = True
subnet_name = ""
for tag in subnet.get("Tags", []):
if tag["Key"] == "Name":
subnet_name = tag["Value"]
# Add it to to list of vpc_subnets and to the VPC object
object = VpcSubnet(
arn=subnet["SubnetArn"],
id=subnet["SubnetId"],
name=subnet_name,
default=subnet["DefaultForAz"],
vpc_id=subnet["VpcId"],
cidr_block=subnet.get("CidrBlock"),
@@ -310,6 +320,7 @@ class VPC(AWSService):
class VpcSubnet(BaseModel):
arn: str
id: str
name: str
default: bool
vpc_id: str
cidr_block: Optional[str]
@@ -324,6 +335,7 @@ class VpcSubnet(BaseModel):
class VPCs(BaseModel):
arn: str
id: str
name: str
default: bool
cidr_block: str
flow_log: bool = False

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

@@ -10,7 +10,9 @@ class vpc_subnet_different_az(Check):
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = f"VPC {vpc.id} has no subnets."
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
@@ -21,12 +23,10 @@ class vpc_subnet_different_az(Check):
and subnet.availability_zone != availability_zone
):
report.status = "PASS"
report.status_extended = f"VPC {vpc.id} has subnets in more than one availability zone."
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has subnets in more than one availability zone."
break
availability_zone = subnet.availability_zone
report.status_extended = (
f"VPC {vpc.id} has only subnets in {availability_zone}."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only subnets in {availability_zone}."
findings.append(report)

8
prowler/providers/aws/services/vpc/vpc_subnet_no_public_ip_by_default/vpc_subnet_no_public_ip_by_default.py
View File

@@ -14,14 +14,10 @@ class vpc_subnet_no_public_ip_by_default(Check):
report.resource_arn = subnet.arn
if subnet.mapPublicIpOnLaunch:
report.status = "FAIL"
report.status_extended = (
f"VPC subnet {subnet.id} assigns public IP by default."
)
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."
else:
report.status = "PASS"
report.status_extended = (
f"VPC subnet {subnet.id} does NOT assign public IP by default."
)
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} does NOT assign public IP by default."
findings.append(report)
return findings

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

@@ -10,7 +10,9 @@ class vpc_subnet_separate_private_public(Check):
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = f"VPC {vpc.id} has no subnets."
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
@@ -19,19 +21,13 @@ class vpc_subnet_separate_private_public(Check):
for subnet in vpc.subnets:
if subnet.public:
public = True
report.status_extended = (
f"VPC {vpc.id} has only public subnets."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only public subnets."
if not subnet.public:
private = True
report.status_extended = (
f"VPC {vpc.id} has only private subnets."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only private subnets."
if public and private:
report.status = "PASS"
report.status_extended = (
f"VPC {vpc.id} has private and public subnets."
)
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has private and public subnets."
findings.append(report)
return findings

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

@@ -102,6 +102,7 @@ class Test_networkfirewall_in_all_vpc:
vpc_client.vpcs = {
VPC_ID_PROTECTED: VPCs(
id=VPC_ID_PROTECTED,
name="",
default=False,
cidr_block="192.168.0.0/16",
flow_log=False,
@@ -111,6 +112,7 @@ class Test_networkfirewall_in_all_vpc:
VpcSubnet(
id="subnet-123456789",
arn="arn_test",
name="",
default=False,
vpc_id=VPC_ID_PROTECTED,
cidr_block="192.168.0.0/24",
@@ -168,6 +170,7 @@ class Test_networkfirewall_in_all_vpc:
vpc_client.vpcs = {
VPC_ID_UNPROTECTED: VPCs(
id=VPC_ID_UNPROTECTED,
name="",
default=False,
cidr_block="192.168.0.0/16",
flow_log=False,
@@ -177,6 +180,7 @@ class Test_networkfirewall_in_all_vpc:
VpcSubnet(
id="subnet-123456789",
arn="arn_test",
name="",
default=False,
vpc_id=VPC_ID_UNPROTECTED,
cidr_block="192.168.0.0/24",
@@ -225,6 +229,74 @@ class Test_networkfirewall_in_all_vpc:
assert result[0].resource_tags == []
assert result[0].resource_arn == "arn_test"
def test_vpcs_with_name_without_firewall(self):
networkfirewall_client = mock.MagicMock
networkfirewall_client.region = AWS_REGION
networkfirewall_client.network_firewalls = []
vpc_client = mock.MagicMock
vpc_client.region = AWS_REGION
vpc_client.vpcs = {
VPC_ID_UNPROTECTED: VPCs(
id=VPC_ID_UNPROTECTED,
name="vpc_name",
default=False,
cidr_block="192.168.0.0/16",
flow_log=False,
region=AWS_REGION,
arn="arn_test",
subnets=[
VpcSubnet(
id="subnet-123456789",
arn="arn_test",
name="",
default=False,
vpc_id=VPC_ID_UNPROTECTED,
cidr_block="192.168.0.0/24",
availability_zone="us-east-1a",
public=False,
nat_gateway=False,
region=AWS_REGION,
tags=[],
mapPublicIpOnLaunch=False,
)
],
tags=[],
)
}
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_in_all_vpc.networkfirewall_in_all_vpc.vpc_client",
new=vpc_client,
):
with mock.patch(
"prowler.providers.aws.services.networkfirewall.networkfirewall_in_all_vpc.networkfirewall_in_all_vpc.networkfirewall_client",
new=networkfirewall_client,
):
# Test Check
from prowler.providers.aws.services.networkfirewall.networkfirewall_in_all_vpc.networkfirewall_in_all_vpc import (
networkfirewall_in_all_vpc,
)
check = networkfirewall_in_all_vpc()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "VPC vpc_name does not have Network Firewall enabled."
)
assert result[0].region == AWS_REGION
assert result[0].resource_id == VPC_ID_UNPROTECTED
assert result[0].resource_tags == []
assert result[0].resource_arn == "arn_test"
def test_vpcs_with_and_without_firewall(self):
networkfirewall_client = mock.MagicMock
networkfirewall_client.region = AWS_REGION
@@ -244,6 +316,7 @@ class Test_networkfirewall_in_all_vpc:
vpc_client.vpcs = {
VPC_ID_UNPROTECTED: VPCs(
id=VPC_ID_UNPROTECTED,
name="",
default=False,
cidr_block="192.168.0.0/16",
flow_log=False,
@@ -253,6 +326,7 @@ class Test_networkfirewall_in_all_vpc:
VpcSubnet(
id="subnet-123456789",
arn="arn_test",
name="",
default=False,
vpc_id=VPC_ID_UNPROTECTED,
cidr_block="192.168.0.0/24",
@@ -268,6 +342,7 @@ class Test_networkfirewall_in_all_vpc:
),
VPC_ID_PROTECTED: VPCs(
id=VPC_ID_PROTECTED,
name="",
default=False,
cidr_block="192.168.0.0/16",
flow_log=False,
@@ -277,6 +352,7 @@ class Test_networkfirewall_in_all_vpc:
VpcSubnet(
id="subnet-123456789",
arn="arn_test",
name="",
default=False,
vpc_id=VPC_ID_PROTECTED,
cidr_block="192.168.0.0/24",

15
tests/providers/aws/services/vpc/vpc_flow_logs_enabled/vpc_flow_logs_enabled_test.py
View File

@@ -71,7 +71,17 @@ class Test_vpc_flow_logs_enabled:
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
vpc = ec2_client.create_vpc(
CidrBlock="10.0.0.0/16",
TagSpecifications=[
{
"ResourceType": "vpc",
"Tags": [
{"Key": "Name", "Value": "vpc_name"},
],
},
],
)["Vpc"]
ec2_client.create_flow_logs(
ResourceType="VPC",
@@ -106,8 +116,7 @@ class Test_vpc_flow_logs_enabled:
if result.resource_id == vpc["VpcId"]:
assert result.status == "PASS"
assert (
result.status_extended
== f"VPC {vpc['VpcId']} Flow logs are enabled."
result.status_extended == "VPC vpc_name Flow logs are enabled."
)
assert result.resource_id == vpc["VpcId"]

17
tests/providers/aws/services/vpc/vpc_subnet_different_az/vpc_subnet_different_az_test.py
View File

@@ -46,7 +46,16 @@ class Test_vpc_subnet_different_az:
def test_vpc_subnet_different_az(self):
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
CidrBlock="172.28.7.0/24",
InstanceTenancy="default",
TagSpecifications=[
{
"ResourceType": "vpc",
"Tags": [
{"Key": "Name", "Value": "vpc_name"},
],
},
],
)
# VPC AZ 1
ec2_client.create_subnet(
@@ -88,10 +97,12 @@ class Test_vpc_subnet_different_az:
assert result.status == "PASS"
assert (
result.status_extended
== f"VPC {vpc['Vpc']['VpcId']} has subnets in more than one availability zone."
== "VPC vpc_name has subnets in more than one availability zone."
)
assert result.resource_id == vpc["Vpc"]["VpcId"]
assert result.resource_tags == []
assert result.resource_tags == [
{"Key": "Name", "Value": "vpc_name"}
]
assert result.region == AWS_REGION
if not found:
assert False

10
tests/providers/aws/services/vpc/vpc_subnet_no_public_ip_by_default/vpc_subnet_no_public_ip_by_default_test.py
View File

@@ -52,6 +52,14 @@ class Test_vpc_subnet_no_public_ip_by_default:
VpcId=vpc["Vpc"]["VpcId"],
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION}a",
TagSpecifications=[
{
"ResourceType": "subnet",
"Tags": [
{"Key": "Name", "Value": "subnet_name"},
],
},
],
)
ec2_client.modify_subnet_attribute(
@@ -83,7 +91,7 @@ class Test_vpc_subnet_no_public_ip_by_default:
assert result.status == "FAIL"
assert (
result.status_extended
== f"VPC subnet {subnet_private['Subnet']['SubnetId']} assigns public IP by default."
== "VPC subnet subnet_name assigns public IP by default."
)
@mock_ec2

17
tests/providers/aws/services/vpc/vpc_subnet_separate_private_public/vpc_subnet_separate_private_public_test.py
View File

@@ -46,7 +46,16 @@ class Test_vpc_subnet_separate_private_public:
def test_vpc_subnet_only_private(self):
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
CidrBlock="172.28.7.0/24",
InstanceTenancy="default",
TagSpecifications=[
{
"ResourceType": "vpc",
"Tags": [
{"Key": "Name", "Value": "vpc_name"},
],
},
],
)
# VPC Private
subnet_private = ec2_client.create_subnet(
@@ -92,10 +101,12 @@ class Test_vpc_subnet_separate_private_public:
assert result.status == "FAIL"
assert (
result.status_extended
== f"VPC {vpc['Vpc']['VpcId']} has only private subnets."
== "VPC vpc_name has only private subnets."
)
assert result.resource_id == vpc["Vpc"]["VpcId"]
assert result.resource_tags == []
assert result.resource_tags == [
{"Key": "Name", "Value": "vpc_name"}
]
assert result.region == AWS_REGION
if not found:
assert False