Repository: salesforce/policy_sentry
Pull request #259 (2020-10-17 17:11:38)

Ensured minimization won't break if resource path contains special characters and improve the unit tests for minimization
## What does this PR do?

#252 introduced a bug where the Sid would be invalid if the resource path had any special characters in it. This PR ensures that does not happen.

## What gif best describes this PR or how it makes you feel?

![image](https://user-images.githubusercontent.com/3422255/96348931-3c615b80-107a-11eb-80df-ca2100d0f0ea.png)

## Completion checklist

- [x] CHANGELOG.md has been updated to reflect the changes
- [x] Additions and changes have unit tests
- [x] [Unit tests, Pylint, security testing, and Integration tests are passing.](https://github.com/salesforce/policy_sentry/actions). GitHub actions does this automatically
- [x] The pull request has been appropriately labeled using the provided PR labels



Patch:
diff --git a/policy_sentry/util/text.py b/policy_sentry/util/text.py
index 727a6751..9b726d4e 100644
--- a/policy_sentry/util/text.py
+++ b/policy_sentry/util/text.py
@@ -10,3 +10,11 @@ def capitalize_first_character(some_string):
     :return:
     """
     return " ".join("".join([w[0].upper(), w[1:].lower()]) for w in some_string.split())
+
+
+def strip_special_characters(some_string):
+    """Remove all special characters, punctuation, and spaces from a string"""
+    # Input: "Special $#! characters   spaces 888323"
+    # Output: 'Specialcharactersspaces888323'
+    result = ''.join(e for e in some_string if e.isalnum())
+    return result
diff --git a/policy_sentry/writing/sid_group.py b/policy_sentry/writing/sid_group.py
index 24ef3a36..a39dbfda 100644
--- a/policy_sentry/writing/sid_group.py
+++ b/policy_sentry/writing/sid_group.py
@@ -14,7 +14,7 @@ from policy_sentry.querying.actions import (
 )
 from policy_sentry.querying.arns import get_resource_type_name_with_raw_arn
 from policy_sentry.util.arns import does_arn_match, get_service_from_arn, parse_arn
-from policy_sentry.util.text import capitalize_first_character
+from policy_sentry.util.text import capitalize_first_character, strip_special_characters
 from policy_sentry.writing.minimize import minimize_statement_actions
 from policy_sentry.writing.validate import check_actions_schema, check_crud_schema
 from policy_sentry.shared.constants import POLICY_LANGUAGE_VERSION
@@ -206,7 +206,11 @@ class SidGroup:
             for stmt in statements:
                 if stmt['Sid'] in sids_to_be_changed:
                     arn_details = parse_arn(stmt['Resource'][0])
-                    stmt['Sid'] = create_policy_sid_namespace(arn_details['service'], arn_details['resource'], arn_details['resource_path'])
+                    resource_path = arn_details.get("resource_path")
+                    resource_sid_segment = strip_special_characters(
+                        f"{arn_details['resource']}{resource_path}"
+                    )
+                    stmt['Sid'] = create_policy_sid_namespace(arn_details['service'], "Mult", resource_sid_segment)
 
         policy = {"Version": POLICY_LANGUAGE_VERSION, "Statement": statements}
         return policy


Test patch:
diff --git a/test/util/test_arns.py b/test/util/test_arns.py
index 6d2988b5..123caba3 100644
--- a/test/util/test_arns.py
+++ b/test/util/test_arns.py
@@ -1,7 +1,5 @@
 import unittest
 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
@@ -15,6 +13,7 @@ import yaml
 # Case 6: arn:partition:service:region:account-id:resourcetype:resource:qualifier
 # Case 7: arn:partition:service:region:account-id:/greengrass/definition/devices/${DeviceDefinitionId}/versions/${VersionId}
 
+
 class ArnsTestCase(unittest.TestCase):
     def test_does_arn_match_case_bucket(self):
         # Case 1: arn:partition:service:region:account-id:resource
diff --git a/test/writing/test_minimize.py b/test/writing/test_minimize.py
index f4c1ee8b..57efe7ac 100644
--- a/test/writing/test_minimize.py
+++ b/test/writing/test_minimize.py
@@ -1,8 +1,9 @@
 import unittest
-
+import json
 from policy_sentry.writing.sid_group import SidGroup
 from policy_sentry.writing.minimize import minimize_statement_actions
 from policy_sentry.querying.all import get_all_actions
+from policy_sentry.util.policy_files import get_sid_names_from_policy, get_statement_from_policy_using_sid
 
 
 class MinimizeWildcardActionsTestCase(unittest.TestCase):
@@ -44,6 +45,46 @@ class MinimizeWildcardActionsTestCase(unittest.TestCase):
             sorted(desired_result),
         )
 
+    def test_minimize_rw_same_one(self):
+        cfg = {
+            "mode": "crud",
+            "name": "",
+            "read": [
+                "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter",
+            ],
+            "write": [
+                "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter",
+            ]
+        }
+        sid_grp = SidGroup()
+        write_format = sid_grp.process_template(cfg, minimize=0)
+
+        """
+        Expected result:
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "SsmMultParametermyparameter",
+                    "Effect": "Allow",
+                    "Action": [
+                        "ssm:getpar*",
+                        "ssm:deletepar*",
+                        "ssm:la*",
+                        "ssm:putp*"
+                    ],
+                    "Resource": [
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter"
+                    ]
+                }
+            ]
+        }
+        """
+        # To future-proof this unit test...
+        # (1) check that there is only one SID so it was combined during minimization
+        sid_names = get_sid_names_from_policy(write_format)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
     def test_minimize_rw_same(self):
         cfg = {
             "mode": "crud",
@@ -58,10 +99,38 @@ class MinimizeWildcardActionsTestCase(unittest.TestCase):
             ]
         }
         sid_grp = SidGroup()
-        write_format = sid_grp.process_template(cfg,minimize=1)
-        write_format.pop('Version')
-        self.assertEqual(len(write_format['Statement']), 1, "More than one statement returned, expected 1")
-        expected_action = ['ssm:getpar*','ssm:deletepar*','ssm:la*','ssm:putp*']
+        write_format = sid_grp.process_template(cfg, minimize=1)
+        # print(json.dumps(write_format, indent=4))
+        """
+        the output will look like:
+
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "SsmParameterMyparameter",
+                    "Effect": "Allow",
+                    "Action": [
+                        "ssm:getpar*",
+                        "ssm:deletepar*",
+                        "ssm:la*",
+                        "ssm:putp*"
+                    ],
+                    "Resource": [
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter",
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter2"
+                    ]
+                }
+            ]
+        }
+        """
+        # To future-proof this unit test...
+        # (1) check that there is only one SID so it was combined during minimization
+        sid_names = get_sid_names_from_policy(write_format)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+        # (2) Check for the presence of certain actions that we know will be there
+        expected_action = ['ssm:getpar*', 'ssm:deletepar*', 'ssm:la*', 'ssm:putp*']
         self.assertEqual(write_format['Statement'][0]['Action'], expected_action, "extra actions are returned")
         self.assertEqual(write_format['Statement'][0]['Resource'], cfg['read'], "Wrong resources were returned")
 
@@ -79,12 +148,132 @@ class MinimizeWildcardActionsTestCase(unittest.TestCase):
             ]
         }
         sid_grp = SidGroup()
-        write_format = sid_grp.process_template(cfg,minimize=1)
-        write_format.pop('Version')
-        self.assertEqual(len(write_format['Statement']), 2, "More than one statement returned, expected 1")
+        write_format = sid_grp.process_template(cfg, minimize=1)
+        """
+        Expected result:
+
+        {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "SsmReadParameter",
+                    "Effect": "Allow",
+                    "Action": [
+                        "ssm:getpar*"
+                    ],
+                    "Resource": [
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter",
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter2"
+                    ]
+                },
+                {
+                    "Sid": "SsmWriteParameter",
+                    "Effect": "Allow",
+                    "Action": [
+                        "ssm:deletepar*",
+                        "ssm:la*",
+                        "ssm:putp*"
+                    ],
+                    "Resource": [
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter",
+                        "arn:aws:ssm:us-east-1:123456789012:parameter/myparameter10"
+                    ]
+                }
+            ]
+        }
+        """
+        self.assertEqual(len(write_format['Statement']), 2, "More than two statements returned, expected 2")
         self.assertEqual(write_format['Statement'][0]['Action'], ['ssm:getpar*'], "extra actions are returned")
         self.assertEqual(write_format['Statement'][0]['Resource'], cfg['read'], "Wrong resources were returned")
         self.assertEqual(write_format['Statement'][1]['Action'], ['ssm:deletepar*', 'ssm:la*', 'ssm:putp*'],
                          "extra actions are returned")
         self.assertEqual(write_format['Statement'][1]['Resource'], cfg['write'], "Wrong resources were returned")
 
+    def test_minimize_arn_case_bucket(self):
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:s3:::bucket_name"],
+            "write": ["arn:aws:s3:::bucket_name"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+
+        # print(json.dumps(results, indent=4))
+        # To future-proof this unit test...
+        # (1) check that there is only one SID so it was combined during minimization
+        sid_names = get_sid_names_from_policy(results)
+        # For S3, it does require iam:PassRole lol because of s3:PutReplicationConfiguration requiring it as an action
+        self.assertEqual(len(sid_names), 2, "More than 2 statements returned, expected 2")
+
+    def test_minimize_arn_case_1(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:codecommit:us-east-1:123456789012:MyDemoRepo"],
+            "write": ["arn:aws:codecommit:us-east-1:123456789012:MyDemoRepo"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+    def test_minimize_arn_case_3(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:kinesis:us-east-1:account-id:firehose/myfirehose/consumer/someconsumer:${ConsumerCreationTimpstamp}"],
+            "write": ["arn:aws:kinesis:us-east-1:account-id:firehose/myfirehose/consumer/someconsumer:${ConsumerCreationTimpstamp}"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+    def test_minimize_arn_case_4(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:batch:region:account-id:job-definition/job-name:revision"],
+            "write": ["arn:aws:batch:region:account-id:job-definition/job-name:revision"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+    def test_minimize_arn_case_5(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:states:region:account-id:stateMachine:stateMachineName"],
+            "write": ["arn:aws:states:region:account-id:stateMachine:stateMachineName"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+    def test_minimize_arn_case_6(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:states:region:account-id:execution:stateMachineName:executionName"],
+            "write": ["arn:aws:states:region:account-id:execution:stateMachineName:executionName"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+
+    def test_minimize_arn_case_7(self):
+        """minimization test with ARN types from test_does_arn_match_case_1"""
+        cfg = {
+            "mode": "crud",
+            "read": ["arn:aws:greengrass:${Region}:${Account}:/greengrass/definition/devices/1234567/versions/1"],
+            "write": ["arn:aws:greengrass:${Region}:${Account}:/greengrass/definition/devices/1234567/versions/1"]
+        }
+        sid_group = SidGroup()
+        results = sid_group.process_template(cfg, minimize=0)
+        sid_names = get_sid_names_from_policy(results)
+        self.assertEqual(len(sid_names), 1, "More than one statement returned, expected 1")
+