feat(aws): new dynamodb_table_cross_account_access check (#3932)

Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
2026-07-23 20:42:02 +00:00 · 2024-05-07 13:36:32 +02:00
parent d9cf113882
commit 6f0dc44975
7 changed files with 505 additions and 2 deletions
@@ -7,7 +7,7 @@
  ],
  "ServiceName": "dynamodb",
  "SubServiceName": "",
-  "ResourceIdTemplate": "arn:partition:dynamodb:region:account-id:certificate/resource-id",
+  "ResourceIdTemplate": "arn:partition:dax:region:account-id:cache/resource-id",
  "Severity": "medium",
  "ResourceType": "AwsDaxCluster",
  "Description": "Check if DynamoDB DAX Clusters are encrypted at rest.",
@@ -1,3 +1,4 @@
+import json
 from typing import Optional

 from botocore.client import ClientError
@@ -17,6 +18,7 @@ class DynamoDB(AWSService):
        self.__threading_call__(self.__list_tables__)
        self.__describe_table__()
        self.__describe_continuous_backups__()
+        self.__get_resource_policy__()
        self.__list_tags_for_resource__()

    def __list_tables__(self, regional_client):
@@ -93,6 +95,35 @@ class DynamoDB(AWSService):
                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
            )

+    def __get_resource_policy__(self):
+        logger.info("DynamoDB - Get Resource Policy...")
+        try:
+            for table in self.tables:
+                try:
+                    regional_client = self.regional_clients[table.region]
+                    response = regional_client.get_resource_policy(
+                        ResourceArn=table.arn
+                    )
+                    table.policy = json.loads(response["Policy"])
+                except ClientError as error:
+                    if error.response["Error"]["Code"] == "ResourceNotFoundException":
+                        logger.warning(
+                            f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+                    elif error.response["Error"]["Code"] == "PolicyNotFoundException":
+                        logger.warning(
+                            f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+                    else:
+                        logger.error(
+                            f"{regional_client.region} -- {error.__class__.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                        )
+                    continue
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
    def __list_tags_for_resource__(self):
        logger.info("DynamoDB - List Tags...")
        try:
@@ -186,6 +217,7 @@ class Table(BaseModel):
    encryption_type: Optional[str]
    kms_arn: Optional[str]
    pitr: bool = False
+    policy: Optional[dict] = None
    region: str
    tags: Optional[list] = []

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "dynamodb_table_cross_account_access",
+  "CheckTitle": "DynamoDB tables should not be accessible from other AWS accounts",
+  "CheckType": [
+    "Infrastructure Security"
+  ],
+  "ServiceName": "dynamodb",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:dynamodb:region:account-id:table/resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsDynamoDBTable",
+  "Description": "This check determines if the DynamoDB table is accessible from other AWS accounts.",
+  "Risk": "If the DynamoDB table is accessible from other AWS accounts, it may lead to unauthorized access to the data stored in the table.",
+  "RelatedUrl": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/access-control-resource-based.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws dynamodb delete-resource-policy --resource-arn <resource-arn>",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that the DynamoDB table is not accessible from other AWS accounts.",
+      "Url": "https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/rbac-bpa-rbp.html"
+    }
+  },
+  "Categories": [
+    "trustboundaries"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,56 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.lib.policy_condition_parser.policy_condition_parser import (
+    is_condition_block_restrictive,
+)
+from prowler.providers.aws.services.dynamodb.dynamodb_client import dynamodb_client
+
+
+class dynamodb_table_cross_account_access(Check):
+    def execute(self):
+        findings = []
+        for table in dynamodb_client.tables:
+            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 = "PASS"
+            report.status_extended = (
+                f"DynamoDB table {table.name} does not have a resource-based policy."
+            )
+            if table.policy:
+                report.status_extended = f"DynamoDB table {table.name} has a resource-based policy but is not cross account."
+                cross_account_access = False
+                policy_statements = table.policy["Statement"]
+                if isinstance(
+                    policy_statements, dict
+                ):  # Normalize single statement to list
+                    policy_statements = [policy_statements]
+                for statement in policy_statements:
+                    if not cross_account_access:
+                        if statement["Effect"] == "Allow":
+                            if "AWS" in statement["Principal"]:
+                                principals = statement["Principal"]["AWS"]
+                                if not isinstance(principals, list):
+                                    principals = [principals]
+                            else:
+                                principals = [statement["Principal"]]
+                            for aws_account in principals:
+                                if (
+                                    dynamodb_client.audited_account not in aws_account
+                                    or "*" == aws_account
+                                ):
+                                    cross_account_access = True
+                                    # Check if the condition block is restrictive
+                                    conditions = statement.get("Condition", {})
+                                    if is_condition_block_restrictive(
+                                        conditions, dynamodb_client.audited_account
+                                    ):
+                                        cross_account_access = False
+                    else:
+                        break
+                if cross_account_access:
+                    report.status = "FAIL"
+                    report.status_extended = f"DynamoDB table {table.name} has a resource-based policy allowing cross account access."
+            findings.append(report)
+        return findings
@@ -7,7 +7,7 @@
  ],
  "ServiceName": "dynamodb",
  "SubServiceName": "",
-  "ResourceIdTemplate": "arn:partition:dynamodb:region:account-id:certificate/resource-id",
+  "ResourceIdTemplate": "arn:partition:dynamodb:region:account-id:table/resource-id",
  "Severity": "medium",
  "ResourceType": "AwsDynamoDBTable",
  "Description": "Check if DynamoDB tables point-in-time recovery (PITR) is enabled.",
@@ -0,0 +1,381 @@
+from unittest import mock
+from uuid import uuid4
+
+from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
+
+test_table_name = str(uuid4())
+test_table_arn = f"arn:aws:dynamodb:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:table/{test_table_name}"
+
+test_restricted_policy = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": {"AWS": {AWS_ACCOUNT_NUMBER}},
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+        }
+    ],
+}
+
+test_public_policy = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+        }
+    ],
+}
+
+test_public_policy_with_condition_same_account_not_valid = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+            "Condition": {
+                "DateGreaterThan": {"aws:CurrentTime": "2009-01-31T12:00Z"},
+                "DateLessThan": {"aws:CurrentTime": "2009-01-31T15:00Z"},
+            },
+        }
+    ],
+}
+
+test_public_policy_with_condition_same_account = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+            "Condition": {
+                "StringEquals": {"aws:SourceAccount": f"{AWS_ACCOUNT_NUMBER}"}
+            },
+        }
+    ],
+}
+
+test_public_policy_with_condition_diff_account = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+            "Condition": {"StringEquals": {"aws:SourceAccount": "111122223333"}},
+        }
+    ],
+}
+
+test_public_policy_with_invalid_condition_block = {
+    "Version": "2012-10-17",
+    "Id": "Table1_Policy_UUID",
+    "Statement": [
+        {
+            "Sid": "Table1_AnonymousAccess_GetItem",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": "dynamodb:BatchGetItem",
+            "Resource": test_table_arn,
+            "Condition": {"DateGreaterThan": {"aws:CurrentTime": "2009-01-31T12:00Z"}},
+        }
+    ],
+}
+
+
+class Test_dynamodb_table_cross_account_access:
+    def test_no_tables(self):
+        dynamodb_client = mock.MagicMock
+        dynamodb_client.tables = []
+        with mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
+            new=dynamodb_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_tables_no_policy(self):
+        dynamodb_client = mock.MagicMock
+        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,
+                name=test_table_name,
+                region=AWS_REGION_EU_WEST_1,
+            )
+        )
+        with mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_service.DynamoDB",
+            new=dynamodb_client,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                f"DynamoDB table {test_table_name} does not have a resource-based policy."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_not_public(self):
+        dynamodb_client = mock.MagicMock
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                f"DynamoDB table {test_table_name} has a resource-based policy but is not cross account."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_public(self):
+        dynamodb_client = mock.MagicMock
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"DynamoDB table {test_table_name} has a resource-based policy allowing cross account access."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_public_with_condition_not_valid(self):
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"DynamoDB table {test_table_name} has a resource-based policy allowing cross account access."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_public_with_condition_valid(self):
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert result[0].status_extended == (
+                f"DynamoDB table {test_table_name} has a resource-based policy but is not cross account."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_public_with_condition_valid_with_other_account(self):
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert result[0].status_extended == (
+                f"DynamoDB table {test_table_name} has a resource-based policy allowing cross account access."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1
+
+    def test_tables_public_with_condition_with_invalid_block(self):
+        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,
+                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,
+        ), mock.patch(
+            "prowler.providers.aws.services.dynamodb.dynamodb_client.dynamodb_client",
+            new=dynamodb_client,
+        ):
+            from prowler.providers.aws.services.dynamodb.dynamodb_table_cross_account_access.dynamodb_table_cross_account_access import (
+                dynamodb_table_cross_account_access,
+            )
+
+            check = dynamodb_table_cross_account_access()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"DynamoDB table {test_table_name} has a resource-based policy allowing cross account access."
+            )
+            assert result[0].resource_id == test_table_name
+            assert result[0].resource_arn == test_table_arn
+            assert result[0].resource_tags == []
+            assert result[0].region == AWS_REGION_EU_WEST_1