fix(s3 tests): audit_info object mocked (#2221)

This commit is contained in:
Nacho Rivera
2023-04-17 10:04:28 +02:00
committed by GitHub
parent 8bdd47f912
commit 4a614855d4
7 changed files with 649 additions and 394 deletions

View File

@@ -1,46 +1,78 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_acl_prohibited:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_ownership(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
check = s3_bucket_acl_prohibited()
result = check.execute()
check = s3_bucket_acl_prohibited()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_without_ownership(self):
@@ -48,35 +80,38 @@ class Test_s3_bucket_acl_prohibited:
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
check = s3_bucket_acl_prohibited()
result = check.execute()
check = s3_bucket_acl_prohibited()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_acl_disabled(self):
@@ -86,32 +121,35 @@ class Test_s3_bucket_acl_prohibited:
Bucket=bucket_name_us, ObjectOwnership="BucketOwnerEnforced"
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_acl_prohibited.s3_bucket_acl_prohibited import (
s3_bucket_acl_prohibited,
)
check = s3_bucket_acl_prohibited()
result = check.execute()
check = s3_bucket_acl_prohibited()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"ACLs disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"ACLs disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"

View File

@@ -1,46 +1,78 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_default_encryption:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_encryption(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption import (
s3_bucket_default_encryption,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption import (
s3_bucket_default_encryption,
)
check = s3_bucket_default_encryption()
result = check.execute()
check = s3_bucket_default_encryption()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"Server Side Encryption is not configured",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"Server Side Encryption is not configured",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_kms_encryption(self):
@@ -63,32 +95,36 @@ class Test_s3_bucket_default_encryption:
s3_client_us_east_1.put_bucket_encryption(
Bucket=bucket_name_us, ServerSideEncryptionConfiguration=sse_config
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption import (
s3_bucket_default_encryption,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_default_encryption.s3_bucket_default_encryption import (
s3_bucket_default_encryption,
)
check = s3_bucket_default_encryption()
result = check.execute()
check = s3_bucket_default_encryption()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"has Server Side Encryption",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"has Server Side Encryption",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"

View File

@@ -1,34 +1,63 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
ACCOUNT_ID = "123456789012"
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_no_mfa_delete:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_no_buckets(self):
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
check = s3_bucket_no_mfa_delete()
result = check.execute()
check = s3_bucket_no_mfa_delete()
result = check.execute()
assert len(result) == 0
assert len(result) == 0
@mock_s3
def test_bucket_without_mfa(self):
@@ -36,34 +65,37 @@ class Test_s3_bucket_no_mfa_delete:
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
check = s3_bucket_no_mfa_delete()
result = check.execute()
check = s3_bucket_no_mfa_delete()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"MFA Delete disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"MFA Delete disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
@mock_s3
def test_bucket_with_mfa(self):
@@ -74,32 +106,35 @@ class Test_s3_bucket_no_mfa_delete:
Bucket=bucket_name_us,
VersioningConfiguration={"MFADelete": "Enabled", "Status": "Enabled"},
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(current_audit_info),
) as service_client:
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete.s3_client",
new=S3(audit_info),
) as service_client:
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_no_mfa_delete.s3_bucket_no_mfa_delete import (
s3_bucket_no_mfa_delete,
)
service_client.buckets[0].mfa_delete = True
check = s3_bucket_no_mfa_delete()
result = check.execute()
service_client.buckets[0].mfa_delete = True
check = s3_bucket_no_mfa_delete()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"MFA Delete enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"MFA Delete enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)

View File

@@ -1,46 +1,78 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_object_versioning:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_object_versioning(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning import (
s3_bucket_object_versioning,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning import (
s3_bucket_object_versioning,
)
check = s3_bucket_object_versioning()
result = check.execute()
check = s3_bucket_object_versioning()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"versioning disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"versioning disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_object_versioning_enabled(self):
@@ -53,32 +85,36 @@ class Test_s3_bucket_object_versioning:
Bucket=bucket_name_us,
VersioningConfiguration={"Status": "Enabled"},
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning import (
s3_bucket_object_versioning,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_object_versioning.s3_bucket_object_versioning import (
s3_bucket_object_versioning,
)
check = s3_bucket_object_versioning()
result = check.execute()
check = s3_bucket_object_versioning()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"versioning enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"versioning enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"

View File

@@ -1,46 +1,78 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_policy_public_write_access:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_policy(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
check = s3_bucket_policy_public_write_access()
result = check.execute()
check = s3_bucket_policy_public_write_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"does not have a bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"does not have a bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_comply_policy(self):
@@ -55,35 +87,38 @@ class Test_s3_bucket_policy_public_write_access:
Bucket=bucket_name_us,
Policy=encryption_policy,
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
check = s3_bucket_policy_public_write_access()
result = check.execute()
check = s3_bucket_policy_public_write_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"does not allow public write access in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"does not allow public write access in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_public_write_policy(self):
@@ -97,32 +132,35 @@ class Test_s3_bucket_policy_public_write_access:
Bucket=bucket_name_us,
Policy=public_write_policy,
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_policy_public_write_access.s3_bucket_policy_public_write_access import (
s3_bucket_policy_public_write_access,
)
check = s3_bucket_policy_public_write_access()
result = check.execute()
check = s3_bucket_policy_public_write_access()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"allows public write access in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"allows public write access in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"

View File

@@ -1,46 +1,78 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_secure_transport_policy:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_policy(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
check = s3_bucket_secure_transport_policy()
result = check.execute()
check = s3_bucket_secure_transport_policy()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"does not have a bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"does not have a bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_comply_policy(self):
@@ -72,35 +104,38 @@ class Test_s3_bucket_secure_transport_policy:
Bucket=bucket_name_us,
Policy=ssl_policy,
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
check = s3_bucket_secure_transport_policy()
result = check.execute()
check = s3_bucket_secure_transport_policy()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"bucket policy to deny requests over insecure transport",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"bucket policy to deny requests over insecure transport",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
@mock_s3
def test_bucket_uncomply_policy(self):
@@ -132,32 +167,35 @@ class Test_s3_bucket_secure_transport_policy:
Bucket=bucket_name_us,
Policy=ssl_policy,
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_secure_transport_policy.s3_bucket_secure_transport_policy import (
s3_bucket_secure_transport_policy,
)
check = s3_bucket_secure_transport_policy()
result = check.execute()
check = s3_bucket_secure_transport_policy()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"allows requests over insecure transport in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"allows requests over insecure transport in the bucket policy",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert result[0].region == "us-east-1"

View File

@@ -1,47 +1,77 @@
from re import search
from unittest import mock
from boto3 import client
from boto3 import client, session
from moto import mock_s3
ACCOUNT_ID = "123456789012"
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
AWS_ACCOUNT_NUMBER = "123456789012"
AWS_REGION = "us-east-1"
class Test_s3_bucket_server_access_logging_enabled:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
region_name=AWS_REGION,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=AWS_REGION,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
)
return audit_info
@mock_s3
def test_bucket_no_logging(self):
s3_client_us_east_1 = client("s3", region_name="us-east-1")
bucket_name_us = "bucket_test_us"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled import (
s3_bucket_server_access_logging_enabled,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled import (
s3_bucket_server_access_logging_enabled,
)
check = s3_bucket_server_access_logging_enabled()
result = check.execute()
check = s3_bucket_server_access_logging_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"server access logging disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"server access logging disabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)
@mock_s3
def test_bucket_with_logging(self):
@@ -103,31 +133,35 @@ class Test_s3_bucket_server_access_logging_enabled:
}
},
)
from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from prowler.providers.aws.services.s3.s3_service import S3
current_audit_info.audited_partition = "aws"
audit_info = self.set_mocked_audit_info()
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled.s3_client",
new=S3(current_audit_info),
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled import (
s3_bucket_server_access_logging_enabled,
)
with mock.patch(
"prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled.s3_client",
new=S3(audit_info),
):
# Test Check
from prowler.providers.aws.services.s3.s3_bucket_server_access_logging_enabled.s3_bucket_server_access_logging_enabled import (
s3_bucket_server_access_logging_enabled,
)
check = s3_bucket_server_access_logging_enabled()
result = check.execute()
check = s3_bucket_server_access_logging_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"server access logging enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{current_audit_info.audited_partition}:s3:::{bucket_name_us}"
)
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"server access logging enabled",
result[0].status_extended,
)
assert result[0].resource_id == bucket_name_us
assert (
result[0].resource_arn
== f"arn:{audit_info.audited_partition}:s3:::{bucket_name_us}"
)