Refactor, add more Rules and use new pycfmodel (#134)

* Refactor, add more Rules and use new pycfmodel * Fix typo * Detect privilege escalation in any policy from any resource (not only IAMPolicies) * Improve tests for privilege escalation * Improve tests and docs * Bump version to 0.21.0 * Fix flake8 issue * Improve ABC implementation of dangerous actions * Update cfripper/rules/base_rules.py Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com> * Update cfripper/rules/sns_topic_policy.py Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com> * Update cfripper/rules/sns_topic_policy.py Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com> * Remove leftover print * Fix test after copy paste error * Add failure when and only when there are dangerous actions Co-authored-by: Oliver Crawford <16978487+ocrawford555@users.noreply.github.com>
Skyscanner · Dec 1, 2020 · a9bd092 · a9bd092
1 parent 825d5ef
commit a9bd092
Show file tree

Hide file tree

Showing 24 changed files with 472 additions and 175 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,17 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [0.21.0] - 2020-11-30
+### Improvements
+- Upgraded to pycfmodel 0.8.1 (this will improve policy action detection)
+- Refactored a few classes to use improvements from new base classes and pycfmodel
+- `PrivilegeEscalationRule` now detects issues in all policies
+### Additions
+- New Rules: `SNSTopicDangerousPolicyActionsRule` and `SQSDangerousPolicyActionsRule`
+- New abstract base rule: BaseDangerousPolicyActions
+### Fixes
+- Various typo fixes
+
 ## [0.20.1] - 2020-10-26
 ### Improvements
 - Added more actions that only allow wildcard as resource

diff --git a/cfripper/__version__.py b/cfripper/__version__.py
@@ -1,3 +1,3 @@
-VERSION = (0, 20, 1)
+VERSION = (0, 21, 0)
 
 __version__ = ".".join(map(str, VERSION))
diff --git a/cfripper/model/result.py b/cfripper/model/result.py
@@ -1,12 +1,9 @@
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from pydantic import BaseModel, Extra
 
 from cfripper.model.enums import RuleMode
 
-if TYPE_CHECKING:
-    from pydantic.typing import AbstractSetIntStr, DictIntStrAny
-
 
 class Failure(BaseModel):
     granularity: str
@@ -49,8 +46,8 @@ def get_properties(cls):
     def dict(
         self,
         *,
-        include: Union["AbstractSetIntStr", "DictIntStrAny"] = None,
-        exclude: Union["AbstractSetIntStr", "DictIntStrAny"] = None,
+        include: Union["AbstractSetIntStr", "DictIntStrAny"] = None,  # noqa: F821
+        exclude: Union["AbstractSetIntStr", "DictIntStrAny"] = None,  # noqa: F821
         by_alias: bool = False,
         skip_defaults: bool = None,
         exclude_unset: bool = False,

diff --git a/cfripper/rules/__init__.py b/cfripper/rules/__init__.py
@@ -1,4 +1,4 @@
-from cfripper.rules.base_rules import PrincipalCheckingRule
+from cfripper.rules.base_rules import BaseDangerousPolicyActions, PrincipalCheckingRule, ResourceSpecificRule
 from cfripper.rules.cloudformation_authentication import CloudFormationAuthenticationRule
 from cfripper.rules.cross_account_trust import (
     CrossAccountCheckingRule,
@@ -20,8 +20,12 @@
 from cfripper.rules.privilege_escalation import PrivilegeEscalationRule
 from cfripper.rules.s3_bucket_policy import S3BucketPolicyPrincipalRule
 from cfripper.rules.s3_public_access import S3BucketPublicReadAclAndListStatementRule, S3BucketPublicReadWriteAclRule
-from cfripper.rules.sns_topic_policy_not_principal import SNSTopicPolicyNotPrincipalRule
-from cfripper.rules.sqs_queue_policy import SQSQueuePolicyNotPrincipalRule, SQSQueuePolicyPublicRule
+from cfripper.rules.sns_topic_policy import SNSTopicDangerousPolicyActionsRule, SNSTopicPolicyNotPrincipalRule
+from cfripper.rules.sqs_queue_policy import (
+    SQSDangerousPolicyActionsRule,
+    SQSQueuePolicyNotPrincipalRule,
+    SQSQueuePolicyPublicRule,
+)
 from cfripper.rules.wildcard_policies import (
     S3BucketPolicyWildcardActionRule,
     SNSTopicPolicyWildcardActionRule,
@@ -41,8 +45,8 @@
         EC2SecurityGroupOpenToWorldRule,
         FullWildcardPrincipalRule,
         HardcodedRDSPasswordRule,
-        IAMRoleWildcardActionOnPolicyRule,
         IAMRolesOverprivilegedRule,
+        IAMRoleWildcardActionOnPolicyRule,
         KMSKeyCrossAccountTrustRule,
         KMSKeyWildcardPrincipalRule,
         ManagedPolicyOnUserRule,
@@ -54,13 +58,18 @@
         S3BucketPublicReadAclAndListStatementRule,
         S3BucketPublicReadWriteAclRule,
         S3CrossAccountTrustRule,
+        SNSTopicDangerousPolicyActionsRule,
         SNSTopicPolicyNotPrincipalRule,
         SNSTopicPolicyWildcardActionRule,
+        SQSDangerousPolicyActionsRule,
         SQSQueuePolicyNotPrincipalRule,
         SQSQueuePolicyPublicRule,
         SQSQueuePolicyWildcardActionRule,
         WildcardResourceRule,
     )
 }
 
-BASE_CLASSES = {rule.__name__: rule for rule in (CrossAccountCheckingRule, PrincipalCheckingRule)}
+BASE_CLASSES = {
+    rule.__name__: rule
+    for rule in (BaseDangerousPolicyActions, CrossAccountCheckingRule, PrincipalCheckingRule, ResourceSpecificRule,)
+}
diff --git a/cfripper/rules/base_rules.py b/cfripper/rules/base_rules.py
@@ -1,8 +1,9 @@
 import logging
 from abc import ABC, abstractmethod
-from typing import Dict, List, Optional, Set
+from typing import Dict, List, Optional, Set, Tuple, Type
 
 from pycfmodel.model.cf_model import CFModel
+from pycfmodel.model.resources.resource import Resource
 
 from cfripper.config.config import Config
 from cfripper.config.rule_config import RuleConfig
@@ -98,7 +99,26 @@ def add_warning_to_result(
             result.add_warning(warning)
 
 
-class PrincipalCheckingRule(Rule):
+class ResourceSpecificRule(Rule):
+    """
+    Base class for rules that only apply to a subset of resource types
+    """
+
+    RESOURCE_TYPES: Tuple[Type] = tuple()
+
+    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
+        result = Result()
+        for logical_id, resource in cfmodel.Resources.items():
+            if isinstance(resource, self.RESOURCE_TYPES):
+                result += self.resource_invoke(resource=resource, logical_id=logical_id)
+        return result
+
+    @abstractmethod
+    def resource_invoke(self, resource: Resource, logical_id: str) -> Result:
+        pass
+
+
+class PrincipalCheckingRule(Rule, ABC):
     """Abstract class for rules that check principals"""
 
     _valid_principals = None
@@ -122,3 +142,34 @@ def valid_principals(self) -> Set[str]:
             if self._config.aws_account_id:
                 self._valid_principals.add(self._config.aws_account_id)
         return self._valid_principals
+
+
+class BaseDangerousPolicyActions(ResourceSpecificRule, ABC):
+    """
+    Base class for dangerous actions. Admits a DANGEROUS_ACTIONS class variable with a list of dangerous actions
+    """
+
+    REASON = "Resource {} should not include the following dangerous actions: {}"
+    RISK_VALUE = RuleRisk.HIGH
+    GRANULARITY = RuleGranularity.ACTION
+
+    @property
+    @classmethod
+    @abstractmethod
+    def DANGEROUS_ACTIONS(cls) -> List[str]:
+        # This is designed to be overwritten as a class variable
+        return NotImplementedError
+
+    def resource_invoke(self, resource: Resource, logical_id: str) -> Result:
+        result = Result()
+        for policy in resource.policy_documents:
+            actions = policy.policy_document.get_allowed_actions()
+            dangerous_actions = set(actions) & set(self.DANGEROUS_ACTIONS)
+            if dangerous_actions:
+                self.add_failure_to_result(
+                    result,
+                    self.REASON.format(logical_id, sorted(dangerous_actions)),
+                    resource_ids={logical_id},
+                    actions=dangerous_actions,
+                )
+        return result
diff --git a/cfripper/rules/ec2_security_group.py b/cfripper/rules/ec2_security_group.py
@@ -4,6 +4,7 @@
     "EC2SecurityGroupOpenToWorldRule",
 ]
 
+from abc import ABC
 from itertools import groupby
 from operator import itemgetter
 from typing import Dict, List, Optional, Tuple, Union
@@ -18,7 +19,7 @@
 from cfripper.rules.base_rules import Rule
 
 
-class SecurityGroupOpenToWorldRule(Rule):
+class SecurityGroupOpenToWorldRule(Rule, ABC):
     """
     Base class not intended to be instantiated, but inherited from.
     This class provides common methods used to detect open ports.

diff --git a/cfripper/rules/privilege_escalation.py b/cfripper/rules/privilege_escalation.py
@@ -1,55 +1,38 @@
 __all__ = ["PrivilegeEscalationRule"]
 
-from typing import Dict, Optional
-
-from pycfmodel.model.cf_model import CFModel
-from pycfmodel.model.resources.iam_policy import IAMPolicy
+from pycfmodel.model.resources.resource import Resource
 
 from cfripper.model.enums import RuleGranularity
-from cfripper.model.result import Result
-from cfripper.rules.base_rules import Rule
+from cfripper.rules.base_rules import BaseDangerousPolicyActions
 
 
-class PrivilegeEscalationRule(Rule):
+class PrivilegeEscalationRule(BaseDangerousPolicyActions):
     """
     Checks for any dangerous IAM actions that could allow privilege escalation and potentially
     represent a large security risk.
     See [current blacklisted IAM actions](https://github.com/Skyscanner/cfripper/blob/master/cfripper/rules/privilege_escalation.py#L29).
 
     Fix:
-        Unless strictly necessary, do not use actions in the IAM action blacklist. CloudFormation files that do require these
-        actions should be added to the whitelist.
+        Unless strictly necessary, do not use actions in the IAM action blacklist. CloudFormation files that do require
+        these actions should be added to the whitelist.
     """
 
-    GRANULARITY = RuleGranularity.RESOURCE
-    REASON = "{} has blacklisted IAM action {}"
-    IAM_BLACKLIST = set(
-        action.lower()
-        for action in [
-            "iam:CreateAccessKey",
-            "iam:CreateLoginProfile",
-            "iam:UpdateLoginProfile",
-            "iam:AttachUserPolicy",
-            "iam:AttachGroupPolicy",
-            "iam:AttachRolePolicy",
-            "iam:PutUserPolicy",
-            "iam:PutGroupPolicy",
-            "iam:PutRolePolicy",
-            "iam:CreatePolicy",
-            "iam:AddUserToGroup",
-            "iam:UpdateAssumeRolePolicy",
-            "iam:CreatePolicyVersion",
-            "iam:SetDefaultPolicyVersion",
-        ]
-    )
-
-    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
-        result = Result()
-        for logical_id, resource in cfmodel.Resources.items():
-            if isinstance(resource, IAMPolicy):
-                policy_actions = set(action.lower() for action in resource.Properties.PolicyDocument.get_iam_actions())
-                for violation in policy_actions.intersection(self.IAM_BLACKLIST):
-                    self.add_failure_to_result(
-                        result, self.REASON.format(logical_id, violation), resource_ids={logical_id}
-                    )
-        return result
+    GRANULARITY = RuleGranularity.ACTION
+    REASON = "{} has blacklisted IAM actions: {}"
+    RESOURCE_TYPES = (Resource,)
+    DANGEROUS_ACTIONS = [
+        "iam:AddUserToGroup",
+        "iam:AttachGroupPolicy",
+        "iam:AttachRolePolicy",
+        "iam:AttachUserPolicy",
+        "iam:CreateAccessKey",
+        "iam:CreateLoginProfile",
+        "iam:CreatePolicy",
+        "iam:CreatePolicyVersion",
+        "iam:PutGroupPolicy",
+        "iam:PutRolePolicy",
+        "iam:PutUserPolicy",
+        "iam:SetDefaultPolicyVersion",
+        "iam:UpdateAssumeRolePolicy",
+        "iam:UpdateLoginProfile",
+    ]
diff --git a/cfripper/rules/s3_bucket_policy.py b/cfripper/rules/s3_bucket_policy.py
@@ -1,19 +1,17 @@
 __all__ = ["S3BucketPolicyPrincipalRule"]
 import logging
-from typing import Dict, Optional
 
-from pycfmodel.model.cf_model import CFModel
 from pycfmodel.model.resources.s3_bucket_policy import S3BucketPolicy
 
 from cfripper.model.enums import RuleGranularity, RuleRisk
 from cfripper.model.result import Result
 from cfripper.model.utils import get_account_id_from_principal
-from cfripper.rules.base_rules import PrincipalCheckingRule
+from cfripper.rules.base_rules import PrincipalCheckingRule, ResourceSpecificRule
 
 logger = logging.getLogger(__file__)
 
 
-class S3BucketPolicyPrincipalRule(PrincipalCheckingRule):
+class S3BucketPolicyPrincipalRule(PrincipalCheckingRule, ResourceSpecificRule):
     """
     Checks for non-whitelisted principals in S3 bucket policies.
 
@@ -26,26 +24,27 @@ class S3BucketPolicyPrincipalRule(PrincipalCheckingRule):
     """
 
     GRANULARITY = RuleGranularity.RESOURCE
+
     REASON = "S3 Bucket {} policy has non-whitelisted principals {}"
     RISK_VALUE = RuleRisk.HIGH
+    RESOURCE_TYPES = (S3BucketPolicy,)
 
-    def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
+    def resource_invoke(self, resource: S3BucketPolicy, logical_id: str) -> Result:
         result = Result()
-        for logical_id, resource in cfmodel.Resources.items():
-            if isinstance(resource, S3BucketPolicy):
-                for statement in resource.Properties.PolicyDocument._statement_as_list():
-                    for principal in statement.get_principal_list():
-                        account_id = get_account_id_from_principal(principal)
-                        if not account_id:
-                            continue
-                        if account_id not in self.valid_principals:
-                            if statement.Condition and statement.Condition.dict():
-                                logger.warning(
-                                    f"Not adding {type(self).__name__} failure in {logical_id} "
-                                    f"because there are conditions: {statement.Condition}"
-                                )
-                            else:
-                                self.add_failure_to_result(
-                                    result, self.REASON.format(logical_id, account_id), resource_ids={logical_id},
-                                )
+
+        for statement in resource.Properties.PolicyDocument._statement_as_list():
+            for principal in statement.get_principal_list():
+                account_id = get_account_id_from_principal(principal)
+                if not account_id:
+                    continue
+                if account_id not in self.valid_principals:
+                    if statement.Condition and statement.Condition.dict():
+                        logger.warning(
+                            f"Not adding {type(self).__name__} failure in {logical_id} "
+                            f"because there are conditions: {statement.Condition}"
+                        )
+                    else:
+                        self.add_failure_to_result(
+                            result, self.REASON.format(logical_id, account_id), resource_ids={logical_id},
+                        )
         return result
diff --git a/cfripper/rules/sns_topic_policy.py b/cfripper/rules/sns_topic_policy.py
@@ -0,0 +1,50 @@
+__all__ = ["SNSTopicPolicyNotPrincipalRule", "SNSTopicDangerousPolicyActionsRule"]
+
+
+from pycfmodel.model.resources.sns_topic_policy import SNSTopicPolicy
+
+from cfripper.model.enums import RuleGranularity, RuleRisk
+from cfripper.model.result import Result
+from cfripper.rules.base_rules import BaseDangerousPolicyActions, ResourceSpecificRule
+
+
+class SNSTopicPolicyNotPrincipalRule(ResourceSpecificRule):
+    """
+    Checks if an SNS topic policy has an Allow + a NotPrincipal.
+
+    Risk:
+        AWS **strongly** recommends against using `NotPrincipal` in the same policy statement as `"Effect": "Allow"`.
+        Doing so grants the permissions specified in the policy statement to all principals except the one named
+        in the `NotPrincipal` element. By doing this, you might grant access to anonymous (unauthenticated) users.
+    """
+
+    GRANULARITY = RuleGranularity.RESOURCE
+    REASON = "SNS Topic policy {} should not allow Allow and NotPrincipal at the same time"
+    RESOURCE_TYPES = (SNSTopicPolicy,)
+
+    def resource_invoke(self, resource: SNSTopicPolicy, logical_id: str) -> Result:
+        result = Result()
+        for statement in resource.Properties.PolicyDocument._statement_as_list():
+            if statement.NotPrincipal:
+                self.add_failure_to_result(result, self.REASON.format(logical_id), resource_ids={logical_id})
+        return result
+
+
+class SNSTopicDangerousPolicyActionsRule(BaseDangerousPolicyActions):
+    """
+    Checks for dangerous permissions in Action statements in an SNS Topic Policy.
+
+    Risk:
+        This is deemed a potential security risk as it could allow privilege escalation.
+    """
+
+    REASON = "SNS Topic policy {} should not not include the following dangerous actions: {}"
+    RISK_VALUE = RuleRisk.MEDIUM
+    RESOURCE_TYPES = (SNSTopicPolicy,)
+
+    DANGEROUS_ACTIONS = [
+        "sns:AddPermission",
+        "sns:RemovePermission",
+        "sns:TagResource",
+        "sns:UntagResource",
+    ]