feat: delete_object and delete_objects

src/s3_encryption/__init__.py

@@ -359,6 +359,93 @@ def put_object(self, **kwargs):
             if hasattr(self._plugin._context, _CTX_ENCRYPTION_CONTEXT):
                 delattr(self._plugin._context, _CTX_ENCRYPTION_CONTEXT)
 
+    ##= specification/s3-encryption/client.md#required-api-operations
+    ##% - DeleteObject MUST be implemented by the S3EC.
+    def delete_object(self, **kwargs):
+        """Delete an object and its associated instruction file from S3.
+
+        Args:
+            **kwargs: Arguments to pass to the S3 client's delete_object method.
+                      Must include Bucket and Key parameters.
+                      May include InstructionFileSuffix to override the default
+                      ".instruction" suffix for instruction file deletion.
+
+        Returns:
+            The response from the S3 client's delete_object call for the object.
+
+        Raises:
+            S3EncryptionClientError: If the delete operation fails.
+        """
+        ##= specification/s3-encryption/data-format/metadata-strategy.md#instruction-file
+        ##= type=implementation
+        ##% The default Instruction File behavior uses the same S3 object key
+        ##% as its associated object suffixed with ".instruction".
+        instruction_file_suffix = kwargs.pop("InstructionFileSuffix", ".instruction")
+
+        try:
+            ##= specification/s3-encryption/client.md#required-api-operations
+            ##= type=implementation
+            ##% - DeleteObject MUST delete the given object key.
+            response = self.wrapped_s3_client.delete_object(**kwargs)
+
+            ##= specification/s3-encryption/client.md#required-api-operations
+            ##= type=implementation
+            ##% - DeleteObject MUST delete the associated instruction file
+            ##%   using the default instruction file suffix.
+            instruction_key = kwargs["Key"] + instruction_file_suffix
+            self.wrapped_s3_client.delete_object(Bucket=kwargs["Bucket"], Key=instruction_key)
+
+            return response
+        except S3EncryptionClientError:
+            raise
+        except Exception as e:
+            raise S3EncryptionClientError(f"Failed to delete object: {str(e)}") from e
+
+    ##= specification/s3-encryption/client.md#required-api-operations
+    ##% - DeleteObjects MUST be implemented by the S3EC.
+    def delete_objects(self, **kwargs):
+        """Delete multiple objects and their associated instruction files from S3.
+
+        2 requests are issued, one for the objects, and one for the instruction files.
+        If either requests fail, the operation fails, and maybe tried again to clean up any missed files.
+
+        Args:
+            **kwargs: Arguments to pass to the S3 client's delete_objects method.
+                      Must include Bucket and Delete (with Objects list) parameters.
+                      May include InstructionFileSuffix to override the default
+                      ".instruction" suffix for instruction file deletion.
+
+        Returns:
+            The response from the S3 client's delete_objects call for the objects.
+
+        Raises:
+            S3EncryptionClientError: If either delete operations fails.
+        """
+        instruction_file_suffix = kwargs.pop("InstructionFileSuffix", ".instruction")
+
+        try:
+            ##= specification/s3-encryption/client.md#required-api-operations
+            ##= type=implementation
+            ##% - DeleteObjects MUST delete each of the given objects.
+            response = self.wrapped_s3_client.delete_objects(**kwargs)
+
+            ##= specification/s3-encryption/client.md#required-api-operations
+            ##= type=implementation
+            ##% - DeleteObjects MUST delete each of the corresponding instruction files
+            ##%   using the default instruction file suffix.
+            instruction_objects = [
+                {"Key": obj["Key"] + instruction_file_suffix} for obj in kwargs["Delete"]["Objects"]
+            ]
+            self.wrapped_s3_client.delete_objects(
+                Bucket=kwargs["Bucket"], Delete={"Objects": instruction_objects}
+            )
+
+            return response
+        except S3EncryptionClientError:
+            raise
+        except Exception as e:
+            raise S3EncryptionClientError(f"Failed to delete objects: {str(e)}") from e
+
     def get_object(self, **kwargs):
         """Download and decrypt an object from S3.
 

test/integration/test_i_s3_encryption_delete_objects.py

@@ -0,0 +1,126 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Integration tests for S3EncryptionClient.delete_objects."""
+
+import os
+from datetime import datetime
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig
+from s3_encryption.materials.kms_keyring import KmsKeyring
+from s3_encryption.materials.materials import AlgorithmSuite, CommitmentPolicy
+
+bucket = os.environ.get("CI_S3_BUCKET", "s3ec-python-github-test-bucket")
+region = os.environ.get("CI_AWS_REGION", "us-west-2")
+kms_key_id = os.environ.get(
+    "CI_KMS_KEY_ALIAS", "arn:aws:kms:us-west-2:370957321024:alias/S3EC-Python-Github-KMS-Key"
+)
+
+ALGORITHM_CONFIGS = [
+    pytest.param(
+        AlgorithmSuite.ALG_AES_256_GCM_IV12_TAG16_NO_KDF,
+        CommitmentPolicy.FORBID_ENCRYPT_ALLOW_DECRYPT,
+        id="AES_GCM",
+    ),
+    pytest.param(
+        AlgorithmSuite.ALG_AES_256_GCM_HKDF_SHA512_COMMIT_KEY,
+        CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT,
+        id="KC_GCM",
+    ),
+]
+
+
+def _make_client(algorithm_suite, commitment_policy):
+    kms_client = boto3.client("kms", region_name=region)
+    keyring = KmsKeyring(kms_client, kms_key_id)
+    wrapped_client = boto3.client("s3")
+    config = S3EncryptionClientConfig(
+        keyring,
+        encryption_algorithm=algorithm_suite,
+        commitment_policy=commitment_policy,
+    )
+    return S3EncryptionClient(wrapped_client, config)
+
+
+def _unique_key(prefix):
+    return prefix + datetime.now().strftime("%Y-%m-%d-%H:%M:%S-%f")
+
+
+def _object_exists(key):
+    """Return True if the object exists in the test bucket."""
+    s3 = boto3.client("s3")
+    try:
+        s3.head_object(Bucket=bucket, Key=key)
+        return True
+    except ClientError as e:
+        if e.response["Error"]["Code"] == "404":
+            return False
+        raise
+
+
+##= specification/s3-encryption/client.md#required-api-operations
+##= type=test
+##% - DeleteObjects MUST delete each of the given objects.
+@pytest.mark.parametrize("algorithm_suite,commitment_policy", ALGORITHM_CONFIGS)
+def test_delete_objects_deletes_objects(algorithm_suite, commitment_policy):
+    """delete_objects removes the encrypted objects from S3."""
+    s3ec = _make_client(algorithm_suite, commitment_policy)
+    keys = [_unique_key("del-objs-"), _unique_key("del-objs-")]
+
+    for key in keys:
+        s3ec.put_object(Bucket=bucket, Key=key, Body=b"data")
+
+    s3ec.delete_objects(
+        Bucket=bucket,
+        Delete={"Objects": [{"Key": k} for k in keys]},
+    )
+
+    for key in keys:
+        assert not _object_exists(key)
+
+
+##= specification/s3-encryption/client.md#required-api-operations
+##= type=test
+##% - DeleteObjects MUST delete each of the corresponding instruction files
+##%   using the default instruction file suffix.
+@pytest.mark.parametrize("algorithm_suite,commitment_policy", ALGORITHM_CONFIGS)
+def test_delete_objects_deletes_instruction_files(algorithm_suite, commitment_policy):
+    """delete_objects also removes the .instruction files from S3."""
+    s3ec = _make_client(algorithm_suite, commitment_policy)
+    keys = [_unique_key("del-objs-instr-"), _unique_key("del-objs-instr-")]
+
+    # Put instruction-file-based objects by uploading instruction files manually
+    plain_s3 = boto3.client("s3")
+    for key in keys:
+        s3ec.put_object(Bucket=bucket, Key=key, Body=b"data")
+        # Also create a fake instruction file to verify it gets deleted
+        plain_s3.put_object(Bucket=bucket, Key=key + ".instruction", Body=b"{}")
+
+    s3ec.delete_objects(
+        Bucket=bucket,
+        Delete={"Objects": [{"Key": k} for k in keys]},
+    )
+
+    for key in keys:
+        assert not _object_exists(key)
+        assert not _object_exists(key + ".instruction")
+
+
+@pytest.mark.parametrize("algorithm_suite,commitment_policy", ALGORITHM_CONFIGS)
+def test_delete_objects_returns_response(algorithm_suite, commitment_policy):
+    """delete_objects returns the S3 response from the object deletion."""
+    s3ec = _make_client(algorithm_suite, commitment_policy)
+    key = _unique_key("del-objs-resp-")
+    s3ec.put_object(Bucket=bucket, Key=key, Body=b"data")
+
+    response = s3ec.delete_objects(
+        Bucket=bucket,
+        Delete={"Objects": [{"Key": key}]},
+    )
+
+    assert "Deleted" in response
+    deleted_keys = [d["Key"] for d in response["Deleted"]]
+    assert key in deleted_keys

