localstack · k-a-il · Nov 14, 2025 · Nov 4, 2025 · Nov 10, 2025
@@ -4,10 +4,13 @@
 import logging
 import os
 
+from cbor2 import loads as cbor2_loads
 from cryptography.exceptions import InvalidTag
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, keywrap
 from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
+from cryptography.hazmat.primitives.serialization import load_der_public_key
 
 from localstack.aws.api import CommonServiceException, RequestContext, handler
 from localstack.aws.api.kms import (
@@ -138,6 +141,7 @@
 from localstack.utils.aws.arns import get_partition, kms_alias_arn, parse_arn
 from localstack.utils.collections import PaginatedList
 from localstack.utils.common import select_attributes
+from localstack.utils.crypto import pkcs7_envelope_encrypt
 from localstack.utils.strings import short_uid, to_bytes, to_str
 
 LOG = logging.getLogger(__name__)
@@ -1075,6 +1079,25 @@ def decrypt(
         self._validate_key_for_encryption_decryption(context, key)
         self._validate_key_state_not_pending_import(key)
 
+        # Handle the recipient field.  This is used by AWS Nitro to re-encrypt the plaintext to the key specified
+        # by the enclave.  Proper support for this will take significant work to figure out how to model enforcing
+        # the attestation measurements; for now, if recipient is specified and has an attestation doc in it including
+        # a public key where it's expected to be, we encrypt to that public key.  This at least allows users to use
+        # localstack as a drop-in replacement for AWS when testing without having to skip the secondary decryption
+        # when using localstack.
+        recipient_pubkey = None
+        if recipient:
+            attestation_document = recipient["AttestationDocument"]
+            # We do all of this in a try/catch and warn if it fails so that if users are currently passing a nonsense
+            # value we don't break it for them.  In the future we could do a breaking change to require a valid attestation
+            # (or at least one that contains the public key).
+            try:
+                recipient_pubkey = self._extract_attestation_pubkey(attestation_document)
+            except Exception as e:
+                logging.warning(
+                    "Unable to extract public key from non-empty attestation document: %s", e
+                )
+
         try:
             # TODO: Extend the implementation to handle additional encryption/decryption scenarios
             # beyond the current support for offline encryption and online decryption using RSA keys if key id exists in
@@ -1088,20 +1111,27 @@ def decrypt(
                 plaintext = key.decrypt(ciphertext, encryption_context)
         except InvalidTag:
             raise InvalidCiphertextException()
+
         # For compatibility, we return EncryptionAlgorithm values expected from AWS. But LocalStack currently always
         # encrypts with symmetric encryption no matter the key settings.
         #
         # We return a key ARN instead of KeyId despite the name of the parameter, as this is what AWS does and states
         # in its docs.
-        # TODO add support for "recipient"
         #  https://docs.aws.amazon.com/kms/latest/APIReference/API_Decrypt.html#API_Decrypt_RequestSyntax
         # TODO add support for "dry_run"
-        return DecryptResponse(
+        response = DecryptResponse(
             KeyId=key.metadata.get("Arn"),
-            Plaintext=plaintext,
             EncryptionAlgorithm=encryption_algorithm,
         )
 
+        # Encrypt to the recipient pubkey if specified.  Otherwise, return the actual plaintext
+        if recipient_pubkey:
+            response["CiphertextForRecipient"] = pkcs7_envelope_encrypt(plaintext, recipient_pubkey)
+        else:
+            response["Plaintext"] = plaintext
+
+        return response
+
     def get_parameters_for_import(
         self,
         context: RequestContext,
@@ -1559,6 +1589,15 @@ def _validate_grant_request(self, data: dict):
                     f" constraint: [Member must satisfy enum value set: {VALID_OPERATIONS}]"
                 )
 
+    def _extract_attestation_pubkey(self, attestation_document: bytes) -> RSAPublicKey:
+        # The attestation document comes as a COSE (CBOR Object Signing and Encryption) object: the CBOR
+        # attestation is signed and then the attestation and signature are again CBOR-encoded.  For now
+        # we don't bother validating the signature, though in the future we could.
+        cose_document = cbor2_loads(attestation_document)
+        attestation = cbor2_loads(cose_document[2])
+        public_key_bytes = attestation["public_key"]
+        return load_der_public_key(public_key_bytes)
+
     def _decrypt_wrapped_key_material(
         self,
         import_state: KeyImportState,

diff --git a/localstack-core/localstack/utils/crypto.py b/localstack-core/localstack/utils/crypto.py
@@ -4,7 +4,13 @@
 import re
 import threading
 
+from asn1crypto import algos, cms, core
+from asn1crypto import x509 as asn1_x509
 from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import padding as sym_padding
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 
 from .files import TMP_FILES, file_exists_not_empty, load_file, new_tmp_file, save_file
@@ -26,6 +32,11 @@
 PEM_KEY_START_REGEX = r"-----BEGIN(.*)PRIVATE KEY-----"
 PEM_KEY_END_REGEX = r"-----END(.*)PRIVATE KEY-----"
 
+OID_AES256_CBC = "2.16.840.1.101.3.4.1.42"
+OID_MGF1 = "1.2.840.113549.1.1.8"
+OID_RSAES_OAEP = "1.2.840.113549.1.1.7"
+OID_SHA256 = "2.16.840.1.101.3.4.2.1"
+
 
 @synchronized(lock=SSL_CERT_LOCK)
 def generate_ssl_cert(
@@ -183,3 +194,101 @@ def decrypt(
     decrypted = decryptor.update(encrypted) + decryptor.finalize()
     decrypted = unpad(decrypted)
     return decrypted
+
+
+def pkcs7_envelope_encrypt(plaintext: bytes, recipient_pubkey: RSAPublicKey) -> bytes:
+    """
+    Create a PKCS7 wrapper of some plaintext decryptable by recipient_pubkey.  Uses RSA-OAEP with SHA-256
+    to encrypt the AES-256-CBC content key.  Hazmat's PKCS7EnvelopeBuilder doesn't support RSA-OAEP with SHA-256,
+    so we need to build the pieces manually and then put them together in an envelope with asn1crypto.
+    """
+
+    # Encrypt the plaintext with an AES session key, then encrypt the session key to the recipient_pubkey
+    session_key = os.urandom(32)
+    iv = os.urandom(16)
+    encrypted_session_key = recipient_pubkey.encrypt(
+        session_key,
+        padding.OAEP(
+            mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None
+        ),
+    )
+    cipher = Cipher(algorithms.AES(session_key), modes.CBC(iv), backend=default_backend())
+    encryptor = cipher.encryptor()
+    padder = sym_padding.PKCS7(algorithms.AES.block_size).padder()
+    padded_plaintext = padder.update(plaintext) + padder.finalize()
+    encrypted_content = encryptor.update(padded_plaintext) + encryptor.finalize()
+
+    # Now put together the envelope.
+    # Add the recipient with their copy of the session key
+    recipient_identifier = cms.RecipientIdentifier(
+        name="issuer_and_serial_number",
+        value=cms.IssuerAndSerialNumber(
+            {
+                "issuer": asn1_x509.Name.build({"common_name": "recipient"}),
+                "serial_number": 1,
+            }
+        ),
+    )
+    key_enc_algorithm = cms.KeyEncryptionAlgorithm(
+        {
+            "algorithm": OID_RSAES_OAEP,
+            "parameters": algos.RSAESOAEPParams(
+                {
+                    "hash_algorithm": algos.DigestAlgorithm(
+                        {
+                            "algorithm": OID_SHA256,
+                        }
+                    ),
+                    "mask_gen_algorithm": algos.MaskGenAlgorithm(
+                        {
+                            "algorithm": OID_MGF1,
+                            "parameters": algos.DigestAlgorithm(
+                                {
+                                    "algorithm": OID_SHA256,
+                                }
+                            ),
+                        }
+                    ),
+                }
+            ),
+        }
+    )
+    recipient_info = cms.KeyTransRecipientInfo(
+        {
+            "version": "v0",
+            "rid": recipient_identifier,
+            "key_encryption_algorithm": key_enc_algorithm,
+            "encrypted_key": encrypted_session_key,
+        }
+    )
+
+    # Add the encrypted content
+    content_enc_algorithm = cms.EncryptionAlgorithm(
+        {
+            "algorithm": OID_AES256_CBC,
+            "parameters": core.OctetString(iv),
+        }
+    )
+    encrypted_content_info = cms.EncryptedContentInfo(
+        {
+            "content_type": "data",
+            "content_encryption_algorithm": content_enc_algorithm,
+            "encrypted_content": encrypted_content,
+        }
+    )
+    enveloped_data = cms.EnvelopedData(
+        {
+            "version": "v0",
+            "recipient_infos": [recipient_info],
+            "encrypted_content_info": encrypted_content_info,
+        }
+    )
+
+    # Finally add a wrapper and return its bytes
+    content_info = cms.ContentInfo(
+        {
+            "content_type": "enveloped_data",
+            "content": enveloped_data,
+        }
+    )
+    return content_info.dump()
@@ -12,6 +12,7 @@ description = "The core library and runtime of LocalStack"
 license = "Apache-2.0"
 requires-python = ">=3.10"
 dependencies = [
+    "asn1crypto>=1.5.1",
     "click>=7.1",
     "cachetools>=5.0",
     "cryptography",

@@ -4,6 +4,8 @@
 #
 #    pip-compile --extra=base-runtime --output-file=requirements-base-runtime.txt --strip-extras --unsafe-package=distribute --unsafe-package=localstack-core --unsafe-package=pip --unsafe-package=setuptools pyproject.toml
 #
+asn1crypto==1.5.1
+    # via localstack-core (pyproject.toml)
 attrs==25.4.0
     # via
     #   jsonschema

@@ -4,6 +4,8 @@
 #
 #    pip-compile --output-file=requirements-basic.txt --strip-extras --unsafe-package=distribute --unsafe-package=localstack-core --unsafe-package=pip --unsafe-package=setuptools pyproject.toml
 #
+asn1crypto==1.5.1
+    # via localstack-core (pyproject.toml)
 cachetools==6.2.1
     # via localstack-core (pyproject.toml)
 certifi==2025.10.5

@@ -18,6 +18,10 @@ apispec==6.8.4
     # via localstack-core
 argparse==1.4.0
     # via kclpy-ext
+asn1crypto==1.5.1
+    # via
+    #   localstack-core
+    #   localstack-core (pyproject.toml)
 attrs==25.4.0
     # via
     #   cattrs

@@ -16,6 +16,10 @@ apispec==6.8.4
     # via localstack-core (pyproject.toml)
 argparse==1.4.0
     # via kclpy-ext
+asn1crypto==1.5.1
+    # via
+    #   localstack-core
+    #   localstack-core (pyproject.toml)
 attrs==25.4.0
     # via
     #   jsonschema

@@ -18,6 +18,10 @@ apispec==6.8.4
     # via localstack-core
 argparse==1.4.0
     # via kclpy-ext
+asn1crypto==1.5.1
+    # via
+    #   localstack-core
+    #   localstack-core (pyproject.toml)
 attrs==25.4.0
     # via
     #   cattrs

@@ -18,6 +18,10 @@ apispec==6.8.4
     # via localstack-core
 argparse==1.4.0
     # via kclpy-ext
+asn1crypto==1.5.1
+    # via
+    #   localstack-core
+    #   localstack-core (pyproject.toml)
 attrs==25.4.0
     # via
     #   cattrs