Add an expected bucket owner test

mountpoint-s3/tests/common/cache.rs

@@ -25,6 +25,8 @@ struct CacheTestWrapperInner<Cache> {
     get_block_invalid_count: AtomicU64,
     /// Number of times the `put_block` was completed
     put_block_count: AtomicU64,
+    /// Number of times the `put_block` was completed with a failure
+    put_block_fail_count: AtomicU64,
 }
 
 impl<Cache> Clone for CacheTestWrapper<Cache> {
@@ -43,6 +45,7 @@ impl<Cache> CacheTestWrapper<Cache> {
                 get_block_hit_count: AtomicU64::new(0),
                 get_block_invalid_count: AtomicU64::new(0),
                 put_block_count: AtomicU64::new(0),
+                put_block_fail_count: AtomicU64::new(0),
             }),
         }
     }
@@ -74,6 +77,11 @@ impl<Cache> CacheTestWrapper<Cache> {
     pub fn put_block_count(&self) -> u64 {
         self.inner.put_block_count.load(Ordering::SeqCst)
     }
+
+    /// Number of times the `put_block` was completed with failure
+    pub fn put_block_fail_count(&self) -> u64 {
+        self.inner.put_block_fail_count.load(Ordering::SeqCst)
+    }
 }
 
 #[async_trait]
@@ -118,6 +126,9 @@ impl<Cache: DataCache + Send + Sync + 'static> DataCache for CacheTestWrapper<Ca
             .put_block(cache_key, block_idx, block_offset, bytes, object_size)
             .await;
         self.inner.put_block_count.fetch_add(1, Ordering::SeqCst);
+        if result.is_err() {
+            self.inner.put_block_fail_count.fetch_add(1, Ordering::SeqCst);
+        }
         result
     }
 

mountpoint-s3/tests/common/s3.rs

@@ -50,6 +50,17 @@ pub fn get_test_region() -> String {
     std::env::var("S3_REGION").expect("Set S3_REGION to run integration tests")
 }
 
+/// Account ID owning buckets specified in `S3_BUCKET_NAME` and `S3_EXPRESS_ONE_ZONE_BUCKET_NAME`
+pub fn get_bucket_owner() -> String {
+    std::env::var("S3_BUCKET_OWNER").expect("Set S3_BUCKET_OWNER to run integration tests")
+}
+
+/// A name of an S3 Express bucket which is owned by a different account (different to `S3_BUCKET_OWNER`)
+pub fn get_external_express_bucket() -> String {
+    std::env::var("S3_EXPRESS_ONE_ZONE_BUCKET_NAME_EXTERNAL")
+        .expect("Set S3_EXPRESS_ONE_ZONE_BUCKET_NAME_EXTERNAL to run integration tests")
+}
+
 /// Optional config for testing against a custom endpoint url
 pub fn get_test_endpoint_url() -> Option<String> {
     if cfg!(feature = "s3express_tests") {

mountpoint-s3/tests/fuse_tests/cache_test.rs

@@ -17,7 +17,9 @@ use tempfile::TempDir;
 use test_case::test_case;
 
 #[cfg(feature = "s3express_tests")]
-use crate::common::s3::{get_express_bucket, get_standard_bucket};
+use crate::common::s3::{
+    get_bucket_owner, get_express_bucket, get_external_express_bucket, get_standard_bucket, get_test_endpoint_config,
+};
 #[cfg(feature = "s3express_tests")]
 use mountpoint_s3::data_cache::{build_prefix, get_s3_key, BlockIndex, ExpressDataCache};
 #[cfg(feature = "s3express_tests")]
@@ -288,6 +290,67 @@ where
     assert!(block.is_none());
 }
 
+/// A test that checks that data is not written to the bucket owned by an unexpected account
+#[test_case(get_express_bucket(), true, true; "bucket owned by the expected account")]
+#[test_case(get_external_express_bucket(), true, false; "bucket owned by another account")]
+#[test_case(get_external_express_bucket(), false, false; "bucket owned by another account, not checked")]
+#[cfg(feature = "s3express_tests")]
+fn express_cache_expected_bucket_owner(cache_bucket: String, owner_checked: bool, owner_matches: bool) {
+    use futures::executor::block_on;
+    use mountpoint_s3_client::config::S3ClientConfig;
+    use mountpoint_s3_client::error::ObjectClientError;
+    use mountpoint_s3_client::S3RequestError;
+
+    let bucket_owner = get_bucket_owner();
+    // Configure the client to enforce the bucket owner
+    let mut client_config = S3ClientConfig::default()
+        .part_size(CLIENT_PART_SIZE)
+        .endpoint_config(get_test_endpoint_config())
+        .read_backpressure(true)
+        .initial_read_window(CLIENT_PART_SIZE);
+    if owner_checked {
+        client_config = client_config.bucket_owner(&bucket_owner);
+    }
+    let client = S3CrtClient::new(client_config).unwrap();
+
+    // Create cache and mount a bucket
+    let bucket = get_standard_bucket();
+    let prefix = get_test_prefix("express_expected_bucket_owner");
+    let cache = ExpressDataCache::new(client.clone(), Default::default(), &bucket, &cache_bucket);
+    let cache_valid = block_on(cache.verify_cache_valid());
+    if owner_checked && !owner_matches {
+        match cache_valid {
+            Err(ObjectClientError::ClientError(S3RequestError::Forbidden(..))) => (),
+            _ => panic!("expected S3RequestError::Forbidden, got: {:?}", cache_valid),
+        }
+    } else {
+        cache_valid.unwrap(); // Ok(()) is expected
+    }
+
+    let cache = CacheTestWrapper::new(cache);
+    let (mount_point, _session) = mount_bucket(client, cache.clone(), &bucket, &prefix);
+
+    // Write an object, no caching happens yet
+    let key = get_random_key(&prefix, "key", 100);
+    let path = mount_point.path().join(&key);
+    let written = random_binary_data(CACHE_BLOCK_SIZE as usize);
+    fs::write(&path, &written).expect("write should succeed");
+
+    // First read should be from the source bucket and be cached
+    let put_block_count = cache.put_block_count();
+    let read = fs::read(&path).expect("read should succeed");
+    assert_eq!(read, written);
+
+    // Cache population is async, wait for it to happen
+    cache.wait_for_put(Duration::from_secs(10), put_block_count);
+
+    if owner_checked && !owner_matches {
+        assert_eq!(cache.put_block_fail_count(), cache.put_block_count());
+    } else {
+        assert_eq!(cache.put_block_fail_count(), 0);
+    }
+}
+
 /// Generates random data of the specified size
 fn random_binary_data(size_in_bytes: usize) -> Vec<u8> {
     let seed = rand::thread_rng().gen();