Retrieve server-side encryption setting on HeadObject (#1143)

## Description of change

Add two new fields to `HeadObjectResult`: 
* `sse_type`: The server-side encryption algorithm used to store the
object (header: "x-amz-server-side-encryption"),
* `sse_kms_key_id`: The ID of the KMS key was used for object
encryption, if present (header:
"x-amz-server-side-encryption-aws-kms-key-id").

## Does this change impact existing behavior?

No. Only adds fields to a non-exhaustive type.

## Does this change need a changelog entry in any of the crates?

Yes: `mountpoint-s3-client`.

---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

---------

Signed-off-by: Alessandro Passaro <[email protected]>

mountpoint-s3-client/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### Other changes
 
+* `HeadObjectResult` now includes the server-side encryption settings used when storing the object.
+  ([#1143](https://github.com/awslabs/mountpoint-s3/pull/1143))
 * Add parameter to request checksum information as part of a `HeadObject` request.
   If specified, the result should contain the checksum for the object if available in the S3 response.
   ([#1083](https://github.com/awslabs/mountpoint-s3/pull/1083))

mountpoint-s3-client/src/mock_client.rs

@@ -732,6 +732,8 @@ impl ObjectClient for MockClient {
                 storage_class: object.storage_class.clone(),
                 restore_status: object.restore_status,
                 checksum,
+                sse_type: None,
+                sse_kms_key_id: None,
             })
         } else {
             Err(ObjectClientError::ServiceError(HeadObjectError::NotFound))

mountpoint-s3-client/src/object_client.rs

@@ -300,6 +300,12 @@ pub struct HeadObjectResult {
     /// HeadObject must explicitly request for this field to be included,
     /// otherwise the values will be empty.
     pub checksum: Checksum,
+
+    /// Server-side encryption type that was used to store the object.
+    pub sse_type: Option<String>,
+
+    /// Server-side encryption KMS key ID that was used to store the object.
+    pub sse_kms_key_id: Option<String>,
 }
 
 /// Errors returned by a [`head_object`](ObjectClient::head_object) request

mountpoint-s3-client/src/s3_crt_client/head_object.rs

@@ -74,6 +74,8 @@ impl HeadObjectResult {
         let etag = headers.get_as_string("Etag")?;
         let storage_class = headers.get_as_optional_string("x-amz-storage-class")?;
         let restore_status = Self::parse_restore_status(headers)?;
+        let sse_type = headers.get_as_optional_string("x-amz-server-side-encryption")?;
+        let sse_kms_key_id = headers.get_as_optional_string("x-amz-server-side-encryption-aws-kms-key-id")?;
         let checksum = parse_checksum(headers)?;
         let result = HeadObjectResult {
             size,
@@ -82,6 +84,8 @@ impl HeadObjectResult {
             restore_status,
             etag: etag.into(),
             checksum,
+            sse_type,
+            sse_kms_key_id,
         };
         Ok(result)
     }

mountpoint-s3-client/tests/head_object.rs

@@ -255,3 +255,70 @@ async fn test_head_object_restored() {
     }
     assert!(!timeout_exceeded, "timeouted while waiting for object become restored");
 }
+
+async fn test_head_object_sse(
+    client: S3CrtClient,
+    bucket: &str,
+    prefix: &str,
+    sse_type: Option<&str>,
+    kms_key_id: Option<String>,
+) {
+    let key = format!("{prefix}hello");
+    let expected_sdk_sse = sse_type.map(|sse| sse.parse().expect("unexpected sse type was used in a test"));
+    let sdk_client = get_test_sdk_client().await;
+    let put_output = sdk_client
+        .put_object()
+        .bucket(bucket)
+        .key(&key)
+        .body(ByteStream::from_static(b"test"))
+        .set_server_side_encryption(expected_sdk_sse)
+        .set_ssekms_key_id(kms_key_id)
+        .send()
+        .await
+        .expect("put object should succeed");
+
+    let result = client
+        .head_object(bucket, &key, &HeadObjectParams::new())
+        .await
+        .expect("head_object failed");
+
+    assert_eq!(
+        result.sse_type.as_deref(),
+        put_output.server_side_encryption().map(|sse| sse.as_str()),
+        "sse_type should match"
+    );
+    assert_eq!(
+        result.sse_kms_key_id, put_output.ssekms_key_id,
+        "kms_key_id should match"
+    );
+}
+
+#[test_case(Some("aws:kms"), Some(get_test_kms_key_id()))]
+#[test_case(Some("aws:kms"), None)]
+#[test_case(Some("aws:kms:dsse"), Some(get_test_kms_key_id()))]
+#[test_case(Some("aws:kms:dsse"), None)]
+#[test_case(None, None)]
+#[test_case(Some("AES256"), None)]
+#[tokio::test]
+#[cfg(not(feature = "s3express_tests"))]
+async fn test_head_object_sse_s3(sse_type: Option<&str>, kms_key_id: Option<String>) {
+    let prefix = get_unique_test_prefix("test_head_object_sse_s3");
+    let bucket = get_test_bucket();
+    let client: S3CrtClient = get_test_client();
+
+    test_head_object_sse(client, &bucket, &prefix, sse_type, kms_key_id).await;
+}
+
+#[tokio::test]
+#[cfg(feature = "s3express_tests")]
+async fn test_head_object_sse_s3express() {
+    // Directory buckets only allow to set sse on the whole bucket. See
+    // [Server-side encryption](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-express-data-protection.html#s3-express-ecnryption) for directory buckets.
+    // We will only test the default here.
+
+    let prefix = get_unique_test_prefix("test_head_object_sse_s3express");
+    let bucket = get_test_bucket();
+    let client: S3CrtClient = get_test_client();
+
+    test_head_object_sse(client, &bucket, &prefix, None, None).await;
+}