Convert more rules to v1 (aws-cloudformation#3250)

* Convert more rules to v1
* Move cfn_path to context
* Switch to Path for tracking all things path

Author: kddejong

src/cfnlint/context/__init__.py

@@ -1,3 +1,3 @@
 __all__ = ["Context", "create_context_for_template"]
 
-from cfnlint.context.context import Context, create_context_for_template
+from cfnlint.context.context import Context, Path, create_context_for_template

src/cfnlint/context/context.py

@@ -38,6 +38,62 @@ def has_language_extensions_transform(self):
         return bool(lang_extensions_transform in self._transforms)
 
 
+@dataclass(frozen=True)
+class Path:
+    """
+    A `Path` keeps track of the different Path values
+    """
+
+    # path keeps track of the path as we move down the template
+    # Example: Resources, MyResource, Properties, Name, ...
+    path: Deque[str | int] = field(init=True, default_factory=deque)
+
+    # value_path is an override of the value if we got it from another place
+    # like a Parameter default value
+    # Example: Parameters, MyParameter, Default, ...
+    value_path: Deque[str | int] = field(init=True, default_factory=deque)
+
+    # cfn_path is a generic path used by cfn-lint to help make
+    # writing rules easier.  The resource name is replaced by the type
+    # lists are replaced with a *
+    # Example: Resources, AWS::S3::Bucket, Properties, Name, ...
+    # Example: Resources, *, Type
+    cfn_path: Deque[str] = field(init=True, default_factory=deque)
+
+    def descend(self, **kwargs):
+        """
+        Create a new Path by appending values
+        """
+        cls = self.__class__
+
+        for f in fields(Path):
+            if kwargs.get(f.name) is not None:
+                kwargs[f.name] = getattr(self, f.name) + deque([kwargs[f.name]])
+            else:
+                kwargs[f.name] = getattr(self, f.name)
+
+        return cls(**kwargs)
+
+    def evolve(self, **kwargs):
+        """
+        Create a new path without appending values
+        """
+        cls = self.__class__
+
+        for f in fields(Path):
+            kwargs.setdefault(f.name, getattr(self, f.name))
+
+        return cls(**kwargs)
+
+    @property
+    def path_string(self):
+        return "/".join(str(p) for p in self.path)
+
+    @property
+    def cfn_path_string(self):
+        return "/".join(self.cfn_path)
+
+
 @dataclass(frozen=True)
 class Context:
     """
@@ -61,12 +117,7 @@ class Context:
     # supported functions at this point in the template
     functions: Sequence[str] = field(init=True, default_factory=list)
 
-    # path keeps track of the path as we move down the template
-    # Example: Resources, MyResource, Properties, Name, ...
-    path: Deque[str] = field(init=True, default_factory=deque)
-    # value_path is an override of the value if we got it from another place
-    # like a Parameter default value
-    value_path: Deque[str] = field(init=True, default_factory=deque)
+    path: Path = field(init=True, default_factory=Path)
 
     # cfn-lint Template class
     parameters: Dict[str, "Parameter"] = field(init=True, default_factory=dict)
@@ -95,21 +146,14 @@ def __post_init__(self) -> None:
 
     def evolve(self, **kwargs) -> "Context":
         """
-        Create a new context merging together attributes
+        Create a new context without merging together attributes
         """
         cls = self.__class__
 
-        if "path" in kwargs and kwargs["path"] is not None:
-            path = self.path.copy()
-            path.append(kwargs["path"])
-            kwargs["path"] = path
-        else:
-            kwargs["path"] = self.path.copy()
-
-        if "ref_values" in kwargs and kwargs["ref_values"] is not None:
-            ref_values = self.ref_values.copy()
-            ref_values.update(kwargs["ref_values"])
-            kwargs["ref_values"] = ref_values
+        if "ref_values" in kwargs:
+            new_ref_values = self.ref_values.copy()
+            new_ref_values.update(kwargs["ref_values"])
+            kwargs["ref_values"] = new_ref_values
 
         for f in fields(Context):
             if f.init:
@@ -134,7 +178,9 @@ def ref_value(self, instance: str) -> Iterator[Tuple[str | List[str], "Context"]
         if instance in self.parameters:
             for v, path in self.parameters[instance].ref(self):
                 yield v, self.evolve(
-                    value_path=deque(["Parameters", instance]) + path,
+                    path=self.path.evolve(
+                        value_path=deque(["Parameters", instance]) + path
+                    ),
                     ref_values={instance: v},
                 )
             return
@@ -450,6 +496,6 @@ def create_context_for_template(cfn):
         transforms=transforms,
         mappings=mappings,
         regions=cfn.regions,
-        path=deque([]),
+        path=Path(),
         functions=["Fn::Transform"],
     )

src/cfnlint/graph.py

@@ -5,6 +5,8 @@
 SPDX-License-Identifier: MIT-0
 """
 
+from __future__ import annotations
+
 import logging
 import warnings
 from typing import Any, List
@@ -43,6 +45,14 @@ class GraphSettings:
     parameter: NodeSetting
     output: NodeSetting
 
+    def _pydot_string_convert(self, value: str | int) -> str | int:
+        if not isinstance(value, str):
+            return value
+        if ":" in value:
+            return f'"{value}"'
+
+        return value
+
     def subgraph_view(self, graph) -> networkx.MultiDiGraph:
         view = networkx.MultiDiGraph(name="template")
         resources: List[str] = [
@@ -52,17 +62,22 @@ def subgraph_view(self, graph) -> networkx.MultiDiGraph:
         # have to add quotes when outputing to dot
         for resource in resources:
             node = graph.nodes[resource]
-            node["type"] = f'"{node["type"]}"'
+            node["label"] = self._pydot_string_convert(node["label"])
+            node["type"] = self._pydot_string_convert(node["type"])
+            del node["resource_type"]
             view.add_node(resource, **node)
 
-        view.add_edges_from(
-            (n, nbr, key, d)
-            for n, nbrs in graph.adj.items()
-            if n in resources
-            for nbr, keydict in nbrs.items()
-            if nbr in resources
-            for key, d in keydict.items()
-        )
+        for edge_1, edge_2, edge_data in graph.edges(data=True):
+            if edge_1 in resources and edge_2 in resources:
+                edge_data["source_paths"] = [
+                    self._pydot_string_convert(p) for p in edge_data["source_paths"]
+                ]
+                view.add_edge(
+                    edge_1,
+                    edge_2,
+                    **edge_data,
+                )
+
         view.graph.update(graph.graph)
         return view
 
@@ -131,7 +146,7 @@ def _add_outputs(self, cfn: Any) -> None:
         # add all outputs in the template as nodes
         for output_id in cfn.template.get("Outputs", {}).keys():
             graph_label = str.format(f'"{output_id}"')
-            self._add_node(output_id, label=graph_label, settings=self.settings.output)
+            self._add_node(output_id, settings=self.settings.output, label=graph_label)
 
     def _add_resources(self, cfn: Any):
         # add all resources in the template as nodes
@@ -141,9 +156,12 @@ def _add_resources(self, cfn: Any):
             type_val = resourceVals.get("Type", "")
             if not isinstance(type_val, str):
                 continue
-            graph_label = str.format(f'"{resourceId}\\n<{type_val}>"')
+            graph_label = str.format(f"{resourceId}\\n<{type_val}>")
             self._add_node(
-                resourceId, label=graph_label, settings=self.settings.resource
+                resourceId,
+                settings=self.settings.resource,
+                label=graph_label,
+                resource_type=type_val,
             )
             target_ids = resourceVals.get("DependsOn", [])
             if isinstance(target_ids, (list, str)):
@@ -248,17 +266,14 @@ def _add_subs(self, cfn: Any) -> None:
                             source_id, sub_parameter, source_path, self.settings.ref
                         )
 
-    def _add_node(self, node_id, label, settings):
+    def _add_node(self, node_id, settings, **attr):
         if settings.node_type in ["Parameter", "Output"]:
             node_id = f"{settings.node_type}-{node_id}"
 
-        self.graph.add_node(
-            node_id,
-            label=label,
-            color=settings.color,
-            shape=settings.shape,
-            type=settings.node_type,
-        )
+        attr.setdefault("color", settings.color)
+        attr.setdefault("shape", settings.shape)
+        attr.setdefault("type", settings.node_type)
+        self.graph.add_node(node_id, **attr)
 
     def _add_edge(self, source_id, target_id, source_path, settings):
         self.graph.add_edge(

src/cfnlint/jsonschema/_filter.py

@@ -3,12 +3,18 @@
 SPDX-License-Identifier: MIT-0
 """
 
+from __future__ import annotations
+
 from dataclasses import dataclass, field, fields
-from typing import Any, Sequence, Tuple
+from typing import TYPE_CHECKING, Any, Sequence, Tuple
 
 from cfnlint.helpers import REGEX_DYN_REF, ToPy
 from cfnlint.jsonschema._utils import ensure_list
 
+if TYPE_CHECKING:
+    from cfnlint.jsonschema.protocols import Validator
+
+
 _all_types = ["array", "boolean", "integer", "number", "object", "string"]
 
 
@@ -53,7 +59,7 @@ class FunctionFilter:
         default=True,
     )
 
-    def _filter_schemas(self, schema, validator: Any) -> Tuple[Any, Any]:
+    def _filter_schemas(self, schema, validator: Validator) -> Tuple[Any, Any]:
         """
         Filter the schemas to only include the ones that are required
         """
@@ -64,8 +70,8 @@ def _filter_schemas(self, schema, validator: Any) -> Tuple[Any, Any]:
         # Example: Typically we want to remove (!Ref AWS::NoValue)
         # to count minItems, maxItems, required properties but if we
         # are in an If we need to be more strict
-        if len(validator.context.path) > 0:
-            if validator.context.path[-1] in ["Fn::If"]:
+        if len(validator.context.path.path) > 0:
+            if validator.context.path.path[-1] in ["Fn::If"]:
                 return schema, None
 
         standard_schema = {}
@@ -78,7 +84,7 @@ def _filter_schemas(self, schema, validator: Any) -> Tuple[Any, Any]:
 
         if self.add_cfn_lint_keyword and "$ref" not in standard_schema:
             standard_schema["cfnLint"] = ensure_list(standard_schema.get("cfnLint", []))
-            standard_schema["cfnLint"].append("/".join(validator.cfn_path))
+            standard_schema["cfnLint"].append("/".join(validator.context.path.cfn_path))
 
         # some times CloudFormation dumps to standard nested "json".
         # it will do by using {"type": "object"} with no properties

src/cfnlint/jsonschema/_keywords.py

@@ -46,7 +46,10 @@ def additionalProperties(
     if validator.is_type(aP, "object"):
         for extra in extras:
             yield from validator.descend(
-                instance[extra], aP, path=extra, property_path="*"
+                instance[extra],
+                aP,
+                path=extra,
+                property_path="*",
             )
     elif not aP and extras:
         if "patternProperties" in schema:
@@ -543,7 +546,7 @@ def type(
         yield ValidationError(f"{instance!r} is not of type {reprs}")
 
 
-def prefixItems(validator, prefixItems, instance, schema):
+def prefixItems(validator: Validator, prefixItems: Any, instance: Any, schema: Any):
     if not validator.is_type(instance, "array"):
         return
 

src/cfnlint/jsonschema/_resolvers_cfn.py

@@ -47,7 +47,11 @@ def find_in_map(validator: Validator, instance: Any) -> ResolutionResult:
             if "DefaultValue" in options:
                 for value, v, _ in validator.resolve_value(options["DefaultValue"]):
                     yield value, v.evolve(
-                        context=v.context.evolve(value_path=deque([4, "DefaultValue"]))
+                        context=v.context.evolve(
+                            path=v.context.path.evolve(
+                                value_path=deque([4, "DefaultValue"])
+                            )
+                        ),
                     ), None
                     default_value = value
 
@@ -66,7 +70,9 @@ def find_in_map(validator: Validator, instance: Any) -> ResolutionResult:
                         not (equal(map_name, each)) for each in mappings
                     ):
                         yield None, map_v.evolve(
-                            context=map_v.context.evolve(value_path=[0])
+                            context=map_v.context.evolve(
+                                path=map_v.context.path.evolve(value_path=deque([0])),
+                            ),
                         ), ValidationError(
                             f"{map_name!r} is not one of {mappings!r}", path=[0]
                         )
@@ -79,7 +85,9 @@ def find_in_map(validator: Validator, instance: Any) -> ResolutionResult:
                         not (equal(top_level_key, each)) for each in top_level_keys
                     ):
                         yield None, top_v.evolve(
-                            context=top_v.context.evolve(value_path=[1])
+                            context=top_v.context.evolve(
+                                path=top_v.context.path.evolve(value_path=deque([1])),
+                            ),
                         ), ValidationError(
                             f"{top_level_key!r} is not one of {top_level_keys!r}",
                             path=[0],
@@ -96,7 +104,11 @@ def find_in_map(validator: Validator, instance: Any) -> ResolutionResult:
                         for each in second_level_keys
                     ):
                         yield None, second_v.evolve(
-                            context=second_v.context.evolve(value_path=[2])
+                            context=second_v.context.evolve(
+                                path=second_v.context.path.evolve(
+                                    value_path=deque([2])
+                                ),
+                            ),
                         ), ValidationError(
                             f"{second_level_key!r} is not one of {second_level_keys!r}",
                             path=[0],
@@ -111,12 +123,16 @@ def find_in_map(validator: Validator, instance: Any) -> ResolutionResult:
                             value,
                             validator.evolve(
                                 context=validator.context.evolve(
-                                    value_path=[
-                                        "Mappings",
-                                        map_name,
-                                        top_level_key,
-                                        second_level_key,
-                                    ]
+                                    path=validator.context.path.evolve(
+                                        value_path=deque(
+                                            [
+                                                "Mappings",
+                                                map_name,
+                                                top_level_key,
+                                                second_level_key,
+                                            ]
+                                        )
+                                    )
                                 )
                             ),
                             None,
@@ -298,7 +314,9 @@ def if_(validator: Validator, instance: Any) -> ResolutionResult:
             yield (
                 value,
                 v.evolve(
-                    context=v.context.evolve(value_path=deque([i])),
+                    context=v.context.evolve(
+                        path=v.context.path.evolve(value_path=deque([i])),
+                    ),
                 ),
                 err,
             )

src/cfnlint/jsonschema/protocols.py

@@ -62,7 +62,7 @@ class Validator(Protocol):
     cfn: Template | None
     context: Context
     function_filter: FunctionFilter
-    cfn_path: deque[str | int]
+    cfn_path: deque[str]
 
     def __init__(
         self,