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_endpoint_connections_trust_boundaries): Handle AWS Account ID as Principal (#2611)

Browse Source
This commit is contained in:
Pepe Fagoaga
2023-08-03 09:16:58 +02:00
committed by GitHub
parent 92c8a440ea
commit 5bf3f70717
4 changed files with 782 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
prowler/providers/aws/lib/policy_condition_parser/policy_condition_parser.py
View File

@@ -1,39 +1,45 @@
# lista de cuentas y te devuelva las válidas
def is_account_only_allowed_in_condition(
condition_statement: dict, source_account: str
):
is_condition_valid = False
valid_condition_options = {
"StringEquals": "aws:SourceAccount",
"ArnLike": "aws:SourceArn",
"ArnEquals": "aws:SourceArn",
"StringEquals": [
"aws:SourceAccount",
"s3:ResourceAccount",
"aws:PrincipalAccount",
],
"StringLike": ["aws:SourceArn", "aws:PrincipalArn"],
"ArnLike": ["aws:SourceArn", "aws:PrincipalArn"],
"ArnEquals": ["aws:SourceArn", "aws:PrincipalArn"],
}
for condition_operator, condition_operator_key in valid_condition_options.items():
if condition_operator in condition_statement:
if condition_operator_key in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][condition_operator_key],
list,
):
# if there is an arn/account without the source account -> we do not consider it safe
# here by default we assume is true and look for false entries
is_condition_valid = True
for item in condition_statement[condition_operator][
condition_operator_key
]:
if source_account not in item:
is_condition_valid = False
break
# value is a string
elif isinstance(
condition_statement[condition_operator][condition_operator_key], str
):
if (
source_account
in condition_statement[condition_operator][
condition_operator_key
]
for value in condition_operator_key:
if value in condition_statement[condition_operator]:
# values are a list
if isinstance(
condition_statement[condition_operator][value],
list,
):
# if there is an arn/account without the source account -> we do not consider it safe
# here by default we assume is true and look for false entries
is_condition_valid = True
for item in condition_statement[condition_operator][value]:
if source_account not in item:
is_condition_valid = False
break
# value is a string
elif isinstance(
condition_statement[condition_operator][value],
str,
):
if (
source_account
in condition_statement[condition_operator][value]
):
is_condition_valid = True
return is_condition_valid

116
prowler/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries.py
View File

@@ -1,5 +1,10 @@
from re import compile
from prowler.config.config import get_config_var
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
is_account_only_allowed_in_condition,
)
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
@@ -8,20 +13,49 @@ class vpc_endpoint_connections_trust_boundaries(Check):
findings = []
# Get trusted account_ids from prowler.config.yaml
trusted_account_ids = get_config_var("trusted_account_ids")
# Always include the same account as trusted
trusted_account_ids.append(vpc_client.audited_account)
for endpoint in vpc_client.vpc_endpoints:
# Check VPC endpoint policy
if endpoint.policy_document:
access_from_trusted_accounts = True
for statement in endpoint.policy_document["Statement"]:
# If one policy allows access from a non-trusted account
if not access_from_trusted_accounts:
break
if "*" == statement["Principal"]:
report = Check_Report_AWS(self.metadata())
report.region = endpoint.region
report.status = "FAIL"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} has full access."
report.resource_id = endpoint.id
report.resource_arn = endpoint.arn
report.resource_tags = endpoint.tags
for account_id in trusted_account_ids:
if (
"Condition" in statement
and is_account_only_allowed_in_condition(
statement["Condition"], account_id
)
):
access_from_trusted_accounts = True
else:
access_from_trusted_accounts = False
break
if (
not access_from_trusted_accounts
or len(trusted_account_ids) == 0
):
access_from_trusted_accounts = False
report.status = "FAIL"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can be accessed from non-trusted accounts."
else:
report.status = "PASS"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can only be accessed from trusted accounts."
findings.append(report)
break
if not access_from_trusted_accounts:
break
else:
if isinstance(statement["Principal"]["AWS"], str):
@@ -29,31 +63,91 @@ class vpc_endpoint_connections_trust_boundaries(Check):
else:
principals = statement["Principal"]["AWS"]
for principal_arn in principals:
report = Check_Report_AWS(self.metadata())
report.region = endpoint.region
if principal_arn == "*":
report.status = "FAIL"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} has full access."
report = Check_Report_AWS(self.metadata())
report.region = endpoint.region
report.resource_id = endpoint.id
report.resource_arn = endpoint.arn
report.resource_tags = endpoint.tags
for account_id in trusted_account_ids:
if (
"Condition" in statement
and is_account_only_allowed_in_condition(
statement["Condition"], account_id
)
):
access_from_trusted_accounts = True
else:
access_from_trusted_accounts = False
break
if (
not access_from_trusted_accounts
or len(trusted_account_ids) == 0
):
access_from_trusted_accounts = False
report.status = "FAIL"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can be accessed from non-trusted accounts."
else:
report.status = "PASS"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can only be accessed from trusted accounts."
findings.append(report)
if not access_from_trusted_accounts:
break
else:
account_id = principal_arn.split(":")[4]
# Account ID can be an ARN or just a 12-digit string
pattern = compile(r"^[0-9]{12}$")
match = pattern.match(principal_arn)
if not match:
account_id = principal_arn.split(":")[4]
else:
account_id = match.string
if (
account_id in trusted_account_ids
or account_id in vpc_client.audited_account
):
report = Check_Report_AWS(self.metadata())
report.region = endpoint.region
report.status = "PASS"
report.status_extended = f"Found trusted account {account_id} in VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id}."
report.resource_id = endpoint.id
report.resource_arn = endpoint.arn
report.resource_tags = endpoint.tags
findings.append(report)
else:
report.status = "FAIL"
report.status_extended = f"Found untrusted account {account_id} in VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id}."
report = Check_Report_AWS(self.metadata())
report.region = endpoint.region
report.resource_id = endpoint.id
report.resource_arn = endpoint.arn
report.resource_tags = endpoint.tags
findings.append(report)
for account_id in trusted_account_ids:
if (
"Condition" in statement
and is_account_only_allowed_in_condition(
statement["Condition"], account_id
)
):
access_from_trusted_accounts = True
else:
access_from_trusted_accounts = False
break
if (
not access_from_trusted_accounts
or len(trusted_account_ids) == 0
):
access_from_trusted_accounts = False
report.status = "FAIL"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can be accessed from non-trusted accounts."
else:
report.status = "PASS"
report.status_extended = f"VPC Endpoint {endpoint.id} in VPC {endpoint.vpc_id} can only be accessed from trusted accounts."
findings.append(report)
if not access_from_trusted_accounts:
break
return findings

