chore(aws): cleanup aws test cases and standardize checks (#4053)

This commit is contained in:
madereddy
2024-05-21 11:49:30 -04:00
committed by GitHub
parent 248c7c51d6
commit 9e934b8e87
22 changed files with 157 additions and 129 deletions
@@ -12,7 +12,7 @@ class organizations_account_part_of_organizations(Check):
if org.status == "ACTIVE":
report.status = "PASS"
report.status_extended = (
f"Account is part of AWS Organization: {org.id}."
f"AWS Organization {org.id} contains this AWS account."
)
else:
report.status = "FAIL"
@@ -30,13 +30,15 @@ class organizations_delegated_administrators(Check):
not in organizations_trusted_delegated_administrators
):
report.status = "FAIL"
report.status_extended = f"Untrusted Delegated Administrators: {delegated_administrator.id}."
report.status_extended = f"AWS Organization {org.id} has an untrusted Delegated Administrator: {delegated_administrator.id}."
else:
report.status = "PASS"
report.status_extended = f"Trusted Delegated Administrator: {delegated_administrator.id}."
report.status_extended = f"AWS Organization {org.id} has a trusted Delegated Administrator: {delegated_administrator.id}."
else:
report.status = "PASS"
report.status_extended = f"No Delegated Administrators: {org.id}."
report.status_extended = (
f"AWS Organization {org.id} has no Delegated Administrators."
)
findings.append(report)
@@ -23,7 +23,7 @@ class organizations_scp_check_deny_regions(Check):
if not org.policies:
report.status = "FAIL"
report.status_extended = (
f"No SCP policies exist at the organization {org.id} level."
f"AWS Organization {org.id} has no SCP policies."
)
else:
# We use this flag if we find a statement that is restricting regions but not all the configured ones:
@@ -57,14 +57,14 @@ class organizations_scp_check_deny_regions(Check):
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"SCP policy {policy.id} restricting all configured regions found."
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"SCP policies exist {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
# Allow if Condition = {"StringEquals": {"aws:RequestedRegion": [region1, region2]}}
if (
@@ -83,18 +83,18 @@ class organizations_scp_check_deny_regions(Check):
):
# All defined regions are restricted, we exit here, no need to continue.
report.status = "PASS"
report.status_extended = f"SCP policy {policy.id} restricting all configured regions found."
report.status_extended = f"AWS Organization {org.id} has SCP policy {policy.id} restricting all configured regions found."
findings.append(report)
return findings
else:
# Regions are restricted, but not the ones defined, we keep this finding, but we continue analyzing:
is_region_restricted_statement = True
report.status = "FAIL"
report.status_extended = f"SCP policies exist {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
report.status_extended = f"AWS Organization {org.id} has SCP policies {policy.id} restricting some AWS Regions, but not all the configured ones, please check config."
if not is_region_restricted_statement:
report.status = "FAIL"
report.status_extended = f"SCP policies exist at the organization {org.id} level but don't restrict AWS Regions."
report.status_extended = f"AWS Organization {org.id} has SCP policies but don't restrict AWS Regions."
else:
report.status = "FAIL"
@@ -13,12 +13,12 @@ class redshift_cluster_audit_logging(Check):
report.resource_tags = cluster.tags
report.status = "PASS"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has audit logging enabled."
f"Redshift Cluster {cluster.id} has audit logging enabled."
)
if not cluster.logging_enabled:
report.status = "FAIL"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has audit logging disabled."
f"Redshift Cluster {cluster.id} has audit logging disabled."
)
findings.append(report)
@@ -13,12 +13,12 @@ class redshift_cluster_automated_snapshot(Check):
report.resource_tags = cluster.tags
report.status = "PASS"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has automated snapshots."
f"Redshift Cluster {cluster.id} has automated snapshots enabled."
)
if not cluster.cluster_snapshots:
report.status = "FAIL"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has automated snapshots disabled."
f"Redshift Cluster {cluster.id} has automated snapshots disabled."
)
findings.append(report)
@@ -13,12 +13,12 @@ class redshift_cluster_automatic_upgrades(Check):
report.resource_tags = cluster.tags
report.status = "PASS"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has AllowVersionUpgrade enabled."
f"Redshift Cluster {cluster.id} has AllowVersionUpgrade enabled."
)
if not cluster.allow_version_upgrade:
report.status = "FAIL"
report.status_extended = (
f"Redshift Cluster {cluster.arn} has AllowVersionUpgrade disabled."
f"Redshift Cluster {cluster.id} has AllowVersionUpgrade disabled."
)
findings.append(report)
@@ -13,11 +13,11 @@ class redshift_cluster_public_access(Check):
report.resource_tags = cluster.tags
report.status = "PASS"
report.status_extended = (
f"Redshift Cluster {cluster.arn} is not publicly accessible."
f"Redshift Cluster {cluster.id} is not publicly accessible."
)
if cluster.endpoint_address and cluster.public_access:
report.status = "FAIL"
report.status_extended = f"Redshift Cluster {cluster.arn} is publicly accessible at endpoint {cluster.endpoint_address}."
report.status_extended = f"Redshift Cluster {cluster.id} is publicly accessible at endpoint {cluster.endpoint_address}."
findings.append(report)
@@ -16,6 +16,6 @@ class s3_bucket_default_encryption(Check):
report.status_extended = f"S3 Bucket {bucket.name} has Server Side Encryption with {bucket.encryption}."
else:
report.status = "FAIL"
report.status_extended = f"Server Side Encryption is not configured for S3 Bucket {bucket.name}."
report.status_extended = f"S3 Bucket {bucket.name} does not have Server Side Encryption enabled."
findings.append(report)
return findings
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -39,9 +38,9 @@ class Test_organizations_account_part_of_organizations:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"AWS Organizations is not in-use for this AWS Account",
result[0].status_extended,
assert (
result[0].status_extended
== "AWS Organizations is not in-use for this AWS Account."
)
assert result[0].resource_id == "AWS Organization"
assert result[0].resource_arn == AWS_ACCOUNT_ARN
@@ -54,6 +53,7 @@ class Test_organizations_account_part_of_organizations:
# Create Organization
conn = client("organizations")
response = conn.create_organization()
org_id = response["Organization"]["Id"]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -73,9 +73,9 @@ class Test_organizations_account_part_of_organizations:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"Account is part of AWS Organization",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} contains this AWS account."
)
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -47,6 +46,7 @@ class Test_organizations_delegated_administrators:
# Create Organization
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
org_id = response["Organization"]["Id"]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -68,9 +68,9 @@ class Test_organizations_delegated_administrators:
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"No Delegated Administrators",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has no Delegated Administrators."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -91,6 +91,8 @@ class Test_organizations_delegated_administrators:
AccountId=account["CreateAccountStatus"]["AccountId"],
ServicePrincipal="config-multiaccountsetup.amazonaws.com",
)
org_id = response["Organization"]["Id"]
account_id = account["CreateAccountStatus"]["AccountId"]
# Set config variable
aws_provider._audit_config = {
@@ -119,9 +121,9 @@ class Test_organizations_delegated_administrators:
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"Trusted Delegated Administrator",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has a trusted Delegated Administrator: {account_id}."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -142,6 +144,8 @@ class Test_organizations_delegated_administrators:
AccountId=account["CreateAccountStatus"]["AccountId"],
ServicePrincipal="config-multiaccountsetup.amazonaws.com",
)
org_id = response["Organization"]["Id"]
account_id = account["CreateAccountStatus"]["AccountId"]
# Set config variable
aws_provider._audit_config = {
@@ -168,8 +172,8 @@ class Test_organizations_delegated_administrators:
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"Untrusted Delegated Administrator",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has an untrusted Delegated Administrator: {account_id}."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -46,9 +45,9 @@ class Test_organizations_scp_check_deny_regions:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"AWS Organizations is not in-use for this AWS Account",
result[0].status_extended,
assert (
result[0].status_extended
== "AWS Organizations is not in-use for this AWS Account."
)
assert result[0].resource_id == "AWS Organization"
assert result[0].resource_arn == AWS_ACCOUNT_ARN
@@ -64,6 +63,7 @@ class Test_organizations_scp_check_deny_regions:
# Create Organization
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
org_id = response["Organization"]["Id"]
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
@@ -85,9 +85,9 @@ class Test_organizations_scp_check_deny_regions:
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"level but don't restrict AWS Regions",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has SCP policies but don't restrict AWS Regions."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -99,12 +99,14 @@ class Test_organizations_scp_check_deny_regions:
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
# Create Policy
conn.create_policy(
response_policy = conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
org_id = response["Organization"]["Id"]
policy_id = response_policy["Policy"]["PolicySummary"]["Id"]
# Set config variable
aws_provider._audit_config = {"organizations_enabled_regions": ["eu-central-1"]}
@@ -129,9 +131,9 @@ class Test_organizations_scp_check_deny_regions:
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"restricting all configured regions found",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has SCP policy {policy_id} restricting all configured regions found."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -143,12 +145,14 @@ class Test_organizations_scp_check_deny_regions:
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
# Create Policy
conn.create_policy(
response_policy = conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
org_id = response["Organization"]["Id"]
policy_id = response_policy["Policy"]["PolicySummary"]["Id"]
# Set config variable
aws_provider._audit_config = {"organizations_enabled_regions": ["us-east-1"]}
@@ -173,9 +177,9 @@ class Test_organizations_scp_check_deny_regions:
assert result[0].status == "FAIL"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"restricting some AWS Regions, but not all the configured ones, please check config.",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has SCP policies {policy_id} restricting some AWS Regions, but not all the configured ones, please check config."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -192,12 +196,14 @@ class Test_organizations_scp_check_deny_regions:
conn = client("organizations", region_name=AWS_REGION_EU_WEST_1)
response = conn.create_organization()
# Create Policy
conn.create_policy(
response_policy = conn.create_policy(
Content=scp_restrict_regions_with_deny(),
Description="Test",
Name="Test",
Type="SERVICE_CONTROL_POLICY",
)
org_id = response["Organization"]["Id"]
policy_id = response_policy["Policy"]["PolicySummary"]["Id"]
# Set config variable
aws_provider._audit_config = {"organizations_enabled_regions": ["eu-central-1"]}
@@ -222,8 +228,8 @@ class Test_organizations_scp_check_deny_regions:
assert result[0].status == "PASS"
assert result[0].resource_id == response["Organization"]["Id"]
assert result[0].resource_arn == response["Organization"]["Arn"]
assert search(
"restricting all configured regions found",
result[0].status_extended,
assert (
result[0].status_extended
== f"AWS Organization {org_id} has SCP policy {policy_id} restricting all configured regions found."
)
assert result[0].region == AWS_REGION_EU_WEST_1
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from uuid import uuid4
@@ -49,7 +48,10 @@ class Test_redshift_cluster_audit_logging:
check = redshift_cluster_audit_logging()
result = check.execute()
assert result[0].status == "FAIL"
assert search("has audit logging disabled", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has audit logging disabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -76,6 +78,9 @@ class Test_redshift_cluster_audit_logging:
check = redshift_cluster_audit_logging()
result = check.execute()
assert result[0].status == "PASS"
assert search("has audit logging enabled", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has audit logging enabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from uuid import uuid4
@@ -49,7 +48,10 @@ class Test_redshift_cluster_automated_snapshot:
check = redshift_cluster_automated_snapshot()
result = check.execute()
assert result[0].status == "FAIL"
assert search("has automated snapshots disabled", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has automated snapshots disabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -75,6 +77,9 @@ class Test_redshift_cluster_automated_snapshot:
check = redshift_cluster_automated_snapshot()
result = check.execute()
assert result[0].status == "PASS"
assert search("has automated snapshots", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has automated snapshots enabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from uuid import uuid4
@@ -49,7 +48,10 @@ class Test_redshift_cluster_automatic_upgrades:
check = redshift_cluster_automatic_upgrades()
result = check.execute()
assert result[0].status == "FAIL"
assert search("has AllowVersionUpgrade disabled", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has AllowVersionUpgrade disabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -75,6 +77,9 @@ class Test_redshift_cluster_automatic_upgrades:
check = redshift_cluster_automatic_upgrades()
result = check.execute()
assert result[0].status == "PASS"
assert search("has AllowVersionUpgrade enabled", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} has AllowVersionUpgrade enabled."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from uuid import uuid4
@@ -50,7 +49,10 @@ class Test_redshift_cluster_public_access:
check = redshift_cluster_public_access()
result = check.execute()
assert result[0].status == "FAIL"
assert search("is publicly accessible", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} is publicly accessible at endpoint 192.192.192.192."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -77,7 +79,10 @@ class Test_redshift_cluster_public_access:
check = redshift_cluster_public_access()
result = check.execute()
assert result[0].status == "PASS"
assert search("is not publicly accessible", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -103,6 +108,9 @@ class Test_redshift_cluster_public_access:
check = redshift_cluster_public_access()
result = check.execute()
assert result[0].status == "PASS"
assert search("is not publicly accessible", result[0].status_extended)
assert (
result[0].status_extended
== f"Redshift Cluster {CLUSTER_ID} is not publicly accessible."
)
assert result[0].resource_id == CLUSTER_ID
assert result[0].resource_arn == CLUSTER_ARN
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -36,9 +35,9 @@ class Test_s3_bucket_acl_prohibited:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has bucket ACLs enabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -75,9 +74,9 @@ class Test_s3_bucket_acl_prohibited:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"ACLs enabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has bucket ACLs enabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -116,9 +115,9 @@ class Test_s3_bucket_acl_prohibited:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"ACLs disabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has bucket ACLs disabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -36,9 +35,9 @@ class Test_s3_bucket_default_encryption:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"Server Side Encryption is not configured",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} does not have Server Side Encryption enabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -91,9 +90,9 @@ class Test_s3_bucket_default_encryption:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"has Server Side Encryption",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has Server Side Encryption with aws:kms."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -60,9 +59,9 @@ class Test_s3_bucket_no_mfa_delete:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"MFA Delete disabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has MFA Delete disabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -102,9 +101,9 @@ class Test_s3_bucket_no_mfa_delete:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"MFA Delete enabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has MFA Delete enabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -36,9 +35,9 @@ class Test_s3_bucket_object_versioning:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"versioning disabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has versioning disabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -81,9 +80,9 @@ class Test_s3_bucket_object_versioning:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"versioning enabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has versioning enabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -194,9 +193,9 @@ class Test_s3_bucket_public_access:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"not public",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} is not public."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -272,9 +271,9 @@ class Test_s3_bucket_public_access:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"public access due to bucket ACL",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has public access due to bucket ACL."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -339,9 +338,9 @@ class Test_s3_bucket_public_access:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"public access due to bucket policy",
result[0].status_extended,
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 (
@@ -390,9 +389,9 @@ class Test_s3_bucket_public_access:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"not public",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} is not public."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -36,9 +35,9 @@ class Test_s3_bucket_secure_transport_policy:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"does not have a bucket policy",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} does not have a bucket policy, thus it allows HTTP requests."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -99,9 +98,9 @@ class Test_s3_bucket_secure_transport_policy:
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].status_extended
== f"S3 Bucket {bucket_name_us} has a bucket policy to deny requests over insecure transport."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -162,9 +161,9 @@ class Test_s3_bucket_secure_transport_policy:
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].status_extended
== f"S3 Bucket {bucket_name_us} allows requests over insecure transport in the bucket policy."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -36,9 +35,9 @@ class Test_s3_bucket_server_access_logging_enabled:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"server access logging disabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has server access logging disabled."
)
assert result[0].resource_id == bucket_name_us
assert (
@@ -129,9 +128,9 @@ class Test_s3_bucket_server_access_logging_enabled:
assert len(result) == 1
assert result[0].status == "PASS"
assert search(
"server access logging enabled",
result[0].status_extended,
assert (
result[0].status_extended
== f"S3 Bucket {bucket_name_us} has server access logging enabled."
)
assert result[0].resource_id == bucket_name_us
assert (