Skip to content

Commit 08afbdf

Browse files
mikkocmikkoc
committed
fix(aws): correct VPC endpoint account trust checks
- Validate policy account values as a subset of trusted accounts - Preserve existing condition handling and implicit audited-account trust - Add focused regression coverage for scalar and list conditions
1 parent c0fdd5b commit 08afbdf

3 files changed

Lines changed: 244 additions & 26 deletions

File tree

‎prowler/changelog.d/vpc-endpoint-principal-account-list.fixed.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
`vpc_endpoint_connections_trust_boundaries` no longer reports trusted `aws:PrincipalAccount` lists as untrusted when every policy account is configured as trusted

‎prowler/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries.py‎

Lines changed: 94 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,92 @@
1+
from fnmatch import fnmatchcase
12
from re import compile
23

34
from prowler.lib.check.models import Check, Check_Report_AWS
45
from prowler.providers.aws.services.iam.lib.policy import is_condition_block_restrictive
56
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
67

78

9+
def _is_condition_restrictive_for_trusted_accounts(
10+
condition_statement: dict, trusted_account_ids: set[str]
11+
) -> bool:
12+
principal_account_groups = []
13+
remaining_conditions = {}
14+
15+
for operator, conditions in condition_statement.items():
16+
remaining_operator_conditions = {}
17+
for condition_key, condition_value in conditions.items():
18+
if (
19+
operator in ("StringEquals", "StringLike")
20+
and condition_key.lower() == "aws:principalaccount"
21+
):
22+
if isinstance(condition_value, str):
23+
values = [condition_value]
24+
elif isinstance(condition_value, list) and all(
25+
isinstance(value, str) for value in condition_value
26+
):
27+
values = condition_value
28+
else:
29+
values = []
30+
31+
principal_account_groups.append((operator, values))
32+
else:
33+
remaining_operator_conditions[condition_key] = condition_value
34+
35+
if remaining_operator_conditions:
36+
remaining_conditions[operator] = remaining_operator_conditions
37+
38+
principal_account_restrictive = False
39+
if principal_account_groups:
40+
exact_groups = [
41+
set(values)
42+
for operator, values in principal_account_groups
43+
if operator == "StringEquals"
44+
]
45+
if exact_groups:
46+
possible_accounts = set.intersection(*exact_groups)
47+
else:
48+
literal_groups = [
49+
set(values)
50+
for _, values in principal_account_groups
51+
if all(
52+
not any(character in value for character in "*?[")
53+
for value in values
54+
)
55+
]
56+
possible_accounts = (
57+
set.intersection(*literal_groups) if literal_groups else None
58+
)
59+
60+
if possible_accounts is not None:
61+
for operator, values in principal_account_groups:
62+
if operator == "StringLike":
63+
possible_accounts = {
64+
account_id
65+
for account_id in possible_accounts
66+
if any(fnmatchcase(account_id, pattern) for pattern in values)
67+
}
68+
69+
principal_account_restrictive = possible_accounts.issubset(
70+
trusted_account_ids
71+
)
72+
73+
remaining_conditions_restrictive = remaining_conditions and any(
74+
is_condition_block_restrictive(remaining_conditions, account_id)
75+
for account_id in trusted_account_ids
76+
)
77+
78+
return bool(principal_account_restrictive or remaining_conditions_restrictive)
79+
80+
881
class vpc_endpoint_connections_trust_boundaries(Check):
982
def execute(self):
1083
findings = []
1184
# Get trusted account_ids from prowler.config.yaml
12-
trusted_account_ids = vpc_client.audit_config.get("trusted_account_ids", [])
85+
trusted_account_ids = set(
86+
vpc_client.audit_config.get("trusted_account_ids", [])
87+
)
1388
# Always include the same account as trusted
14-
trusted_account_ids.append(vpc_client.audited_account)
89+
trusted_account_ids.add(vpc_client.audited_account)
1590
for endpoint in vpc_client.vpc_endpoints:
1691
# Check VPC endpoint policy and avoid "com.amazonaws.vpce" endpoints since the policy cannot be modified
1792
if (
@@ -30,14 +105,11 @@ def execute(self):
30105
)
31106

