Add S3ObjectVersioning rule to CFRipper (#193)

* Add S3ObjectVersioning rule to CFRipper

* Change version

* Use new pycfmodels from version 0.11.0 in this rule and previous S3 rules

Co-authored-by: Oliver Crawford <oliver.crawford@skyscanner.net>

CHANGELOG.md

@@ -1,6 +1,12 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [1.1.0] - 2021-09-XX
+### Improvements
+- Add `S3ObjectVersioning` rule
+- Update `pycfmodel` to `0.11.0`
+  - This includes model support for S3 Buckets. Rules against these resources have been updated (alongside tests).
+
 ## [1.0.9] - 2021-09-10
 ### Improvements
 - Update valid AWS Account IDs that might be included as principals on policies.

cfripper/__version__.py

@@ -1,3 +1,3 @@
-VERSION = (1, 0, 9)
+VERSION = (1, 1, 0)
 
 __version__ = ".".join(map(str, VERSION))

cfripper/rules/__init__.py

@@ -22,6 +22,7 @@
 from cfripper.rules.privilege_escalation import PrivilegeEscalationRule
 from cfripper.rules.s3_bucket_policy import S3BucketPolicyPrincipalRule
 from cfripper.rules.s3_lifecycle_configuration import S3LifecycleConfigurationRule
+from cfripper.rules.s3_object_versioning import S3ObjectVersioningRule
 from cfripper.rules.s3_public_access import (
     S3BucketPublicReadAclAndListStatementRule,
     S3BucketPublicReadAclRule,
@@ -68,6 +69,7 @@
         S3BucketPublicReadWriteAclRule,
         S3BucketPublicReadAclRule,
         S3CrossAccountTrustRule,
+        S3ObjectVersioningRule,
         SNSTopicDangerousPolicyActionsRule,
         SNSTopicPolicyNotPrincipalRule,
         SNSTopicPolicyWildcardActionRule,

cfripper/rules/s3_lifecycle_configuration.py

@@ -2,14 +2,14 @@
 
 from typing import Dict, Optional
 
-from pycfmodel.model.cf_model import CFModel
+from pycfmodel.model.resources.s3_bucket import S3Bucket
 
 from cfripper.model.enums import RuleGranularity, RuleRisk
 from cfripper.model.result import Result
-from cfripper.rules.base_rules import Rule
+from cfripper.rules.base_rules import ResourceSpecificRule
 
 
-class S3LifecycleConfigurationRule(Rule):
+class S3LifecycleConfigurationRule(ResourceSpecificRule):
     """
     Checks for the presence of `LifecycleConfiguration` on S3 buckets.
     These rules can help with security, compliance, and reduce AWS Costs. The rule does not
@@ -42,22 +42,21 @@ class S3LifecycleConfigurationRule(Rule):
         |`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 |
-        |`resource`     | `S3BucketPolicy`   | Resource that is being addressed                               |
-        |`bucket_name`  | str                | Name of the S3 bucket being analysed                           |
+        |`resource`     | `S3Bucket`         | Resource that is being addressed                               |
     """
 
     GRANULARITY = RuleGranularity.RESOURCE
     REASON = "S3 Bucket {} is required to contain a LifecycleConfiguration property"
+    RESOURCE_TYPES = (S3Bucket,)
     RISK_VALUE = RuleRisk.LOW
 
-    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
+    def resource_invoke(self, resource: S3Bucket, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        for logical_id, resource in cfmodel.resources_filtered_by_type(("AWS::S3::Bucket",)).items():
-            if hasattr(resource, "Properties") and resource.Properties.get("LifecycleConfiguration") is None:
-                self.add_failure_to_result(
-                    result,
-                    self.REASON.format(logical_id),
-                    resource_ids={logical_id},
-                    context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
-                )
+        if not resource.Properties.LifecycleConfiguration:
+            self.add_failure_to_result(
+                result,
+                self.REASON.format(logical_id),
+                resource_ids={logical_id},
+                context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
+            )
         return result

cfripper/rules/s3_object_versioning.py

@@ -0,0 +1,62 @@
+__all__ = ["S3ObjectVersioningRule"]
+
+from typing import Dict, Optional
+
+from pycfmodel.model.resources.s3_bucket import S3Bucket
+
+from cfripper.model.enums import RuleGranularity, RuleRisk
+from cfripper.model.result import Result
+from cfripper.rules.base_rules import ResourceSpecificRule
+
+
+class S3ObjectVersioningRule(ResourceSpecificRule):
+    """
+    Checks if the S3 bucket has object versioning enabled or not.
+
+    Risk:
+        Not having this property enabled could make the bucket more vulnerable to ransomware attacks.
+        Bucket versioning allows the automatic creation of multiple versions of an object.
+        When an object is deleted with versioning turned on, it is only marked as deleted but is still retrievable.
+
+    Fix:
+        Add `VersioningConfiguration` property with the value `Enabled` to bucket as defined in the
+        [AWS documentation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-s3-bucket-versioningconfig.html).
+
+    Code for fix:
+        ````yml
+        Resources:
+          S3Bucket:
+            Type: AWS::S3::Bucket
+            Properties:
+              ...
+              VersioningConfiguration:
+                Status: Enabled
+              ...
+        ````
+
+    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 |
+        |`resource`     | `S3Bucket`         | Resource that is being addressed                               |
+    """
+
+    ENABLED_STATUS = "Enabled"
+    GRANULARITY = RuleGranularity.RESOURCE
+    REASON = "S3 Bucket {} is required to have object versioning enabled"
+    RESOURCE_TYPES = (S3Bucket,)
+    RISK_VALUE = RuleRisk.LOW
+
+    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:
+            self.add_failure_to_result(
+                result,
+                self.REASON.format(logical_id),
+                resource_ids={logical_id},
+                context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
+            )
+        return result

cfripper/rules/s3_public_access.py

@@ -5,11 +5,12 @@
 from typing import Dict, Optional
 
 from pycfmodel.model.cf_model import CFModel
+from pycfmodel.model.resources.s3_bucket import S3Bucket
 from pycfmodel.model.resources.s3_bucket_policy import S3BucketPolicy
 
 from cfripper.model.enums import RuleGranularity, RuleRisk
 from cfripper.model.result import Result
-from cfripper.rules.base_rules import Rule
+from cfripper.rules.base_rules import ResourceSpecificRule, Rule
 
 logger = logging.getLogger(__file__)
 
@@ -50,7 +51,7 @@ def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
                     bucket_name = bucket_name[len("UNDEFINED_PARAM_") :]
 
                 bucket = cfmodel.Resources.get(bucket_name)
-                if bucket and bucket.Properties.get("AccessControl") == "PublicRead":
+                if bucket and bucket.Properties.AccessControl == "PublicRead":
                     self.add_failure_to_result(
                         result,
                         self.REASON.format(logical_id),
@@ -66,7 +67,7 @@ def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
         return result
 
 
-class S3BucketPublicReadWriteAclRule(Rule):
+class S3BucketPublicReadWriteAclRule(ResourceSpecificRule):
     """
     Checks if any S3 bucket policy has access control set to `PublicReadWrite`.
 
@@ -83,31 +84,27 @@ class S3BucketPublicReadWriteAclRule(Rule):
         |`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 |
-        |`resource`     | `Resource`         | S3 Bucket that is being addressed                              |
+        |`resource`     | `S3Bucket`         | S3 Bucket that is being addressed                              |
     """
 
     GRANULARITY = RuleGranularity.RESOURCE
     REASON = "S3 Bucket {} should not have a public read-write acl"
+    RESOURCE_TYPES = (S3Bucket,)
     RISK_VALUE = RuleRisk.HIGH
 
-    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
+    def resource_invoke(self, resource: S3Bucket, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        for logical_id, resource in cfmodel.Resources.items():
-            if (
-                resource.Type == "AWS::S3::Bucket"
-                and hasattr(resource, "Properties")
-                and resource.Properties.get("AccessControl") == "PublicReadWrite"
-            ):
-                self.add_failure_to_result(
-                    result,
-                    self.REASON.format(logical_id),
-                    resource_ids={logical_id},
-                    context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
-                )
+        if resource.Properties.AccessControl == "PublicReadWrite":
+            self.add_failure_to_result(
+                result,
+                self.REASON.format(logical_id),
+                resource_ids={logical_id},
+                context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
+            )
         return result
 
 
-class S3BucketPublicReadAclRule(Rule):
+class S3BucketPublicReadAclRule(ResourceSpecificRule):
     """
     Checks if any S3 bucket policy has access control set to `PublicRead`.
 
@@ -124,21 +121,21 @@ class S3BucketPublicReadAclRule(Rule):
         |`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 |
-        |`resource`     | `Resource`         | S3 Bucket that is being addressed                              |
+        |`resource`     | `S3Bucket`         | S3 Bucket that is being addressed                              |
     """
 
     GRANULARITY = RuleGranularity.RESOURCE
     REASON = "S3 Bucket {} should not have a public-read acl"
+    RESOURCE_TYPES = (S3Bucket,)
     RISK_VALUE = RuleRisk.HIGH
 
-    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
+    def resource_invoke(self, resource: S3Bucket, logical_id: str, extras: Optional[Dict] = None) -> Result:
         result = Result()
-        for logical_id, resource in cfmodel.resources_filtered_by_type(("AWS::S3::Bucket",)).items():
-            if hasattr(resource, "Properties") and resource.Properties.get("AccessControl") == "PublicRead":
-                self.add_failure_to_result(
-                    result,
-                    self.REASON.format(logical_id),
-                    resource_ids={logical_id},
-                    context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
-                )
+        if resource.Properties.AccessControl == "PublicRead":
+            self.add_failure_to_result(
+                result,
+                self.REASON.format(logical_id),
+                resource_ids={logical_id},
+                context={"config": self._config, "extras": extras, "logical_id": logical_id, "resource": resource},
+            )
         return result

requirements.txt

@@ -8,10 +8,9 @@ boto3==1.17.86
 botocore==1.20.86
 cfn-flip==1.2.3
 click==7.1.2
-importlib-metadata==4.4.0
 jmespath==0.10.0
 pluggy==0.13.1
-pycfmodel==0.10.4
+pycfmodel==0.11.0
 pydantic==1.8.2
 pydash==4.7.6
 python-dateutil==2.8.1
@@ -20,4 +19,3 @@ s3transfer==0.4.2
 six==1.16.0
 typing-extensions==3.10.0.0
 urllib3==1.26.5
-zipp==3.4.1

setup.py

@@ -11,7 +11,7 @@
     "cfn_flip>=1.2.0",
     "click~=7.1.1",
     "pluggy~=0.13.1",
-    "pycfmodel>=0.10.4",
+    "pycfmodel>=0.11.0",
     "pydash~=4.7.6",
     "PyYAML>=4.2b1",
 ]

tests/rules/test_S3ObjectVersioningRule.py

@@ -0,0 +1,41 @@
+import pytest
+
+from cfripper.config.config import Config
+from cfripper.model.enums import RuleGranularity, RuleMode, RuleRisk
+from cfripper.model.result import Failure
+from cfripper.rules import S3ObjectVersioningRule
+from tests.utils import compare_lists_of_failures, get_cfmodel_from
+
+
+def test_rule_passing():
+    template_path = "rules/S3ObjectVersioning/good_template.yaml"
+    rule = S3ObjectVersioningRule(None)
+    result = rule.invoke(get_cfmodel_from(template_path).resolve())
+
+    assert result.valid
+    assert compare_lists_of_failures(result.failures, [])
+
+
+@pytest.mark.parametrize(
+    "template_path",
+    ["rules/S3ObjectVersioning/status_suspended.yaml", "rules/S3ObjectVersioning/no_versioning_defined.yaml"],
+)
+def test_failures_are_raised(template_path):
+    rule = S3ObjectVersioningRule(Config())
+    result = rule.invoke(get_cfmodel_from(template_path).resolve())
+
+    assert not result.valid
+    assert compare_lists_of_failures(
+        result.failures,
+        [
+            Failure(
+                granularity=RuleGranularity.RESOURCE,
+                reason="S3 Bucket VersionBucket is required to have object versioning enabled",
+                risk_value=RuleRisk.LOW,
+                rule="S3ObjectVersioningRule",
+                rule_mode=RuleMode.BLOCKING,
+                actions=None,
+                resource_ids={"VersionBucket"},
+            )
+        ],
+    )

tests/test_templates/rules/S3CrossAccountTrustRule/s3_bucket_cross_account.json

@@ -4,10 +4,7 @@
     "AWSAccount": {
       "Default": "sandbox",
       "Description": "Which AWS Account?",
-      "AllowedValues": [
-        "sandbox",
-        "prod"
-      ],
+      "AllowedValues": ["sandbox", "prod"],
       "Type": "String"
     },
     "ProjectName": {
@@ -40,26 +37,8 @@
       "Type": "AWS::S3::Bucket",
       "Properties": {
         "AccessControl": "PublicRead",
-        "BucketName": [
-          "-",
-          [
-            "company",
-            "AWSAccount",
-            "BucketPostFix"
-          ]
-        ],
+        "BucketName": "test",
         "Tags": [
-          {
-            "Key": "Name",
-            "Value": [
-              "-",
-              [
-                "company",
-                "AWSAccount",
-                "BucketPostFix"
-              ]
-            ]
-          },
           {
             "Key": "Project",
             "Value": "ProjectName"
@@ -85,22 +64,16 @@
           "Version": "2012-10-17",
           "Statement": {
             "Sid": "AllowAccessFromSandbox",
-            "Action": [
-              "s3:PutObject"
-            ],
+            "Action": ["s3:PutObject"],
             "Effect": "Allow",
-            "Resource": [
-              "arn:aws:s3:::S3Bucket/*"
-            ],
+            "Resource": ["arn:aws:s3:::S3Bucket/*"],
             "Principal": {
-              "AWS": [
-                "arn:aws:iam::987654321:root"
-              ]
+              "AWS": ["arn:aws:iam::987654321:root"]
             }
           }
         }
       }
     }
   },
   "AWSTemplateFormatVersion": "2010-09-09"
-}
+}