Resolve Fn::FindInMap when a string is returned as part of a foreach (#2814)

kddejong · web-flow · commit d6a46c5b36cd · 2023-07-26T12:52:38.000-07:00
* Resolve Fn::FindInMap when a string is returned as part of a foreach
* Update rule to handle ForEach false positives
diff --git a/src/cfnlint/conditions/conditions.py b/src/cfnlint/conditions/conditions.py
@@ -271,6 +271,8 @@ def build_scenerios_on_region(
             return
         cnf_region = self._cnf.copy()
         found_region = False
+        if condition_name not in self._conditions:
+            return
         for eql in self._conditions[condition_name].equals:
             is_region, equal_region = eql.is_region
             if is_region:
diff --git a/src/cfnlint/rules/functions/FindInMap.py b/src/cfnlint/rules/functions/FindInMap.py
@@ -259,7 +259,7 @@ def validate_enhanced_find_in_map(self, find_in_map, cfn):
 
     def _match_with_language_extensions(self, cfn):
         matches = []
-        find_in_maps = cfn.search_deep_keys("Fn::FindInMap")
+        find_in_maps = cfn.transform_pre["Fn::FindInMap"]
         for in_map in find_in_maps:
             matches.extend(
                 self.validate_enhanced_find_in_map(find_in_map=in_map, cfn=cfn)
diff --git a/src/cfnlint/rules/parameters/Used.py b/src/cfnlint/rules/parameters/Used.py
@@ -35,11 +35,19 @@ def isparaminref(self, subs, parameter):
     def match(self, cfn):
         matches = []
 
+        le_refs = None
+        if cfn.has_language_extensions_transform():
+            le_refs = cfn.search_deep_keys("Ref")
+
         reftrees = cfn.transform_pre.get("Ref")
         subtrees = cfn.transform_pre.get("Fn::Sub")
         refs = []
         for reftree in reftrees:
             refs.append(reftree[-1])
+        if le_refs:
+            for le_ref in le_refs:
+                if le_ref[-1] not in refs:
+                    refs.append(le_ref[-1])
         subs = []
         for subtree in subtrees:
             if isinstance(subtree[-1], list):
diff --git a/src/cfnlint/template/template.py b/src/cfnlint/template/template.py
@@ -844,7 +844,7 @@ def is_resource_available(self, path, resource):
         resource_condition = (
             self.template.get("Resources", {}).get(resource, {}).get("Condition")
         )
-        if resource_condition:
+        if isinstance(resource_condition, str):
             # if path conditions are empty that means its always true
             if not path_conditions:
                 return [{resource_condition: False}]
@@ -1133,7 +1133,7 @@ def get_conditions_from_path(
             if len(path) >= 2:
                 if path[0] in ["Resources", "Outputs"]:
                     condition = text.get(path[0], {}).get(path[1], {}).get("Condition")
-                    if condition:
+                    if isinstance(condition, str):
                         if not results.get(condition):
                             results[condition] = set()
                         results[condition].add(True)
diff --git a/src/cfnlint/template/transforms/_language_extensions.py b/src/cfnlint/template/transforms/_language_extensions.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import logging
 import random
 import string
 from copy import deepcopy
@@ -16,6 +17,8 @@
 from cfnlint.helpers import FUNCTION_FOR_EACH
 from cfnlint.template.transforms._types import TransformResult
 
+LOGGER = logging.getLogger("cfnlint")
+
 # initializing size of string
 _N = 7
 
@@ -51,14 +54,14 @@ def language_extension(cfn: Any) -> TransformResult:
 
         message = "Error transforming template: {0}"
         if hasattr(e.key, "start_mark"):
-            sm_line = e.key.start_mark.line + 1
-            sm_column = e.key.start_mark.column + 1
+            sm_line = e.key.start_mark[0] + 1
+            sm_column = e.key.start_mark[1] + 1
         else:
             sm_line = 1
             sm_column = 1
         if hasattr(e.key, "end_mark"):
-            em_line = e.key.end_mark.line + 1
-            em_column = e.key.end_mark.column + 1
+            em_line = e.key.end_mark[0] + 1
+            em_column = e.key.end_mark[1] + 1
         else:
             em_line = 1
             em_column = 1
@@ -76,6 +79,7 @@ def language_extension(cfn: Any) -> TransformResult:
         ], None
     except Exception as e:  # pylint: disable=broad-exception-caught
         # pylint: disable=import-outside-toplevel
+        from cfnlint.match import Match  # pylint: disable=cyclic-import
         from cfnlint.rules import TransformError  # pylint: disable=cyclic-import
 
         message = "Error transforming template: {0}"
@@ -141,6 +145,17 @@ def _walk(self, item: Any, params: MutableMapping[str, Any], cfn: Any):
                         )
                         if only_string:
                             return obj[k][0]
+                elif k == "Fn::FindInMap":
+                    try:
+                        mapping = _ForEachValueFnFindInMap(get_hash(v), v)
+                        map_value = mapping.value(cfn, params)
+                        if map_value is None:
+                            del obj[k]
+                            continue
+                        if isinstance(map_value, str):
+                            return map_value
+                    except Exception as e:  # pylint: disable=broad-exception-caught
+                        LOGGER.debug("Transform and Fn::FindInMap error: %s", {str(e)})
                 elif k == "Ref":
                     if isinstance(v, str):
                         if v in params:
@@ -152,7 +167,11 @@ def _walk(self, item: Any, params: MutableMapping[str, Any], cfn: Any):
                                 return params[r]
                         obj[k] = r
                 else:
-                    obj[k] = self._walk(v, params, cfn)
+                    sub_value = self._walk(v, params, cfn)
+                    if sub_value is None:
+                        del obj[k]
+                    else:
+                        obj[k] = self._walk(v, params, cfn)
         elif isinstance(obj, list):
             for i, v in enumerate(obj):
                 obj[i] = self._walk(v, params, cfn)
@@ -195,45 +214,89 @@ def create(obj: Any) -> _ForEachValue:
         raise _TypeError(f"Unsupported value {obj!r}", obj)
 
     # pylint: disable=unused-argument
-    def value(self, cfn):
+    def value(
+        self, cfn, params: Optional[Mapping[str, Any]] = None, only_params: bool = False
+    ):
         return self._value
 
     @property
     def hash(self):
         return self._hash
 
 
+class _FnFindInMapDefaultValue(_ForEachValue):
+    def __init__(self, _hash: str, value: Any = None) -> None:
+        super().__init__(_hash, value)
+        if not isinstance(value, dict):
+            raise _TypeError(
+                "Fn::FindInMap parameter must be an object with key 'DefaultValue'",
+                value,
+            )
+        if len(value) != 1:
+            raise _ValueError(
+                "Fn::FindInMap parameter only supports 'DefaultValue'", value
+            )
+
+        for k, v in value.items():
+            if k != "DefaultValue":
+                raise _ValueError(
+                    "Fn::FindInMap parameter only supports 'DefaultValue'", value
+                )
+            self._value = _ForEachValue.create(v)
+
+    def value(
+        self, cfn, params: Optional[Mapping[str, Any]] = None, only_params: bool = False
+    ):
+        if params is None:
+            params = {}
+
+        return self._value.value(cfn, params, only_params)
+
+
 class _ForEachValueFnFindInMap(_ForEachValue):
     def __init__(self, _hash: str, obj: Any) -> None:
         super().__init__(_hash)
         if not isinstance(obj, list):
             raise _TypeError("Fn::FindInMap should be a list", obj)
 
-        if len(obj) != 3:
-            raise _ValueError("Fn::FindInMap requires a list of 3 values", obj)
+        if len(obj) not in [3, 4]:
+            raise _ValueError("Fn::FindInMap requires a list of 3 or 4 values", obj)
 
         self._map = [
             _ForEachValue.create(obj[0]),
             _ForEachValue.create(obj[1]),
             _ForEachValue.create(obj[2]),
         ]
+        if len(obj) == 4:
+            self._map.append(_FnFindInMapDefaultValue(get_hash(obj[3]), obj[3]))
 
         self._obj = obj
 
-    def value(self, cfn: Any) -> Any:
+    def value(
+        self,
+        cfn: Any,
+        params: Optional[Mapping[str, Any]] = None,
+        only_params: bool = False,
+    ) -> Any:
+        if params is None:
+            params = {}
         t_map = deepcopy(self._map)
         mapping = None
 
         try:
-            mapping = cfn.template.get("Mappings", {}).get(t_map[0].value(cfn))
+            mapping = cfn.template.get("Mappings", {}).get(
+                t_map[0].value(cfn, params, only_params)
+            )
         except Exception:  # pylint: disable=broad-exception-caught
             if len(cfn.template.get("Mappings", {}).keys()) == 1:
                 mapping = cfn.template.get("Mappings", {}).get(
                     list(cfn.template.get("Mappings", {}).keys())[0]
                 )
 
         try:
-            if mapping is None and isinstance(t_map[1].value(cfn), str):
+            if mapping is None and isinstance(
+                t_map[1].value(cfn, params, only_params), str
+            ):
                 for k, v in cfn.template.get("Mappings", {}).items():
                     if isinstance(v, dict):
                         if t_map[1].value(cfn) in v:
@@ -244,12 +307,14 @@ def value(self, cfn: Any) -> Any:
             pass
 
         try:
-            if mapping is None and isinstance(t_map[2].value(cfn), str):
+            if mapping is None and isinstance(
+                t_map[2].value(cfn, params, only_params), str
+            ):
                 for m1, mv1 in cfn.template.get("Mappings", {}).items():
                     if isinstance(mv1, dict):
                         for k1, kv1 in mv1.items():
                             if isinstance(kv1, dict):
-                                if t_map[2].value(cfn) in kv1:
+                                if t_map[2].value(cfn, params, only_params) in kv1:
                                     t_map[1] = _ForEachValue.create(k1)
                                     t_map[0] = _ForEachValue.create(m1)
                                     mapping = mv1
@@ -265,17 +330,23 @@ def value(self, cfn: Any) -> Any:
                     t_map[2].value(cfn)
                     for k, v in mapping.items():
                         if isinstance(v, dict):
-                            if t_map[2].value(cfn) in v:
+                            if t_map[2].value(cfn, params, only_params) in v:
                                 t_map[1] = _ForEachValue.create(k)
                 except _ResolveError:
                     pass
 
         if mapping:
             try:
-                return mapping.get(t_map[1].value(cfn)).get(t_map[2].value(cfn))
+                return mapping.get(t_map[1].value(cfn, params, only_params), {}).get(
+                    t_map[2].value(cfn, params, only_params)
+                )
             except _ResolveError as e:
+                if len(self._map) == 4:
+                    return self._map[3].value(cfn, params, only_params)
                 raise _ResolveError("Can't resolve Fn::FindInMap", self._obj) from e
 
+        if len(self._map) == 4:
+            return self._map[3].value(cfn, params, only_params)
         raise _ResolveError("Can't resolve Fn::FindInMap", self._obj)
 
 
@@ -289,12 +360,25 @@ def __init__(self, _hash: str, obj: Any) -> None:
         self._obj = obj
 
     # pylint: disable=too-many-return-statements
-    def value(self, cfn: Any) -> Any:
+    def value(
+        self,
+        cfn: Any,
+        params: Optional[Mapping[str, Any]] = None,
+        only_params: bool = False,
+    ) -> Any:
+        if params is None:
+            params = {}
         v = self._ref.value(cfn)
 
         if not isinstance(v, str):
             raise _ResolveError("Can't resolve Fn::Ref", self._obj)
 
+        if v in params:
+            return params[v]
+
+        if only_params:
+            raise _ResolveError("Can't resolve Fn::Ref", self._obj)
+
         region = cfn.regions[0]
         account_id = "123456789012"
         partition = "aws"
@@ -373,15 +457,15 @@ def values(
         if self._collection:
             for item in self._collection:
                 try:
-                    yield item.value(cfn)
+                    yield item.value(cfn, {}, False)
                 except _ResolveError:
                     v = "".join(random.choices(string.ascii_letters, k=_N))  # nosec
                     collection_cache[item.hash] = v
                     yield v
             return
         if self._fn:
             try:
-                values = self._fn.value(cfn)
+                values = self._fn.value(cfn, {}, False)
                 if values:
                     if isinstance(values, list):
                         for value in values:
diff --git a/src/cfnlint/template/transforms/transform.py b/src/cfnlint/template/transforms/transform.py
@@ -4,6 +4,7 @@
 """
 from __future__ import annotations
 
+import logging
 from typing import Any, Mapping
 
 from cfnlint.conditions import Conditions
@@ -12,6 +13,8 @@
 from cfnlint.template.transforms._protocols import Transformer
 from cfnlint.template.transforms._sam import sam
 
+LOGGER = logging.getLogger("cfnlint")
+
 
 class Transform:
     def __init__(self) -> None:
@@ -48,4 +51,5 @@ def transform(self, cfn: Any):
 
         cfn.graph = Graph(cfn)
         cfn.conditions = Conditions(cfn)
+        LOGGER.info("Transformed template: %s", cfn.template)
         return matches
diff --git a/test/fixtures/templates/good/parameters/used_transform_language_extension.json b/test/fixtures/templates/good/parameters/used_transform_language_extension.json
diff --git a/test/unit/module/template/transforms/test_language_extensions.py b/test/unit/module/template/transforms/test_language_extensions.py
diff --git a/test/unit/rules/parameters/test_used.py b/test/unit/rules/parameters/test_used.py
