fix: validate encryption materials from cmm are compatible with kc policy

src/aws_encryption_sdk/streaming_client.py

@@ -553,6 +553,8 @@ def _prep_message(self):
             request=encryption_materials_request
         )
 
+        validate_commitment_policy_on_encrypt(self.config.commitment_policy, self._encryption_materials.algorithm)
+
         if self.config.algorithm is not None and self._encryption_materials.algorithm != self.config.algorithm:
             raise ActionNotAllowedError(
                 (

test/functional/test_f_commitment.py

@@ -225,3 +225,59 @@ def test_encrypt_with_uncommitting_algorithm_require_decrypt():
     with pytest.raises(ActionNotAllowedError) as excinfo:
         decrypting_client.decrypt(source=ciphertext, key_provider=key_provider)
     excinfo.match("Configuration conflict. Cannot decrypt due to .* requiring only committed messages")
+
+
+def test_encrypt_with_require_policy_fail_when_retrieving_invalid_cmm_materials():
+    """Tests that when a client with a require policy shares a cache with a client with a forbid policy
+    an error gets thrown due to invalid materials retrieved from cmm"""
+    forbid_encrypting_client = aws_encryption_sdk.EncryptionSDKClient(
+        commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT
+    )
+    required_encrypting_client = aws_encryption_sdk.EncryptionSDKClient(
+        commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
+    )
+
+    provider = StaticRawMasterKeyProvider(
+        wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING,
+        encryption_key_type=EncryptionKeyType.SYMMETRIC,
+        key_bytes=b"\00" * 32,
+    )
+    provider.add_master_key("KeyId")
+    cache = aws_encryption_sdk.LocalCryptoMaterialsCache(capacity=10)
+    ccmm = aws_encryption_sdk.CachingCryptoMaterialsManager(
+        master_key_provider=provider, cache=cache, max_age=3600.0, max_messages_encrypted=5
+    )
+    plaintext = b"Yellow Submarine"
+
+    _, _ = forbid_encrypting_client.encrypt(source=plaintext, materials_manager=ccmm)
+    with pytest.raises(ActionNotAllowedError) as excinfo:
+        required_encrypting_client.encrypt(source=plaintext, materials_manager=ccmm)
+    excinfo.match("Configuration conflict. Cannot encrypt due to .* requiring only committed messages")
+
+
+def test_encrypt_with_forbid_policy_fail_when_retrieving_invalid_cmm_materials():
+    """Tests that when a client with a forbid policy shares a cache with a client with a require policy
+    an error gets thrown due to invalid materials retrieved from cmm"""
+    forbid_encrypting_client = aws_encryption_sdk.EncryptionSDKClient(
+        commitment_policy=CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT
+    )
+    required_encrypting_client = aws_encryption_sdk.EncryptionSDKClient(
+        commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
+    )
+
+    provider = StaticRawMasterKeyProvider(
+        wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING,
+        encryption_key_type=EncryptionKeyType.SYMMETRIC,
+        key_bytes=b"\00" * 32,
+    )
+    provider.add_master_key("KeyId")
+    cache = aws_encryption_sdk.LocalCryptoMaterialsCache(capacity=10)
+    ccmm = aws_encryption_sdk.CachingCryptoMaterialsManager(
+        master_key_provider=provider, cache=cache, max_age=3600.0, max_messages_encrypted=5
+    )
+    plaintext = b"Yellow Submarine"
+
+    _, _ = required_encrypting_client.encrypt(source=plaintext, materials_manager=ccmm)
+    with pytest.raises(ActionNotAllowedError) as excinfo:
+        forbid_encrypting_client.encrypt(source=plaintext, materials_manager=ccmm)
+    excinfo.match("Configuration conflict. Cannot encrypt due to .* requiring only non-committed messages.")