Upgrade more rules to v1

src/cfnlint/helpers.py

@@ -280,6 +280,21 @@
     "AWS::SSM::Parameter::Value<List<String>>",
 ]
 
+TEMPLATED_PROPERTY_CFN_PATHS = [
+    "Resources/AWS::ApiGateway::RestApi/Properties/BodyS3Location",
+    "Resources/AWS::Lambda::Function/Properties/Code",
+    "Resources/AWS::Lambda::LayerVersion/Properties/Content",
+    "Resources/AWS::ElasticBeanstalk::ApplicationVersion/Properties/SourceBundle",
+    "Resources/AWS::StepFunctions::StateMachine/Properties/DefinitionS3Location",
+    "Resources/AWS::AppSync::GraphQLSchema/Properties/DefinitionS3Location",
+    "Resources/AWS::AppSync::Resolver/Properties/RequestMappingTemplateS3Location",
+    "Resources/AWS::AppSync::Resolver/Properties/ResponseMappingTemplateS3Location",
+    "Resources/AWS::AppSync::FunctionConfiguration/Properties/RequestMappingTemplateS3Location",
+    "Resources/AWS::AppSync::FunctionConfiguration/Properties/ResponseMappingTemplateS3Location",
+    "Resources/AWS::CloudFormation::Stack/Properties/TemplateURL",
+    "Resources/AWS::CodeCommit::Repository/Properties/Code/S3",
+]
+
 VALID_PARAMETER_TYPES = VALID_PARAMETER_TYPES_SINGLE + VALID_PARAMETER_TYPES_LIST
 
 BOOLEAN_STRINGS_TRUE = frozenset(["true", "True"])

src/cfnlint/jsonschema/_keywords.py

