Repository: salesforce/policy_sentry
Pull request #216 (2020-08-29 21:08:13)

Fix for #215 (DynamoDB ARN matching)


Patch:
diff --git a/policy_sentry/util/arns.py b/policy_sentry/util/arns.py
index 5345bccc..cf2321dc 100644
--- a/policy_sentry/util/arns.py
+++ b/policy_sentry/util/arns.py
@@ -1,12 +1,141 @@
 """
 Functions to use for parsing ARNs, matching ARN types, and getting the right fragment/component from an ARN string,
 """
+#     Case 1: arn:partition:service:region:account-id:resource
+#     Case 2: arn:partition:service:region:account-id:resourcetype/resource
+#     Case 3: arn:partition:service:region:account-id:resourcetype/resource/qualifier
+#     Case 4: arn:partition:service:region:account-id:resourcetype/resource:qualifier
+#     Case 5: arn:partition:service:region:account-id:resourcetype:resource
+#     Case 6: arn:partition:service:region:account-id:resourcetype:resource:qualifier
+#     Source: https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-arns
 import logging
 import re
 
 logger = logging.getLogger(__name__)
 
 
+# pylint: disable=too-many-instance-attributes
+class ARN:
+    """Class that helps to match ARN resource type formats neatly"""
+
+    def __init__(self, provided_arn):
+        self.arn = provided_arn
+        follows_arn_format = re.search(
+            r"^arn:([^:]*):([^:]*):([^:]*):([^:]*):(.+)$", provided_arn
+        )
+
+        if not follows_arn_format:
+            raise Exception("The provided value does not follow required ARN formatting.")
+        elements = self.arn.split(":", 5)
+        self.partition = elements[1]
+        self.service_prefix = elements[2]
+        self.region = elements[3]
+        self.account = elements[4]
+        self.resource = elements[5]
+
+        if "/" in self.resource:
+            self.resource, self.resource_path = self.resource.split("/", 1)
+        elif ":" in self.resource:
+            self.resource, self.resource_path = self.resource.split(":", 1)
+        self.resource_string = self._resource_string()
+
+    # pylint: disable=too-many-return-statements
+    def _resource_string(self):
+        """
+        Given an ARN, return the string after the account ID, no matter the ARN format.
+        Return:
+            String: The resource string, like `resourcetype/resource`
+        """
+        # Resource string will equal:
+        #     Case 1: resource
+        #     Case 2: resourcetype/resource
+        #     Case 3: resourcetype/resource/qualifier
+        #     Case 4: resourcetype/resource:qualifier
+        #     Case 5: resourcetype:resource
+        #     Case 6: resourcetype:resource:qualifier
+        split_arn = self.arn.split(":")
+        resource_string = ":".join(split_arn[5:])
+        return resource_string
+
+    def same_resource_type(self, arn_in_database):
+        """Given an arn, see if it has the same resource type"""
+
+        # 1. If the RAW ARN in the database is *, then it doesn't have a resource type
+        if arn_in_database == "*":
+            return False
+
+        # 2. ARNs should have the same service
+        elements = arn_in_database.split(":", 5)
+        if self.service_prefix != elements[2]:
+            return False
+
+        # 3. Add support for resource type wildcards, per #204
+        # When specifying an ARN and setting a * on the resource type (e.g. instead of db:instance it is *:*),
+        #   multiple ARNs can match.
+        #   Previously, this would fail and return empty results.
+        #   Now it correctly returns the full list of matching ARNs and corresponding actions.
+        resource_type_arn_to_test = parse_arn_for_resource_type(self.arn)
+        if resource_type_arn_to_test == '*':
+            return True
+
+        # 4. Match patterns for complicated resource strings, leveraging the standardized format of the Raw ARN format
+        # table/${TableName} should not match `table/${TableName}/backup/${BackupName}`
+        resource_string_arn_in_database = get_resource_string(arn_in_database)
+
+        split_resource_string_in_database = re.split(':|/', resource_string_arn_in_database)
+        # logger.debug(str(split_resource_string_in_database))
+        arn_format_list = []
+        for elem in split_resource_string_in_database:
+            if "${" not in elem:
+                arn_format_list.append(elem)
+            else:
+                # If an element says something like ${TableName}, normalize it to an empty string
+                arn_format_list.append("")
+
+        split_resource_string_to_test = re.split(':|/', self.resource_string)
+        # 4b: If we have a confusing resource string, the length of the split resource string list
+        #  should at least be the same
+        # Again, table/${TableName} (len of 2) should not match `table/${TableName}/backup/${BackupName}` (len of 4)
+        if len(split_resource_string_to_test) != len(arn_format_list):
+            return False
+
+        # 4c: See if the non-normalized fields match
+        for i in range(len(arn_format_list)):
+            # If the field is not normalized to empty string, then make sure the resource type segments match
+            # So, using table/${TableName}/backup/${BackupName} as an example:
+            # table should match, backup should match,
+            # and length of the arn_format_list should be the same as split_resource_string_to_test
+            # If all conditions match, then the ARN format is the same.
+            if arn_format_list[i] != "":
+                if arn_format_list[i] == split_resource_string_to_test[i]:
+                    pass
+                elif split_resource_string_to_test[i] == "*":
+                    pass
+                else:
+                    return False
+
+        # 4. Special type for S3 bucket objects and CodeCommit repos
+        # Note: Each service can only have one of these, so these are definitely exceptions
+        exclusion_list = ["${ObjectName}", "${RepositoryName}", "${BucketName}"]
+        resource_path_arn_in_database = elements[5]
+        if resource_path_arn_in_database in exclusion_list:
+            logger.debug("Special type: %s", resource_path_arn_in_database)
+            # If we've made it this far, then it is a special type
+            # return True
+            # Presence of / would mean it's an object in both so it matches
+            if "/" in self.resource_string and "/" in elements[5]:
+                return True
+            # / not being present in either means it's a bucket in both so it matches
+            elif "/" not in self.resource_string and "/" not in elements[5]:
+                return True
+            # If there is a / in one but not in the other, it does not match
+            else:
+                return False
+
+        # 5. If we've made it this far, then it should pass
+        return True
+
+
 def parse_arn(arn):
     """
     Given an ARN, split up the ARN into the ARN namespacing schema dictated by the AWS docs.
@@ -28,36 +157,12 @@ def parse_arn(arn):
     return result
 
 
-# def get_string_arn(arn):
-#     result = "{0}".format(str(arn))
-#     after = result.strip('((,\'')
-#     return after
-
-
-def get_partition_from_arn(arn):
-    """Given an ARN string, return the partition string. This is usually `aws` unless you are in C2S or
-    AWS GovCloud."""
-    result = parse_arn(arn)
-    return result["partition"]
-
-
 def get_service_from_arn(arn):
     """Given an ARN string, return the service """
     result = parse_arn(arn)
     return result["service"]
 
 
-def get_region_from_arn(arn):
-    """Given an ARN, return the region in the ARN, if it is available. In certain cases like S3 it is not"""
-    result = parse_arn(arn)
-    # Support S3 buckets with no values under region
-    if result["region"] is None:
-        result = ""
-    else:
-        result = result["region"]
-    return result
-
-
 def get_account_from_arn(arn):
     """Given an ARN, return the account ID in the ARN, if it is available. In certain cases like S3 it is not"""
     result = parse_arn(arn)
@@ -69,13 +174,6 @@ def get_account_from_arn(arn):
     return result
 
 
-def get_resource_from_arn(arn):
-    """Given an ARN, parse it according to ARN namespacing and return the resource. See
-    http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html for more details on ARN namespacing."""
-    result = parse_arn(arn)
-    return result["resource"]
-
-
 def get_resource_path_from_arn(arn):
     """Given an ARN, parse it according to ARN namespacing and return the resource path. See
     http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html for more details on ARN namespacing."""
@@ -83,28 +181,6 @@ def get_resource_path_from_arn(arn):
     return result["resource_path"]
 
 
-# pylint: disable=simplifiable-if-statement
-def arn_has_slash(arn):
-    """Given an ARN, determine if the ARN has a stash in it. Just useful for the hacky methods for
-    parsing ARN namespaces. See
-    http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html for more details on ARN namespacing."""
-    if arn.count("/") > 0:
-        return True
-    else:
-        return False
-
-
-# pylint: disable=simplifiable-if-statement
-def arn_has_colons(arn):
-    """Given an ARN, determine if the ARN has colons in it. Just useful for the hacky methods for
-    parsing ARN namespaces. See
-    http://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html for more details on ARN namespacing."""
-    if arn.count(":") > 0:
-        return True
-    else:
-        return False
-
-
 def get_resource_string(arn):
     """
     Given an ARN, return the string after the account ID, no matter the ARN format.
@@ -119,7 +195,6 @@ def get_resource_string(arn):
     return resource_string
 
 
-# TODO: query all arns in the service then find the ARN that matches the pattern here. This will match for buckets, for example.
 # In the meantime, we have to skip this pylint check (consider this as tech debt)
 # pylint: disable=inconsistent-return-statements
 def parse_arn_for_resource_type(arn):
@@ -132,13 +207,6 @@ def parse_arn_for_resource_type(arn):
         String: The resource type, like `bucket` or `object`
     """
     split_arn = arn.split(":")
