Add foreach compatibility (#2807)

Author: kddejong

src/cfnlint/__init__.py

@@ -20,7 +20,7 @@
 from cfnlint.rules import TransformError as _TransformError
 from cfnlint.runner import Runner as _Runner
 from cfnlint.template import Template as _Template
-from cfnlint.transform import Transform
+from cfnlint.template.transforms._sam import Transform
 
 LOGGER = logging.getLogger(__name__)
 

src/cfnlint/decode/__init__.py

@@ -10,3 +10,4 @@
     decode,
     decode_str,
 )
+from cfnlint.decode.utils import convert_dict

src/cfnlint/decode/cfn_yaml.py

@@ -192,6 +192,18 @@ def __init__(self, stream, filename):
         Resolver.__init__(self)
         NodeConstructor.__init__(self, filename)
 
+    def construct_getatt(self, node):
+        """
+        Reconstruct !GetAtt into a list
+        """
+
+        if isinstance(node.value, (str)):
+            return list_node(node.value.split(".", 1), node.start_mark, node.end_mark)
+        if isinstance(node.value, list):
+            return [self.construct_object(child, deep=False) for child in node.value]
+
+        raise ValueError(f"Unexpected node type: {type(node.value)}")
+
 
 def multi_constructor(loader, tag_suffix, node):
     """
@@ -203,7 +215,7 @@ def multi_constructor(loader, tag_suffix, node):
 
     constructor = None
     if tag_suffix == "Fn::GetAtt":
-        constructor = construct_getatt
+        constructor = loader.construct_getatt
     elif isinstance(node, ScalarNode):
         constructor = loader.construct_scalar
     elif isinstance(node, SequenceNode):
@@ -219,19 +231,6 @@ def multi_constructor(loader, tag_suffix, node):
     return dict_node({tag_suffix: constructor(node)}, node.start_mark, node.end_mark)
 
 
-def construct_getatt(node):
-    """
-    Reconstruct !GetAtt into a list
-    """
-
-    if isinstance(node.value, (str)):
-        return list_node(node.value.split(".", 1), node.start_mark, node.end_mark)
-    if isinstance(node.value, list):
-        return list_node([s.value for s in node.value], node.start_mark, node.end_mark)
-
-    raise ValueError(f"Unexpected node type: {type(node.value)}")
-
-
 def loads(yaml_string, fname=None):
     """
     Load the given YAML string

src/cfnlint/decode/decode.py

@@ -11,7 +11,7 @@
 from yaml.scanner import ScannerError
 
 from cfnlint.decode import cfn_json, cfn_yaml
-from cfnlint.rules import Match, ParseError
+from cfnlint.match import Match
 
 LOGGER = logging.getLogger(__name__)
 
