mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
chore(s3): reduce false positive in s3 public check (#4281)
This commit is contained in:
Rubén De la Torre Vico
committed by
GitHub
GitHub
parent
bae224c891
commit
093738c65f
@@ -29,6 +29,8 @@ def is_condition_block_restrictive(
|
||||
"aws:principalaccount",
|
||||
"aws:resourceaccount",
|
||||
"aws:sourcearn",
|
||||
"aws:sourcevpc",
|
||||
"aws:sourcevpce",
|
||||
],
|
||||
"StringLike": [
|
||||
"aws:sourceaccount",
|
||||
@@ -37,6 +39,8 @@ def is_condition_block_restrictive(
|
||||
"aws:principalarn",
|
||||
"aws:resourceaccount",
|
||||
"aws:principalaccount",
|
||||
"aws:sourcevpc",
|
||||
"aws:sourcevpce",
|
||||
],
|
||||
"ArnLike": ["aws:sourcearn", "aws:principalarn"],
|
||||
"ArnEquals": ["aws:sourcearn", "aws:principalarn"],
|
||||
|
||||
@@ -1,3 +1,8 @@
|
||||
from ipaddress import ip_address, ip_network
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
|
||||
|
||||
def is_policy_cross_account(policy: dict, audited_account: str) -> bool:
|
||||
"""
|
||||
is_policy_cross_account checks if the policy allows cross-account access.
|
||||
@@ -68,3 +73,55 @@ def is_policy_public(policy: dict) -> bool:
|
||||
) and "Condition" not in statement:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def is_condition_restricting_from_private_ip(condition_statement: dict) -> bool:
|
||||
"""Check if the policy condition is coming from a private IP address.
|
||||
|
||||
Keyword arguments:
|
||||
condition_statement -- The policy condition to check. For example:
|
||||
{
|
||||
"IpAddress": {
|
||||
"aws:SourceIp": "X.X.X.X"
|
||||
}
|
||||
}
|
||||
"""
|
||||
try:
|
||||
CONDITION_OPERATOR = "IpAddress"
|
||||
CONDITION_KEY = "aws:sourceip"
|
||||
|
||||
is_from_private_ip = False
|
||||
|
||||
if condition_statement.get(CONDITION_OPERATOR, {}):
|
||||
# We need to transform the condition_statement into lowercase
|
||||
condition_statement[CONDITION_OPERATOR] = {
|
||||
k.lower(): v for k, v in condition_statement[CONDITION_OPERATOR].items()
|
||||
}
|
||||
|
||||
if condition_statement[CONDITION_OPERATOR].get(CONDITION_KEY, ""):
|
||||
if not isinstance(
|
||||
condition_statement[CONDITION_OPERATOR][CONDITION_KEY], list
|
||||
):
|
||||
condition_statement[CONDITION_OPERATOR][CONDITION_KEY] = [
|
||||
condition_statement[CONDITION_OPERATOR][CONDITION_KEY]
|
||||
]
|
||||
|
||||
for ip in condition_statement[CONDITION_OPERATOR][CONDITION_KEY]:
|
||||
# Select if IP address or IP network searching in the string for '/'
|
||||
if "/" in ip:
|
||||
if not ip_network(ip, strict=False).is_private:
|
||||
break
|
||||
else:
|
||||
if not ip_address(ip).is_private:
|
||||
break
|
||||
else:
|
||||
is_from_private_ip = True
|
||||
|
||||
except ValueError:
|
||||
logger.error(f"Invalid IP: {ip}")
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
return is_from_private_ip
|
||||
|
||||
+31
-16
@@ -1,4 +1,10 @@
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
|
||||
is_condition_block_restrictive,
|
||||
)
|
||||
from prowler.providers.aws.services.iam.lib.policy import (
|
||||
is_condition_restricting_from_private_ip,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_client import s3_client
|
||||
from prowler.providers.aws.services.s3.s3control_client import s3control_client
|
||||
|
||||
@@ -46,26 +52,35 @@ class s3_bucket_public_access(Check):
|
||||
|
||||
# 4. Check bucket policy
|
||||
if bucket.policy:
|
||||
for statement in bucket.policy["Statement"]:
|
||||
for statement in bucket.policy.get("Statement", []):
|
||||
if (
|
||||
"Principal" in statement
|
||||
and "*" == statement["Principal"]
|
||||
and statement["Effect"] == "Allow"
|
||||
and not is_condition_block_restrictive(
|
||||
statement.get("Condition", {}), "", True
|
||||
)
|
||||
and (
|
||||
not is_condition_restricting_from_private_ip(
|
||||
statement.get("Condition", {})
|
||||
)
|
||||
if statement.get("Condition", {}).get(
|
||||
"IpAddress", {}
|
||||
)
|
||||
else True
|
||||
)
|
||||
):
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
|
||||
else:
|
||||
if (
|
||||
"Principal" in statement
|
||||
and "AWS" in statement["Principal"]
|
||||
and statement["Effect"] == "Allow"
|
||||
):
|
||||
if isinstance(
|
||||
statement["Principal"]["AWS"], str
|
||||
):
|
||||
principals = [statement["Principal"]["AWS"]]
|
||||
else:
|
||||
principals = statement["Principal"]["AWS"]
|
||||
if "*" == statement["Principal"]:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"S3 Bucket {bucket.name} has public access due to bucket policy."
|
||||
elif "AWS" in statement["Principal"]:
|
||||
principals = (
|
||||
statement["Principal"]["AWS"]
|
||||
if isinstance(
|
||||
statement["Principal"]["AWS"], list
|
||||
)
|
||||
else [statement["Principal"]["AWS"]]
|
||||
)
|
||||
|
||||
for principal_arn in principals:
|
||||
if principal_arn == "*":
|
||||
report.status = "FAIL"
|
||||
|
||||
@@ -1366,3 +1366,26 @@ class Test_policy_condition_parser:
|
||||
assert not is_condition_block_restrictive(
|
||||
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
|
||||
)
|
||||
|
||||
def test_condition_parser_string_equals_vpc(self):
|
||||
condition_statement = {"StringEquals": {"aws:SourceVpc": "vpc-123456"}}
|
||||
|
||||
assert is_condition_block_restrictive(
|
||||
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
|
||||
)
|
||||
|
||||
def test_condition_parser_string_equals_vpc_list(self):
|
||||
condition_statement = {"StringEquals": {"aws:sourcevpc": ["vpc-123456"]}}
|
||||
|
||||
assert is_condition_block_restrictive(
|
||||
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
|
||||
)
|
||||
|
||||
def test_condition_parser_string_equals_vpc_list_not_valid(self):
|
||||
condition_statement = {
|
||||
"StringEquals": {"aws:SourceVpc": ["vpc-123456", "vpc-654321"]}
|
||||
}
|
||||
|
||||
assert is_condition_block_restrictive(
|
||||
condition_statement, TRUSTED_AWS_ACCOUNT_NUMBER, True
|
||||
)
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
from prowler.providers.aws.services.iam.lib.policy import (
|
||||
is_condition_restricting_from_private_ip,
|
||||
is_policy_cross_account,
|
||||
is_policy_public,
|
||||
)
|
||||
@@ -89,3 +90,91 @@ class Test_Policy:
|
||||
assert is_policy_public(policy2)
|
||||
assert not is_policy_public(policy3)
|
||||
assert not is_policy_public(policy4)
|
||||
|
||||
def test_is_condition_restricting_from_private_ip_no_condition(self):
|
||||
assert not is_condition_restricting_from_private_ip({})
|
||||
|
||||
def test_is_condition_restricting_from_private_ip(self):
|
||||
condition_from_private_ip = {
|
||||
"IpAddress": {"aws:SourceIp": "10.0.0.22"},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(condition_from_private_ip)
|
||||
|
||||
def test_is_condition_restricting_from_public_ip(self):
|
||||
condition_not_from_private_ip = {
|
||||
"IpAddress": {"aws:SourceIp": "1.2.3.4"},
|
||||
}
|
||||
assert not is_condition_restricting_from_private_ip(
|
||||
condition_not_from_private_ip
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_private_ipv6(self):
|
||||
condition_from_private_ipv6 = {
|
||||
"IpAddress": {"aws:SourceIp": "fd00::1"},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(condition_from_private_ipv6)
|
||||
|
||||
def test_is_condition_restricting_from_public_ipv6(self):
|
||||
condition_not_from_private_ipv6 = {
|
||||
"IpAddress": {"aws:SourceIp": "2001:0db8::1"},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(condition_not_from_private_ipv6)
|
||||
|
||||
def test_is_condition_restricting_from_private_ip_network(self):
|
||||
condition_from_private_ip_network = {
|
||||
"IpAddress": {"aws:SourceIp": "10.0.0.0/24"},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(
|
||||
condition_from_private_ip_network
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_public_ip_network(self):
|
||||
condition_from_public_ip_network = {
|
||||
"IpAddress": {"aws:SourceIp": "1.2.3.0/24"},
|
||||
}
|
||||
|
||||
assert not is_condition_restricting_from_private_ip(
|
||||
condition_from_public_ip_network
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_private_ipv6_network(self):
|
||||
condition_from_private_ipv6_network = {
|
||||
"IpAddress": {"aws:SourceIp": "fd00::/8"},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(
|
||||
condition_from_private_ipv6_network
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_private_ip_array(self):
|
||||
condition_from_private_ip_array = {
|
||||
"IpAddress": {"aws:SourceIp": ["10.0.0.22", "192.168.1.1"]},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(condition_from_private_ip_array)
|
||||
|
||||
def test_is_condition_restricting_from_private_ipv6_array(self):
|
||||
condition_from_private_ipv6_array = {
|
||||
"IpAddress": {"aws:SourceIp": ["fd00::1", "fe80::1"]},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(
|
||||
condition_from_private_ipv6_array
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_mixed_ip_array(self):
|
||||
condition_from_mixed_ip_array = {
|
||||
"IpAddress": {"aws:SourceIp": ["10.0.0.22", "2001:0db8::1"]},
|
||||
}
|
||||
assert is_condition_restricting_from_private_ip(condition_from_mixed_ip_array)
|
||||
|
||||
def test_is_condition_restricting_from_mixed_ip_array_not_private(self):
|
||||
condition_from_mixed_ip_array_not_private = {
|
||||
"IpAddress": {"aws:SourceIp": ["1.2.3.4", "2001:0db8::1"]},
|
||||
}
|
||||
assert not is_condition_restricting_from_private_ip(
|
||||
condition_from_mixed_ip_array_not_private
|
||||
)
|
||||
|
||||
def test_is_condition_restricting_from_private_ip_from_invalid_ip(self):
|
||||
condition_from_invalid_ip = {
|
||||
"IpAddress": {"aws:SourceIp": "256.256.256.256"},
|
||||
}
|
||||
assert not is_condition_restricting_from_private_ip(condition_from_invalid_ip)
|
||||
|
||||
+328
@@ -349,6 +349,267 @@ class Test_s3_bucket_public_access:
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_not_public_due_to_policy_conditions_from_vpc(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
bucket_name_us = "bucket_test_us"
|
||||
s3_client.create_bucket(Bucket=bucket_name_us)
|
||||
# Generate S3Control Client
|
||||
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
|
||||
s3control_client.put_public_access_block(
|
||||
AccountId=AWS_ACCOUNT_NUMBER,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
s3_client.put_public_access_block(
|
||||
Bucket=bucket_name_us,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"StringEquals": {"aws:SourceVpc": "vpc-123456"}}}]}'
|
||||
s3_client.put_bucket_policy(
|
||||
Bucket=bucket_name_us,
|
||||
Policy=public_write_policy,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
|
||||
|
||||
aws_provider = set_mocked_aws_provider([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.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
|
||||
new=S3(aws_provider),
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
|
||||
new=S3Control(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
|
||||
s3_bucket_public_access,
|
||||
)
|
||||
|
||||
check = s3_bucket_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"S3 Bucket {bucket_name_us} is not public."
|
||||
)
|
||||
assert result[0].resource_id == bucket_name_us
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:s3:::{bucket_name_us}"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_not_public_due_to_policy_conditions_from_private_ip(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
bucket_name_us = "bucket_test_us"
|
||||
s3_client.create_bucket(Bucket=bucket_name_us)
|
||||
# Generate S3Control Client
|
||||
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
|
||||
s3control_client.put_public_access_block(
|
||||
AccountId=AWS_ACCOUNT_NUMBER,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
s3_client.put_public_access_block(
|
||||
Bucket=bucket_name_us,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": "10.0.0.25"}}}]}'
|
||||
s3_client.put_bucket_policy(
|
||||
Bucket=bucket_name_us,
|
||||
Policy=public_write_policy,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
|
||||
new=S3(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
|
||||
new=S3Control(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
|
||||
s3_bucket_public_access,
|
||||
)
|
||||
|
||||
check = s3_bucket_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"S3 Bucket {bucket_name_us} is not public."
|
||||
)
|
||||
assert result[0].resource_id == bucket_name_us
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:s3:::{bucket_name_us}"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_public_due_to_policy_conditions_from_public_ip(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
bucket_name_us = "bucket_test_us"
|
||||
s3_client.create_bucket(Bucket=bucket_name_us)
|
||||
# Generate S3Control Client
|
||||
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
|
||||
s3control_client.put_public_access_block(
|
||||
AccountId=AWS_ACCOUNT_NUMBER,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
s3_client.put_public_access_block(
|
||||
Bucket=bucket_name_us,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": "1.2.3.4"}}}]}'
|
||||
s3_client.put_bucket_policy(
|
||||
Bucket=bucket_name_us,
|
||||
Policy=public_write_policy,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
|
||||
new=S3(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
|
||||
new=S3Control(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
|
||||
s3_bucket_public_access,
|
||||
)
|
||||
|
||||
check = s3_bucket_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
|
||||
)
|
||||
assert result[0].resource_id == bucket_name_us
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:s3:::{bucket_name_us}"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_public_due_to_policy_conditions_from_public_and_private_ips(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
bucket_name_us = "bucket_test_us"
|
||||
s3_client.create_bucket(Bucket=bucket_name_us)
|
||||
# Generate S3Control Client
|
||||
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
|
||||
s3control_client.put_public_access_block(
|
||||
AccountId=AWS_ACCOUNT_NUMBER,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
s3_client.put_public_access_block(
|
||||
Bucket=bucket_name_us,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": "*","Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*","Condition": {"IpAddress": {"aws:SourceIp": ["192.168.10.0/24", "2001:DB8:1234:5678::/64", "1.2.3.4"]}}}]}'
|
||||
s3_client.put_bucket_policy(
|
||||
Bucket=bucket_name_us,
|
||||
Policy=public_write_policy,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
|
||||
with mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=aws_provider,
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
|
||||
new=S3(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
|
||||
new=S3Control(aws_provider),
|
||||
):
|
||||
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
|
||||
s3_bucket_public_access,
|
||||
)
|
||||
|
||||
check = s3_bucket_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
|
||||
)
|
||||
assert result[0].resource_id == bucket_name_us
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:s3:::{bucket_name_us}"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_not_public(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
@@ -442,3 +703,70 @@ class Test_s3_bucket_public_access:
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_bucket_public_with_aws_principals(self):
|
||||
s3_client = client("s3", region_name=AWS_REGION_US_EAST_1)
|
||||
bucket_name_us = "bucket_test_us"
|
||||
s3_client.create_bucket(Bucket=bucket_name_us)
|
||||
# Generate S3Control Client
|
||||
s3control_client = client("s3control", region_name=AWS_REGION_US_EAST_1)
|
||||
s3control_client.put_public_access_block(
|
||||
AccountId=AWS_ACCOUNT_NUMBER,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
s3_client.put_public_access_block(
|
||||
Bucket=bucket_name_us,
|
||||
PublicAccessBlockConfiguration={
|
||||
"BlockPublicAcls": False,
|
||||
"IgnorePublicAcls": False,
|
||||
"BlockPublicPolicy": False,
|
||||
"RestrictPublicBuckets": False,
|
||||
},
|
||||
)
|
||||
public_write_policy = '{"Version": "2012-10-17","Id": "PutObjPolicy","Statement": [{"Sid": "PublicWritePolicy","Effect": "Allow","Principal": {"AWS": ["arn:aws:iam::123456789012:root", "*"]},"Action": "s3:PutObject","Resource": "arn:aws:s3:::bucket_test_us/*"}]}'
|
||||
s3_client.put_bucket_policy(
|
||||
Bucket=bucket_name_us,
|
||||
Policy=public_write_policy,
|
||||
)
|
||||
from prowler.providers.aws.services.s3.s3_service import S3, S3Control
|
||||
|
||||
aws_provider = set_mocked_aws_provider([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.s3.s3_bucket_public_access.s3_bucket_public_access.s3_client",
|
||||
new=S3(aws_provider),
|
||||
):
|
||||
with mock.patch(
|
||||
"prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access.s3control_client",
|
||||
new=S3Control(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.s3.s3_bucket_public_access.s3_bucket_public_access import (
|
||||
s3_bucket_public_access,
|
||||
)
|
||||
|
||||
check = s3_bucket_public_access()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
result[0].status_extended
|
||||
== f"S3 Bucket {bucket_name_us} has public access due to bucket policy."
|
||||
)
|
||||
assert result[0].resource_id == bucket_name_us
|
||||
assert (
|
||||
result[0].resource_arn
|
||||
== f"arn:{aws_provider.identity.partition}:s3:::{bucket_name_us}"
|
||||
)
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
|
||||
Reference in New Issue
Block a user