Added SGR validation to cfn validate

setup.py

@@ -56,6 +56,7 @@ def find_version(*file_paths):
         "cfn_flip>=1.2.3",
         "nested-lookup",
         "botocore>=1.31.17",
+        "resource-schema-guard-rail>=0.0.12",
     ],
     entry_points={
         "console_scripts": ["cfn-cli = rpdk.core.cli:main", "cfn = rpdk.core.cli:main"]

src/rpdk/core/data_loaders.py

@@ -3,6 +3,7 @@
 import os
 import re
 import shutil
+from copy import deepcopy
 from io import TextIOWrapper
 from pathlib import Path
 
@@ -12,6 +13,9 @@
 from jsonschema.exceptions import RefResolutionError, ValidationError
 from nested_lookup import nested_lookup
 
+from rpdk.guard_rail.core.data_types import Stateful, Stateless
+from rpdk.guard_rail.core.runner import exec_compliance
+
 from .exceptions import InternalError, SpecValidationError
 from .jsonutils.flattener import JsonSchemaFlattener
 from .jsonutils.inliner import RefInliner
@@ -144,10 +148,36 @@ def get_file_base_uri(file):
     return path.resolve().as_uri()
 
 
-def load_resource_spec(resource_spec_file):  # pylint: disable=R # noqa: C901
+def sgr_stateless_eval(schema):
+    schema_copy = deepcopy(schema)
+    result = exec_compliance(Stateless([schema_copy], []))[0]
+    result.display()
+
+
+def sgr_stateful_eval(schema, original_schema):
+    result = exec_compliance(
+        Stateful(current_schema=schema, previous_schema=original_schema, rules=[])
+    )[0]
+    result.display()
+
+
+def load_resource_spec(
+    resource_spec_file, original_schema_raw=None
+):  # pylint: disable=R # noqa: C901
     """Load a resource provider definition from a file, and validate it."""
+    original_resource_spec = None
     try:
         resource_spec = json.load(resource_spec_file)
+        if original_schema_raw:
+            print(
+                "Type Exists in CloudFormation Registry. Evaluating Resource Schema Backward Compatibility Compliance"
+            )
+            original_resource_spec = json.loads(original_schema_raw)
+            sgr_stateful_eval(resource_spec, original_resource_spec)
+
+        print("Evaluating Resource Schema Compliance")
+        sgr_stateless_eval(resource_spec)
+
     except ValueError as e:
         LOG.debug("Resource spec decode failed", exc_info=True)
         raise SpecValidationError(str(e)) from e
@@ -156,6 +186,7 @@ def load_resource_spec(resource_spec_file):  # pylint: disable=R # noqa: C901
     additional_properties_validator = (
         make_resource_validator_with_additional_properties_check()
     )
+
     try:
         validator.validate(resource_spec)
     except ValidationError as e:

src/rpdk/core/generate.py

@@ -11,7 +11,7 @@
 
 def generate(args):
     project = Project()
-    project.load()
+    project.load(args)
     project.generate(
         args.endpoint_url,
         args.region,

src/rpdk/core/project.py

@@ -275,7 +275,7 @@ def validate_and_load_resource_settings(self, raw_settings):
         self.entrypoint = raw_settings["entrypoint"]
         self.test_entrypoint = raw_settings["testEntrypoint"]
         self.executable_entrypoint = raw_settings.get("executableEntrypoint")
-        self._plugin = load_plugin(raw_settings["language"])
+        # self._plugin = load_plugin(raw_settings["language"])
         self.settings = raw_settings.get("settings", {})
 
     def _write_example_schema(self):
@@ -421,14 +421,36 @@ def load_hook_schema(self):
         with self.schema_path.open("r", encoding="utf-8") as f:
             self.schema = load_hook_spec(f)
 
-    def load_schema(self):
+    def load_schema(self, args=None):
         if not self.type_info:
             msg = "Internal error (Must load settings first)"
             LOG.critical(msg)
             raise InternalError(msg)
 
+        type_name = f"{self.type_info[0]}::{self.type_info[1]}::{self.type_info[2]}"
+
+        original_schema = self._retrieve_global_schema(type_name, args)
+
         with self.schema_path.open("r", encoding="utf-8") as f:
-            self.schema = load_resource_spec(f)
+            self.schema = load_resource_spec(f, original_schema)
+
+    def _retrieve_global_schema(self, type_name, args):
+        try:
+            session = create_sdk_session(args.region, args.profile)
+            cfn_client = session.client(
+                "cloudformation", endpoint_url=args.endpoint_url
+            )
+            _response = cfn_client.describe_type(Type="RESOURCE", TypeName=type_name)
+            return _response["Schema"]
+        except ClientError as e:
+            LOG.warning(
+                f"Attempted to retrieve latest schema from registry for ResourceType {type_name}"
+            )
+            LOG.warning(str(e))
+            return None
+        except Exception as ex:
+            print(str(ex))
+            return None
 
     def load_configuration_schema(self):
         if not self.schema:
@@ -546,7 +568,7 @@ def generate(
 
         self._plugin.generate(self)
 
-    def load(self):
+    def load(self, args=None):
         try:
             self.load_settings()
         except FileNotFoundError as e:
@@ -561,12 +583,12 @@ def load(self):
         elif self.artifact_type == ARTIFACT_TYPE_HOOK:
             self._load_hooks_project()
         else:
-            self._load_resources_project()
+            self._load_resources_project(args)
 
-    def _load_resources_project(self):
+    def _load_resources_project(self, args):
         LOG.info("Validating your resource specification...")
         try:
-            self.load_schema()
+            self.load_schema(args)
             self.load_configuration_schema()
             LOG.warning("Resource schema is valid.")
         except FileNotFoundError as e:

src/rpdk/core/submit.py

@@ -11,7 +11,7 @@
 
 def submit(args):
     project = Project()
-    project.load()
+    project.load(args)
     # Use CLI override opposed to config file if use-docker or no-docker switch used
     if args.use_docker or args.no_docker:
         project.settings["use_docker"] = args.use_docker

src/rpdk/core/validate.py

@@ -11,9 +11,14 @@
 
 def validate(_args):
     project = Project()
-    project.load()
+    project.load(_args)
 
 
 def setup_subparser(subparsers, parents):
+
     parser = subparsers.add_parser("validate", description=__doc__, parents=parents)
+    parser.add_argument("--endpoint-url", help="CloudFormation endpoint to use.")
+    parser.add_argument("--region", help="AWS Region to submit the resource type.")
+    parser.add_argument("--profile", help="AWS profile to use.")
+
     parser.set_defaults(command=validate)