-    # Resource string will equal:
-    #     Case 1: resource
-    #     Case 2: resourcetype/resource
-    #     Case 3: resourcetype/resource/qualifier
-    #     Case 4: resourcetype/resource:qualifier
-    #     Case 5: resourcetype:resource
-    #     Case 6: resourcetype:resource:qualifier
     resource_string = ":".join(split_arn[5:])
     split_resource = re.split("/|:", resource_string)
     if len(split_resource) == 1:
@@ -148,13 +216,6 @@ def parse_arn_for_resource_type(arn):
         return split_resource[0]
 
 
-#     Case 1: arn:partition:service:region:account-id:resource
-#     Case 2: arn:partition:service:region:account-id:resourcetype/resource
-#     Case 3: arn:partition:service:region:account-id:resourcetype/resource/qualifier
-#     Case 4: arn:partition:service:region:account-id:resourcetype/resource:qualifier
-#     Case 5: arn:partition:service:region:account-id:resourcetype:resource
-#     Case 6: arn:partition:service:region:account-id:resourcetype:resource:qualifier
-#     Source: https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-arns
 def does_arn_match(arn_to_test, arn_in_database):
     """
     Given two ARNs, determine if they have the same resource type.
@@ -166,35 +227,5 @@ def does_arn_match(arn_to_test, arn_in_database):
     Returns:
         Boolean: result of whether or not the ARNs match
     """
