From 7777c8f135103e8f518ee749dff62baae20d23dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= Date: Mon, 13 Jan 2025 15:49:10 +0100 Subject: [PATCH] fix(vpc): add new principal wildcard verification (#6461) --- ...ces_allowed_principals_trust_boundaries.py | 28 ++++---- ...llowed_principals_trust_boundaries_test.py | 64 +++++++++++++++++++ 2 files changed, 77 insertions(+), 15 deletions(-) diff --git a/prowler/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries.py b/prowler/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries.py index 9f04cf54da..7dd7bb919f 100644 --- a/prowler/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries.py +++ b/prowler/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries.py @@ -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 diff --git a/tests/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries_test.py b/tests/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries_test.py index c7993423a3..eda3aafa12 100644 --- a/tests/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries_test.py +++ b/tests/providers/aws/services/vpc/vpc_endpoint_services_allowed_principals_trust_boundaries/vpc_endpoint_services_allowed_principals_trust_boundaries_test.py @@ -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