KMS mocking fails with AccessDenied in subsequent tests when trying to decrypt

[ruhrohraggy]

I'm working on writing unit tests for a service that encrypts a payload using the aws_encryption_sdk using a standard master key. I've tried a few different approaches, but I think I've finally run into an issue stemming from Moto as opposed to my mocking/scoping setup.

If I have a set of unit tests, say 2 for this example, and both try to execute the service that does encryption, the first test is able to successfully decrypt the cipher text, while the second test fails when attempting to decrypt the data key with an AccessDeniedException, stating that the master key (the same Moto-created KMS key) is unable to decrypt the data key. The below example should reproduce this, but I tried to keep it as short as possible so some fill in the blanks may be required. It assumes an implementation similar to this example from AWS. Also note - the setup would normally be done once for the whole class but I kept hitting issues with that, so I moved to global key, provider, etc.

@mock_aws
class TestClass:
    def setUp(self) -> None:
        """Setup for test."""
        global key_arn, key_id, test_key_provider, kms_configured
        if not kms_configured:
            key_creation = self.kms_client.create_key()
            key_id = key_creation["KeyMetadata"]["KeyId"]
            key_arn = key_creation["KeyMetadata"]["Arn"]
            self.kms_client.create_alias(AliasName="alias/my_123456789012_kms", TargetKeyId=key_id)
            test_key_provider = aws_encryption_sdk.StrictAwsKmsMasterKeyProvider(
                key_ids=[key_arn], region_names=["us-east-1"]
            )
            kms_configured = True

    def test1(self):
        encrypted_payload = encryption_service.encrypt(b'payload')
        decrypted_payload, decryptor_header = self.encryption_client.decrypt(
                source=encrypted_payload,
                key_provider=test_key_provider
            )
        assert decrypted_payload == b'payload'
    
    def test2(self):
        encrypted_payload = encryption_service.encrypt(b'payload-2')
        decrypted_payload, decryptor_header = self.encryption_client.decrypt(
                source=encrypted_payload,
                key_provider=test_key_provider
            )
        assert decrypted_payload == b'payload-2'

The first test succeeds without issue (the encryption service is instantiated only once and pulls the master key created in the setUp). The second test fails when trying to call the kms service (via Moto) to decrypt the data key, and returns an AccessDeniedException response. The POST request created is the exact same according to the DEBUG logs between the two calls, both in metadata and in the POST body (Key-id, CipherText, etc.) The MasterKeyInfo object is the exact same between the two. I've tried pulling the key provider from the encryption service itself as opposed to creating a "test" version against the same key.

Expected behavior:
Subsequent tests pass when mocking KMS, where data keys are able to be decrypted using the same master key multiple times.

Environment:
Poetry Python3.11 project with Moto @ 5.0.9

[bblommers]

Hi @ruhrohraggy, can you share the encryption service as well? Does that do anything special?

Can you reproduce this problem if, instead of encrypting/decrypting, you make other calls to to Moto?
E.g.:

def test1(self):
    self.kms_client.create_alias(AliasName="alias/my_123456789012_kms2", TargetKeyId=key_id)

def test2(self):
    self.kms_client.create_alias(AliasName="alias/my_123456789012_kms3", TargetKeyId=key_id)

[ruhrohraggy]

Hey @bblommers -- appreciate you jumping in. The encryption service is the encryption SDK provided by AWS. I'm retrieving a master key from KMS in another account, using it to do encryption, and then decrypting to ensure that I successfully retrieved the key from KMS, etc.

As far as reproducing, I'll get back to you on Monday -- it's been a long week and I'll need a clear head to dig back into that stuff! 😁