-    score = 0
-    # arn_in_database = 'arn:aws:ssm:${Region}:${Account}:parameter/${FullyQualifiedParameterName}'
-    # arn_to_test = 'arn:aws:ssm:us-east-1:123456789012:parameter/test'
-    exclusion_list = ["${ObjectName}"]
-    if arn_in_database == "*":
-        return False
-    else:
-        resource_string_arn_in_database = get_resource_string(arn_in_database)
-        resource_string_arn_to_test = get_resource_string(arn_to_test)
-        resource_type_arn_in_database = parse_arn_for_resource_type(arn_in_database)
-        resource_type_arn_to_test = parse_arn_for_resource_type(arn_to_test)
-        if get_service_from_arn(arn_in_database) != get_service_from_arn(arn_to_test):
-            score += 10
-            return False
-        if resource_type_arn_to_test in (resource_type_arn_in_database, '*'):
-            return True
-        else:
-            score += 1
-        if (
-            resource_string_arn_in_database.count("/") > 0
-            and resource_string_arn_to_test.count("/") > 0
-        ):
-            # Example: SSM `parameter/`
-            if get_resource_from_arn(arn_in_database) != get_resource_from_arn(
-                arn_to_test
-            ):
-                # Some exclusions, like ObjectId for S3 buckets
-                if get_resource_path_from_arn(arn_in_database) in exclusion_list:
-                    return True
-                else:
-                    return False
-    return score < 1
+    this_arn = ARN(arn_to_test)
+    return this_arn.same_resource_type(arn_in_database)


Test patch:
diff --git a/test/command/test_write_policy.py b/test/command/test_write_policy.py
index 8b957af5..6201345f 100644
--- a/test/command/test_write_policy.py
+++ b/test/command/test_write_policy.py
@@ -160,6 +160,91 @@ class WildcardOnlyServiceLevelTestCase(unittest.TestCase):
         }
         self.assertDictEqual(result, expected_results)
 