@@ -227,7 +227,7 @@ def exclusiveMinimum(
 
     if t_instance <= m:
         yield ValidationError(
-            f"{instance!r} is less than or equal to " f"the minimum of {m!r}",
+            f"{instance!r} is less than or equal to the minimum of {m!r}",
         )
 
 

src/cfnlint/jsonschema/_resolvers_cfn.py

@@ -243,8 +243,7 @@ def _sub_string(validator: Validator, string: str) -> ResolutionResult:
     sub_regex = re.compile(r"(\${([^!].*?)})")
 
     def _replace(matchobj):
-        if matchobj.group(2) in validator.context.ref_values:
-            value = validator.context.ref_values[matchobj.group(2)]
+        for value, _ in validator.context.ref_value(matchobj.group(2).strip()):
             if not isinstance(value, (str, int, float, bool)):
                 raise ValueError(f"Parameter {matchobj.group(2)!r} has wrong type")
             return value
@@ -254,6 +253,7 @@ def _replace(matchobj):
         yield re.sub(sub_regex, _replace, string), validator.evolve(), None
     except ValueError:
         return
+        yield string, validator, ValidationError(f"Unable to resolve {string!r}")
 
 
 def sub(validator: Validator, instance: Any) -> ResolutionResult:

src/cfnlint/jsonschema/validators.py

@@ -144,10 +144,19 @@ def is_valid(self, instance: Any) -> bool:
         def resolve_value(self, instance: Any) -> ResolutionResult:
             if self.is_type(instance, "object"):
                 if len(instance) == 1:
-                    for k, v in instance.items():
-                        if k in self.fn_resolvers:
-                            for value, v, _ in self.resolve_value(v):
-                                yield from self.fn_resolvers[k](v, value)
+                    for key, value in instance.items():
+                        if key in self.fn_resolvers:
+                            for (
+                                resolved_value,
+                                validator,
+                                resolve_errs,
+                            ) in self.resolve_value(value):
+                                if resolve_errs:
+                                    continue
+                                yield from self.fn_resolvers[key](
+                                    validator, resolved_value  # type: ignore
+                                )
+
                             return
 
             # The return type is a Protocol and we are returning an instance
@@ -242,15 +251,12 @@ def descend(
             schema_path: str | None = None,
             property_path: str | int | None = None,
         ) -> ValidationResult:
-            cfn_path = self.cfn_path.copy()
-            if property_path:
-                cfn_path.append(property_path)
             for error in self.evolve(
                 schema=schema,
                 context=self.context.evolve(
                     path=path,
                 ),
-                cfn_path=cfn_path,
+                cfn_path=deque([property_path]) if property_path else deque([]),
             ).iter_errors(instance):
                 if path is not None:
                     error.path.appendleft(path)
@@ -273,7 +279,12 @@ def evolve(self, **kwargs) -> "Validator":
             cls = self.__class__
             for f in fields(Validator):
                 if f.init:
-                    kwargs.setdefault(f.name, getattr(self, f.name))
+                    if f.name == "cfn_path":
+                        cfn_path = self.cfn_path.copy()
+                        cfn_path.extend(kwargs.get(f.name, []))
+                        kwargs["cfn_path"] = cfn_path
+                    else:
+                        kwargs.setdefault(f.name, getattr(self, f.name))
 
             return cls(**kwargs)
 

src/cfnlint/rules/functions/Sub.py

@@ -78,23 +78,27 @@ def _clean_error(
         err.validator = self.fn.py
         return err
 
-    def _validate_string(self, validator: Validator, instance: Any) -> ValidationResult:
+    def _validate_string(
+        self, validator: Validator, key: str, instance: Any
+    ) -> ValidationResult:
         params = re.findall(REGEX_SUB_PARAMETERS, instance)
         validator = validator.evolve(
             context=validator.context.evolve(
                 functions=self._functions,
-            )
+            ),
+            cfn_path=deque([key]),
         )
         for param in params:
             param = param.strip()
             if "." in param:
                 for err in validator.descend(
-                    {"Fn::GetAtt": param},
-                    {"type": ["string"]},
+                    instance={"Fn::GetAtt": param},
+                    schema={"type": ["string"]},
+                    path=key,
                 ):
                     yield self._clean_error(err, {"Fn::GetAtt": param}, param)
             else:
-                for err in validator.descend({"Ref": param}, {"type": ["string"]}):
+                for err in validator.descend({"Ref": param}, {"type": ["string"]}, key):
                     yield self._clean_error(err, {"Ref": param}, param)
 
     def fn_sub(
@@ -105,7 +109,7 @@ def fn_sub(
         )
         yield from super().validate(validator, s, instance, schema)
 
-        _, value = self.key_value(instance)
+        key, value = self.key_value(instance)
         if validator.is_type(value, "array"):
             if len(value) != 2:
                 return
@@ -121,7 +125,7 @@ def fn_sub(
         else:
             return
 
-        errs = list(self._validate_string(validator_string, value))
+        errs = list(self._validate_string(validator_string, key, value))
         if errs:
             yield from iter(errs)
             return

src/cfnlint/rules/resources/events/RuleScheduleExpression.py

@@ -3,10 +3,11 @@
 SPDX-License-Identifier: MIT-0
 """
 
-from cfnlint.rules import CloudFormationLintRule, RuleMatch
+from cfnlint.jsonschema import ValidationError
+from cfnlint.rules.jsonschema.CfnLintKeyword import CfnLintKeyword
 
 
-class RuleScheduleExpression(CloudFormationLintRule):
+class RuleScheduleExpression(CfnLintKeyword):
     """Validate AWS Events Schedule expression format"""
 
     id = "E3027"
@@ -17,121 +18,92 @@ class RuleScheduleExpression(CloudFormationLintRule):
 
     def initialize(self, cfn):
         """Initialize the rule"""
-        self.resource_property_types = ["AWS::Events::Rule"]
+        self.__init__(["Resources/AWS::Events::Rule/Properties/ScheduleExpression"])
 
-    def check_rate(self, value, path):
+    def validate(self, validator, keywords, instance, schema):
+        # Value is either "cron()" or "rate()"
+        if not validator.is_type(instance, "string"):
+            return
+
+        if instance.startswith("rate(") and instance.endswith(")"):
+            yield from self._check_rate(instance)
+        elif instance.startswith("cron(") and instance.endswith(")"):
+            yield from self._check_cron(instance)
+        else:
+            yield ValidationError(f"{instance!r} has to be either 'cron()' or 'rate()'")
+
+    def _check_rate(self, value):
         """Check Rate configuration"""
-        matches = []
         # Extract the expression from rate(XXX)
         rate_expression = value[value.find("(") + 1 : value.find(")")]
 
         if not rate_expression:
-            return [RuleMatch(path, "Rate value of ScheduleExpression cannot be empty")]
+            yield ValidationError(
+                message=f"{rate_expression!r} is not of type 'string'",
+            )
+            return
 
         # Rate format: rate(Value Unit)
         items = rate_expression.split(" ")
 
         if len(items) != 2:
-            message = (
-                "Rate expression must contain 2 elements "
-                "(Value Unit), rate contains {} elements"
+            yield ValidationError(
+                f"{rate_expression!r} has to be of format rate(Value Unit)"
             )
-            matches.append(RuleMatch(path, message.format(len(items))))
-            return [RuleMatch(path, message.format(len(items)))]
+            return
 
         # Check the Value
         if not items[0].isdigit():
-            message = "Rate Value ({}) should be of type Integer."
             extra_args = {
                 "actual_type": type(items[0]).__name__,
                 "expected_type": int.__name__,
             }
-            return [RuleMatch(path, message.format(items[0]), **extra_args)]
+            yield ValidationError(
+                message=f"{items[0]!r} is not of type 'integer'",
+                extra_args=extra_args,
+            )
+            return
 
         if float(items[0]) <= 0:
-            return [
-                RuleMatch(path, f"Rate Value {items[0]!r} should be greater than 0.")
-            ]
+            yield ValidationError(f"{items[0]!r} is less than the minimum of {0!r}")
 
         if float(items[0]) <= 1:
             valid_periods = ["minute", "hour", "day"]
         elif float(items[0]) > 1:
             valid_periods = ["minutes", "hours", "days"]
         # Check the Unit
         if items[1] not in valid_periods:
-            return [
-                RuleMatch(
-                    path, f"Rate Unit {items[1]!r} should be one of {valid_periods!r}."
-                )
-            ]
-
-        return []
+            yield ValidationError(f"{items[1]!r} is not one of {valid_periods!r}")
 
-    def check_cron(self, value, path):
+    def _check_cron(self, value):
         """Check Cron configuration"""
-        matches = []
         # Extract the expression from cron(XXX)
         cron_expression = value[value.find("(") + 1 : value.find(")")]
 
         if not cron_expression:
-            matches.append(
-                RuleMatch(path, "Cron value of ScheduleExpression cannot be empty")
+            yield ValidationError(
+                message=f"{cron_expression!r} is not of type 'string'",
             )
+            return
         else:
             # Rate format: cron(Minutes Hours Day-of-month Month Day-of-week Year)
             items = cron_expression.split(" ")
 
             if len(items) != 6:
-                message = (
-                    "Cron expression must contain 6 elements (Minutes Hours"
-                    " Day-of-month Month Day-of-week Year), cron contains {} elements"
+                yield ValidationError(
+                    message=(
+                        f"{items[0]!r} is not of length {6!r}. "
+                        "(Minutes Hours Day-of-month Month Day-of-week Year)"
+                    ),
                 )
-                matches.append(RuleMatch(path, message.format(len(items))))
-                return matches
+                return
 
             _, _, day_of_month, _, day_of_week, _ = cron_expression.split(" ")
             if day_of_month != "?" and day_of_week != "?":
-                matches.append(
-                    RuleMatch(
-                        path,
-                        (
-                            "Don't specify the Day-of-month and Day-of-week fields in"
-                            " the same cron expression"
-                        ),
-                    )
+                yield ValidationError(
+                    message=(
+                        f"{cron_expression[0]!r} specifies both "
+                        "Day-of-month and Day-of-week. "
+                        "(Minutes Hours Day-of-month Month Day-of-week Year)"
+                    ),
                 )
-
-        return matches
-
-    def check_value(self, value, path):
-        """Count ScheduledExpression value"""
-        matches = []
-
-        # Value is either "cron()" or "rate()"
-        if value.startswith("rate(") and value.endswith(")"):
-            matches.extend(self.check_rate(value, path))
-        elif value.startswith("cron(") and value.endswith(")"):
-            matches.extend(self.check_cron(value, path))
-        else:
-            message = (
-                "Invalid ScheduledExpression specified ({}). Value has to be either"
-                " cron() or rate()"
-            )
-            matches.append(RuleMatch(path, message.format(value)))
-
-        return matches
-
-    def match_resource_properties(self, properties, _, path, cfn):
-        """Check CloudFormation Properties"""
-        matches = []
-
-        matches.extend(
-            cfn.check_value(
-                obj=properties,
-                key="ScheduleExpression",
-                path=path[:],
-                check_value=self.check_value,
-            )
-        )
-
-        return matches

src/cfnlint/rules/resources/properties/Properties.py

@@ -91,7 +91,9 @@ def cfnresourceproperties(self, validator: Validator, _, instance: Any, schema):
                             context=validator.context.evolve(
                                 regions=[region], path="Properties"
                             ),
-                            cfn_path=deque(["Resources", t, "Properties"]),
+                        )
+                        region_validator.cfn_path = deque(
+                            ["Resources", t, "Properties"]
                         )
                         for err in self.validate(
                             region_validator, t, properties, schema.json_schema
@@ -109,8 +111,8 @@ def cfnresourceproperties(self, validator: Validator, _, instance: Any, schema):
                 context=validator.context.evolve(
                     regions=cached_regions, path="Properties"
                 ),
-                cfn_path=deque(["Resources", t, "Properties"]),
             )
+            region_validator.cfn_path = deque(["Resources", t, "Properties"])
             for err in self.validate(
                 region_validator, t, properties, schema.json_schema
             ):