s3: test GetObject with PartNumber and SSE-C encryption

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/s3tests_boto3/functional/test_s3.py b/s3tests_boto3/functional/test_s3.py
index b73a18b79372f16f789ef87af23229162ab5608c..eefa8d225da3aaf6d772db25261732a686e0ce54 100644 (file)
--- a/s3tests_boto3/functional/test_s3.py
+++ b/s3tests_boto3/functional/test_s3.py
@@ -6459,6 +6459,61 @@ def test_multipart_get_part():
     assert status == 400
     assert error_code == 'InvalidPart'
 
+@pytest.mark.encryption
+@pytest.mark.fails_on_dbstore
+def test_multipart_sse_c_get_part():
+    bucket_name = get_new_bucket()
+    client = get_client()
+    key = "mymultipart"
+
+    part_size = 5*1024*1024
+    part_sizes = 3 * [part_size] + [1*1024*1024]
+    part_count = len(part_sizes)
+    total_size = sum(part_sizes)
+
+    enc_headers = {
+        'x-amz-server-side-encryption-customer-algorithm': 'AES256',
+        'x-amz-server-side-encryption-customer-key': 'pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=',
+        'x-amz-server-side-encryption-customer-key-md5': 'DWygnHRtgiJ77HCm+1rvHw==',
+    }
+    get_args = {
+        'SSECustomerAlgorithm': 'AES256',
+        'SSECustomerKey': 'pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=',
+        'SSECustomerKeyMD5': 'DWygnHRtgiJ77HCm+1rvHw==',
+    }
+
+    (upload_id, data, parts) = _multipart_upload_enc(client, bucket_name, key, total_size,
+            part_size, init_headers=enc_headers, part_headers=enc_headers, metadata=None, resend_parts=[2])
+
+    # request part before complete
+    e = assert_raises(ClientError, client.get_object, Bucket=bucket_name, Key=key, PartNumber=1)
+    status, error_code = _get_status_and_error_code(e.response)
+    assert status == 404
+    assert error_code == 'NoSuchKey'
+
+    client.complete_multipart_upload(Bucket=bucket_name, Key=key, UploadId=upload_id, MultipartUpload={'Parts': parts})
+    assert len(parts) == part_count
+
+    for part, size in zip(parts, part_sizes):
+        response = client.head_object(Bucket=bucket_name, Key=key, PartNumber=part['PartNumber'], **get_args)
+        assert response['PartsCount'] == part_count
+        assert response['ETag'] == '"{}"'.format(part['ETag'])
+
+        response = client.get_object(Bucket=bucket_name, Key=key, PartNumber=part['PartNumber'], **get_args)
+        assert response['PartsCount'] == part_count
+        assert response['ETag'] == '"{}"'.format(part['ETag'])
+        assert response['ContentLength'] == size
+        # compare contents
+        for chunk in response['Body'].iter_chunks():
+            assert chunk.decode() == data[0:len(chunk)]
+            data = data[len(chunk):]
+
+    # request PartNumber out of range
+    e = assert_raises(ClientError, client.get_object, Bucket=bucket_name, Key=key, PartNumber=5)
+    status, error_code = _get_status_and_error_code(e.response)
+    assert status == 400
+    assert error_code == 'InvalidPart'
+
 @pytest.mark.fails_on_dbstore
 def test_multipart_single_get_part():
     bucket_name = get_new_bucket()
@@ -6521,6 +6576,33 @@ def test_non_multipart_get_part():
     assert response['ETag'] == etag
     assert _get_body(response) == 'body'
 
+@pytest.mark.encryption
+@pytest.mark.fails_on_dbstore
+def test_non_multipart_sse_c_get_part():
+    bucket_name = get_new_bucket()
+    client = get_client()
+    key = "singlepart"
+
+    sse_args = {
+        'SSECustomerAlgorithm': 'AES256',
+        'SSECustomerKey': 'pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=',
+        'SSECustomerKeyMD5': 'DWygnHRtgiJ77HCm+1rvHw=='
+    }
+
+    response = client.put_object(Bucket=bucket_name, Key=key, Body='body', **sse_args)
+    etag = response['ETag']
+
+    # request for PartNumber > 1 results in InvalidPart
+    e = assert_raises(ClientError, client.get_object, Bucket=bucket_name, Key=key, PartNumber=2, **sse_args)
+    status, error_code = _get_status_and_error_code(e.response)
+    assert status == 400
+    assert error_code == 'InvalidPart'
+
+    # request for PartNumber = 1 gives back the entire object
+    response = client.get_object(Bucket=bucket_name, Key=key, PartNumber=1, **sse_args)
+    assert response['ETag'] == etag
+    assert _get_body(response) == 'body'
+
 
 def _simple_http_req_100_cont(host, port, is_secure, method, resource):
     """