diff --git a/crates/ecstore/src/client/api_get_object.rs b/crates/ecstore/src/client/api_get_object.rs index 598297b4a..14c9b44f4 100644 --- a/crates/ecstore/src/client/api_get_object.rs +++ b/crates/ecstore/src/client/api_get_object.rs @@ -30,7 +30,7 @@ use tokio_util::io::StreamReader; use crate::client::{ api_error_response::err_invalid_argument, api_get_options::GetObjectOptions, - transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info}, + transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info_for_provider}, }; use futures_util::StreamExt; use http_body_util::BodyExt; @@ -77,7 +77,8 @@ impl TransitionClient { ) .await?; - let object_stat = to_object_info(bucket_name, object_name, resp.headers())?; + let object_stat = + to_object_info_for_provider(bucket_name, object_name, resp.headers(), self.provider_version_capabilities())?; let h = resp.headers().clone(); diff --git a/crates/ecstore/src/client/api_put_object_multipart.rs b/crates/ecstore/src/client/api_put_object_multipart.rs index 8597b62b0..daebe97c2 100644 --- a/crates/ecstore/src/client/api_put_object_multipart.rs +++ b/crates/ecstore/src/client/api_put_object_multipart.rs @@ -41,7 +41,7 @@ use crate::client::{ transition_api::{ReaderImpl, RequestMetadata, TransitionClient, UploadInfo}, }; use rustfs_utils::path::trim_etag; -use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID}; +use s3s::header::X_AMZ_EXPIRATION; impl TransitionClient { pub async fn put_object_multipart( @@ -417,11 +417,7 @@ impl TransitionClient { bucket: complete_multipart_upload_result.bucket, key: complete_multipart_upload_result.key, etag: trim_etag(&complete_multipart_upload_result.etag), - version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) { - h_x_amz_version_id.to_str().unwrap_or("").to_string() - } else { - "".to_string() - }, + version_id: self.raw_version_id(&h)?.unwrap_or_default().to_string(), location: complete_multipart_upload_result.location, expiration: exp_time, expiration_rule_id: rule_id, diff --git a/crates/ecstore/src/client/api_put_object_streaming.rs b/crates/ecstore/src/client/api_put_object_streaming.rs index ac1766954..fdb1e820e 100644 --- a/crates/ecstore/src/client/api_put_object_streaming.rs +++ b/crates/ecstore/src/client/api_put_object_streaming.rs @@ -44,7 +44,7 @@ use crate::client::{ use crate::client::utils::base64_encode; use rustfs_utils::path::trim_etag; -use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID}; +use s3s::header::X_AMZ_EXPIRATION; /// Read exactly `want` bytes for a single multipart part, or fewer if the reader /// reaches EOF first. Advances the reader so the next call returns the following @@ -567,11 +567,7 @@ impl TransitionClient { key: object_name.to_string(), etag: trim_etag(h.get("ETag").and_then(|v| v.to_str().ok()).unwrap_or("")), - version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) { - h_x_amz_version_id.to_str().unwrap_or("").to_string() - } else { - "".to_string() - }, + version_id: self.raw_version_id(h)?.unwrap_or_default().to_string(), size, expiration: exp_time, expiration_rule_id: rule_id, diff --git a/crates/ecstore/src/client/api_remove.rs b/crates/ecstore/src/client/api_remove.rs index f192087da..24575ff7c 100644 --- a/crates/ecstore/src/client/api_remove.rs +++ b/crates/ecstore/src/client/api_remove.rs @@ -201,12 +201,7 @@ impl TransitionClient { object_name: object_name.to_string(), object_version_id: opts.version_id, delete_marker: resp.headers().get(X_AMZ_DELETE_MARKER).map_or(false, |v| v == "true"), - delete_marker_version_id: resp - .headers() - .get(X_AMZ_VERSION_ID) - .and_then(|v| v.to_str().ok()) - .unwrap_or_default() - .to_string(), + delete_marker_version_id: self.raw_version_id(resp.headers())?.unwrap_or_default().to_string(), ..Default::default() }) } diff --git a/crates/ecstore/src/client/api_stat.rs b/crates/ecstore/src/client/api_stat.rs index 621c619b7..c8557b4ce 100644 --- a/crates/ecstore/src/client/api_stat.rs +++ b/crates/ecstore/src/client/api_stat.rs @@ -31,7 +31,7 @@ use uuid::Uuid; use crate::client::{ api_error_response::{ErrorResponse, err_invalid_argument, http_resp_to_error_response}, api_get_options::GetObjectOptions, - transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info}, + transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info_for_provider}, }; use s3s::{ dto::VersioningConfiguration, @@ -206,20 +206,20 @@ impl TransitionClient { ..Default::default() }; return Ok(ObjectInfo { - version_id: h - .get(X_AMZ_VERSION_ID) - .and_then(|v| v.to_str().ok()) - .and_then(|s| Uuid::from_str(s).ok()), + version_id: self + .raw_version_id(h)? + .and_then(|s| Uuid::from_str(s).ok()) + .filter(|v| !v.is_nil()), is_delete_marker: delete_marker, ..Default::default() }); //err_resp } return Ok(ObjectInfo { - version_id: h - .get(X_AMZ_VERSION_ID) - .and_then(|v| v.to_str().ok()) - .and_then(|s| Uuid::from_str(s).ok()), + version_id: self + .raw_version_id(h)? + .and_then(|s| Uuid::from_str(s).ok()) + .filter(|v| !v.is_nil()), is_delete_marker: delete_marker, replication_ready: replication_ready, ..Default::default() @@ -227,7 +227,7 @@ impl TransitionClient { //http_resp_to_error_response(resp, bucket_name, object_name) } - Ok(to_object_info(bucket_name, object_name, h).expect("operation should succeed")) + to_object_info_for_provider(bucket_name, object_name, h, self.provider_version_capabilities()) } Err(err) => { return Err(std::io::Error::other(err)); diff --git a/crates/ecstore/src/client/mod.rs b/crates/ecstore/src/client/mod.rs index 78f4807f9..eebbe42cc 100644 --- a/crates/ecstore/src/client/mod.rs +++ b/crates/ecstore/src/client/mod.rs @@ -38,6 +38,7 @@ pub mod constants; pub mod credentials; pub mod object_api_utils; pub mod object_handlers_common; +pub(crate) mod provider_versions; pub(crate) mod runtime_sources; pub mod signer_error; pub mod transition_api; diff --git a/crates/ecstore/src/client/provider_versions.rs b/crates/ecstore/src/client/provider_versions.rs new file mode 100644 index 000000000..cad80ed4e --- /dev/null +++ b/crates/ecstore/src/client/provider_versions.rs @@ -0,0 +1,233 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::io::{Error, ErrorKind}; + +use http::HeaderMap; + +const X_AMZ_VERSION_ID: &str = "x-amz-version-id"; +const X_OSS_VERSION_ID: &str = "x-oss-version-id"; +const X_COS_VERSION_ID: &str = "x-cos-version-id"; +const X_OBS_VERSION_ID: &str = "x-obs-version-id"; +const MAX_REMOTE_VERSION_ID_LEN: usize = 1024; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum BucketVersioningState { + Unknown, + Disabled, + Suspended, + Enabled, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) enum RemoteVersion { + Unknown, + Disabled, + SuspendedNull, + Exact(String), +} + +impl RemoteVersion { + pub(crate) fn exact_id(&self) -> Option<&str> { + match self { + Self::SuspendedNull => Some("null"), + Self::Exact(version_id) => Some(version_id), + Self::Unknown | Self::Disabled => None, + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct ProviderVersionCapabilities { + raw_version_header: Option<&'static str>, + pub(crate) exact_get_delete: bool, +} + +impl ProviderVersionCapabilities { + pub(crate) fn for_tier_type(tier_type: &str) -> Self { + if tier_type.eq_ignore_ascii_case("s3") + || tier_type.eq_ignore_ascii_case("rustfs") + || tier_type.eq_ignore_ascii_case("minio") + || tier_type.eq_ignore_ascii_case("r2") + { + Self { + raw_version_header: Some(X_AMZ_VERSION_ID), + exact_get_delete: true, + } + } else if tier_type.eq_ignore_ascii_case("aliyun") { + Self { + raw_version_header: Some(X_OSS_VERSION_ID), + exact_get_delete: true, + } + } else if tier_type.eq_ignore_ascii_case("tencent") { + Self { + raw_version_header: Some(X_COS_VERSION_ID), + exact_get_delete: true, + } + } else if tier_type.eq_ignore_ascii_case("huaweicloud") { + Self { + raw_version_header: Some(X_OBS_VERSION_ID), + exact_get_delete: true, + } + } else { + Self { + raw_version_header: None, + exact_get_delete: false, + } + } + } + + pub(crate) fn raw_version_id(self, headers: &HeaderMap) -> Result, Error> { + let Some(header_name) = self.raw_version_header else { + return Ok(None); + }; + let Some(value) = headers.get(header_name) else { + return Ok(None); + }; + let value = value + .to_str() + .map_err(|_| Error::new(ErrorKind::InvalidData, "remote object version id is not valid ASCII"))?; + validate_remote_version_id(value)?; + Ok(Some(value)) + } + + pub(crate) fn remote_version(self, headers: &HeaderMap, versioning: BucketVersioningState) -> Result { + let Some(value) = self.raw_version_id(headers)? else { + return Ok(match versioning { + BucketVersioningState::Disabled => RemoteVersion::Disabled, + BucketVersioningState::Unknown | BucketVersioningState::Suspended | BucketVersioningState::Enabled => { + RemoteVersion::Unknown + } + }); + }; + if value == "null" { + return Ok(RemoteVersion::SuspendedNull); + } + Ok(RemoteVersion::Exact(value.to_string())) + } +} + +fn validate_remote_version_id(version_id: &str) -> Result<(), Error> { + if version_id.is_empty() { + return Err(Error::new( + ErrorKind::InvalidData, + "remote tier returned an empty object version id header", + )); + } + if version_id.len() > MAX_REMOTE_VERSION_ID_LEN { + return Err(Error::new( + ErrorKind::InvalidData, + "remote tier returned an oversized object version id header", + )); + } + if version_id.chars().any(char::is_control) { + return Err(Error::new( + ErrorKind::InvalidData, + "remote tier returned an object version id containing control characters", + )); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::{BucketVersioningState, ProviderVersionCapabilities, RemoteVersion}; + use http::{HeaderMap, HeaderValue}; + + #[test] + fn provider_version_header_matrix_preserves_opaque_versions() { + for (tier_type, header_name) in [ + ("s3", "x-amz-version-id"), + ("S3", "x-amz-version-id"), + ("rustfs", "x-amz-version-id"), + ("RustFS", "x-amz-version-id"), + ("minio", "x-amz-version-id"), + ("MinIO", "x-amz-version-id"), + ("r2", "x-amz-version-id"), + ("R2", "x-amz-version-id"), + ("aliyun", "x-oss-version-id"), + ("Aliyun", "x-oss-version-id"), + ("tencent", "x-cos-version-id"), + ("Tencent", "x-cos-version-id"), + ("huaweicloud", "x-obs-version-id"), + ("Huaweicloud", "x-obs-version-id"), + ] { + let mut headers = HeaderMap::new(); + headers.insert(header_name, HeaderValue::from_static("opaque.version_01")); + let capabilities = ProviderVersionCapabilities::for_tier_type(tier_type); + + assert_eq!(capabilities.raw_version_id(&headers).expect("raw version"), Some("opaque.version_01")); + assert_eq!( + capabilities + .remote_version(&headers, BucketVersioningState::Enabled) + .expect("remote version"), + RemoteVersion::Exact("opaque.version_01".to_string()) + ); + } + } + + #[test] + fn provider_version_header_matrix_does_not_cross_read_sibling_headers() { + let mut headers = HeaderMap::new(); + headers.insert("x-amz-version-id", HeaderValue::from_static("aws-version")); + headers.insert("x-cos-version-id", HeaderValue::from_static("cos-version")); + + assert_eq!( + ProviderVersionCapabilities::for_tier_type("s3") + .raw_version_id(&headers) + .expect("aws raw version"), + Some("aws-version") + ); + assert_eq!( + ProviderVersionCapabilities::for_tier_type("tencent") + .raw_version_id(&headers) + .expect("cos raw version"), + Some("cos-version") + ); + } + + #[test] + fn provider_version_missing_header_is_unknown_until_bucket_state_is_known() { + let headers = HeaderMap::new(); + let capabilities = ProviderVersionCapabilities::for_tier_type("aliyun"); + + assert_eq!( + capabilities + .remote_version(&headers, BucketVersioningState::Unknown) + .expect("unknown versioning"), + RemoteVersion::Unknown + ); + assert_eq!( + capabilities + .remote_version(&headers, BucketVersioningState::Disabled) + .expect("disabled versioning"), + RemoteVersion::Disabled + ); + } + + #[test] + fn provider_version_rejects_empty_or_oversized_headers() { + let oversized = "v".repeat(1025); + for bad in ["", oversized.as_str()] { + let mut headers = HeaderMap::new(); + headers.insert("x-oss-version-id", HeaderValue::from_str(bad).expect("test header value")); + + assert!( + ProviderVersionCapabilities::for_tier_type("aliyun") + .raw_version_id(&headers) + .is_err() + ); + } + } +} diff --git a/crates/ecstore/src/client/transition_api.rs b/crates/ecstore/src/client/transition_api.rs index 2ca654444..75fbcb206 100644 --- a/crates/ecstore/src/client/transition_api.rs +++ b/crates/ecstore/src/client/transition_api.rs @@ -31,6 +31,7 @@ use crate::client::{ }, constants::{UNSIGNED_PAYLOAD, UNSIGNED_PAYLOAD_TRAILER}, credentials::{CredContext, Credentials, SignatureType, Static}, + provider_versions::{BucketVersioningState, ProviderVersionCapabilities}, signer_error, }; use crate::{client::checksum::ChecksumMode, object_api::GetObjectReader}; @@ -309,6 +310,14 @@ impl TransitionClient { self.endpoint_url.clone() } + pub(crate) fn provider_version_capabilities(&self) -> ProviderVersionCapabilities { + ProviderVersionCapabilities::for_tier_type(&self.tier_type) + } + + pub(crate) fn raw_version_id<'a>(&self, headers: &'a HeaderMap) -> Result, std::io::Error> { + self.provider_version_capabilities().raw_version_id(headers) + } + fn trace_errors_only_off(&self) { if let Ok(mut trace_errors_only) = self.trace_errors_only.lock() { *trace_errors_only = false; @@ -1153,6 +1162,15 @@ impl Default for UploadInfo { /// This function parses various S3 response headers to construct an ObjectInfo struct /// containing metadata about an S3 object. pub fn to_object_info(bucket_name: &str, object_name: &str, h: &HeaderMap) -> Result { + to_object_info_for_provider(bucket_name, object_name, h, ProviderVersionCapabilities::for_tier_type("s3")) +} + +pub(crate) fn to_object_info_for_provider( + bucket_name: &str, + object_name: &str, + h: &HeaderMap, + version_capabilities: ProviderVersionCapabilities, +) -> Result { // Helper function to get header value as string let get_header = |name: &str| -> String { h.get(name).and_then(|val| val.to_str().ok()).unwrap_or("").to_string() }; @@ -1270,14 +1288,11 @@ pub fn to_object_info(bucket_name: &str, object_name: &str, h: &HeaderMap) -> Re }; // Extract version ID - let version_id = { - let version_id_str = get_header("x-amz-version-id"); - if !version_id_str.is_empty() { - Some(Uuid::parse_str(&version_id_str).unwrap_or_else(|_| Uuid::nil())) - } else { - None - } - }; + let version_id = version_capabilities + .remote_version(h, BucketVersioningState::Unknown)? + .exact_id() + .and_then(|version_id| Uuid::parse_str(version_id).ok()) + .filter(|version_id| !version_id.is_nil()); // Check if it's a delete marker let is_delete_marker = get_header("x-amz-delete-marker") == "true"; @@ -1392,8 +1407,13 @@ pub struct CreateBucketConfiguration { #[cfg(test)] mod tests { - use super::{SignatureType, build_tls_config, signer_error_to_io_error, validate_header_values, with_rustls_init_guard}; + use super::{ + SignatureType, build_tls_config, signer_error_to_io_error, to_object_info_for_provider, validate_header_values, + with_rustls_init_guard, + }; + use crate::client::provider_versions::ProviderVersionCapabilities; use http::{HeaderMap, HeaderValue}; + use uuid::Uuid; #[test] fn rustls_guard_converts_panics_to_io_errors() { @@ -1475,4 +1495,39 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); assert!(err.to_string().contains("x-amz-meta-invalid")); } + + #[test] + fn object_info_uses_provider_version_header_without_uuid_coercion() { + let mut headers = HeaderMap::new(); + headers.insert("x-cos-version-id", HeaderValue::from_static("opaque.version_01")); + + let info = + to_object_info_for_provider("bucket", "object", &headers, ProviderVersionCapabilities::for_tier_type("tencent")) + .expect("opaque provider version should parse"); + + assert_eq!(info.version_id, None); + assert_eq!( + info.metadata.get("x-cos-version-id").and_then(|value| value.to_str().ok()), + Some("opaque.version_01") + ); + } + + #[test] + fn object_info_keeps_s3_uuid_version_id_and_filters_nil() { + let version_id = Uuid::new_v4(); + let mut headers = HeaderMap::new(); + headers.insert( + "x-amz-version-id", + HeaderValue::from_str(&version_id.to_string()).expect("uuid header should be valid"), + ); + + let info = to_object_info_for_provider("bucket", "object", &headers, ProviderVersionCapabilities::for_tier_type("s3")) + .expect("s3 uuid version should parse"); + assert_eq!(info.version_id, Some(version_id)); + + headers.insert("x-amz-version-id", HeaderValue::from_static("00000000-0000-0000-0000-000000000000")); + let info = to_object_info_for_provider("bucket", "object", &headers, ProviderVersionCapabilities::for_tier_type("s3")) + .expect("nil s3 uuid version should parse"); + assert_eq!(info.version_id, None); + } }