+    def test_dynamodb_arn_policy_gh_215(self):
+        """test_dynamodb_arn_matching_gh_215: Test writing a policy with DynamoDB"""
+        template_file_path = os.path.abspath(
+            os.path.join(
+                os.path.dirname(__file__),
+                os.path.pardir,
+                "files",
+                "dynamodb_gh_215.yml",
+            )
+        )
+        cfg = read_yaml_file(template_file_path)
+        results = write_policy_with_template(cfg)
+        print(json.dumps(results, indent=4))
+        expected_results = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "MultMultNone",
+                    "Effect": "Allow",
+                    "Action": [
+                        "dynamodb:DescribeLimits",
+                        "dynamodb:DescribeReservedCapacity",
+                        "dynamodb:DescribeReservedCapacityOfferings",
+                        "dynamodb:ListStreams",
+                        "dynamodb:ListBackups",
+                        "dynamodb:ListContributorInsights",
+                        "dynamodb:ListGlobalTables",
+                        "dynamodb:ListTables"
+                    ],
+                    "Resource": [
+                        "*"
+                    ]
+                },
+                {
+                    "Sid": "DynamodbReadTable",
+                    "Effect": "Allow",
+                    "Action": [
+                        "dynamodb:BatchGetItem",
+                        "dynamodb:ConditionCheckItem",
+                        "dynamodb:DescribeContinuousBackups",
+                        "dynamodb:DescribeContributorInsights",
+                        "dynamodb:DescribeTable",
+                        "dynamodb:DescribeTableReplicaAutoScaling",
+                        "dynamodb:DescribeTimeToLive",
+                        "dynamodb:GetItem",
+                        "dynamodb:ListTagsOfResource",
+                        "dynamodb:Query",
+                        "dynamodb:Scan"
+                    ],
+                    "Resource": [
+                        "arn:aws:dynamodb:us-east-1:123456789123:table/mytable"
+                    ]
+                },
+                {
+                    "Sid": "DynamodbWriteTable",
+                    "Effect": "Allow",
+                    "Action": [
+                        "dynamodb:BatchWriteItem",
+                        "dynamodb:CreateBackup",
+                        "dynamodb:CreateGlobalTable",
+                        "dynamodb:CreateTable",
+                        "dynamodb:CreateTableReplica",
+                        "dynamodb:DeleteItem",
+                        "dynamodb:DeleteTable",
+                        "dynamodb:DeleteTableReplica",
+                        "dynamodb:PutItem",
+                        "dynamodb:RestoreTableFromBackup",
+                        "dynamodb:RestoreTableToPointInTime",
+                        "dynamodb:UpdateContinuousBackups",
+                        "dynamodb:UpdateContributorInsights",
+                        "dynamodb:UpdateGlobalTable",
+                        "dynamodb:UpdateGlobalTableSettings",
+                        "dynamodb:UpdateItem",
+                        "dynamodb:UpdateTable",
+                        "dynamodb:UpdateTableReplicaAutoScaling",
+                        "dynamodb:UpdateTimeToLive"
+                    ],
+                    "Resource": [
+                        "arn:aws:dynamodb:us-east-1:123456789123:table/mytable"
+                    ]
+                }
+            ]
+        }
+        self.assertDictEqual(results, expected_results)
+
 
 class RdsWritingTestCase(unittest.TestCase):
     def test_rds_policy_read_only(self):
