diff --git a/quickwit/quickwit-storage/src/prefix_storage.rs b/quickwit/quickwit-storage/src/prefix_storage.rs index 312998ac91f..3242f24ea49 100644 --- a/quickwit/quickwit-storage/src/prefix_storage.rs +++ b/quickwit/quickwit-storage/src/prefix_storage.rs @@ -111,23 +111,37 @@ impl Storage for PrefixStorage { /// Makes listed paths relative to this storage root again, undoing the prefix added by /// [`PrefixStorage::list`]. fn strip_prefix_from_objects( - mut objects: Vec, + objects: Vec, prefix: &Path, ) -> StorageResult> { if prefix == Path::new("") { return Ok(objects); } - for object in &mut objects { - let relative_path = object.path.strip_prefix(prefix).map_err(|error| { - StorageErrorKind::Internal.with_error(anyhow::anyhow!( - "listed object `{}` is not under storage prefix `{}`: {error}", - object.path.display(), - prefix.display() - )) - })?; - object.path = relative_path.to_path_buf(); + let prefix_bytes = prefix.as_os_str().as_encoded_bytes(); + let mut relative_objects = Vec::with_capacity(objects.len()); + for mut object in objects { + match object.path.strip_prefix(prefix) { + Ok(relative_path) => { + object.path = relative_path.to_path_buf(); + relative_objects.push(object); + } + Err(error) => { + let is_under_prefix = object + .path + .as_os_str() + .as_encoded_bytes() + .starts_with(prefix_bytes); + if !is_under_prefix { + return Err(StorageErrorKind::Internal.with_error(anyhow::anyhow!( + "listed object `{}` is not under storage prefix `{}`: {error}", + object.path.display(), + prefix.display() + ))); + } + } + } } - Ok(objects) + Ok(relative_objects) } let storage_prefix = self.prefix.clone(); @@ -257,6 +271,64 @@ mod tests { assert_eq!(pages[0][0].size, bytesize::ByteSize(11)); } + #[tokio::test] + async fn test_prefix_storage_list_filters_lexical_siblings() { + let mut mock_storage = MockStorage::default(); + mock_storage.expect_list().times(1).returning(|prefix| { + assert_eq!(prefix, Path::new("ram:///indexes")); + let objects = vec![ + ObjectMetadata { + path: PathBuf::from("ram:///indexes/foo.split"), + size: bytesize::ByteSize(11), + last_modified: SystemTime::UNIX_EPOCH, + }, + ObjectMetadata { + path: PathBuf::from("ram:///indexes-old/unrelated.split"), + size: bytesize::ByteSize(13), + last_modified: SystemTime::UNIX_EPOCH, + }, + ]; + stream::once(async move { Ok(objects) }).boxed() + }); + let storage = add_prefix_to_storage( + Arc::new(mock_storage), + PathBuf::from("ram:///indexes"), + Uri::for_test("ram:///indexes"), + ); + let pages: Vec> = + storage.list(Path::new("")).try_collect().await.unwrap(); + + assert_eq!(pages.len(), 1); + assert_eq!(pages[0].len(), 1); + assert_eq!(pages[0][0].path, Path::new("foo.split")); + } + + #[tokio::test] + async fn test_prefix_storage_list_rejects_unrelated_paths() { + let mut mock_storage = MockStorage::default(); + mock_storage.expect_list().times(1).returning(|prefix| { + assert_eq!(prefix, Path::new("ram:///indexes")); + let objects = vec![ObjectMetadata { + path: PathBuf::from("ram:///unrelated/foo.split"), + size: bytesize::ByteSize(11), + last_modified: SystemTime::UNIX_EPOCH, + }]; + stream::once(async move { Ok(objects) }).boxed() + }); + let storage = add_prefix_to_storage( + Arc::new(mock_storage), + PathBuf::from("ram:///indexes"), + Uri::for_test("ram:///indexes"), + ); + let error = storage + .list(Path::new("")) + .try_collect::>>() + .await + .unwrap_err(); + + assert_eq!(error.kind(), StorageErrorKind::Internal); + } + #[test] fn test_strip_prefix_from_error() { {