Update rules to support latest pycfmodel 0.16.2 (#205)

* [broken tests] update rules to support pycfmodel 0.16.0 * update pycfmodel version * change to getattr * updated CHANGELOG.md and version Co-authored-by: Oliver Crawford <oliver.crawford@skyscanner.net> Co-authored-by: Ramon <ramon.guimera@skyscanner.net>
Skyscanner · Feb 21, 2022 · 061dcce · 061dcce
1 parent 914ac98
commit 061dcce
Show file tree

Hide file tree

Showing 9 changed files with 53 additions and 41 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,22 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [1.4.0] - 2022-2-21
+### Fixes
+- Fix CI, updated tests to work with `pycfmodel` latest version which includes the use of the `Generic`.
+### Improvements
+- Bump and fixed required dependency `pycfmodel` to be at least `0.16.2`.
+- Bump several dependencies: 
+  - `boto3` to `1.21.2`
+  - `botocore` to `1.24.2`
+  - `cfn-flip` to `1.3.0`
+  - `pydantic` to `1.9.0`
+  - `python-dateutil` to `2.8.2`
+  - `pyyaml` to `6.0`
+  - `s3transfer` to `0.5.1`
+  - `typing-extensions` to `4.1.1`
+  - `urllib3` to `1.26.8`
+
 ## [1.3.3] - 2022-2-3
 ### Fixes
 - Fix CI, force `pycfmodel` to use version `0.13.0`.

diff --git a/cfripper/__version__.py b/cfripper/__version__.py
@@ -1,3 +1,3 @@
-VERSION = (1, 3, 3)
+VERSION = (1, 4, 0)
 
 __version__ = ".".join(map(str, VERSION))
diff --git a/cfripper/rules/ebs_volume_has_sse.py b/cfripper/rules/ebs_volume_has_sse.py
@@ -49,7 +49,9 @@ def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
         result = Result()
         for logical_id, resource in cfmodel.Resources.items():
             if resource.Type == "AWS::EC2::Volume":
-                if resource.Properties.get("Encrypted") != "true":
+                encrypted_status = getattr(resource.Properties, "Encrypted", None)
+
+                if encrypted_status is None or encrypted_status.lower() != "true":
                     self.add_failure_to_result(
                         result,
                         self.REASON.format(logical_id),

diff --git a/cfripper/rules/hardcoded_RDS_password.py b/cfripper/rules/hardcoded_RDS_password.py
@@ -78,10 +78,9 @@ def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
 
         # check each instance with the context of clusters.
         for logical_id, resource in instances_to_check:
-            if resource.Properties.get("DBClusterIdentifier") and any(
-                cluster_id in resource.Properties.get("DBClusterIdentifier")
-                for cluster_id in password_protected_cluster_ids
-            ):
+            db_cluster_id = getattr(resource.Properties, "DBClusterIdentifier", None)
+
+            if db_cluster_id and any(cluster_id in db_cluster_id for cluster_id in password_protected_cluster_ids):
                 continue
 
             self._failure_added(result, logical_id, resource, extras)
@@ -90,7 +89,7 @@ def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
     def _failure_added(
         self, result: Result, logical_id: str, resource: GenericResource, extras: Optional[Dict] = None
     ) -> bool:
-        master_user_password = resource.Properties.get("MasterUserPassword", Parameter.NO_ECHO_NO_DEFAULT)
+        master_user_password = getattr(resource.Properties, "MasterUserPassword", Parameter.NO_ECHO_NO_DEFAULT)
         resource_type = resource.Type.replace("AWS::RDS::DB", "")
         if master_user_password == Parameter.NO_ECHO_WITH_DEFAULT:
             self.add_failure_to_result(

diff --git a/cfripper/rules/s3_object_versioning.py b/cfripper/rules/s3_object_versioning.py
@@ -51,8 +51,8 @@ class S3ObjectVersioningRule(ResourceSpecificRule):
 
     def resource_invoke(self, resource: S3Bucket, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        version_configuration = resource.Properties.VersioningConfiguration
-        if version_configuration is None or version_configuration.get("Status") != self.ENABLED_STATUS:
+        version_configuration_status = getattr(resource.Properties.VersioningConfiguration, "Status", None)
+        if version_configuration_status != self.ENABLED_STATUS:
             self.add_failure_to_result(
                 result,
                 self.REASON.format(logical_id),

diff --git a/cfripper/rules/wildcard_resource_rule.py b/cfripper/rules/wildcard_resource_rule.py
@@ -1,6 +1,4 @@
-__all__ = [
-    "WildcardResourceRule",
-]
+__all__ = ["WildcardResourceRule"]
 import json
 import logging
 from typing import Dict, Optional
@@ -58,22 +56,21 @@ def resource_invoke(self, resource: Resource, logical_id: str, extras: Optional[
         elif isinstance(resource, KMSKey):
             self._check_policy_document(result, logical_id, resource.Properties.KeyPolicy, None, extras)
         elif isinstance(resource, GenericResource):
-            if hasattr(resource, "Properties"):
-                policy_document = resource.Properties.get("PolicyDocument")
-                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
-                        )
-                        self._check_policy_document(
-                            result, logical_id, PolicyDocument(**formatted_policy_document), None, extras
-                        )
-                    except Exception:
-                        logger.warning(
-                            f"Could not process the PolicyDocument {policy_document} on {logical_id}", stack_info=True
-                        )
+            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
+                    )
+                    self._check_policy_document(
+                        result, logical_id, PolicyDocument(**formatted_policy_document), None, extras
+                    )
+                except Exception:
+                    logger.warning(
+                        f"Could not process the PolicyDocument {policy_document} on {logical_id}", stack_info=True
+                    )
 
         return result
 

diff --git a/requirements.txt b/requirements.txt
@@ -4,18 +4,18 @@
 #
 #    make freeze
 #
-boto3==1.17.86
-botocore==1.20.86
-cfn-flip==1.2.3
+boto3==1.21.2
+botocore==1.24.2
+cfn-flip==1.3.0
 click==7.1.2
 jmespath==0.10.0
 pluggy==0.13.1
-pycfmodel==0.13.0
-pydantic==1.8.2
+pycfmodel==0.16.2
+pydantic==1.9.0
 pydash==4.7.6
-python-dateutil==2.8.1
-pyyaml==5.4.1
-s3transfer==0.4.2
+python-dateutil==2.8.2
+pyyaml==6.0
+s3transfer==0.5.1
 six==1.16.0
-typing-extensions==3.10.0.0
-urllib3==1.26.5
+typing-extensions==4.1.1
+urllib3==1.26.8
diff --git a/setup.py b/setup.py
@@ -11,7 +11,7 @@
     "cfn_flip>=1.2.0",
     "click~=7.1.1",
     "pluggy~=0.13.1",
-    "pycfmodel==0.13.0",
+    "pycfmodel>=0.16.2",
     "pydash~=4.7.6",
     "PyYAML>=4.2b1",
 ]

diff --git a/tests/test_templates/rules/PrivilegeEscalationRule/privilege_escalation_role.yaml b/tests/test_templates/rules/PrivilegeEscalationRule/privilege_escalation_role.yaml
@@ -22,5 +22,3 @@ Resources:
             Action: sts:AssumeRoleWithSAML
             Principal:
               Federated: !Sub "arn:aws:iam::${AWS::AccountId}:saml-provider/saml-provider"
-              Action:
-                - sts:AssumeRole