Update WildcardResourceRule for a better use with GenericResource (#210)

* renamed file * delete case for KMS Key, since pycfmodel 0.17.0 allows obtaining policies for that resource * updated rule documentation * renamed method to improve readability * updated version and changelog * revert method renaming * stopped using `_statement_as_list()` when retrieving statements in several rules in favor of `statement_as_list()`. Co-authored-by: Ramon <[email protected]>
Skyscanner · Mar 14, 2022 · 60c7069 · 60c7069
1 parent c06d73c
commit 60c7069
Show file tree

Hide file tree

Showing 7 changed files with 25 additions and 16 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,12 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [1.5.2]
+### Updates
+- Updates `WildcardResourceRule` for a better use with the `GenericResource`.
+### Fixes
+- Stopped using `_statement_as_list()` when retrieving statements in several rules in favor of `statement_as_list()`.
+
 ## [1.5.1]
 ### Updates
 - Created `GenericResourceWildcardPolicyRule` in order to check for WildcardPolicy issues in generic resources.

diff --git a/cfripper/__version__.py b/cfripper/__version__.py
@@ -1,3 +1,3 @@
-VERSION = (1, 5, 1)
+VERSION = (1, 5, 2)
 
 __version__ = ".".join(map(str, VERSION))
diff --git a/cfripper/rules/sns_topic_policy.py b/cfripper/rules/sns_topic_policy.py
@@ -34,7 +34,7 @@ class SNSTopicPolicyNotPrincipalRule(ResourceSpecificRule):
 
     def resource_invoke(self, resource: SNSTopicPolicy, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        for statement in resource.Properties.PolicyDocument._statement_as_list():
+        for statement in resource.Properties.PolicyDocument.statement_as_list():
             if statement.NotPrincipal:
                 self.add_failure_to_result(
                     result,

diff --git a/cfripper/rules/sqs_queue_policy.py b/cfripper/rules/sqs_queue_policy.py
@@ -38,7 +38,7 @@ class SQSQueuePolicyNotPrincipalRule(ResourceSpecificRule):
 
     def resource_invoke(self, resource: SQSQueuePolicy, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        for statement in resource.Properties.PolicyDocument._statement_as_list():
+        for statement in resource.Properties.PolicyDocument.statement_as_list():
             if statement.NotPrincipal:
                 self.add_failure_to_result(
                     result,
@@ -80,7 +80,7 @@ class SQSQueuePolicyPublicRule(ResourceSpecificRule):
     def resource_invoke(self, resource: SQSQueuePolicy, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
         if resource.Properties.PolicyDocument.allowed_principals_with(REGEX_HAS_STAR_OR_STAR_AFTER_COLON):
-            for statement in resource.Properties.PolicyDocument._statement_as_list():
+            for statement in resource.Properties.PolicyDocument.statement_as_list():
                 if statement.Effect == "Allow" and statement.principals_with(REGEX_HAS_STAR_OR_STAR_AFTER_COLON):
                     if statement.Condition and statement.Condition.dict():
                         # Ignoring condition checks since they will get reviewed in other rules and future improvements

diff --git a/cfripper/rules/wildcard_resource_rule.py b/cfripper/rules/wildcard_resource_rule.py
@@ -5,7 +5,6 @@
 
 from pycfmodel.model.resources.generic_resource import GenericResource
 from pycfmodel.model.resources.iam_role import IAMRole
-from pycfmodel.model.resources.kms_key import KMSKey
 from pycfmodel.model.resources.properties.policy_document import PolicyDocument
 from pycfmodel.model.resources.properties.statement import Statement
 from pycfmodel.model.resources.resource import Resource
@@ -32,9 +31,9 @@ class WildcardResourceRule(ResourceSpecificRule):
     Filters context:
         | Parameter    | Type             | Description                                                     |
         |:------------:|:----------------:|:---------------------------------------------------------------:|
-        |`config`      | str              | `config` variable available inside the rule                     |
-        |`extras`      | str              | `extras` variable available inside the rule                     |
-        |`logical_id`  | str              | ID used in Cloudformation to refer the resource being analysed  |
+        |`config`      | `str`            | `config` variable available inside the rule                     |
+        |`extras`      | `str`            | `extras` variable available inside the rule                     |
+        |`logical_id`  | `str`            | ID used in CloudFormation to refer the resource being analysed  |
         |`policy_name` | `Optional[str]`  | If available, the policy name                                   |
         |`statement`   | `Statement`      | Statement being checked found in the Resource                   |
         |`action`      | `Optional[str]`  | Action that has a wildcard resource. If None, means all actions |
@@ -47,20 +46,24 @@ class WildcardResourceRule(ResourceSpecificRule):
     REASON_ALL_ACTIONS_WITHOUT_POLICY_NAME = '"{}" is using a wildcard resource allowing all actions'
 
     def resource_invoke(self, resource: Resource, logical_id: str, extras: Optional[Dict] = None) -> Result:
+        """
+        Checks each policy of a given resource.
+        If it's an IAMRole, it will check its AssumeRolePolicyDocument as well.
+        There are some cases where GenericResource contains a property called PolicyDocument that can be a str and
+        therefore, it's not being retrieved in the initial for loop.
+        For those cases, we run another check transforming the str to a PolicyDocument.
+        """
         result = Result()
 
         for policy in resource.policy_documents:
             self._check_policy_document(result, logical_id, policy.policy_document, policy.name, extras)
+
         if isinstance(resource, IAMRole):
             self._check_policy_document(result, logical_id, resource.Properties.AssumeRolePolicyDocument, None, extras)
-        elif isinstance(resource, KMSKey):
-            self._check_policy_document(result, logical_id, resource.Properties.KeyPolicy, None, extras)
         elif isinstance(resource, GenericResource):
             policy_document = getattr(resource.Properties, "PolicyDocument", None)
             if policy_document:
                 try:
-                    # PolicyDocument requires a dict. If we receive a string, attempt a conversion to dict.
-                    # If this conversion fails, show the appropriate warning and continue.
                     formatted_policy_document = (
                         json.loads(policy_document) if isinstance(policy_document, str) else policy_document
                     )
@@ -99,8 +102,8 @@ def _check_statement(
                 if action in CLOUDFORMATION_ACTIONS_ONLY_ACCEPTS_WILDCARD:
                     logger.info(f"Action {action} only accepts wildcard, ignoring...")
                 elif action.lower().startswith("kms:"):
-                    # When KMS Key policies use * in the resource, that * will only apply this policy to the KMS Key being created
-                    # so we must not flag this
+                    # When KMS Key policies use * in the resource, that * will only apply this policy to the KMS Key
+                    # being created so, we must not flag this
                     # Source: https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html
                     logger.info(f"KMS Action {action} only accepts wildcard, ignoring...")
                 elif statement.Condition:

diff --git a/tests/rules/test_WildcardResourceRule.py b/tests/rules/test_WildcardResourceRule.py
@@ -89,7 +89,7 @@ def test_user_with_inline_policy_with_wildcard_resource_is_detected(user_with_wi
 
 def test_kms_key_with_wildcard_resource_not_allowed_is_not_flagged(kms_key_with_wildcard_policy):
     # When KMS Key policies use * in the resource, that * will only apply this policy to the KMS Key being created
-    # so we must not flag this
+    # so, we must not flag this
     # Source: https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html
     rule = WildcardResourceRule(None)
     rule._config.stack_name = "stack3"
@@ -734,5 +734,5 @@ def test_policy_document_with_wildcard_resource_without_policy_name_is_detected(
 def test_policy_document_with_wildcard_resource_and_wildcard_action_without_policy_name_is_detected():
     with pytest.raises(pydantic.ValidationError):
         get_cfmodel_from(
-            "rules/WildcardResourceRule/iam_policy_with_wildcard_resource_and_wilcard_action_without_policy_name.json"
+            "rules/WildcardResourceRule/iam_policy_with_wildcard_resource_and_wildcard_action_without_policy_name.json"
         )
diff --git a/...d_wilcard_action_without_policy_name.json → ..._wildcard_action_without_policy_name.json b/...d_wilcard_action_without_policy_name.json → ..._wildcard_action_without_policy_name.json