fix(vpc): add new principal wildcard verification (#6461)

This commit is contained in:
Rubén De la Torre Vico
2025-01-13 15:49:10 +01:00
committed by GitHub
parent 2386490002
commit 7777c8f135
2 changed files with 77 additions and 15 deletions
@@ -10,16 +10,17 @@ class vpc_endpoint_services_allowed_principals_trust_boundaries(Check):
# Get trusted account_ids from prowler.config.yaml
trusted_account_ids = vpc_client.audit_config.get("trusted_account_ids", [])
for service in vpc_client.vpc_endpoint_services:
report = Check_Report_AWS(self.metadata())
report.region = service.region
report.resource_id = service.id
report.resource_arn = service.arn
report.resource_tags = service.tags
if not service.allowed_principals:
report = Check_Report_AWS(self.metadata())
report.region = service.region
report.status = "PASS"
report.status_extended = (
f"VPC Endpoint Service {service.id} has no allowed principals."
)
report.resource_id = service.id
report.resource_arn = service.arn
report.resource_tags = service.tags
findings.append(report)
else:
for principal in service.allowed_principals:
@@ -27,27 +28,24 @@ class vpc_endpoint_services_allowed_principals_trust_boundaries(Check):
pattern = compile(r"^[0-9]{12}$")
match = pattern.match(principal)
if not match:
account_id = principal.split(":")[4]
account_id = (
principal.split(":")[4] if principal != "*" else "*"
)
else:
account_id = match.string
report = Check_Report_AWS(self.metadata())
report.region = service.region
if (
account_id in trusted_account_ids
or account_id in vpc_client.audited_account
):
report.status = "PASS"
report.status_extended = f"Found trusted account {account_id} in VPC Endpoint Service {service.id}."
report.resource_id = service.id
report.resource_arn = service.arn
report.resource_tags = service.tags
else:
report.status = "FAIL"
report.status_extended = f"Found untrusted account {account_id} in VPC Endpoint Service {service.id}."
report.resource_id = service.id
report.resource_arn = service.arn
report.resource_tags = service.tags
if account_id == "*":
report.status_extended = f"Wildcard principal found in VPC Endpoint Service {service.id}."
else:
report.status_extended = f"Found untrusted account {account_id} in VPC Endpoint Service {service.id}."
findings.append(report)
return findings
@@ -367,3 +367,67 @@ class Test_vpc_endpoint_services_allowed_principals_trust_boundaries:
assert result[0].resource_arn == endpoint_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_vpc_endpoint_service_with_principal_wildcard(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
elbv2_client = client("elbv2", region_name=AWS_REGION_US_EAST_1)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
subnet = ec2_client.create_subnet(
VpcId=vpc["Vpc"]["VpcId"],
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
)
lb_name = "lb_vpce-test"
lb_arn = elbv2_client.create_load_balancer(
Name=lb_name,
Subnets=[subnet["Subnet"]["SubnetId"]],
Scheme="internal",
Type="network",
)["LoadBalancers"][0]["LoadBalancerArn"]
endpoint_id = ec2_client.create_vpc_endpoint_service_configuration(
NetworkLoadBalancerArns=[lb_arn]
)["ServiceConfiguration"]["ServiceId"]
# Add allowed principals
ec2_client.modify_vpc_endpoint_service_permissions(
ServiceId=endpoint_id, AddAllowedPrincipals=["*"]
)
endpoint_arn = f"arn:aws:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:vpc-endpoint-service/{endpoint_id}"
from prowler.providers.aws.services.vpc.vpc_service import VPC
aws_provider = set_mocked_aws_provider(audited_regions=[AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_services_allowed_principals_trust_boundaries.vpc_endpoint_services_allowed_principals_trust_boundaries.vpc_client",
new=VPC(aws_provider),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_services_allowed_principals_trust_boundaries.vpc_endpoint_services_allowed_principals_trust_boundaries import (
vpc_endpoint_services_allowed_principals_trust_boundaries,
)
check = vpc_endpoint_services_allowed_principals_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Wildcard principal found in VPC Endpoint Service {endpoint_id}."
)
assert result[0].resource_id == endpoint_id
assert result[0].resource_arn == endpoint_arn
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1