Identifier Utils Module (#126)

Author: ammokhov

python/rpdk/python/templates/handlers.py

@@ -9,6 +9,7 @@
     Resource,
     SessionProxy,
     exceptions,
+    identifier_utils,
 )
 
 from .models import ResourceHandlerRequest, ResourceModel
@@ -36,6 +37,19 @@ def create_handler(
 
     # Example:
     try:
+
+        # primary identifier from example
+        primary_identifier = None
+
+        # setting up random primary identifier compliant with cfn standard
+        if primary_identifier is None:
+            primary_identifier = identifier_utils.generate_resource_identifier(
+                stack_id_or_name=request.stackId,
+                logical_resource_id=request.logicalResourceIdentifier,
+                client_request_token=request.clientRequestToken,
+                max_length=255
+                )
+
         if isinstance(session, SessionProxy):
             client = session.client("s3")
         # Setting Status to success will signal to cfn that the operation is complete

src/cloudformation_cli_python_lib/identifier_utils.py

@@ -0,0 +1,59 @@
+import hashlib
+import re
+from typing import Optional
+
+STACK_ARN_PATTERN = "^[a-z0-9-:]*stack/[-a-z0-9A-Z]*/[-a-z0-9A-Z]*"
+
+MIN_PHYSICAL_RESOURCE_ID_LENGTH = 15
+MIN_PREFERRED_LENGTH = 17
+HASH_LENGTH = 12
+
+
+def _get_hash(client_request_token: str) -> str:
+    return hashlib.sha1(str.encode(client_request_token)).hexdigest()  # nosec
+
+
+def generate_resource_identifier(
+    stack_id_or_name: Optional[str],
+    logical_resource_id: Optional[str],
+    client_request_token: Optional[str],
+    max_length: int,
+) -> str:
+    if max_length < MIN_PHYSICAL_RESOURCE_ID_LENGTH:
+        raise Exception(
+            f"Cannot generate resource IDs shorter than\
+               {MIN_PHYSICAL_RESOURCE_ID_LENGTH} characters."
+        )
+
+    strinct_logical_resource_id: str = logical_resource_id or ""
+    strinct_client_request_token: str = client_request_token or ""
+
+    stack_name: str = stack_id_or_name or ""
+
+    pattern = re.compile(STACK_ARN_PATTERN)
+
+    if pattern.match(stack_name):
+        stack_name = stack_name.split("/")[1]
+
+    separate: bool = max_length > MIN_PREFERRED_LENGTH
+
+    clean_stack_name: str = stack_name.replace("^-+", "", 1).replace("--", "-")
+    free_chars: int = max_length - (HASH_LENGTH + 1) - (1 if separate else 0)
+
+    chars_for_resource_name: int = min(
+        free_chars // 2, len(strinct_logical_resource_id)
+    )
+    chars_for_stack_name: int = min(
+        free_chars - chars_for_resource_name, len(clean_stack_name)
+    )
+
+    hash_value: str = _get_hash(strinct_client_request_token)
+
+    identifier: str = (
+        clean_stack_name[:chars_for_stack_name]
+        + ("-" if separate else "")
+        + strinct_logical_resource_id[:chars_for_resource_name]
+        + "-"
+        + hash_value[:HASH_LENGTH]
+    )
+    return identifier

src/cloudformation_cli_python_lib/interface.py

@@ -138,3 +138,4 @@ class BaseResourceHandlerRequest:
     nextToken: Optional[str]
     region: Optional[str]
     awsPartition: Optional[str]
+    stackId: Optional[str]

src/cloudformation_cli_python_lib/resource.py

@@ -163,6 +163,7 @@ def _cast_resource_request(
                 systemTags=request.requestData.systemTags,
                 awsAccountId=request.awsAccountId,
                 logicalResourceIdentifier=request.requestData.logicalResourceId,
+                stackId=request.stackId,
                 region=request.region,
             ).to_modelled(self._model_cls)
         except Exception as e:  # pylint: disable=broad-except

src/cloudformation_cli_python_lib/utils.py

@@ -124,6 +124,7 @@ class UnmodelledRequest:
     awsAccountId: Optional[str] = None
     logicalResourceIdentifier: Optional[str] = None
     nextToken: Optional[str] = None
+    stackId: Optional[str] = None
     region: Optional[str] = None
 
     def to_modelled(self, model_cls: Type[BaseModel]) -> BaseResourceHandlerRequest:
@@ -138,6 +139,7 @@ def to_modelled(self, model_cls: Type[BaseModel]) -> BaseResourceHandlerRequest:
             awsAccountId=self.awsAccountId,
             logicalResourceIdentifier=self.logicalResourceIdentifier,
             nextToken=self.nextToken,
+            stackId=self.stackId,
             region=self.region,
             awsPartition=self.get_partition(self.region),
         )

tests/lib/identifier_utils_test.py

@@ -0,0 +1,57 @@
+import pytest
+from cloudformation_cli_python_lib.identifier_utils import generate_resource_identifier
+
+
+def test_generated_name_with_stack_name_and_long_logical_id():
+    result: str = generate_resource_identifier(
+        stack_id_or_name="my-custom-stack-name",
+        logical_resource_id="my-long-long-long-long-long-logical-id-name",
+        client_request_token="123456789",
+        max_length=36,
+    )
+    assert len(result) == 36
+    assert result.startswith("my-custom-s-my-long-lon-")
+
+
+def test_generated_name_with_stack_id_and_long_logical_id():
+    result: str = generate_resource_identifier(
+        stack_id_or_name="arn:aws:cloudformation:us-east-1:123456789012:stack/my-stack-name/084c0bd1-082b-11eb-afdc-0a2fadfa68a5",  # noqa: B950 # pylint: disable=line-too-long
+        logical_resource_id="my-long-long-long-long-long-logical-id-name",
+        client_request_token="123456789",
+        max_length=36,
+    )
+    assert len(result) == 36
+    assert result.startswith("my-stack-na-my-long-lon-")
+
+
+def test_generated_name_with_short_stack_name_and_short_logical_id():
+    result: str = generate_resource_identifier(
+        stack_id_or_name="abc",
+        logical_resource_id="abc",
+        client_request_token="123456789",
+        max_length=255,
+    )
+    assert len(result) == 20  # "abc" + "-" + "abc" + "-" + 12 char hash
+    assert result.startswith("abc-abc-")
+
+
+def test_generated_name_with_max_len_shorter_than_preferred():
+    result: str = generate_resource_identifier(
+        stack_id_or_name="abc",
+        logical_resource_id="abc",
+        client_request_token="123456789",
+        max_length=16,
+    )
+    assert len(result) == 16
+    assert result.startswith("aba-f7c3bc1d808e")
+
+
+def test_generated_name_with_invalid_len():
+    with pytest.raises(Exception) as excinfo:
+        generate_resource_identifier(
+            stack_id_or_name="my-stack-name",
+            logical_resource_id="my-logical-id",
+            client_request_token="123456789",
+            max_length=13,
+        )
+    assert "Cannot generate resource IDs shorter than" in str(excinfo.value)