@@ -118,6 +118,9 @@ def _decode(
         matches = [create_match_file_error(filename, str(err))]
 
     if not isinstance(template, dict) and not matches:
+        # pylint: disable=import-outside-toplevel
+        from cfnlint.rules import ParseError
+
         # Template isn't a dict which means nearly nothing will work
         matches = [
             Match(
@@ -135,6 +138,9 @@ def _decode(
 
 def create_match_yaml_parser_error(parser_error, filename):
     """Create a Match for a parser error"""
+    # pylint: disable=import-outside-toplevel
+    from cfnlint.rules import ParseError
+
     lineno = parser_error.problem_mark.line + 1
     colno = parser_error.problem_mark.column + 1
     msg = parser_error.problem
@@ -143,6 +149,9 @@ def create_match_yaml_parser_error(parser_error, filename):
 
 def create_match_file_error(filename, msg):
     """Create a Match for a parser error"""
+    # pylint: disable=import-outside-toplevel
+    from cfnlint.rules import ParseError
+
     return Match(
         linenumber=1,
         columnnumber=1,
@@ -156,6 +165,9 @@ def create_match_file_error(filename, msg):
 
 def create_match_json_parser_error(parser_error, filename):
     """Create a Match for a parser error"""
+    # pylint: disable=import-outside-toplevel
+    from cfnlint.rules import ParseError
+
     lineno = parser_error.lineno
     colno = parser_error.colno
     msg = parser_error.msg

src/cfnlint/decode/utils.py

@@ -0,0 +1,29 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfnlint.decode.node import dict_node, list_node, str_node
+
+
+def convert_dict(template, start_mark=(0, 0), end_mark=(0, 0)):
+    """Convert dict to template"""
+    if isinstance(template, dict):
+        if not isinstance(template, dict_node):
+            template = dict_node(template, start_mark, end_mark)
+        for k, v in template.copy().items():
+            k_start_mark = start_mark
+            k_end_mark = end_mark
+            if isinstance(k, str_node):
+                k_start_mark = k.start_mark
+                k_end_mark = k.end_mark
+            new_k = str_node(k, k_start_mark, k_end_mark)
+            del template[k]
+            template[new_k] = convert_dict(v, k_start_mark, k_end_mark)
+    elif isinstance(template, list):
+        if not isinstance(template, list_node):
+            template = list_node(template, start_mark, end_mark)
+        for i, v in enumerate(template):
+            template[i] = convert_dict(v, start_mark, end_mark)
+
+    return template

src/cfnlint/formatters/__init__.py

@@ -13,7 +13,8 @@
 from jschema_to_python.to_json import to_json
 from junit_xml import TestCase, TestSuite, to_xml_report_string
 
-from cfnlint.rules import Match, ParseError, RuleError, RulesCollection, TransformError
+from cfnlint.match import Match
+from cfnlint.rules import ParseError, RuleError, RulesCollection, TransformError
 from cfnlint.version import __version__
 
 Matches = List[Match]

src/cfnlint/helpers.py

@@ -22,7 +22,6 @@
 import regex as re
 
 from cfnlint.data import CloudSpecs
-from cfnlint.decode.node import dict_node, list_node, str_node
 
 LOGGER = logging.getLogger(__name__)
 
@@ -82,7 +81,6 @@
     r"^.*{{resolve:ssm-secure:[a-zA-Z0-9_\.\-/]+(:\d+)?}}.*$"
 )
 
-
 FUNCTIONS = [
     "Fn::Base64",
     "Fn::GetAtt",
@@ -111,6 +109,7 @@
 FUNCTION_OR = "Fn::Or"
 FUNCTION_NOT = "Fn::Not"
 FUNCTION_EQUALS = "Fn::Equals"
+FUNCTION_FOR_EACH = re.compile(r"^Fn::ForEach::[a-zA-Z0-9]+$")
 
 PSEUDOPARAMS = [
     "AWS::AccountId",
@@ -543,29 +542,6 @@ def onerror(os_error):
     return result
 
 
-def convert_dict(template, start_mark=(0, 0), end_mark=(0, 0)):
-    """Convert dict to template"""
-    if isinstance(template, dict):
-        if not isinstance(template, dict_node):
-            template = dict_node(template, start_mark, end_mark)
-        for k, v in template.copy().items():
-            k_start_mark = start_mark
-            k_end_mark = end_mark
-            if isinstance(k, str_node):
-                k_start_mark = k.start_mark
-                k_end_mark = k.end_mark
-            new_k = str_node(k, k_start_mark, k_end_mark)
-            del template[k]
-            template[new_k] = convert_dict(v, k_start_mark, k_end_mark)
-    elif isinstance(template, list):
-        if not isinstance(template, list_node):
-            template = list_node(template, start_mark, end_mark)
-        for i, v in enumerate(template):
-            template[i] = convert_dict(v, start_mark, end_mark)
-
-    return template
-
-
 def override_specs(override_spec_file):
     """Override specs file"""
     try:

src/cfnlint/match.py

@@ -0,0 +1,53 @@
+"""
+Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+
+class Match:
+    """Match Classes"""
+
+    def __init__(
+        self,
+        linenumber,
+        columnnumber,
+        linenumberend,
+        columnnumberend,
+        filename,
+        rule,
+        message=None,
+        rulematch_obj=None,
+    ):
+        """Init"""
+        self.linenumber = linenumber
+        """Starting line number of the region this match spans"""
+        self.columnnumber = columnnumber
+        """Starting line number of the region this match spans"""
+        self.linenumberend = linenumberend
+        """Ending line number of the region this match spans"""
+        self.columnnumberend = columnnumberend
+        """Ending column number of the region this match spans"""
+        self.filename = filename
+        """Name of the filename associated with this match, or None if there is no such file"""
+        self.rule = rule
+        """The rule of this match"""
+        self.message = message  # or rule.shortdesc
+        """The message of this match"""
+        if rulematch_obj:
+            for k, v in vars(rulematch_obj).items():
+                if not hasattr(self, k):
+                    setattr(self, k, v)
+
+    def __repr__(self):
+        """Represent"""
+        file_str = self.filename + ":" if self.filename else ""
+        return f"[{self.rule}] ({self.message}) matched {file_str}{self.linenumber}"
+
+    def __eq__(self, item):
+        """Override equal to compare matches"""
+        return (self.linenumber, self.columnnumber, self.rule.id, self.message) == (
+            item.linenumber,
+            item.columnnumber,
+            item.rule.id,
+            item.message,
+        )

src/cfnlint/rules/__init__.py

@@ -13,6 +13,7 @@
 import cfnlint.rules.custom
 from cfnlint.decode.node import TemplateAttributeError
 from cfnlint.exceptions import DuplicateRuleError
+from cfnlint.match import Match
 from cfnlint.template import Template
 
 LOGGER = logging.getLogger(__name__)
@@ -628,55 +629,6 @@ def __hash__(self):
         return hash((self.path, self.message))
 
 
-class Match:  # pylint: disable=R0902
-    """Match Classes"""
-
-    def __init__(
-        self,
-        linenumber,
-        columnnumber,
-        linenumberend,
-        columnnumberend,
-        filename,
-        rule,
-        message=None,
-        rulematch_obj=None,
-    ):
-        """Init"""
-        self.linenumber = linenumber
-        """Starting line number of the region this match spans"""
-        self.columnnumber = columnnumber
-        """Starting line number of the region this match spans"""
-        self.linenumberend = linenumberend
-        """Ending line number of the region this match spans"""
-        self.columnnumberend = columnnumberend
-        """Ending column number of the region this match spans"""
-        self.filename = filename
-        """Name of the filename associated with this match, or None if there is no such file"""
-        self.rule = rule
-        """The rule of this match"""
-        self.message = message  # or rule.shortdesc
-        """The message of this match"""
-        if rulematch_obj:
-            for k, v in vars(rulematch_obj).items():
-                if not hasattr(self, k):
-                    setattr(self, k, v)
-
-    def __repr__(self):
-        """Represent"""
-        file_str = self.filename + ":" if self.filename else ""
-        return f"[{self.rule}] ({self.message}) matched {file_str}{self.linenumber}"
-
-    def __eq__(self, item):
-        """Override equal to compare matches"""
-        return (self.linenumber, self.columnnumber, self.rule.id, self.message) == (
-            item.linenumber,
-            item.columnnumber,
-            item.rule.id,
-            item.message,
-        )
-
-
 class ParseError(CloudFormationLintRule):
     """Parse Lint Rule"""
 

src/cfnlint/rules/functions/ForEach.py

@@ -0,0 +1,21 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+import logging
+
+from cfnlint.rules import CloudFormationLintRule
+
+LOGGER = logging.getLogger("cfnlint")
+
+
+class ForEach(CloudFormationLintRule):
+    id = "E1032"
+    shortdesc = "Validates ForEach functions"
+    description = "Validates that ForEach parameters have a valid configuration"
+    source_url = "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/intrinsic-function-reference-getatt.html"
+    tags = ["functions", "foreach"]
+
+    # pylint: disable=unused-argument
+    def match(self, cfn):
+        return []

src/cfnlint/rules/outputs/Value.py

@@ -41,7 +41,7 @@ def match(self, cfn):
                         objtype = (
                             template.get("Resources", {}).get(obj[0], {}).get("Type")
                         )
-                        if objtype:
+                        if objtype and isinstance(obj[1], str):
                             attribute = (
                                 self.resourcetypes.get(objtype, {})
                                 .get("Attributes", {})

src/cfnlint/rules/resources/iam/Permissions.py

@@ -5,7 +5,8 @@
 import json
 
 from cfnlint.data import AdditionalSpecs
-from cfnlint.helpers import convert_dict, load_resource
+from cfnlint.decode import convert_dict
+from cfnlint.helpers import load_resource
 from cfnlint.rules import CloudFormationLintRule, RuleMatch
 
 

src/cfnlint/rules/resources/iam/Policy.py

@@ -5,7 +5,8 @@
 import json
 from datetime import date
 
-from cfnlint.helpers import FUNCTIONS_SINGLE, convert_dict
+from cfnlint.decode import convert_dict
+from cfnlint.helpers import FUNCTIONS_SINGLE
 from cfnlint.rules import CloudFormationLintRule, RuleMatch