fix(ecstore): read provider tier version headers (#5041)

Refs rustfs/backlog#1358

Co-authored-by: heihutu <heihutu@gmail.com>

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();
 

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) {
+            version_id: self.raw_version_id(&h)?.unwrap_or_default().to_string(),
-                h_x_amz_version_id.to_str().unwrap_or("").to_string()
-            } else {
-                "".to_string()
-            },
             location: complete_multipart_upload_result.location,
             expiration: exp_time,
             expiration_rule_id: rule_id,

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) {
+            version_id: self.raw_version_id(h)?.unwrap_or_default().to_string(),
-                h_x_amz_version_id.to_str().unwrap_or("").to_string()
-            } else {
-                "".to_string()
-            },
             size,
             expiration: exp_time,
             expiration_rule_id: rule_id,

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
+            delete_marker_version_id: self.raw_version_id(resp.headers())?.unwrap_or_default().to_string(),
-                .headers()
-                .get(X_AMZ_VERSION_ID)
-                .and_then(|v| v.to_str().ok())
-                .unwrap_or_default()
-                .to_string(),
             ..Default::default()
         })
     }

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
+                            version_id: self
-                                .get(X_AMZ_VERSION_ID)
+                                .raw_version_id(h)?
-                                .and_then(|v| v.to_str().ok())
+                                .and_then(|s| Uuid::from_str(s).ok())
-                                .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
+                        version_id: self
-                            .get(X_AMZ_VERSION_ID)
+                            .raw_version_id(h)?
-                            .and_then(|v| v.to_str().ok())
+                            .and_then(|s| Uuid::from_str(s).ok())
-                            .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));

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;

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<Option<&str>, 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<RemoteVersion, Error> {
+        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()
+            );
+        }
+    }
+}

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<Option<&'a str>, 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<ObjectInfo, std::io::Error> {
+    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<ObjectInfo, std::io::Error> {
     // 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 = version_capabilities
-        let version_id_str = get_header("x-amz-version-id");
+        .remote_version(h, BucketVersioningState::Unknown)?
-        if !version_id_str.is_empty() {
+        .exact_id()
-            Some(Uuid::parse_str(&version_id_str).unwrap_or_else(|_| Uuid::nil()))
+        .and_then(|version_id| Uuid::parse_str(version_id).ok())
-        } else {
+        .filter(|version_id| !version_id.is_nil());
-            None
-        }
-    };
 
     // 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);
+    }
 }