chore(tests): add security tests (#174)

* chore: add tests around downgrade and EC tampering

Author: kessplas

src/s3_encryption/decryptor.py

@@ -72,8 +72,11 @@ def finalize(self, data: bytes) -> bytes:
             plaintext = self._decryptor.update(data) if data else b""
             plaintext += self._decryptor.finalize()
             return self._unpadder.update(plaintext) + self._unpadder.finalize()
-        except Exception as e:
-            raise S3EncryptionClientSecurityError(f"Failed to decrypt CBC content: {e}") from e
+        except Exception:
+            # Use a fixed message for all CBC failures to prevent padding oracle attacks.
+            # Different failure modes (bad padding, truncated ciphertext, wrong key) MUST
+            # produce identical error responses so an attacker cannot distinguish them.
+            raise S3EncryptionClientSecurityError("Failed to decrypt CBC content.") from None
 
 
 @define

src/s3_encryption/materials/kms_keyring.py

@@ -200,6 +200,16 @@ def on_decrypt(self, dec_materials, encrypted_data_keys=None):
                         f"Enable legacy wrapping algorithms to use legacy key wrapping "
                         f"algorithm: {edk.key_provider_info}"
                     )
+                # The KmsV1 wrapping algorithm does not support caller-provided
+                # encryption context. If the caller provided encryption context,
+                # the client MUST reject the request. This prevents a downgrade
+                # from kms+context to kms from bypassing context validation.
+                if dec_materials.encryption_context_from_request:
+                    raise S3EncryptionClientError(
+                        "Encryption context is not supported with the KmsV1 (kms) "
+                        "wrapping algorithm. Use kms+context wrapping algorithm to "
+                        "use encryption context."
+                    )
             else:
                 raise S3EncryptionClientError(
                     f"{edk.key_provider_info} is not a valid key wrapping algorithm!"

src/s3_encryption/pipelines.py

@@ -316,6 +316,30 @@ def decrypt(
         # Determine the algorithm suite from the metadata
         algorithm_suite = self._determine_algorithm_suite(metadata)
 
+        # Reject metadata that contains keys from multiple format versions.
+        # This prevents format confusion attacks where an attacker injects
+        # V2 keys via an instruction file to bypass V3 key-commitment verification.
+        if metadata.has_exclusive_key_collision():
+            raise S3EncryptionClientError(
+                "Object metadata contains keys from multiple format versions. "
+                "The object or its instruction file may have been tampered with."
+            )
+
+        # Also reject V2 format metadata that contains V3 content keys.
+        # In the instruction file injection scenario, the attacker replaces
+        # V3 EDK keys with V2 keys, but V3 content keys (x-amz-c, x-amz-d,
+        # x-amz-i) remain from the object metadata. This combination is
+        # never produced by legitimate encryption.
+        if metadata.is_v2_format() and (
+            metadata.content_cipher_v3 is not None
+            or metadata.key_commitment_v3 is not None
+            or metadata.message_id_v3 is not None
+        ):
+            raise S3EncryptionClientError(  # pragma: no cover — only reachable via instruction file merge; covered by TestInstructionFileFormatConfusion
+                "Object metadata contains V2 format keys alongside V3 content keys. "
+                "The object or its instruction file may have been tampered with."
+            )
+
         # Determine which format we're dealing with and get decryption materials
         if metadata.is_v1_format():
             dec_materials = self._decrypt_v1(metadata, encryption_context)
@@ -590,7 +614,13 @@ def _decrypt_v3(self, metadata, encryption_context) -> DecryptionMaterials:
 
         # Map V3 compressed wrapping algorithm to canonical key_provider_info
         raw_wrap_alg = metadata.encrypted_data_key_algorithm_v3 or "12"
-        wrap_alg = self._V3_WRAP_ALG_MAP.get(raw_wrap_alg, raw_wrap_alg)
+        wrap_alg = self._V3_WRAP_ALG_MAP.get(raw_wrap_alg)
+        if wrap_alg is None:
+            raise S3EncryptionClientError(
+                f"Unknown V3 wrapping algorithm: '{raw_wrap_alg}'. "
+                f"Valid values are: {list(self._V3_WRAP_ALG_MAP.keys())}. "
+                f"The object metadata may have been tampered with."
+            )
 
         encrypted_data_key = EncryptedDataKey(
             key_provider_id=b"S3Keyring",
@@ -607,8 +637,13 @@ def _decrypt_v3(self, metadata, encryption_context) -> DecryptionMaterials:
         stored_context = {}
         if wrap_alg == "kms+context":
             raw_ctx = metadata.encryption_context_v3
-        else:
+        elif wrap_alg in ("AES/GCM", "RSA-OAEP-SHA1"):
             raw_ctx = metadata.mat_desc_v3
+        else:
+            raise S3EncryptionClientError(  # pragma: no cover — defense in depth, unreachable
+                f"Unexpected V3 wrapping algorithm for context selection: '{wrap_alg}'. "
+                f"The object metadata may have been tampered with."
+            )
 
         if raw_ctx is not None:
             if isinstance(raw_ctx, dict):

test-server/java-tests/src/it/java/software/amazon/encryption/s3/RoundTripTests.java

@@ -332,6 +332,10 @@ public void kmsV1Legacy(TestUtils.LanguageServerTarget language) {
     @ParameterizedTest(name = "{displayName} for Encrypt: Java, Decrypt: {0}")
     @MethodSource("software.amazon.encryption.s3.TestUtils#clientsForTest")
     public void kmsV1LegacyWithEncCtx(TestUtils.LanguageServerTarget language) {
+        if (KMSV1_ENCRYPTION_CONTEXT_UNSUPPORTED.contains(language.getLanguageName())) {
+            throw new TestAbortedException(
+                "KmsV1 with encryption context not supported for: " + language.getLanguageName());
+        }
         S3ECTestServerClient client = testServerClientFor(language);
         final String objectKey = appendTestSuffix("test-key-kms-v1-with-enc-ctx-" + language);
         final String input = "simple-test-input";

test-server/java-tests/src/it/java/software/amazon/encryption/s3/TestUtils.java

@@ -98,6 +98,11 @@ public class TestUtils {
     public static final Set<String> ENCRYPTION_CONTEXT_ON_ENCRYPT_UNSUPPORTED =
       Set.of(NET_V3_TRANSITION, NET_V4);
 
+    // Languages that reject caller-provided encryption context when the
+    // wrapping algorithm is KmsV1 ("kms").
+    public static final Set<String> KMSV1_ENCRYPTION_CONTEXT_UNSUPPORTED =
+      Set.of(PYTHON_V3);
+
     public static final Set<String> RE_ENCRYPT_SUPPORTED =
       Set.of(JAVA_V3_TRANSITION, JAVA_V4);