diff --git a/test/files/dynamodb_gh_215.yml b/test/files/dynamodb_gh_215.yml
new file mode 100644
index 00000000..911bee70
--- /dev/null
+++ b/test/files/dynamodb_gh_215.yml
@@ -0,0 +1,34 @@
+mode: crud
+name: ''
+# Specify resource ARNs
+read:
+- 'arn:aws:dynamodb:us-east-1:123456789123:table/mytable'
+write:
+- 'arn:aws:dynamodb:us-east-1:123456789123:table/mytable'
+list:
+- 'arn:aws:dynamodb:us-east-1:123456789123:table/mytable'
+tagging:
+- ''
+permissions-management:
+- ''
+# Actions that do not support resource constraints
+wildcard-only:
+  single-actions: # standalone actions
+  - ''
+  # Service-wide - like 's3' or 'ec2'
+  service-read:
+  - 'dynamodb'
+  service-write:
+  - ''
+  service-list:
+  - 'dynamodb'
+  service-tagging:
+  - ''
+  service-permissions-management:
+  - ''
+# Skip resource constraint requirements by listing actions here.
+skip-resource-constraints:
+- ''
+# Exclude actions from the output by specifying them here. Accepts wildcards, like kms:Delete*
+exclude-actions:
+- ''
diff --git a/test/util/test_arns.py b/test/util/test_arns.py
index 9839bdc9..6d2988b5 100644
--- a/test/util/test_arns.py
+++ b/test/util/test_arns.py
@@ -1,5 +1,7 @@
 import unittest
-from policy_sentry.util.arns import does_arn_match
+from policy_sentry.util.arns import does_arn_match, ARN
+import os
+import yaml
 
 # "Does Arn Match" tests
 # See docs for this list: # https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html#genref-arns
@@ -18,7 +20,8 @@ class ArnsTestCase(unittest.TestCase):
         # Case 1: arn:partition:service:region:account-id:resource
         arn_to_test = "arn:aws:s3:::bucket_name"
         arn_in_database = "arn:${Partition}:s3:::${BucketName}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_1(self):
         # Case 1: arn:partition:service:region:account-id:resource
@@ -26,33 +29,37 @@ class ArnsTestCase(unittest.TestCase):
         arn_in_database = (
             "arn:${Partition}:codecommit:${Region}:${Account}:${RepositoryName}"
         )
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_2(self):
         # Case 2: arn:partition:service:region:account-id:resourcetype/resource
         arn_to_test = "arn:aws:ssm:us-east-1:123456789012:parameter/test"
         arn_in_database = "arn:${Partition}:ssm:${Region}:${Account}:parameter/${FullyQualifiedParameterName}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
-    # This one is failing
-    # def test_does_arn_match_case_3(self):
-    #     # Case 3: arn:partition:service:region:account-id:resourcetype/resource/qualifier
-    #     arn_to_test = "arn:aws:kinesis:us-east-1:account-id:firehose/myfirehose/consumer/someconsumer:${ConsumerCreationTimpstamp}"
-    #     arn_in_database = "arn:aws:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}"
-    #     # https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ConsumerDescription.html
-    #     self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+    def test_does_arn_match_case_3(self):
+        # Case 3: arn:partition:service:region:account-id:resourcetype/resource/qualifier
+        arn_to_test = "arn:aws:kinesis:us-east-1:account-id:firehose/myfirehose/consumer/someconsumer:${ConsumerCreationTimpstamp}"
+        arn_in_database = "arn:aws:kinesis:${Region}:${Account}:${StreamType}/${StreamName}/consumer/${ConsumerName}:${ConsumerCreationTimpstamp}"
+        # https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ConsumerDescription.html
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_4(self):
         # Case 4: arn:partition:service:region:account-id:resourcetype/resource:qualifier
         arn_to_test = "arn:aws:batch:region:account-id:job-definition/job-name:revision"
         arn_in_database = "arn:${Partition}:batch:${Region}:${Account}:job-definition/${JobDefinitionName}:${Revision}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_5(self):
         # Case 5: arn:partition:service:region:account-id:resourcetype:resource
         arn_to_test = "arn:aws:states:region:account-id:stateMachine:stateMachineName"
         arn_in_database = "arn:${Partition}:states:${Region}:${Account}:stateMachine:${StateMachineName}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_6(self):
         # Case 6: arn:partition:service:region:account-id:resourcetype:resource:qualifier
@@ -60,13 +67,15 @@ class ArnsTestCase(unittest.TestCase):
             "arn:aws:states:region:account-id:execution:stateMachineName:executionName"
         )
         arn_in_database = "arn:${Partition}:states:${Region}:${Account}:execution:${StateMachineName}:${ExecutionId}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_case_greengrass(self):
         # Undocumented case: AWS Greengrass: arn:aws:greengrass:${Region}:${Account}:/greengrass/definition/devices/${DeviceDefinitionId}/versions/${VersionId}
         arn_to_test = "arn:aws:greengrass:${Region}:${Account}:/greengrass/definition/devices/1234567/versions/1"
         arn_in_database = "arn:aws:greengrass:${Region}:${Account}:/greengrass/definition/devices/${DeviceDefinitionId}/versions/${VersionId}"
