localstack · Morijarti · Oct 10, 2023 · Oct 6, 2023 · Oct 6, 2023 · Oct 6, 2023
diff --git a/localstack/services/cloudformation/resource_provider.py b/localstack/services/cloudformation/resource_provider.py
@@ -67,6 +67,8 @@
     "AWS::CloudWatch::Alarm": "ResourceProvider",
     "AWS::CloudWatch::CompositeAlarm": "ResourcePro
8000
vider",
     # "AWS::ECR::Repository": "ResourceProvider",  # FIXME: add full -ext provider & override logic for -ext
+    "AWS::KMS::Key": "ResourceProvider",
+    "AWS::KMS::Alias": "ResourceProvider",
 }
 
 

diff --git a/localstack/services/kms/resource_providers/__init__.py b/localstack/services/kms/resource_providers/__init__.py
diff --git a/localstack/services/kms/resource_providers/aws_kms_alias.py b/localstack/services/kms/resource_providers/aws_kms_alias.py
@@ -0,0 +1,106 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class KMSAliasProperties(TypedDict):
+    AliasName: Optional[str]
+    TargetKeyId: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class KMSAliasProvider(ResourceProvider[KMSAliasProperties]):
+
+    TYPE = "AWS::KMS::Alias"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[KMSAliasProperties],
+    ) -> ProgressEvent[KMSAliasProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/AliasName
+
+        Required properties:
+          - AliasName
+          - TargetKeyId
+
+        Create-only properties:
+          - /properties/AliasName
+
+
+
+        IAM permissions required:
+          - kms:CreateAlias
+
+        """
+        model = request.desired_state
+        kms = request.aws_client_factory.kms
+
+        kms.create_alias(AliasName=model["AliasName"], TargetKeyId=model["TargetKeyId"])
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def read(
+        self,
+        request: ResourceRequest[KMSAliasProperties],
+    ) -> ProgressEvent[KMSAliasProperties]:
+        """
+        Fetch resource information
+
+        IAM permissions required:
+          - kms:ListAliases
+        """
+        raise NotImplementedError
+    def delete(
+        self,
+        request: ResourceRequest[KMSAliasProperties],
+    ) -> ProgressEvent[KMSAliasProperties]:
+        """
+        Delete a resource
+
+        IAM permissions required:
+          - kms:DeleteAlias
+        """
+        model = request.desired_state
+        kms = request.aws_client_factory.kms
+
+        kms.delete_alias(AliasName=model["AliasName"])
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def update(
+        self,
+        request: ResourceRequest[KMSAliasProperties],
+    ) -> ProgressEvent[KMSAliasProperties]:
+        """
+        Update a resource
+
+        IAM permissions required:
+          - kms:UpdateAlias
+        """
+        raise NotImplementedError
diff --git a/localstack/services/kms/resource_providers/aws_kms_alias.schema.json b/localstack/services/kms/resource_providers/aws_kms_alias.schema.json
@@ -0,0 +1,61 @@
+{
+  "typeName": "AWS::KMS::Alias",
+  "description": "The AWS::KMS::Alias resource specifies a display name for an AWS KMS key in AWS Key Management Service (AWS KMS). You can use an alias to identify an AWS KMS key in cryptographic operations.",
+  "sourceUrl": "https://github.com/aws-cloudformation/aws-cloudformation-rpdk.git",
+  "properties": {
+    "AliasName&quo
8000
t;: {
+      "description": "Specifies the alias name. This value must begin with alias/ followed by a name, such as alias/ExampleAlias. The alias name cannot begin with alias/aws/. The alias/aws/ prefix is reserved for AWS managed keys.",
+      "type": "string",
+      "pattern": "^(alias/)[a-zA-Z0-9:/_-]+$",
+      "minLength": 1,
+      "maxLength": 256
+    },
+    "TargetKeyId": {
+      "description": "Identifies the AWS KMS key to which the alias refers. Specify the key ID or the Amazon Resource Name (ARN) of the AWS KMS key. You cannot specify another alias. For help finding the key ID and ARN, see Finding the Key ID and ARN in the AWS Key Management Service Developer Guide.",
+      "type": "string",
+      "minLength": 1,
+      "maxLength": 256
+    }
+  },
+  "additionalProperties": false,
+  "required": [
+    "AliasName",
+    "TargetKeyId"
+  ],
+  "createOnlyProperties": [
+    "/properties/AliasName"
+  ],
+  "primaryIdentifier": [
+    "/properties/AliasName"
+  ],
+  "tagging": {
+    "taggable": false
+  },
+  "handlers": {
+    "create": {
+      "permissions": [
+        "kms:CreateAlias"
+      ]
+    },
+    "read": {
+      "permissions": [
+        "kms:ListAliases"
+      ]
+    },
+    "update": {
+      "permissions": [
+        "kms:UpdateAlias"
+      ]
+    },
+    "delete": {
+      "permissions": [
+        "kms:DeleteAlias"
+      ]
+    },
+    "list": {
+      "permissions": [
+        "kms:ListAliases"
+      ]
+    }
+  }
+}
diff --git a/localstack/services/kms/resource_providers/aws_kms_alias_plugin.py b/localstack/services/kms/resource_providers/aws_kms_alias_plugin.py
@@ -0,0 +1,18 @@
+from typing import Optional, Type
+
+from localstack.services.cloudformation.resource_provider import (
+    CloudFormationResourceProviderPlugin,
+    ResourceProvider,
+)
+
+
+class KMSAliasProviderPlugin(CloudFormationResourceProviderPlugin):
+    name = "AWS::KMS::Alias"
+
+    def __init__(self):
+        self.factory: Optional[Type[ResourceProvider]] = None
+
+    def load(self):
+        from localstack.services.kms.resource_providers.aws_kms_alias import KMSAliasProvider
+
+        self.factory = KMSAliasProvider
diff --git a/localstack/services/kms/resource_providers/aws_kms_key.py b/localstack/services/kms/resource_providers/aws_kms_key.py
@@ -0,0 +1,158 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class KMSKeyProperties(TypedDict):
+    KeyPolicy: Optional[dict | str]
+    Arn: Optional[str]
+    Description: Optional[str]
+    EnableKeyRotation: Optional[bool]
+    Enabled: Optional[bool]
+    KeyId: Optional[str]
+    KeySpec: Optional[str]
+    KeyUsage: Optional[str]
+    MultiRegion: Optional[bool]
+    PendingWindowInDays: Optional[int]
+    Tags: Optional[list[Tag]]
+
+
+class Tag(TypedDict):
+    Key: Optional[str]
+    Value: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class KMSKeyProvider(ResourceProvider[KMSKeyProperties]):
+
+    TYPE = "AWS::KMS::Key"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[KMSKeyProperties],
+    ) -> ProgressEvent[KMSKeyProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/KeyId
+
+        Required properties:
+          - KeyPolicy
+
+
+
+        Read-only properties:
+          - /properties/Arn
+          - /properties/KeyId
+
+        IAM permissions required:
+          - kms:CreateKey
+          - kms:EnableKeyRotation
+          - kms:DisableKey
+          - kms:TagResource
+
+        """
+        model = request.desired_state
+        kms = request.aws_client_factory.kms
+
+        params = util.select_attributes(model, ["Description", "KeySpec", "KeyUsage"])
+
+        if model.get("KeyPolicy"):
+            params["Policy"] = json.dumps(model["KeyPolicy"])
+
+        if model.get("Tags"):
+            params["Tags"] = [
+                {"TagKey": tag["Key"], "TagValue": tag["Value"]} for tag in model.get("Tags", [])
+            ]
+        response = kms.create_key(**params)
+        model["KeyId"] = response["KeyMetadata"]["KeyId"]
+        model["Arn"] = response["KeyMetadata"]["Arn"]
+
+        # key is created but some fields map to separate api calls
+        if model.get("EnableKeyRotation", False):
+            kms.enable_key_rotation(KeyId=model["KeyId"])
+        else:
+            kms.disable_key_rotation(KeyId=model["KeyId"])
+
+        if model.get("Enabled", True):
+            kms.enable_key(KeyId=model["KeyId"])
+        else:
+            kms.disable_key(KeyId=model["KeyId"])
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def read(
+        self,
+        request: ResourceRequest[KMSKeyProperties],
+    ) -> ProgressEvent[KMSKeyProperties]:
+        """
+        Fetch resource information
+
+        IAM permissions required:
+          - kms:DescribeKey
+          - kms:GetKeyPolicy
+          - kms:GetKeyRotationStatus
+          - kms:ListResourceTags
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[KMSKeyProperties],
+    ) -> ProgressEvent[KMSKeyProperties]:
+        """
+        Delete a resource
+
+        IAM permissions required:
+          - kms:DescribeKey
+          - kms:ScheduleKeyDeletion
+        """
+    <
81A8
span class='blob-code-inner blob-code-marker ' data-code-marker="+">        model = request.desired_state
+        kms = request.aws_client_factory.kms
+
+        kms.schedule_key_deletion(KeyId=model["KeyId"])
+
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def update(
+        self,
+        request: ResourceRequest[KMSKeyProperties],
+    ) -> ProgressEvent[KMSKeyProperties]:
+        """
+        Update a resource
+
+        IAM permissions required:
+          - kms:DescribeKey
+          - kms:DisableKey
+          - kms:DisableKeyRotation
+          - kms:EnableKey
+          - kms:EnableKeyRotation
+          - kms:PutKeyPolicy
+          - kms:TagResource
+          - kms:UntagResource
+          - kms:UpdateKeyDescription
+        """
+        raise NotImplementedError