mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 21:11:53 +00:00
feat(dynamodb): add new check dynamodb_table_autoscaling_enabled (#5129)
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
committed by
GitHub
parent
2511df1732
commit
2c2dd82d0c
@@ -0,0 +1,6 @@
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
)
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
applicationautoscaling_client = ApplicationAutoScaling(Provider.get_global_provider())
|
||||
@@ -5,7 +5,6 @@ from prowler.lib.scan_filters.scan_filters import is_resource_filtered
|
||||
from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
|
||||
################## AutoScaling
|
||||
class AutoScaling(AWSService):
|
||||
def __init__(self, provider):
|
||||
# Call AWSService's __init__
|
||||
@@ -74,6 +73,49 @@ class AutoScaling(AWSService):
|
||||
)
|
||||
|
||||
|
||||
# Global list for service namespaces needed for Describe Scalable Targets
|
||||
SERVICE_NAMESPACES = ["dynamodb"]
|
||||
|
||||
|
||||
class ApplicationAutoScaling(AWSService):
|
||||
def __init__(self, provider):
|
||||
super().__init__("application-autoscaling", provider)
|
||||
self.scalable_targets = []
|
||||
self.__threading_call__(self._describe_scalable_targets)
|
||||
|
||||
def _describe_scalable_targets(self, regional_client):
|
||||
logger.info("ApplicationAutoScaling - Describing Scalable Targets...")
|
||||
try:
|
||||
describe_scalable_targets_paginator = regional_client.get_paginator(
|
||||
"describe_scalable_targets"
|
||||
)
|
||||
for service_namespace in SERVICE_NAMESPACES:
|
||||
logger.info(f"Processing ServiceNamespace: {service_namespace}")
|
||||
for page in describe_scalable_targets_paginator.paginate(
|
||||
ServiceNamespace=service_namespace
|
||||
):
|
||||
for target in page.get("ScalableTargets", []):
|
||||
if not self.audit_resources or (
|
||||
is_resource_filtered(
|
||||
target["ScalableTargetARN"],
|
||||
self.audit_resources,
|
||||
)
|
||||
):
|
||||
self.scalable_targets.append(
|
||||
ScalableTarget(
|
||||
arn=target.get("ScalableTargetARN", ""),
|
||||
resource_id=target.get("ResourceId"),
|
||||
service_namespace=target.get("ServiceNamespace"),
|
||||
scalable_dimension=target.get("ScalableDimension"),
|
||||
region=regional_client.region,
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
class LaunchConfiguration(BaseModel):
|
||||
arn: str
|
||||
name: str
|
||||
@@ -88,3 +130,11 @@ class Group(BaseModel):
|
||||
region: str
|
||||
availability_zones: list
|
||||
tags: list = []
|
||||
|
||||
|
||||
class ScalableTarget(BaseModel):
|
||||
arn: str
|
||||
resource_id: str
|
||||
service_namespace: str
|
||||
scalable_dimension: str
|
||||
region: str
|
||||
|
||||
@@ -11,7 +11,6 @@ from prowler.providers.aws.lib.service.service import AWSService
|
||||
|
||||
class DynamoDB(AWSService):
|
||||
def __init__(self, provider):
|
||||
# Call AWSService's __init__
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.tables = {}
|
||||
self.__threading_call__(self._list_tables)
|
||||
@@ -49,6 +48,9 @@ class DynamoDB(AWSService):
|
||||
properties = regional_client.describe_table(TableName=table.name)[
|
||||
"Table"
|
||||
]
|
||||
table.billing_mode = properties.get("BillingModeSummary", {}).get(
|
||||
"BillingMode", "PROVISIONED"
|
||||
)
|
||||
if "SSEDescription" in properties:
|
||||
if "SSEType" in properties["SSEDescription"]:
|
||||
table.encryption_type = properties["SSEDescription"]["SSEType"]
|
||||
@@ -152,7 +154,6 @@ class DynamoDB(AWSService):
|
||||
|
||||
class DAX(AWSService):
|
||||
def __init__(self, provider):
|
||||
# Call AWSService's __init__
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.clusters = []
|
||||
self.__threading_call__(self._describe_clusters)
|
||||
@@ -217,6 +218,7 @@ class DAX(AWSService):
|
||||
|
||||
class Table(BaseModel):
|
||||
name: str
|
||||
billing_mode: str = "PROVISIONED"
|
||||
encryption_type: Optional[str]
|
||||
kms_arn: Optional[str]
|
||||
pitr: bool = False
|
||||
|
||||
+32
@@ -0,0 +1,32 @@
|
||||
{
|
||||
"Provider": "aws",
|
||||
"CheckID": "dynamodb_table_autoscaling_enabled",
|
||||
"CheckTitle": "Check if DynamoDB tables automatically scale capacity with demand.",
|
||||
"CheckType": [
|
||||
"Software and Configuration Checks/AWS Security Best Practices"
|
||||
],
|
||||
"ServiceName": "dynamodb",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "arn:aws:dynamodb:region:account-id:table/table-name",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "AwsDynamoDbTable",
|
||||
"Description": "This check ensures that DynamoDB tables can scale their read and write capacity as needed, either using on-demand capacity mode or provisioned mode with auto scaling configured.",
|
||||
"Risk": "If DynamoDB tables do not automatically scale capacity with demand, they may experience throttling exceptions, leading to reduced availability and performance of applications.",
|
||||
"RelatedUrl": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/AutoScaling.Console.html#AutoScaling.Console.ExistingTable",
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "aws dynamodb update-table --table-name <table-name> --billing-mode PAY_PER_REQUEST",
|
||||
"NativeIaC": "",
|
||||
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dynamodb-controls.html#dynamodb-1",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable DynamoDB automatic scaling on existing tables by configuring on-demand capacity mode or provisioned mode with auto scaling.",
|
||||
"Url": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/AutoScaling.Console.html#AutoScaling.Console.ExistingTable"
|
||||
}
|
||||
},
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
from prowler.lib.check.models import Check, Check_Report_AWS
|
||||
from prowler.providers.aws.services.autoscaling.applicationautoscaling_client import (
|
||||
applicationautoscaling_client,
|
||||
)
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_client
|
||||
|
||||
|
||||
class dynamodb_table_autoscaling_enabled(Check):
|
||||
def execute(self):
|
||||
findings = []
|
||||
scalable_targets = applicationautoscaling_client.scalable_targets
|
||||
dynamodb_scalable_targets = [
|
||||
target
|
||||
for target in scalable_targets
|
||||
if target.service_namespace == "dynamodb"
|
||||
and target.resource_id.startswith("table/")
|
||||
]
|
||||
autoscaling_mapping = {}
|
||||
for target in dynamodb_scalable_targets:
|
||||
table_name = target.resource_id.split("/")[1]
|
||||
if table_name not in autoscaling_mapping:
|
||||
autoscaling_mapping[table_name] = {}
|
||||
autoscaling_mapping[table_name][target.scalable_dimension] = target
|
||||
|
||||
for table_arn, table in dynamodb_client.tables.items():
|
||||
report = Check_Report_AWS(self.metadata())
|
||||
report.region = table.region
|
||||
report.resource_id = table.name
|
||||
report.resource_arn = table_arn
|
||||
report.resource_tags = table.tags
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"DynamoDB table {table.name} automatically scales capacity on demand."
|
||||
)
|
||||
if table.billing_mode == "PROVISIONED":
|
||||
read_autoscaling = False
|
||||
write_autoscaling = False
|
||||
|
||||
if table.name in autoscaling_mapping:
|
||||
if (
|
||||
"dynamodb:table:ReadCapacityUnits"
|
||||
in autoscaling_mapping[table.name]
|
||||
):
|
||||
read_autoscaling = True
|
||||
if (
|
||||
"dynamodb:table:WriteCapacityUnits"
|
||||
in autoscaling_mapping[table.name]
|
||||
):
|
||||
write_autoscaling = True
|
||||
|
||||
if read_autoscaling and write_autoscaling:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"DynamoDB table {table.name} is in provisioned mode with auto scaling enabled for both read and write capacity units."
|
||||
else:
|
||||
missing_autoscaling = []
|
||||
if not read_autoscaling:
|
||||
missing_autoscaling.append("read")
|
||||
if not write_autoscaling:
|
||||
missing_autoscaling.append("write")
|
||||
|
||||
if missing_autoscaling:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"DynamoDB table {table.name} is in provisioned mode without auto scaling enabled for {', '.join(missing_autoscaling)}."
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -4,7 +4,10 @@ from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from prowler.config.config import encoding_format_utf_8
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import AutoScaling
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
AutoScaling,
|
||||
)
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_ACCOUNT_NUMBER,
|
||||
AWS_REGION_US_EAST_1,
|
||||
@@ -127,3 +130,42 @@ class Test_AutoScaling_Service:
|
||||
"Value": "value_test",
|
||||
}
|
||||
]
|
||||
|
||||
# Test Application AutoScaling Describe Scalable Targets
|
||||
@mock_aws
|
||||
def test_application_auto_scaling_scalable_targets(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)["TableDescription"]
|
||||
|
||||
autoscaling_client = client(
|
||||
"application-autoscaling", region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
autoscaling_client.register_scalable_target(
|
||||
ServiceNamespace="dynamodb",
|
||||
ResourceId=f"table/{table['TableName']}",
|
||||
ScalableDimension="dynamodb:table:ReadCapacityUnits",
|
||||
MinCapacity=1,
|
||||
MaxCapacity=10,
|
||||
)
|
||||
|
||||
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
|
||||
autoscaling = ApplicationAutoScaling(aws_provider)
|
||||
assert len(autoscaling.scalable_targets) == 1
|
||||
assert autoscaling.scalable_targets[0].service_namespace == "dynamodb"
|
||||
assert autoscaling.scalable_targets[0].resource_id == "table/test1"
|
||||
assert (
|
||||
autoscaling.scalable_targets[0].scalable_dimension
|
||||
== "dynamodb:table:ReadCapacityUnits"
|
||||
)
|
||||
|
||||
@@ -71,7 +71,8 @@ class Test_DynamoDB_Service:
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)
|
||||
# DynamoDB client for this test class
|
||||
aws_provider = set_mocked_aws_provider()
|
||||
@@ -82,6 +83,9 @@ class Test_DynamoDB_Service:
|
||||
assert "test2" in table_names
|
||||
for table in dynamo.tables.values():
|
||||
assert table.region == AWS_REGION_US_EAST_1
|
||||
table_billing = [table.billing_mode for table in dynamo.tables.values()]
|
||||
assert "PAY_PER_REQUEST" in table_billing
|
||||
assert "PROVISIONED" in table_billing
|
||||
|
||||
# Test DynamoDB Describe Table
|
||||
@mock_aws
|
||||
@@ -116,6 +120,7 @@ class Test_DynamoDB_Service:
|
||||
assert tables.tags == [
|
||||
{"Key": "test", "Value": "test"},
|
||||
]
|
||||
assert tables.billing_mode == "PAY_PER_REQUEST"
|
||||
assert tables.deletion_protection
|
||||
|
||||
# Test DynamoDB Describe Continuous Backups
|
||||
|
||||
+340
@@ -0,0 +1,340 @@
|
||||
from unittest import mock
|
||||
|
||||
from boto3 import client
|
||||
from moto import mock_aws
|
||||
|
||||
from tests.providers.aws.utils import (
|
||||
AWS_REGION_EU_WEST_1,
|
||||
AWS_REGION_US_EAST_1,
|
||||
set_mocked_aws_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_dynamodb_table_autoscaling_enabled:
|
||||
@mock_aws
|
||||
def test_dynamodb_no_tables(self):
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 0
|
||||
|
||||
@mock_aws
|
||||
def test_dynamodb_table_on_demand(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)["TableDescription"]
|
||||
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].status_extended == (
|
||||
"DynamoDB table test1 automatically scales capacity on demand."
|
||||
)
|
||||
assert result[0].resource_id == table["TableName"]
|
||||
assert result[0].resource_arn == table["TableArn"]
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
@mock_aws
|
||||
def test_dynamodb_table_provisioned_with_autoscaling(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)["TableDescription"]
|
||||
|
||||
autoscaling_client = client(
|
||||
"application-autoscaling", region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
autoscaling_client.register_scalable_target(
|
||||
ServiceNamespace="dynamodb",
|
||||
ResourceId=f"table/{table['TableName']}",
|
||||
ScalableDimension="dynamodb:table:ReadCapacityUnits",
|
||||
MinCapacity=1,
|
||||
MaxCapacity=10,
|
||||
)
|
||||
autoscaling_client.register_scalable_target(
|
||||
ServiceNamespace="dynamodb",
|
||||
ResourceId=f"table/{table['TableName']}",
|
||||
ScalableDimension="dynamodb:table:WriteCapacityUnits",
|
||||
MinCapacity=1,
|
||||
MaxCapacity=10,
|
||||
)
|
||||
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
)
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.applicationautoscaling_client",
|
||||
new=ApplicationAutoScaling(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert result[0].status_extended == (
|
||||
"DynamoDB table test1 is in provisioned mode with auto scaling enabled for both read and write capacity units."
|
||||
)
|
||||
assert result[0].resource_id == table["TableName"]
|
||||
assert result[0].resource_arn == table["TableArn"]
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
@mock_aws
|
||||
def test_dynamodb_table_provisioned_only_with_read_autoscaling(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)["TableDescription"]
|
||||
|
||||
autoscaling_client = client(
|
||||
"application-autoscaling", region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
autoscaling_client.register_scalable_target(
|
||||
ServiceNamespace="dynamodb",
|
||||
ResourceId=f"table/{table['TableName']}",
|
||||
ScalableDimension="dynamodb:table:ReadCapacityUnits",
|
||||
MinCapacity=1,
|
||||
MaxCapacity=10,
|
||||
)
|
||||
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
)
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.applicationautoscaling_client",
|
||||
new=ApplicationAutoScaling(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].status_extended == (
|
||||
"DynamoDB table test1 is in provisioned mode without auto scaling enabled for write."
|
||||
)
|
||||
assert result[0].resource_id == table["TableName"]
|
||||
assert result[0].resource_arn == table["TableArn"]
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
@mock_aws
|
||||
def test_dynamodb_table_provisioned_only_with_write_autoscaling(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)["TableDescription"]
|
||||
|
||||
autoscaling_client = client(
|
||||
"application-autoscaling", region_name=AWS_REGION_US_EAST_1
|
||||
)
|
||||
autoscaling_client.register_scalable_target(
|
||||
ServiceNamespace="dynamodb",
|
||||
ResourceId=f"table/{table['TableName']}",
|
||||
ScalableDimension="dynamodb:table:WriteCapacityUnits",
|
||||
MinCapacity=1,
|
||||
MaxCapacity=10,
|
||||
)
|
||||
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
)
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.applicationautoscaling_client",
|
||||
new=ApplicationAutoScaling(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].status_extended == (
|
||||
"DynamoDB table test1 is in provisioned mode without auto scaling enabled for read."
|
||||
)
|
||||
assert result[0].resource_id == table["TableName"]
|
||||
assert result[0].resource_arn == table["TableArn"]
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
|
||||
@mock_aws
|
||||
def test_dynamodb_table_provisioned_without_autoscaling(self):
|
||||
dynamodb_client = client("dynamodb", region_name=AWS_REGION_US_EAST_1)
|
||||
table = dynamodb_client.create_table(
|
||||
TableName="test1",
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "client", "AttributeType": "S"},
|
||||
{"AttributeName": "app", "AttributeType": "S"},
|
||||
],
|
||||
KeySchema=[
|
||||
{"AttributeName": "client", "KeyType": "HASH"},
|
||||
{"AttributeName": "app", "KeyType": "RANGE"},
|
||||
],
|
||||
BillingMode="PROVISIONED",
|
||||
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
|
||||
)["TableDescription"]
|
||||
|
||||
from prowler.providers.aws.services.autoscaling.autoscaling_service import (
|
||||
ApplicationAutoScaling,
|
||||
)
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
|
||||
|
||||
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_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.dynamodb_client",
|
||||
new=DynamoDB(aws_provider),
|
||||
), mock.patch(
|
||||
"prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled.applicationautoscaling_client",
|
||||
new=ApplicationAutoScaling(aws_provider),
|
||||
):
|
||||
# Test Check
|
||||
from prowler.providers.aws.services.dynamodb.dynamodb_table_autoscaling_enabled.dynamodb_table_autoscaling_enabled import (
|
||||
dynamodb_table_autoscaling_enabled,
|
||||
)
|
||||
|
||||
check = dynamodb_table_autoscaling_enabled()
|
||||
result = check.execute()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert result[0].status_extended == (
|
||||
"DynamoDB table test1 is in provisioned mode without auto scaling enabled for read, write."
|
||||
)
|
||||
assert result[0].resource_id == table["TableName"]
|
||||
assert result[0].resource_arn == table["TableArn"]
|
||||
assert result[0].region == AWS_REGION_US_EAST_1
|
||||
assert result[0].resource_tags == []
|
||||
Reference in New Issue
Block a user