-        self.assertTrue(does_arn_match(arn_to_test, arn_in_database))
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_rds(self):
         arns_to_test = [
@@ -88,26 +97,52 @@ class ArnsTestCase(unittest.TestCase):
         ]
         arn_in_database = "arn:${Partition}:rds:${Region}:${Account}:db:${DbInstanceName}"
         for arn in arns_to_test:
-            decision = does_arn_match(arn, arn_in_database)
-            self.assertFalse(decision)
+            this_arn = ARN(arn)
+            self.assertFalse(this_arn.same_resource_type(arn_in_database))
 
         arn_to_test = "arn:${Partition}:rds:${Region}:${Account}:cluster:${DbClusterInstanceName}"
-        decision = does_arn_match(arn_to_test, arn_in_database)
-        self.assertFalse(decision)
+        this_arn = ARN(arn_to_test)
+        self.assertFalse(this_arn.same_resource_type(arn_in_database))
 
         arn_to_test = "arn:${Partition}:rds:${Region}:${Account}:db:${DbInstanceName}"
         arn_in_database = "arn:${Partition}:rds:${Region}:${Account}:db:${DbInstanceName}"
-        decision = does_arn_match(arn_to_test, arn_in_database)
-        self.assertTrue(decision)
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
     def test_does_arn_match_resource_wildcard(self):
         arn_to_test = "arn:${Partition}:rds:${Region}:${Account}:*:*"
         arn_in_database = "arn:${Partition}:rds:${Region}:${Account}:db:${DbInstanceName}"
-        decision = does_arn_match(arn_to_test, arn_in_database)
-        self.assertTrue(decision)
+        this_arn = ARN(arn_to_test)
+        self.assertTrue(this_arn.same_resource_type(arn_in_database))
 
         # Make sure wrong service yields False
         arn_to_test = "arn:${Partition}:s3:${Region}:${Account}:*:*"
         arn_in_database = "arn:${Partition}:rds:${Region}:${Account}:db:${DbInstanceName}"
-        decision = does_arn_match(arn_to_test, arn_in_database)
-        self.assertFalse(decision)
+        this_arn = ARN(arn_to_test)
+        self.assertFalse(this_arn.same_resource_type(arn_in_database))
+
+    def test_dynamodb_arn_matching_gh_215(self):
+        """test_dynamodb_arn_matching_gh_215: Validate fix for DynamoDB arn mismatch in GitHub issue #215"""
+        index = "arn:${Partition}:dynamodb:${Region}:${Account}:table/${TableName}/index/${IndexName}"
+        stream = "arn:${Partition}:dynamodb:${Region}:${Account}:table/${TableName}/stream/${StreamLabel}"
+        table = "arn:${Partition}:dynamodb:${Region}:${Account}:table/${TableName}"
+        backup = "arn:${Partition}:dynamodb:${Region}:${Account}:table/${TableName}/backup/${BackupName}"
+        global_table = "arn:${Partition}:dynamodb::${Account}:global-table/${GlobalTableName}"
+
+        this_arn = ARN("arn:aws:dynamodb:us-east-1:123456789123:table/mytable")
+        self.assertTrue(this_arn.same_resource_type(table))
+        result = this_arn.same_resource_type(index)
+        self.assertFalse(result)
+        result = this_arn.same_resource_type(stream)
+        self.assertFalse(result)
+        result = this_arn.same_resource_type(backup)
+        self.assertFalse(result)
+        result = this_arn.same_resource_type(global_table)
+        self.assertFalse(result)
+
+        this_arn = "arn:aws:dynamodb:us-east-1:123456789123:table/mytable"
+        self.assertTrue(does_arn_match(this_arn, table))
+        self.assertFalse(does_arn_match(this_arn, index))
+        self.assertFalse(does_arn_match(this_arn, stream))
+        self.assertFalse(does_arn_match(this_arn, backup))
+        self.assertFalse(does_arn_match(this_arn, global_table))