32107
if "Condition" in statement:
33-
for account_id in trusted_account_ids:
34-
if is_condition_block_restrictive(
35-
statement["Condition"], account_id
36-
):
37-
access_from_trusted_accounts = True
38-
else:
39-
access_from_trusted_accounts = False
40-
break
108+
access_from_trusted_accounts = (
109+
_is_condition_restrictive_for_trusted_accounts(
110+
statement["Condition"], trusted_account_ids
111+
)
112+
)
41113

42114
if not access_from_trusted_accounts:
43115
report.status = "FAIL"
@@ -67,14 +139,12 @@ def execute(self):
67139
if principal_arn == "*":
68140
access_from_trusted_accounts = False
69141
if "Condition" in statement:
70-
for account_id in trusted_account_ids:
71-
if is_condition_block_restrictive(
72-
statement["Condition"], account_id
73-
):
74-
access_from_trusted_accounts = True
75-
else:
76-
access_from_trusted_accounts = False
77-
break
142+
access_from_trusted_accounts = (
143+
_is_condition_restrictive_for_trusted_accounts(
144+
statement["Condition"],
145+
trusted_account_ids,
146+
)
147+
)
78148

79149
if not access_from_trusted_accounts:
80150
report.status = "FAIL"
@@ -99,14 +169,12 @@ def execute(self):
99169
access_from_trusted_accounts = False
100170

101171
if "Condition" in statement:
102-
for account_id in trusted_account_ids:
103-
if is_condition_block_restrictive(
104-
statement["Condition"], account_id
105-
):
106-
access_from_trusted_accounts = True
107-
else:
108-
access_from_trusted_accounts = False
109-
break
172+
access_from_trusted_accounts = (
173+
_is_condition_restrictive_for_trusted_accounts(
174+
statement["Condition"],
175+
trusted_account_ids,
176+
)
177+
)
110178

111179
if not access_from_trusted_accounts:
112180
report.status = "FAIL"

‎tests/providers/aws/services/vpc/vpc_endpoint_connections_trust_boundaries/vpc_endpoint_connections_trust_boundaries_test.py‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,56 @@
1414
NON_TRUSTED_AWS_ACCOUNT_NUMBER = "000011112222"
1515

1616

