ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-09 21:14:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(aws): add new check dynamodb_accelerator_cluster_multi_az (#5493)

Browse Source 
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
sansns-aws
2024-10-23 10:16:56 -07:00
committed by GitHub
co-authored by Sergio
parent 41cdc2bcc7
commit 7e7d86f14a
9 changed files with 199 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_encryption_enabled/dynamodb_accelerator_cluster_encryption_enabled_test.py
+9 -5
View File
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -47,6 +46,7 @@ class Test_dynamodb_accelerator_cluster_encryption_enabled:
NodeType="dax.t3.small",
ReplicationFactor=3,
IamRoleArn=iam_role_arn,
ClusterEndpointEncryptionType="TLS",
)["Cluster"]
from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
@@ -71,9 +71,9 @@ class Test_dynamodb_accelerator_cluster_encryption_enabled:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search(
"does not have encryption at rest enabled",
result[0].status_extended,
assert (
result[0].status_extended
== "DAX cluster daxcluster does not have encryption at rest enabled."
)
assert result[0].resource_id == cluster["ClusterName"]
assert result[0].resource_arn == cluster["ClusterArn"]
@@ -89,6 +89,7 @@ class Test_dynamodb_accelerator_cluster_encryption_enabled:
NodeType="dax.t3.small",
ReplicationFactor=3,
IamRoleArn=iam_role_arn,
ClusterEndpointEncryptionType="TLS",
SSESpecification={"Enabled": True},
)["Cluster"]
from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
@@ -114,7 +115,10 @@ class Test_dynamodb_accelerator_cluster_encryption_enabled:
assert len(result) == 1
assert result[0].status == "PASS"
assert search("has encryption at rest enabled", result[0].status_extended)
assert (
result[0].status_extended
== "DAX cluster daxcluster has encryption at rest enabled."
)
assert result[0].resource_id == cluster["ClusterName"]
assert result[0].resource_arn == cluster["ClusterArn"]
assert result[0].region == AWS_REGION_US_EAST_1
tests/providers/aws/services/dynamodb/dynamodb_accelerator_cluster_multi_az/dynamodb_accelerator_cluster_multi_az_test.py
+125
View File
@@ -0,0 +1,125 @@
from unittest import mock
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
class Test_dynamodb_accelerator_cluster_multi_az:
@mock_aws
def test_dax_no_clusters(self):
from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, 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.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az.dax_client",
new=DAX(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az import (
dynamodb_accelerator_cluster_multi_az,
)
check = dynamodb_accelerator_cluster_multi_az()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_dax_cluster_no_multi_az(self):
dax_client = client("dax", region_name=AWS_REGION_US_EAST_1)
iam_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/aws-service-role/dax.amazonaws.com/AWSServiceRoleForDAX"
cluster = dax_client.create_cluster(
ClusterName="daxcluster",
NodeType="dax.t3.small",
ReplicationFactor=3,
IamRoleArn=iam_role_arn,
)["Cluster"]
from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, 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.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az.dax_client",
new=DAX(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az import (
dynamodb_accelerator_cluster_multi_az,
)
check = dynamodb_accelerator_cluster_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "DAX cluster daxcluster does not have nodes in multiple availability zones."
)
assert result[0].resource_id == cluster["ClusterName"]
assert result[0].resource_arn == cluster["ClusterArn"]
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
@mock_aws
def test_dax_cluster_with_multi_az(self):
dax_client = client("dax", region_name=AWS_REGION_US_EAST_1)
iam_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/aws-service-role/dax.amazonaws.com/AWSServiceRoleForDAX"
cluster = dax_client.create_cluster(
ClusterName="daxcluster",
NodeType="dax.t3.small",
ReplicationFactor=3,
IamRoleArn=iam_role_arn,
ClusterEndpointEncryptionType="TLS",
)["Cluster"]
from prowler.providers.aws.services.dynamodb.dynamodb_service import DAX
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, 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.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az.dax_client",
new=DAX(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_accelerator_cluster_multi_az.dynamodb_accelerator_cluster_multi_az import (
dynamodb_accelerator_cluster_multi_az,
)
# Setting node_azs manually as Moto does not support that yet.
service_client.clusters[0].node_azs = ["us-east-1a", "us-east-1b"]
check = dynamodb_accelerator_cluster_multi_az()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "DAX cluster daxcluster has nodes in multiple availability zones."
)
assert result[0].resource_id == cluster["ClusterName"]
assert result[0].resource_arn == cluster["ClusterArn"]
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_tags == []
tests/providers/aws/services/dynamodb/dynamodb_service_test.py
-1
View File
@@ -195,7 +195,6 @@ class Test_DynamoDB_Service:
{"Key": "test", "Value": "test"},
]
assert dax.clusters[0].tls_encryption
assert dax.clusters[1].name == "daxcluster2"
assert dax.clusters[1].region == AWS_REGION_US_EAST_1
assert dax.clusters[1].encryption
tests/providers/aws/services/dynamodb/dynamodb_tables_kms_cmk_encryption_enabled/dynamodb_tables_kms_cmk_encryption_enabled_test.py
+8 -3
View File
@@ -1,4 +1,3 @@
from re import search
from unittest import mock
from boto3 import client
@@ -76,7 +75,10 @@ class Test_dynamodb_tables_kms_cmk_encryption_enabled:
assert len(result) == 1
assert result[0].status == "PASS"
assert search("KMS encryption enabled", result[0].status_extended)
assert (
result[0].status_extended
== "DynamoDB table test1 has KMS encryption enabled with key custom-kms-key."
)
assert result[0].resource_id == table["TableName"]
assert result[0].resource_arn == table["TableArn"]
assert result[0].region == AWS_REGION_US_EAST_1
@@ -120,7 +122,10 @@ class Test_dynamodb_tables_kms_cmk_encryption_enabled:
assert len(result) == 1
assert result[0].status == "FAIL"
assert search("DEFAULT encryption enabled", result[0].status_extended)
assert (
result[0].status_extended
== "DynamoDB table test1 is using DEFAULT encryption."
)
assert result[0].resource_id == table["TableName"]
assert result[0].resource_arn == table["TableArn"]
assert result[0].region == AWS_REGION_US_EAST_1