feat(dynamodb): add new check dynamodb_table_protected_by_backup_plan (#5175)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Daniel Barranquero
2024-09-24 18:45:12 +02:00
committed by GitHub
parent 980b9b4770
commit e4890f9d9d
10 changed files with 411 additions and 65 deletions
@@ -14,7 +14,7 @@ class DynamoDB(AWSService):
def __init__(self, provider):
# Call AWSService's __init__
super().__init__(__class__.__name__, provider)
self.tables = []
self.tables = {}
self.__threading_call__(self._list_tables)
self._describe_table()
self._describe_continuous_backups()
@@ -31,14 +31,11 @@ class DynamoDB(AWSService):
if not self.audit_resources or (
is_resource_filtered(arn, self.audit_resources)
):
self.tables.append(
Table(
arn=arn,
name=table,
encryption_type=None,
kms_arn=None,
region=regional_client.region,
)
self.tables[arn] = Table(
name=table,
encryption_type=None,
kms_arn=None,
region=regional_client.region,
)
except Exception as error:
logger.error(
@@ -48,7 +45,7 @@ class DynamoDB(AWSService):
def _describe_table(self):
logger.info("DynamoDB - Describing Table...")
try:
for table in self.tables:
for table in self.tables.values():
regional_client = self.regional_clients[table.region]
properties = regional_client.describe_table(TableName=table.name)[
"Table"
@@ -66,7 +63,7 @@ class DynamoDB(AWSService):
def _describe_continuous_backups(self):
logger.info("DynamoDB - Describing Continuous Backups...")
try:
for table in self.tables:
for table in self.tables.values():
try:
regional_client = self.regional_clients[table.region]
properties = regional_client.describe_continuous_backups(
@@ -98,11 +95,11 @@ class DynamoDB(AWSService):
def _get_resource_policy(self):
logger.info("DynamoDB - Get Resource Policy...")
try:
for table in self.tables:
for table_arn, table in self.tables.items():
try:
regional_client = self.regional_clients[table.region]
response = regional_client.get_resource_policy(
ResourceArn=table.arn
ResourceArn=table_arn
)
table.policy = json.loads(response["Policy"])
except ClientError as error:
@@ -127,11 +124,11 @@ class DynamoDB(AWSService):
def _list_tags_for_resource(self):
logger.info("DynamoDB - List Tags...")
try:
for table in self.tables:
for table_arn, table in self.tables.items():
try:
regional_client = self.regional_clients[table.region]
response = regional_client.list_tags_of_resource(
ResourceArn=table.arn
ResourceArn=table_arn
)["Tags"]
table.tags = response
except ClientError as error:
@@ -212,7 +209,6 @@ class DAX(AWSService):
class Table(BaseModel):
arn: str
name: str
encryption_type: Optional[str]
kms_arn: Optional[str]
@@ -8,10 +8,10 @@ from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_cli
class dynamodb_table_cross_account_access(Check):
def execute(self):
findings = []
for table in dynamodb_client.tables:
for table_arn, table in dynamodb_client.tables.items():
report = Check_Report_AWS(self.metadata())
report.resource_id = table.name
report.resource_arn = table.arn
report.resource_arn = table_arn
report.resource_tags = table.tags
report.region = table.region
report.status = "PASS"
@@ -0,0 +1,34 @@
{
"Provider": "aws",
"CheckID": "dynamodb_table_protected_by_backup_plan",
"CheckTitle": "Check if DynamoDB tables are included in a backup plan.",
"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 control checks whether an Amazon DynamoDB table is covered by a backup plan. The control fails if the DynamoDB table isn't included in a backup plan.",
"Risk": "If a DynamoDB table is not covered by a backup plan, data loss may occur due to accidental deletion, corruption, or unexpected failure, compromising the resilience of your application.",
"RelatedUrl": "https://docs.aws.amazon.com/aws-backup/latest/devguide/assigning-resources.html",
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/dynamodb-controls.html#dynamodb-4",
"Terraform": ""
},
"Recommendation": {
"Text": "Ensure that all active DynamoDB tables are included in a backup plan to safeguard against data loss.",
"Url": "https://docs.aws.amazon.com/aws-backup/latest/devguide/assigning-resources.html"
}
},
"Categories": [
"redundancy"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,26 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.backup.backup_client import backup_client
from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_client
class dynamodb_table_protected_by_backup_plan(Check):
def execute(self):
findings = []
for table_arn, table in dynamodb_client.tables.items():
report = Check_Report_AWS(self.metadata())
report.resource_id = table.name
report.resource_arn = table_arn
report.resource_tags = table.tags
report.region = table.region
report.status = "FAIL"
report.status_extended = (
f"DynamoDB table {table.name} is not protected by a backup plan."
)
if table_arn in backup_client.protected_resources:
report.status = "PASS"
report.status_extended = (
f"DynamoDB table {table.name} is protected by a backup plan."
)
findings.append(report)
return findings
@@ -5,10 +5,10 @@ from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_cli
class dynamodb_tables_kms_cmk_encryption_enabled(Check):
def execute(self):
findings = []
for table in dynamodb_client.tables:
for table_arn, table in dynamodb_client.tables.items():
report = Check_Report_AWS(self.metadata())
report.resource_id = table.name
report.resource_arn = table.arn
report.resource_arn = table_arn
report.resource_tags = table.tags
report.region = table.region
report.status = "FAIL"
@@ -5,10 +5,10 @@ from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_cli
class dynamodb_tables_pitr_enabled(Check):
def execute(self):
findings = []
for table in dynamodb_client.tables:
for table_arn, table in dynamodb_client.tables.items():
report = Check_Report_AWS(self.metadata())
report.resource_id = table.name
report.resource_arn = table.arn
report.resource_arn = table_arn
report.resource_tags = table.tags
report.region = table.region
report.status = "FAIL"
@@ -77,10 +77,11 @@ class Test_DynamoDB_Service:
aws_provider = set_mocked_aws_provider()
dynamo = DynamoDB(aws_provider)
assert len(dynamo.tables) == 2
assert dynamo.tables[0].name == "test1"
assert dynamo.tables[1].name == "test2"
assert dynamo.tables[0].region == AWS_REGION_US_EAST_1
assert dynamo.tables[1].region == AWS_REGION_US_EAST_1
table_names = [table.name for table in dynamo.tables.values()]
assert "test1" in table_names
assert "test2" in table_names
for table in dynamo.tables.values():
assert table.region == AWS_REGION_US_EAST_1
# Test DynamoDB Describe Table
@mock_aws
@@ -107,10 +108,11 @@ class Test_DynamoDB_Service:
aws_provider = set_mocked_aws_provider()
dynamo = DynamoDB(aws_provider)
assert len(dynamo.tables) == 1
assert dynamo.tables[0].arn == table["TableArn"]
assert dynamo.tables[0].name == "test1"
assert dynamo.tables[0].region == AWS_REGION_US_EAST_1
assert dynamo.tables[0].tags == [
tables_arn, tables = next(iter(dynamo.tables.items()))
assert tables_arn == table["TableArn"]
assert tables.name == "test1"
assert tables.region == AWS_REGION_US_EAST_1
assert tables.tags == [
{"Key": "test", "Value": "test"},
]
@@ -140,10 +142,11 @@ class Test_DynamoDB_Service:
aws_provider = set_mocked_aws_provider()
dynamo = DynamoDB(aws_provider)
assert len(dynamo.tables) == 1
assert dynamo.tables[0].arn == table["TableArn"]
assert dynamo.tables[0].name == "test1"
assert dynamo.tables[0].pitr
assert dynamo.tables[0].region == AWS_REGION_US_EAST_1
tables_arn, tables = next(iter(dynamo.tables.items()))
assert tables_arn == table["TableArn"]
assert tables.name == "test1"
assert tables.pitr
assert tables.region == AWS_REGION_US_EAST_1
# Test DAX Describe Clusters
@mock_aws
@@ -103,7 +103,7 @@ test_public_policy_with_invalid_condition_block = {
class Test_dynamodb_table_cross_account_access:
def test_no_tables(self):
dynamodb_client = mock.MagicMock
dynamodb_client.tables = []
dynamodb_client.tables = {}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -124,14 +124,15 @@ class Test_dynamodb_table_cross_account_access:
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables = []
dynamodb_client.tables.append(
Table(
arn=test_table_arn,
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -160,15 +161,15 @@ class Test_dynamodb_table_cross_account_access:
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables = []
dynamodb_client.tables.append(
Table(
arn=test_table_arn,
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_restricted_policy,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -197,15 +198,16 @@ class Test_dynamodb_table_cross_account_access:
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables = []
dynamodb_client.tables.append(
Table(
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=test_table_arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_public_policy,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -234,16 +236,17 @@ class Test_dynamodb_table_cross_account_access:
dynamodb_client = mock.MagicMock
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = []
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables.append(
Table(
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=test_table_arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_public_policy_with_condition_same_account_not_valid,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -272,16 +275,17 @@ class Test_dynamodb_table_cross_account_access:
dynamodb_client = mock.MagicMock
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = []
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables.append(
Table(
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=test_table_arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_public_policy_with_condition_same_account,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -309,16 +313,17 @@ class Test_dynamodb_table_cross_account_access:
dynamodb_client = mock.MagicMock
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = []
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables.append(
Table(
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=test_table_arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_public_policy_with_condition_diff_account,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -346,16 +351,17 @@ class Test_dynamodb_table_cross_account_access:
dynamodb_client = mock.MagicMock
from prowler.providers.aws.services.dynamodb.dynamodb_service import Table
dynamodb_client.tables = []
dynamodb_client.audited_account = AWS_ACCOUNT_NUMBER
dynamodb_client.tables.append(
Table(
arn = test_table_arn
dynamodb_client.tables = {
arn: Table(
arn=test_table_arn,
name=test_table_name,
region=AWS_REGION_EU_WEST_1,
policy=test_public_policy_with_invalid_condition_block,
)
)
}
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
new=dynamodb_client,
@@ -0,0 +1,281 @@
from unittest import mock
from unittest.mock import patch
import botocore
from boto3 import client
from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
)
make_api_call = botocore.client.BaseClient._make_api_call
def mock_make_api_call(self, operation_name, kwarg):
if operation_name == "CreateBackupSelection":
return {
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1",
],
}
elif operation_name == "ListProtectedResources":
return {
"Results": [
{
"ResourceArn": f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/test1",
"ResourceType": "DynamoDB",
"LastBackupTime": "2023-08-23T00:00:00Z",
}
]
}
elif operation_name == "ListBackupPlans":
return {
"BackupPlans": [
{
"BackupPlanId": "test-backup-plan-id",
"BackupPlanName": "test-backup-plan",
}
]
}
return make_api_call(self, operation_name, kwarg)
class Test_dynamodb_table_protected_by_backup_plan:
@mock_aws
def test_dynamodb_no_tables(self):
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=DynamoDB(aws_provider),
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 0
@mock_aws
def test_dynamodb_table_no_existing_backup_plans(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"},
],
DeletionProtectionEnabled=True,
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=DynamoDB(aws_provider),
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "DynamoDB table test1 is not protected by a backup plan."
)
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_without_backup_plan(self):
backup = client("backup", region_name=AWS_REGION_US_EAST_1)
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"},
],
DeletionProtectionEnabled=True,
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]
backup.create_backup_plan(
BackupPlan={
"BackupPlanName": "test-backup-plan",
"Rules": [
{
"RuleName": "DailyBackup",
"TargetBackupVaultName": "test-vault",
"ScheduleExpression": "cron(0 12 * * ? *)",
"Lifecycle": {"DeleteAfterDays": 30},
"RecoveryPointTags": {
"Type": "Daily",
},
},
],
}
)
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.dynamodb.dynamodb_service import DynamoDB
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=DynamoDB(aws_provider),
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "DynamoDB table test1 is not protected by a backup plan."
)
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_with_backup_plan(self):
with patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call):
backup = client("backup", region_name=AWS_REGION_US_EAST_1)
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"},
],
DeletionProtectionEnabled=True,
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]
backup.create_backup_plan(
BackupPlan={
"BackupPlanName": "test-backup-plan",
"Rules": [
{
"RuleName": "DailyBackup",
"TargetBackupVaultName": "test-vault",
"ScheduleExpression": "cron(0 12 * * ? *)",
"Lifecycle": {"DeleteAfterDays": 30},
"RecoveryPointTags": {
"Type": "Daily",
},
},
],
}
)
backup.create_backup_selection(
BackupPlanID={
backup.list_backup_plans()["BackupPlans"][0]["BackupPlanId"]
},
BackupPlanSelection={
"SelectionName": "test-backup-selection",
"IamRoleArn": "arn:aws:iam::123456789012:role/backup-role",
"Resources": [
f"{table['TableArn']}",
],
},
)
from prowler.providers.aws.services.backup.backup_service import Backup
from prowler.providers.aws.services.dynamodb.dynamodb_service import (
DynamoDB,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.dynamodb_client",
new=DynamoDB(aws_provider),
), mock.patch(
"prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan.backup_client",
new=Backup(aws_provider),
):
# Test Check
from prowler.providers.aws.services.dynamodb.dynamodb_table_protected_by_backup_plan.dynamodb_table_protected_by_backup_plan import (
dynamodb_table_protected_by_backup_plan,
)
check = dynamodb_table_protected_by_backup_plan()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "DynamoDB table test1 is protected by a backup plan."
)
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 == []