Add mpr validations to rpdk with testing

src/rpdk/core/data_loaders.py

@@ -21,6 +21,7 @@
 
 TIMEOUT_IN_SECONDS = 10
 STDIN_NAME = "<stdin>"
+MAX_CONFIGURATION_SCHEMA_LENGTH = 60 * 1024  # 60 KiB
 
 
 def resource_stream(package_name, resource_name, encoding="utf-8"):
@@ -152,6 +153,12 @@ def load_resource_spec(resource_spec_file):  # pylint: disable=R # noqa: C901
         LOG.debug("Resource spec decode failed", exc_info=True)
         raise SpecValidationError(str(e)) from e
 
+    # check TypeConfiguration schema size
+    if len(json.dumps(resource_spec).encode("utf-8")) > MAX_CONFIGURATION_SCHEMA_LENGTH:
+        raise SpecValidationError(
+            "TypeConfiguration schema exceeds maximum length of 60 KiB"
+        )
+
     validator = make_resource_validator()
     additional_properties_validator = (
         make_resource_validator_with_additional_properties_check()

src/rpdk/core/project.py

@@ -95,6 +95,10 @@
 # https://docs.aws.amazon.com/IAM/latest/APIReference/API_CreateRole.html
 MIN_ROLE_TIMEOUT_SECONDS = 3600  # 1 hour
 MAX_ROLE_TIMEOUT_SECONDS = 43200  # 12 hours
+MAX_RPDK_CONFIG_LENGTH = 10 * 1024  # 10 KiB
+MAX_CONFIGURATION_SCHEMA_LENGTH = 60 * 1024  # 60 KiB
+
+PROTOCOL_VERSION_VALUES = frozenset({"1.0.0", "2.0.0"})
 
 CFN_METADATA_FILENAME = ".cfn_metadata.json"
 
@@ -282,6 +286,25 @@ def load_settings(self):
                 f"Project file '{self.settings_path}' is invalid", e
             )
 
+        # check size of RPDK config
+        if len(json.dumps(raw_settings).encode("utf-8")) > MAX_RPDK_CONFIG_LENGTH:
+            raise InvalidProjectError(
+                f"Project file '{self.settings_path}' exceeds maximum length of 10 KiB."
+            )
+        # validate protocol version, if specified
+        if "settings" in raw_settings and "protocolVersion" in raw_settings["settings"]:
+            protocol_version = raw_settings["settings"]["protocolVersion"]
+            if protocol_version not in PROTOCOL_VERSION_VALUES:
+                raise InvalidProjectError(
+                    f"Invalid 'protocolVersion' settings in '{self.settings_path}"
+                )
+        else:
+            LOG.warning(
+                "No protovolVersion found: this will default to version 1.0.0 during registration. "
+                "Please consider upgrading to CFN-CLI 2.0 following the guide: "
+                "https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/what-is-cloudformation-cli.html"
+            )
+
         # backward compatible
         if "artifact_type" not in raw_settings:
             raw_settings["artifact_type"] = ARTIFACT_TYPE_RESOURCE
@@ -844,13 +867,15 @@ def generate_docs(self):
         target_names = (
             self.target_info.keys()
             if self.target_info
-            else {
-                target_name
-                for handler in self.schema.get("handlers", {}).values()
-                for target_name in handler.get("targetNames", [])
-            }
-            if self.artifact_type == ARTIFACT_TYPE_HOOK
-            else []
+            else (
+                {
+                    target_name
+                    for handler in self.schema.get("handlers", {}).values()
+                    for target_name in handler.get("targetNames", [])
+                }
+                if self.artifact_type == ARTIFACT_TYPE_HOOK
+                else []
+            )
         )
 
         LOG.debug("Removing generated docs: %s", docs_path)

src/rpdk/core/validate.py

@@ -9,6 +9,7 @@
 LOG = logging.getLogger(__name__)
 
 
+# validations for cfn validate are done in both project.py and data_loaders.py
 def validate(_args):
     project = Project()
     project.load()

tests/test_project.py

@@ -173,6 +173,36 @@ def test_load_settings_invalid_hooks_settings(project):
     mock_open.assert_called_once_with("r", encoding="utf-8")
 
 
+def test_load_settings_invalid_protocol_version(project):
+    with patch_settings(
+        project, '{"settings": {"protocolVersion": "3.0.0"}}'
+    ) as mock_open:
+        with pytest.raises(InvalidProjectError):
+            project.load_settings()
+    mock_open.assert_called_once_with("r", encoding="utf-8")
+
+
+def test_load_settings_missing_protocol_version(project):
+    plugin = object()
+    data = json.dumps(
+        {"artifact_type": "MODULE", "typeName": MODULE_TYPE_NAME, "settings": {}}
+    )
+    patch_load = patch(
+        "rpdk.core.project.load_plugin", autospec=True, return_value=plugin
+    )
+
+    with patch_settings(project, data) as mock_open, patch_load as mock_load:
+        project.load_settings()
+    mock_open.assert_called_once_with("r", encoding="utf-8")
+    mock_load.assert_not_called()
+    assert project.type_info == ("AWS", "Color", "Red", "MODULE")
+    assert project.type_name == MODULE_TYPE_NAME
+    assert project.language is None
+    assert project.artifact_type == ARTIFACT_TYPE_MODULE
+    assert project._plugin is None
+    assert project.settings == {}
+
+
 def test_load_settings_valid_json_for_resource(project):
     plugin = object()
     data = json.dumps(
@@ -292,6 +322,57 @@ def test_load_settings_valid_json_for_hook(project):
     assert project.settings == {}
 
 
+def test_load_settings_valid_protocol_version(project):
+    plugin = object()
+    data = json.dumps(
+        {
+            "artifact_type": "MODULE",
+            "typeName": MODULE_TYPE_NAME,
+            "settings": {"protocolVersion": "2.0.0"},
+        }
+    )
+    patch_load = patch(
+        "rpdk.core.project.load_plugin", autospec=True, return_value=plugin
+    )
+
+    with patch_settings(project, data) as mock_open, patch_load as mock_load:
+        project.load_settings()
+
+    mock_open.assert_called_once_with("r", encoding="utf-8")
+    mock_load.assert_not_called()
+    assert project.type_info == ("AWS", "Color", "Red", "MODULE")
+    assert project.type_name == MODULE_TYPE_NAME
+    assert project.language is None
+    assert project.artifact_type == ARTIFACT_TYPE_MODULE
+    assert project._plugin is None
+    assert project.settings == {"protocolVersion": "2.0.0"}
+
+
+def test_load_settings_missing_settings(project):
+    plugin = object()
+    data = json.dumps(
+        {
+            "artifact_type": "MODULE",
+            "typeName": MODULE_TYPE_NAME,
+        }
+    )
+    patch_load = patch(
+        "rpdk.core.project.load_plugin", autospec=True, return_value=plugin
+    )
+
+    with patch_settings(project, data) as mock_open, patch_load as mock_load:
+        project.load_settings()
+
+    mock_open.assert_called_once_with("r", encoding="utf-8")
+    mock_load.assert_not_called()
+    assert project.type_info == ("AWS", "Color", "Red", "MODULE")
+    assert project.type_name == MODULE_TYPE_NAME
+    assert project.language is None
+    assert project.artifact_type == ARTIFACT_TYPE_MODULE
+    assert project._plugin is None
+    assert project.settings == {}
+
+
 def test_load_schema_settings_not_loaded(project):
     with pytest.raises(InternalError):
         project.load_schema()