17+
def _execute_check_with_principal_account_condition(
18+
principal_accounts,
19+
trusted_account_ids,
20+
operator="StringEquals",
21+
condition=None,
22+
):
23+
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
24+
vpc = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
25+
route_table = ec2_client.create_route_table(VpcId=vpc["VpcId"])["RouteTable"]
26+
ec2_client.create_vpc_endpoint(
27+
VpcId=vpc["VpcId"],
28+
ServiceName="com.amazonaws.us-east-1.s3",
29+
RouteTableIds=[route_table["RouteTableId"]],
30+
VpcEndpointType="Gateway",
31+
PolicyDocument=json.dumps(
32+
{
33+
"Statement": [
34+
{
35+
"Action": "*",
36+
"Effect": "Allow",
37+
"Principal": "*",
38+
"Resource": "*",
39+
"Condition": condition
40+
or {operator: {"aws:PrincipalAccount": principal_accounts}},
41+
}
42+
]
43+
}
44+
),
45+
)
46+
47+
from prowler.providers.aws.services.vpc.vpc_service import VPC
48+
49+
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
50+
aws_provider._audit_config = {"trusted_account_ids": trusted_account_ids}
51+
52+
with mock.patch(
53+
"prowler.providers.common.provider.Provider.get_global_provider",
54+
return_value=aws_provider,
55+
):
56+
with mock.patch(
57+
"prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries.vpc_client",
58+
new=VPC(aws_provider),
59+
):
60+
from prowler.providers.aws.services.vpc.vpc_endpoint_connections_trust_boundaries.vpc_endpoint_connections_trust_boundaries import (
61+
vpc_endpoint_connections_trust_boundaries,
62+
)
63+
64+
return vpc_endpoint_connections_trust_boundaries().execute()
65+
66+
1767
class Test_vpc_endpoint_connections_trust_boundaries:
1868
@mock_aws
1969
def test_vpc_no_endpoints(self):
@@ -711,3 +761,102 @@ def test_vpc_endpoint_with_aws_principal_all_but_restricted_condition_with_Princ
711761
== vpc_endpoint["VpcEndpoint"]["VpcEndpointId"]
712762
)
713763
assert result[0].region == AWS_REGION_US_EAST_1
764+
765+
@mock_aws
766+
def test_principal_account_condition_allows_trusted_subset(self):
767+
result = _execute_check_with_principal_account_condition(
768+
TRUSTED_AWS_ACCOUNT_NUMBER,
769+
[TRUSTED_AWS_ACCOUNT_NUMBER, "444455556666"],
770+
)
771+
772+
assert len(result) == 1
773+
assert result[0].status == "PASS"
774+
775+
@mock_aws
776+
def test_principal_account_condition_allows_scalar_trusted_account(self):
777+
result = _execute_check_with_principal_account_condition(
778+
TRUSTED_AWS_ACCOUNT_NUMBER,
779+
[TRUSTED_AWS_ACCOUNT_NUMBER],
780+
)
781+
782+
assert len(result) == 1
783+
assert result[0].status == "PASS"
784+
785+
@mock_aws
786+
def test_principal_account_condition_allows_multiple_trusted_accounts(self):
787+
result = _execute_check_with_principal_account_condition(
788+
[TRUSTED_AWS_ACCOUNT_NUMBER, AWS_ACCOUNT_NUMBER],
789+
[TRUSTED_AWS_ACCOUNT_NUMBER],
790+
)
791+
792+
assert len(result) == 1
793+
assert result[0].status == "PASS"
794+
795+
@mock_aws
796+
def test_principal_account_condition_rejects_mixed_trust_list(self):
797+
result = _execute_check_with_principal_account_condition(
798+
[TRUSTED_AWS_ACCOUNT_NUMBER, NON_TRUSTED_AWS_ACCOUNT_NUMBER],
799+
[TRUSTED_AWS_ACCOUNT_NUMBER],
800+
)
801+
802+
assert len(result) == 1
803+
assert result[0].status == "FAIL"
804+
805+
@mock_aws
806+
def test_principal_account_condition_implicitly_trusts_audited_account(self):
807+
result = _execute_check_with_principal_account_condition(
808+
AWS_ACCOUNT_NUMBER,
809+
[],
810+
)
811+
812+
assert len(result) == 1
813+
assert result[0].status == "PASS"
814+
815+
@mock_aws
816+
def test_principal_account_string_like_allows_scalar_trusted_account(self):
817+
result = _execute_check_with_principal_account_condition(
818+
TRUSTED_AWS_ACCOUNT_NUMBER,
819+
[TRUSTED_AWS_ACCOUNT_NUMBER],
820+
operator="StringLike",
821+
)
822+
823+
assert len(result) == 1
824+
assert result[0].status == "PASS"
825+
826+
@mock_aws
827+
def test_principal_account_condition_preserves_operator_intersection(self):
828+
result = _execute_check_with_principal_account_condition(
829+
None,
830+
[TRUSTED_AWS_ACCOUNT_NUMBER],
831+
condition={
832+
"StringEquals": {
833+
"aws:PrincipalAccount": [
834+
TRUSTED_AWS_ACCOUNT_NUMBER,
835+
NON_TRUSTED_AWS_ACCOUNT_NUMBER,
836+
]
837+
},
838+
"StringLike": {"aws:PrincipalAccount": TRUSTED_AWS_ACCOUNT_NUMBER},
839+
},
840+
)
841+
842+
assert len(result) == 1
843+
assert result[0].status == "PASS"
844+
845+
@mock_aws
846+
def test_mixed_principal_accounts_with_other_restrictive_condition(self):
847+
result = _execute_check_with_principal_account_condition(
848+
None,
849+
[TRUSTED_AWS_ACCOUNT_NUMBER],
850+
condition={
851+
"StringEquals": {
852+
"aws:PrincipalAccount": [
853+
TRUSTED_AWS_ACCOUNT_NUMBER,
854+
NON_TRUSTED_AWS_ACCOUNT_NUMBER,
855+
],
856+
"aws:SourceAccount": TRUSTED_AWS_ACCOUNT_NUMBER,
857+
}
858+
},
859+
)
860+
861+
assert len(result) == 1
862+
assert result[0].status == "PASS"

0 commit comments

Comments
 (0)