Verify object metadata in Express cache (#1125)

## Description of change

- Verify S3 Express cache objects have valid object metadata which
matches the keys
- Verifies the CRC32C of the object content post-download from S3
  - If checksum is missing, return `BlockChecksumMissing`.

Relevant issues: N/A

## Does this change impact existing behavior?

Changes S3 Express cache to require object metadata and CRC32C to be
present. Old caches will not be used.

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

No

---

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: Simon Beal <[email protected]>

mountpoint-s3-client/src/mock_client.rs

@@ -797,13 +797,21 @@ impl ObjectClient for MockClient {
             return Err(ObjectClientError::ServiceError(PutObjectError::NoSuchBucket));
         }
 
+        let content_crc32 = crc32c::checksum(contents.as_ref());
         let mut object: MockObject = contents.into();
         object.set_storage_class(params.storage_class.clone());
         object.set_object_metadata(params.object_metadata.clone());
         if let Some(upload_checksum) = &params.checksum {
             let mut checksum = Checksum::empty();
             match upload_checksum {
-                UploadChecksum::Crc32c(crc32c) => checksum.checksum_crc32c = Some(crc32c_to_base64(crc32c)),
+                UploadChecksum::Crc32c(crc32c) => {
+                    if crc32c != &content_crc32 {
+                        return Err(ObjectClientError::ClientError(MockClientError(
+                            "crc32c specified did not match data content".into(),
+                        )));
+                    }
+                    checksum.checksum_crc32c = Some(crc32c_to_base64(crc32c));
+                }
             }
             object.set_checksum(checksum);
         }

mountpoint-s3/Cargo.toml

@@ -46,6 +46,7 @@ tracing-log = "0.2.0"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 async-stream = "0.3.6"
 humansize = "2.1.3"
+base64ct = "1.6.0"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 procfs = { version = "0.17.0", default-features = false }

mountpoint-s3/src/data_cache.rs

@@ -30,6 +30,10 @@ pub type BlockIndex = u64;
 pub enum DataCacheError {
     #[error("IO error when reading or writing from cache: {0}")]
     IoFailure(#[source] anyhow::Error),
+    #[error("Block header was not valid: {0}")]
+    InvalidBlockHeader(String),
+    #[error("Block checksum was not present")]
+    BlockChecksumMissing,
     #[error("Block content was not valid/readable")]
     InvalidBlockContent,
     #[error("Block offset does not match block index")]