test/test_s3_encryption_client_delete.py

@@ -0,0 +1,108 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Unit tests for S3EncryptionClient.delete_object."""
+
+from unittest.mock import Mock, call
+
+import pytest
+
+from s3_encryption import S3EncryptionClient, S3EncryptionClientConfig
+from s3_encryption.exceptions import S3EncryptionClientError
+from s3_encryption.materials.keyring import S3Keyring
+
+
+def _make_client():
+    """Create an S3EncryptionClient with a mocked wrapped S3 client."""
+    mock_keyring = Mock(spec=S3Keyring)
+    mock_s3 = Mock()
+    mock_s3.meta.events = Mock()
+    config = S3EncryptionClientConfig(keyring=mock_keyring)
+    s3ec = S3EncryptionClient(wrapped_s3_client=mock_s3, config=config)
+    return s3ec, mock_s3
+
+
+class TestDeleteObject:
+    ##= specification/s3-encryption/client.md#required-api-operations
+    ##= type=test
+    ##% - DeleteObject MUST delete the given object key.
+    def test_deletes_object(self):
+        """delete_object forwards the call to the wrapped client."""
+        s3ec, mock_s3 = _make_client()
+        mock_s3.delete_object.return_value = {"DeleteMarker": True}
+
+        response = s3ec.delete_object(Bucket="bucket", Key="key")
+
+        assert response == {"DeleteMarker": True}
+        assert mock_s3.delete_object.call_args_list[0] == call(Bucket="bucket", Key="key")
+
+    ##= specification/s3-encryption/client.md#required-api-operations
+    ##= type=test
+    ##% - DeleteObject MUST delete the associated instruction file
+    ##%   using the default instruction file suffix.
+    def test_deletes_instruction_file(self):
+        """delete_object also deletes the instruction file with default suffix."""
+        s3ec, mock_s3 = _make_client()
+
+        s3ec.delete_object(Bucket="bucket", Key="key")
+
+        assert mock_s3.delete_object.call_count == 2
+        assert mock_s3.delete_object.call_args_list[1] == call(
+            Bucket="bucket", Key="key.instruction"
+        )
+
+    def test_returns_object_delete_response(self):
+        """delete_object returns the response from the object deletion, not the instruction file."""
+        s3ec, mock_s3 = _make_client()
+        object_response = {"DeleteMarker": True, "VersionId": "v1"}
+        instruction_response = {"DeleteMarker": False, "VersionId": "v2"}
+        mock_s3.delete_object.side_effect = [object_response, instruction_response]
+
+        response = s3ec.delete_object(Bucket="bucket", Key="key")
+
+        assert response == object_response
+
+    def test_wraps_unexpected_errors(self):
+        """delete_object wraps unexpected errors in S3EncryptionClientError."""
+        s3ec, mock_s3 = _make_client()
+        mock_s3.delete_object.side_effect = RuntimeError("network error")
+
+        with pytest.raises(S3EncryptionClientError, match="Failed to delete object"):
+            s3ec.delete_object(Bucket="bucket", Key="key")
+
+    def test_reraises_s3_encryption_client_error(self):
+        """delete_object re-raises S3EncryptionClientError without wrapping."""
+        s3ec, mock_s3 = _make_client()
+        mock_s3.delete_object.side_effect = S3EncryptionClientError("original error")
+
+        with pytest.raises(S3EncryptionClientError, match="original error"):
+            s3ec.delete_object(Bucket="bucket", Key="key")
+
+    def test_passes_extra_kwargs(self):
+        """delete_object forwards extra kwargs like VersionId to the wrapped client."""
+        s3ec, mock_s3 = _make_client()
+
+        s3ec.delete_object(Bucket="bucket", Key="key", VersionId="abc123")
+
+        assert mock_s3.delete_object.call_args_list[0] == call(
+            Bucket="bucket", Key="key", VersionId="abc123"
+        )
+
+    def test_custom_instruction_file_suffix(self):
+        """delete_object uses a custom instruction file suffix when provided."""
+        s3ec, mock_s3 = _make_client()
+
+        s3ec.delete_object(Bucket="bucket", Key="key", InstructionFileSuffix=".custom-suffix")
+
+        assert mock_s3.delete_object.call_count == 2
+        assert mock_s3.delete_object.call_args_list[1] == call(
+            Bucket="bucket", Key="key.custom-suffix"
+        )
+
+    def test_instruction_file_suffix_not_forwarded_to_s3(self):
+        """InstructionFileSuffix is popped from kwargs and not sent to S3."""
+        s3ec, mock_s3 = _make_client()
+
+        s3ec.delete_object(Bucket="bucket", Key="key", InstructionFileSuffix=".custom")
+
+        # First call (object delete) should not contain InstructionFileSuffix
+        assert mock_s3.delete_object.call_args_list[0] == call(Bucket="bucket", Key="key")