268
tests/providers/aws/lib/policy_condition_parser/policy_condition_parser_test.py
View File

@@ -6,19 +6,19 @@ AWS_ACCOUNT_NUMBER = "123456789012"
class Test_policy_condition_parser:
def test_condition_parser_string_equals_list(self):
def test_condition_parser_string_equals_aws_SourceAccount_list(self):
condition_statement = {"StringEquals": {"aws:SourceAccount": ["123456789012"]}}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_str(self):
def test_condition_parser_string_equals_aws_SourceAccount_str(self):
condition_statement = {"StringEquals": {"aws:SourceAccount": "123456789012"}}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_list_not_valid(self):
def test_condition_parser_string_equals_aws_SourceAccount_list_not_valid(self):
condition_statement = {
"StringEquals": {"aws:SourceAccount": ["123456789012", "111222333444"]}
}
@@ -26,13 +26,67 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_str_not_valid(self):
def test_condition_parser_string_equals_aws_SourceAccount_str_not_valid(self):
condition_statement = {"StringEquals": {"aws:SourceAccount": "111222333444"}}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnlike_list(self):
def test_condition_parser_string_equals_s3_ResourceAccount_list(self):
condition_statement = {"StringEquals": {"s3:ResourceAccount": ["123456789012"]}}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_s3_ResourceAccount_str(self):
condition_statement = {"StringEquals": {"s3:ResourceAccount": "123456789012"}}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_s3_ResourceAccount_list_not_valid(self):
condition_statement = {
"StringEquals": {"s3:ResourceAccount": ["123456789012", "111222333444"]}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_s3_ResourceAccount_str_not_valid(self):
condition_statement = {"StringEquals": {"s3:ResourceAccount": "111222333444"}}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_aws_PrincipalAccount_list(self):
condition_statement = {
"StringEquals": {"aws:PrincipalAccount": ["123456789012"]}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_aws_PrincipalAccount_str(self):
condition_statement = {"StringEquals": {"aws:PrincipalAccount": "123456789012"}}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_aws_PrincipalAccount_list_not_valid(self):
condition_statement = {
"StringEquals": {"aws:PrincipalAccount": ["123456789012", "111222333444"]}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_equals_aws_PrincipalAccount_str_not_valid(self):
condition_statement = {"StringEquals": {"aws:PrincipalAccount": "111222333444"}}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_like_aws_SourceArn_list(self):
condition_statement = {
"ArnLike": {"aws:SourceArn": ["arn:aws:cloudtrail:*:123456789012:trail/*"]}
}
@@ -41,7 +95,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnlike_list_not_valid(self):
def test_condition_parser_arn_like_aws_SourceArn_list_not_valid(self):
condition_statement = {
"ArnLike": {
"aws:SourceArn": [
@@ -55,7 +109,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnlike_str(self):
def test_condition_parser_arn_like_aws_SourceArn_str(self):
condition_statement = {
"ArnLike": {"aws:SourceArn": "arn:aws:cloudtrail:*:123456789012:trail/*"}
}
@@ -64,7 +118,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnlike_str_not_valid(self):
def test_condition_parser_arn_like_aws_SourceArn_str_not_valid(self):
condition_statement = {
"ArnLike": {"aws:SourceArn": "arn:aws:cloudtrail:*:111222333444:trail/*"}
}
@@ -73,7 +127,50 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnequals_list(self):
def test_condition_parser_arn_like_aws_PrincipalArn_list(self):
condition_statement = {
"ArnLike": {
"aws:PrincipalArn": ["arn:aws:cloudtrail:*:123456789012:trail/*"]
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_like_aws_PrincipalArn_list_not_valid(self):
condition_statement = {
"ArnLike": {
"aws:PrincipalArn": [
"arn:aws:cloudtrail:*:123456789012:trail/*",
"arn:aws:cloudtrail:*:111222333444:trail/*",
]
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_like_aws_PrincipalArn_str(self):
condition_statement = {
"ArnLike": {"aws:PrincipalArn": "arn:aws:cloudtrail:*:123456789012:trail/*"}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_like_aws_PrincipalArn_str_not_valid(self):
condition_statement = {
"ArnLike": {"aws:PrincipalArn": "arn:aws:cloudtrail:*:111222333444:trail/*"}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_equals_aws_SourceArn_list(self):
condition_statement = {
"ArnEquals": {
"aws:SourceArn": [
@@ -86,7 +183,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnequals_list_not_valid(self):
def test_condition_parser_arn_equals_aws_SourceArn_list_not_valid(self):
condition_statement = {
"ArnEquals": {
"aws:SourceArn": [
@@ -100,7 +197,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnequals_str(self):
def test_condition_parser_arn_equals_aws_SourceArn_str(self):
condition_statement = {
"ArnEquals": {
"aws:SourceArn": "arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
@@ -111,7 +208,7 @@ class Test_policy_condition_parser:
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arnequals_str_not_valid(self):
def test_condition_parser_arn_equals_aws_SourceArn_str_not_valid(self):
condition_statement = {
"ArnEquals": {
"aws:SourceArn": "arn:aws:cloudtrail:eu-west-1:111222333444:trail/test"
@@ -121,3 +218,150 @@ class Test_policy_condition_parser:
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_equals_aws_PrincipalArn_list(self):
condition_statement = {
"ArnEquals": {
"aws:PrincipalArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
]
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_equals_aws_PrincipalArn_list_not_valid(self):
condition_statement = {
"ArnEquals": {
"aws:PrincipalArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test",
"arn:aws:cloudtrail:eu-west-1:111222333444:trail/test",
]
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_equals_aws_PrincipalArn_str(self):
condition_statement = {
"ArnEquals": {
"aws:PrincipalArn": "arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_arn_equals_aws_PrincipalArn_str_not_valid(self):
condition_statement = {
"ArnEquals": {
"aws:PrincipalArn": "arn:aws:cloudtrail:eu-west-1:111222333444:trail/test"
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_SourceArn_list(self):
condition_statement = {
"StringLike": {
"aws:SourceArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
]
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_SourceArn_list_not_valid(self):
condition_statement = {
"StringLike": {
"aws:SourceArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test",
"arn:aws:cloudtrail:eu-west-1:111222333444:trail/test",
]
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_SourceArn_str(self):
condition_statement = {
"StringLike": {
"aws:SourceArn": "arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_SourceArn_str_not_valid(self):
condition_statement = {
"StringLike": {
"aws:SourceArn": "arn:aws:cloudtrail:eu-west-1:111222333444:trail/test"
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_PrincipalArn_list(self):
condition_statement = {
"StringLike": {
"aws:PrincipalArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
]
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_PrincipalArn_list_not_valid(self):
condition_statement = {
"StringLike": {
"aws:PrincipalArn": [
"arn:aws:cloudtrail:eu-west-1:123456789012:trail/test",
"arn:aws:cloudtrail:eu-west-1:111222333444:trail/test",
]
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_PrincipalArn_str(self):
condition_statement = {
"StringLike": {
"aws:PrincipalArn": "arn:aws:cloudtrail:eu-west-1:123456789012:trail/test"
}
}
assert is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)
def test_condition_parser_string_like_aws_PrincipalArn_str_not_valid(self):
condition_statement = {
"StringLike": {
"aws:PrincipalArn": "arn:aws:cloudtrail:eu-west-1:111222333444:trail/test"
}
}
assert not is_account_only_allowed_in_condition(
condition_statement, AWS_ACCOUNT_NUMBER
)

397
tests/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries_test.py
View File

@@ -9,11 +9,13 @@ from prowler.providers.common.models import Audit_Metadata
AWS_REGION = "us-east-1"
AWS_ACCOUNT_NUMBER = "123456789012"
TRUSTED_AWS_ACCOUNT_NUMBER = "111122223333"
NON_TRUSTED_AWS_ACCOUNT_NUMBER = "000011112222"
def mock_get_config_var(config_var):
if config_var == "trusted_account_ids":
return ["123456789010"]
return [TRUSTED_AWS_ACCOUNT_NUMBER]
return []
@@ -124,7 +126,7 @@ class Test_vpc_endpoint_connections_trust_boundaries:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} has full access."
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can be accessed from non-trusted accounts."
)
assert (
result[0].resource_id
@@ -133,7 +135,7 @@ class Test_vpc_endpoint_connections_trust_boundaries:
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_trusted_account(self):
def test_vpc_endpoint_with_trusted_account_arn(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
@@ -150,7 +152,67 @@ class Test_vpc_endpoint_connections_trust_boundaries:
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789012:root"},
"Principal": {
"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
},
"Action": "*",
"Resource": "*",
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Found trusted account {AWS_ACCOUNT_NUMBER} in VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']}."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_trusted_account_id(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": AWS_ACCOUNT_NUMBER},
"Action": "*",
"Resource": "*",
}
@@ -208,7 +270,9 @@ class Test_vpc_endpoint_connections_trust_boundaries:
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789010:root"},
"Principal": {
"AWS": f"arn:aws:iam::{NON_TRUSTED_AWS_ACCOUNT_NUMBER}:root"
},
"Action": "*",
"Resource": "*",
}
@@ -241,7 +305,7 @@ class Test_vpc_endpoint_connections_trust_boundaries:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"Found untrusted account 123456789010 in VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']}."
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can be accessed from non-trusted accounts."
)
assert (
result[0].resource_id
@@ -249,7 +313,7 @@ class Test_vpc_endpoint_connections_trust_boundaries:
)
@mock_ec2
def test_vpc_endpoint_with_config_trusted_account(self):
def test_vpc_endpoint_with_config_trusted_account_with_arn(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
@@ -266,7 +330,9 @@ class Test_vpc_endpoint_connections_trust_boundaries:
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::123456789010:root"},
"Principal": {
"AWS": f"arn:aws:iam::{TRUSTED_AWS_ACCOUNT_NUMBER}:root"
},
"Action": "*",
"Resource": "*",
}
@@ -302,10 +368,323 @@ class Test_vpc_endpoint_connections_trust_boundaries:
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Found trusted account 123456789010 in VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']}."
== f"Found trusted account {TRUSTED_AWS_ACCOUNT_NUMBER} in VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']}."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_config_trusted_account(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": [TRUSTED_AWS_ACCOUNT_NUMBER]},
"Action": "*",
"Resource": "*",
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.get_config_var",
new=mock_get_config_var,
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Found trusted account {TRUSTED_AWS_ACCOUNT_NUMBER} in VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']}."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_two_account_ids_one_trusted_one_not(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Effect": "Allow",
"Principal": {
"AWS": [
NON_TRUSTED_AWS_ACCOUNT_NUMBER,
TRUSTED_AWS_ACCOUNT_NUMBER,
]
},
"Action": "*",
"Resource": "*",
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can be accessed from non-trusted accounts."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_aws_principal_all(self):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Effect": "Allow",
"Principal": {"AWS": "*"},
"Action": "*",
"Resource": "*",
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can be accessed from non-trusted accounts."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_aws_principal_all_but_restricted_condition_with_SourceAccount(
self,
):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Action": "*",
"Effect": "Allow",
"Principal": "*",
"Resource": "*",
"Condition": {
"StringEquals": {
"aws:SourceAccount": AWS_ACCOUNT_NUMBER
}
},
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can only be accessed from trusted accounts."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION
@mock_ec2
def test_vpc_endpoint_with_aws_principal_all_but_restricted_condition_with_PrincipalAccount(
self,
):
# Create VPC Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION)
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
vpc_endpoint = ec2_client.create_vpc_endpoint(
VpcId=vpc["VpcId"],
ServiceName="com.amazonaws.us-east-1.s3",
RouteTableIds=[route_table["RouteTableId"]],
VpcEndpointType="Gateway",
PolicyDocument=json.dumps(
{
"Statement": [
{
"Action": "*",
"Effect": "Allow",
"Principal": "*",
"Resource": "*",
"Condition": {
"StringEquals": {
"aws:PrincipalAccount": AWS_ACCOUNT_NUMBER
}
},
}
]
}
),
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
new=VPC(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
vpc_endpoint_connections_trust_boundaries,
)
check = vpc_endpoint_connections_trust_boundaries()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"VPC Endpoint {vpc_endpoint['VpcEndpoint']['VpcEndpointId']} in VPC {vpc['VpcId']} can only be accessed from trusted accounts."
)
assert (
result[0].resource_id
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
)
assert result[0].region == AWS_REGION