How to use the cfripper.model.rule.Rule function in cfripper

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
from pycfmodel.model.resources.iam_role import IAMRole

from ..config.regex import REGEX_WILDCARD_POLICY_ACTION
from ..model.rule import Rule


class IAMRoleWildcardActionOnPermissionsPolicyRule(Rule):

    REASON = "IAM role {} should not allow * action on its permissions policy {}"

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, IAMRole):
                for policy in resource.Properties.Policies:
                    if policy.PolicyDocument.allowed_actions_with(REGEX_WILDCARD_POLICY_ACTION):
                        self.add_failure(type(self).__name__, self.REASON.format(logical_id, policy.PolicyName))

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
from pycfmodel.model.resources.security_group import SecurityGroup

from ..model.enums import RuleMode
from ..model.rule import Rule


class SecurityGroupMissingEgressRule(Rule):

    REASON = (
        "Missing egress rule in {} means all traffic is allowed outbound. Make this explicit if it is desired "
        "configuration"
    )
    RULE_MODE = RuleMode.MONITOR

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, SecurityGroup) and not resource.Properties.SecurityGroupEgress:
                self.add_failure(type(self).__name__, self.REASON.format(logical_id))

under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
import logging
import re

from cfripper.model.enums import RuleMode
from cfripper.model.rule import Rule

logger = logging.getLogger(__file__)

POLICY_DOCUMENT_TYPES = ("policy_document", "key_policy", "assume_role_policy_document")


class GenericWildcardPrincipal(Rule):

    REASON_WILCARD_PRINCIPAL = "{} should not allow wildcard in principals or account-wide principals (principal: '{}')"
    REASON_NOT_ALLOWED_PRINCIPAL = "{} contains an unknown principal: {}"
    RULE_MODE = RuleMode.MONITOR
    IAM_PATTERN = r"arn:aws:iam::(\d*|\*):.*"

    FULL_REGEX = r"^((\w*:){0,1}\*|arn:aws:iam::(\d*|\*):.*)$"

    def invoke(self, resources, parameters):
        for resource_list in resources.values():
            for resource in resource_list:
                # Resource has policies
                for policy in getattr(resource, "policies", []):
                    self.check_for_wildcards(
                        resource=policy, resource_id=getattr(policy, "policy_name", resource.logical_id)
                    )

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
from pycfmodel.model.resources.sqs_queue_policy import SQSQueuePolicy

from ..model.enums import RuleMode
from ..model.rule import Rule


class SQSQueuePolicyNotPrincipalRule(Rule):
    REASON = "SQS Queue {} policy should not allow Allow and NotPrincipal at the same time"
    RULE_MODE = RuleMode.MONITOR

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, SQSQueuePolicy):
                for statement in resource.Properties.PolicyDocument._statement_as_list():
                    if statement.NotPrincipal:
                        self.add_failure(type(self).__name__, self.REASON.format(logical_id))

Licensed under the Apache License, Version 2.0 (the "License"); you may not use
this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
from ..model.enums import RuleRisk
from ..model.rule import Rule


class S3BucketPublicReadWriteAclRule(Rule):
    REASON = "S3 Bucket {} should not have a public read-write acl"
    RISK_VALUE = RuleRisk.HIGH

    def invoke(self, cfmodel):
        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(type(self).__name__, self.REASON.format(logical_id))

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
import re

from pycfmodel.model.resources.sqs_queue_policy import SQSQueuePolicy

from ..model.rule import Rule


class SQSQueuePolicyWildcardActionRule(Rule):

    REASON = "SQS Queue policy {} should not allow * action"

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, SQSQueuePolicy) and resource.Properties.PolicyDocument.allowed_actions_with(
                re.compile(r"^(\w*:){0,1}\*$")
            ):
                self.add_failure(type(self).__name__, self.REASON.format(logical_id))

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
__all__ = ["IAMManagedPolicyWildcardActionRule"]
from pycfmodel.model.resources.iam_managed_policy import IAMManagedPolicy

from cfripper.config.regex import REGEX_WILDCARD_POLICY_ACTION
from cfripper.model.enums import RuleGranularity
from cfripper.model.rule import Rule


class IAMManagedPolicyWildcardActionRule(Rule):
    """
    This rule checks for wildcards in IAM Managed policies
    """

    GRANULARITY = RuleGranularity.RESOURCE
    REASON = "IAM managed policy {} should not allow * action"

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, IAMManagedPolicy) and resource.Properties.PolicyDocument.allowed_actions_with(
                REGEX_WILDCARD_POLICY_ACTION
            ):
                self.add_failure(type(self).__name__, self.REASON.format(logical_id), resource_ids={logical_id})

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
import re

from pycfmodel.model.resources.s3_bucket_policy import S3BucketPolicy

from ..model.rule import Rule


class S3BucketPolicyWildcardActionRule(Rule):
    REASON = "S3 Bucket policy {} should not allow * action"

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, S3BucketPolicy) and resource.Properties.PolicyDocument.allowed_actions_with(
                re.compile(r"^(\w*:){0,1}\*$")
            ):
                self.add_failure(type(self).__name__, self.REASON.format(logical_id))

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed
under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
CONDITIONS OF ANY KIND, either express or implied. See the License for the
specific language governing permissions and limitations under the License.
"""
from pycfmodel.model.resources.iam_role import IAMRole

from ..config.regex import REGEX_WILDCARD_POLICY_ACTION
from ..model.rule import Rule


class IAMRoleWildcardActionOnTrustPolicyRule(Rule):

    REASON = "IAM role {} should not allow * action on its trust policy"

    def invoke(self, cfmodel):
        for logical_id, resource in cfmodel.Resources.items():
            if isinstance(resource, IAMRole) and resource.Properties.AssumeRolePolicyDocument.allowed_actions_with(
                REGEX_WILDCARD_POLICY_ACTION
            ):
                self.add_failure(type(self).__name__, self.REASON.format(logical_id))