Merge branch 'main' of https://github.com/rustfs/rustfs into feat/odm-azure-gcs-sources

# Conflicts:
#	crates/ecstore/src/bucket/on_demand_migration/source_client.rs

Cargo.lock

@@ -9792,6 +9792,7 @@ dependencies = [
  "path-absolutize",
  "pin-project-lite",
  "proptest",
+ "quick-xml",
  "rand 0.10.2",
  "ratelimit",
  "rcgen",

crates/ecstore/Cargo.toml

@@ -215,6 +215,7 @@ serde_urlencoded.workspace = true
 google-cloud-storage = { workspace = true }
 google-cloud-auth = { workspace = true }
 faster-hex = { workspace = true }
+quick-xml = { workspace = true }
 ratelimit = { workspace = true }
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["rustls-aws-lc"] }
 

crates/ecstore/src/api/mod.rs

@@ -153,12 +153,13 @@ pub mod bucket {
             LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup,
             OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason,
             PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS,
-            SourceLatencySnapshot, source_client_spec,
+            SourceLatencySnapshot, source_backend_spec, source_client_spec,
         };
         pub use crate::bucket::on_demand_migration::{
-            ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
-            OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
-            SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+            AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
+            ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig,
+            Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig,
+            ValidationContext,
         };
         pub use crate::bucket::on_demand_migration::{
             EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS,
@@ -184,9 +185,9 @@ pub mod bucket {
         }
         pub mod source_client {
             pub use crate::bucket::on_demand_migration::source_client::{
-                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
-                SourceProbe, SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
-                resolve_path_style,
+                AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError,
+                SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse,
+                SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style,
             };
         }
     }

crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs

@@ -0,0 +1,172 @@
+// 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.
+
+//! One contract every [`SourceBackend`] implementation must satisfy.
+//!
+//! The migration pipeline talks to a source only through the trait, so a new
+//! provider is correct exactly when it answers the same questions the same way:
+//! the same head fields, the same range semantics, the same page shape, the
+//! same error classes. Each backend supplies a fixture that answers this fixed
+//! corpus in its own dialect and then runs [`assert_backend_contract`], so a
+//! provider-specific mapping bug shows up as a contract failure rather than as
+//! a surprise in the pull pipeline.
+//!
+//! Backends differ in two documented ways, declared through
+//! [`BackendCapabilities`]: whether the provider's ETag is a content digest,
+//! and whether the provider can resume a listing from a key.
+
+use super::source_client::{SourceBackend, SourceError, SourceListRequest};
+use crate::storage_api_contracts::range::HTTPRangeSpec;
+use std::collections::HashMap;
+
+/// The single object every fixture serves.
+pub(super) const OBJECT_KEY: &str = "dir/a.txt";
+pub(super) const OBJECT_BODY: &[u8] = b"hello";
+/// MD5 of [`OBJECT_BODY`]; the ETag of the object on a digest provider.
+pub(super) const OBJECT_MD5: &str = "5d41402abc4b2a76b9719d911017c592";
+/// The second key the fixture's listing returns, on its second page.
+pub(super) const SECOND_KEY: &str = "dir/b.txt";
+pub(super) const COMMON_PREFIX: &str = "dir/sub/";
+pub(super) const LIST_CURSOR: &str = "cursor-1";
+/// A key the fixture answers with the provider's "no such object".
+pub(super) const MISSING_KEY: &str = "missing";
+/// A key the fixture answers with the provider's "not authorized".
+pub(super) const FORBIDDEN_KEY: &str = "secret";
+
+/// Where backends are allowed to differ.
+#[derive(Clone, Copy, Debug)]
+pub(super) struct BackendCapabilities {
+    /// The provider's ETag is an opaque token, not a digest of the bytes.
+    pub(super) etag_is_opaque: bool,
+    /// The provider can resume a listing from a key rather than only from an
+    /// opaque cursor.
+    pub(super) supports_start_after: bool,
+    /// The provider has an object-tagging concept at all. GCS does not, and
+    /// answers with an empty map instead of failing a pull.
+    pub(super) supports_tagging: bool,
+}
+
+/// Drives `backend` through the shared corpus. Fixtures are scripted in
+/// request order, so the call order here is part of the contract.
+pub(super) async fn assert_backend_contract(backend: &dyn SourceBackend, caps: BackendCapabilities) {
+    // 1. HEAD maps the object's shared fields.
+    let head = backend.head(OBJECT_KEY).await.expect("HEAD of the fixture object");
+    assert_eq!(head.size, OBJECT_BODY.len() as u64, "HEAD reports the object size");
+    assert_eq!(head.content_type.as_deref(), Some("text/plain"));
+    assert_eq!(
+        head.user_metadata,
+        HashMap::from([("owner".to_string(), "alice".to_string())]),
+        "user metadata is keyed without the provider prefix"
+    );
+    assert!(head.storage_class.is_some(), "the provider's tier is recorded");
+    assert!(head.last_modified.is_some(), "the provider's timestamp is parsed");
+    assert!(head.sse.is_none(), "the fixture object is not server-side encrypted");
+    assert!(!head.is_multipart_etag);
+    assert_eq!(head.etag_is_opaque, caps.etag_is_opaque);
+    match caps.etag_is_opaque {
+        false => assert_eq!(head.etag.as_deref(), Some(OBJECT_MD5), "a digest ETag is mapped verbatim"),
+        true => assert!(head.etag.is_some(), "an opaque ETag is still recorded"),
+    }
+
+    // 2. An unranged GET streams the whole object and reports no range.
+    let got = backend.get(OBJECT_KEY, None).await.expect("unranged GET");
+    assert_eq!(got.head.size, OBJECT_BODY.len() as u64);
+    assert!(got.content_range.is_none(), "an unranged GET has no content-range");
+    assert_eq!(got.head.etag_is_opaque, caps.etag_is_opaque, "GET and HEAD agree about the ETag");
+    let body = got.body.collect().await.expect("body streams").into_bytes();
+    assert_eq!(body.as_ref(), OBJECT_BODY);
+
+    // 3. A ranged GET returns exactly the requested interval, and `size` is
+    //    the length of the returned bytes rather than of the object.
+    let range = HTTPRangeSpec {
+        is_suffix_length: false,
+        start: 1,
+        end: 3,
+    };
+    let got = backend.get(OBJECT_KEY, Some(&range)).await.expect("ranged GET");
+    assert_eq!(got.head.size, 3, "a ranged GET reports the range length");
+    assert_eq!(got.content_range.as_deref(), Some("bytes 1-3/5"));
+    let body = got.body.collect().await.expect("body streams").into_bytes();
+    assert_eq!(body.as_ref(), &OBJECT_BODY[1..=3]);
+
+    // 4. A delimiter listing rolls prefixes up and hands back a cursor.
+    let page = backend
+        .list(&SourceListRequest {
+            prefix: Some("dir/"),
+            delimiter: Some("/"),
+            max_keys: 2,
+            ..Default::default()
+        })
+        .await
+        .expect("first listing page");
+    assert_eq!(page.objects.len(), 1, "the first page holds one object");
+    assert_eq!(page.objects[0].key, OBJECT_KEY, "listing keys are in the source namespace");
+    assert_eq!(page.objects[0].size, OBJECT_BODY.len() as u64);
+    assert!(page.objects[0].last_modified.is_some());
+    assert_eq!(page.common_prefixes, vec![COMMON_PREFIX.to_string()]);
+    assert!(page.is_truncated);
+    assert_eq!(page.next_continuation_token.as_deref(), Some(LIST_CURSOR));
+
+    // 5. The cursor is passed back verbatim and the last page ends the walk.
+    let page = backend
+        .list(&SourceListRequest {
+            prefix: Some("dir/"),
+            delimiter: Some("/"),
+            continuation_token: Some(LIST_CURSOR),
+            max_keys: 2,
+            ..Default::default()
+        })
+        .await
+        .expect("second listing page");
+    assert_eq!(page.objects.len(), 1);
+    assert_eq!(page.objects[0].key, SECOND_KEY);
+    assert!(!page.is_truncated);
+    assert!(page.next_continuation_token.is_none(), "a complete listing carries no cursor");
+
+    // 6. Tags come back as a flat map, empty on a provider without tags.
+    let tags = backend.tagging(OBJECT_KEY).await.expect("object tags");
+    match caps.supports_tagging {
+        true => assert_eq!(tags, HashMap::from([("env".to_string(), "prod".to_string())])),
+        false => assert!(tags.is_empty(), "a provider without tags reports none: {tags:?}"),
+    }
+
+    // 7. The probe confirms the bucket or container answers.
+    backend.probe().await.expect("probe of the fixture bucket");
+
+    // 8. A missing object is `NotFound`, and never retried.
+    let err = backend.head(MISSING_KEY).await.expect_err("a missing object must fail");
+    assert!(matches!(err, SourceError::NotFound), "{err:?}");
+    assert_eq!(err.class_label(), "not_found");
+    assert!(!err.is_retryable());
+
+    // 9. A denied object is `AccessDenied`, and never retried.
+    let err = backend.head(FORBIDDEN_KEY).await.expect_err("a denied object must fail");
+    assert!(matches!(err, SourceError::AccessDenied), "{err:?}");
+    assert_eq!(err.class_label(), "access_denied");
+    assert!(!err.is_retryable());
+
+    // 10. A provider without a key cursor must refuse one instead of listing
+    //     from the wrong position. This issues no request either way.
+    if !caps.supports_start_after {
+        let err = backend
+            .list(&SourceListRequest {
+                start_after: Some(OBJECT_KEY),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect_err("a backend without a key cursor must refuse start_after");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/config.rs

@@ -30,6 +30,10 @@ pub const ON_DEMAND_MIGRATION_CONFIG_VERSION: u32 = 1;
 const REDACTED: &str = "REDACTED";
 const AUTO_REGION: &str = "auto";
 const AUTO_REGION_FALLBACK: &str = "us-east-1";
+/// Public Azure Blob host suffix; the account name is the first label.
+pub const AZURE_BLOB_SUFFIX: &str = "blob.core.windows.net";
+/// Public Google Cloud Storage endpoint for the native provider.
+pub const GCS_DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com";
 
 const KIB: u64 = 1024;
 const MIB: u64 = 1024 * KIB;
@@ -75,14 +79,25 @@ pub struct SourceConfig {
     pub bucket: String,
     #[serde(default)]
     pub path_style: PathStyle,
-    /// `None` means anonymous access to a public source bucket.
+    /// `None` means anonymous access to a public source bucket. Only the
+    /// SigV4 providers read it; `azure` and `gcs_native` carry their own
+    /// credentials in `azure` / `gcs`.
     #[serde(default)]
     pub credentials: Option<SourceCredentials>,
     #[serde(default)]
     pub tls: TlsConfig,
+    /// Required for [`Provider::Azure`] and rejected for every other
+    /// provider.
+    #[serde(default)]
+    pub azure: Option<AzureSourceConfig>,
+    /// Required for [`Provider::GcsNative`] and rejected for every other
+    /// provider. [`Provider::Gcs`] keeps using `credentials` because it
+    /// speaks the S3 interoperability API.
+    #[serde(default)]
+    pub gcs: Option<GcsSourceConfig>,
 }
 
-/// Source vendor family. `azure` is deliberately absent from this version.
+/// Source vendor family.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Provider {
@@ -94,6 +109,12 @@ pub enum Provider {
     R2,
     /// GCS XML interoperability API with HMAC keys.
     Gcs,
+    /// Native Azure Blob service; parameters in `source.azure`.
+    Azure,
+    /// Native GCS JSON API with a service-account key; parameters in
+    /// `source.gcs`.
+    #[serde(rename = "gcs_native")]
+    GcsNative,
 }
 
 impl Provider {
@@ -105,13 +126,22 @@ impl Provider {
             Provider::Rustfs => "rustfs",
             Provider::R2 => "r2",
             Provider::Gcs => "gcs",
+            Provider::Azure => "azure",
+            Provider::GcsNative => "gcs_native",
         }
     }
 
+    /// Providers that do not speak S3 and therefore ignore `region`,
+    /// `path_style` and `credentials`.
+    pub fn is_native(&self) -> bool {
+        matches!(self, Provider::Azure | Provider::GcsNative)
+    }
+
     /// Providers whose SDKs accept `region = "auto"`; RustFS maps it to
-    /// `us-east-1` for signing.
+    /// `us-east-1` for signing. The native providers never sign with a
+    /// region, so they accept it as well.
     fn accepts_auto_region(&self) -> bool {
-        matches!(self, Provider::R2 | Provider::Minio | Provider::Rustfs)
+        matches!(self, Provider::R2 | Provider::Minio | Provider::Rustfs) || self.is_native()
     }
 }
 
@@ -164,6 +194,73 @@ impl fmt::Debug for SourceCredentials {
     }
 }
 
+/// Native Azure Blob source parameters. The container is `source.bucket`,
+/// so a config never carries two names for the same container. Exactly one
+/// of `account_key` and `sas_token` must be set: the account key signs with
+/// Shared Key, the SAS token is appended to every request URL.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct AzureSourceConfig {
+    /// Storage account name; also derives the default `blob.core.windows.net`
+    /// endpoint when `source.endpoint` is absent.
+    pub account: String,
+    /// Base64 shared key of the storage account.
+    #[serde(default)]
+    pub account_key: Option<String>,
+    /// SAS query string without the leading `?`.
+    #[serde(default)]
+    pub sas_token: Option<String>,
+}
+
+impl AzureSourceConfig {
+    /// A copy safe to return to admin clients or log: both secrets are
+    /// replaced by `REDACTED`, and whether each is set stays visible.
+    pub fn redacted(&self) -> Self {
+        Self {
+            account: self.account.clone(),
+            account_key: self.account_key.as_ref().map(|_| REDACTED.to_string()),
+            sas_token: self.sas_token.as_ref().map(|_| REDACTED.to_string()),
+        }
+    }
+}
+
+impl fmt::Debug for AzureSourceConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("AzureSourceConfig")
+            .field("account", &self.account)
+            .field("account_key", &self.account_key.as_ref().map(|_| REDACTED))
+            .field("sas_token", &self.sas_token.as_ref().map(|_| REDACTED))
+            .finish()
+    }
+}
+
+/// Native Google Cloud Storage source parameters. The bucket is
+/// `source.bucket`; only the service-account key lives here.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct GcsSourceConfig {
+    /// Service-account key JSON, verbatim as downloaded from Google Cloud.
+    pub service_account_json: String,
+}
+
+impl GcsSourceConfig {
+    /// A copy safe to return to admin clients or log: the whole key JSON is
+    /// a secret (it embeds the private key), so it is replaced wholesale.
+    pub fn redacted(&self) -> Self {
+        Self {
+            service_account_json: REDACTED.to_string(),
+        }
+    }
+}
+
+impl fmt::Debug for GcsSourceConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GcsSourceConfig")
+            .field("service_account_json", &REDACTED)
+            .finish()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct TlsConfig {
@@ -354,6 +451,14 @@ pub enum OnDemandMigrationConfigError {
     InvalidBucket(&'static str),
     #[error("source credentials field {0} must not be empty")]
     EmptyCredential(&'static str),
+    #[error("source.{0} is required for provider {1}")]
+    MissingProviderBlock(&'static str, Provider),
+    #[error("source.{0} is not valid for provider {1}")]
+    UnexpectedProviderBlock(&'static str, Provider),
+    /// Carries only the reason: the block holds account keys, SAS tokens and
+    /// service-account JSON, so no value of it is ever echoed.
+    #[error("source.{0} is invalid: {1}")]
+    InvalidProviderBlock(&'static str, &'static str),
     #[error("source tls.ca_cert_pem is not a PEM certificate")]
     InvalidCaCert,
     #[error("filter.{0} must be null or a non-empty string")]
@@ -388,6 +493,8 @@ impl OnDemandMigrationConfig {
     pub fn redacted(&self) -> Self {
         let mut copy = self.clone();
         copy.source.credentials = self.source.credentials.as_ref().map(SourceCredentials::redacted);
+        copy.source.azure = self.source.azure.as_ref().map(AzureSourceConfig::redacted);
+        copy.source.gcs = self.source.gcs.as_ref().map(GcsSourceConfig::redacted);
         copy
     }
 
@@ -433,6 +540,12 @@ impl SourceConfig {
         match (&self.endpoint, self.provider) {
             (Some(endpoint), _) => endpoint.clone(),
             (None, Provider::Aws) => format!("https://s3.{}.amazonaws.com", self.region),
+            (None, Provider::Azure) => self
+                .azure
+                .as_ref()
+                .map(|azure| format!("https://{}.{AZURE_BLOB_SUFFIX}", azure.account))
+                .unwrap_or_default(),
+            (None, Provider::GcsNative) => GCS_DEFAULT_ENDPOINT.to_string(),
             (None, _) => String::new(),
         }
     }
@@ -448,6 +561,8 @@ impl SourceConfig {
     }
 
     fn validate(&self) -> Result<(), OnDemandMigrationConfigError> {
+        self.validate_provider_block()?;
+
         if self.region.is_empty() {
             return Err(OnDemandMigrationConfigError::EmptyRegion);
         }
@@ -466,6 +581,9 @@ impl SourceConfig {
                     ));
                 }
             }
+            // Both native providers derive a fixed endpoint; Azure's is built
+            // from the account name, already checked by `validate_provider_block`.
+            None if self.provider.is_native() => {}
             None => return Err(OnDemandMigrationConfigError::MissingEndpoint(self.provider)),
         }
 
@@ -496,6 +614,84 @@ impl SourceConfig {
 
         Ok(())
     }
+
+    /// The provider-specific block must be present for exactly its own
+    /// provider: a stray `azure` block on an `s3` source would otherwise be
+    /// accepted, stored, and silently ignored by the client builder.
+    fn validate_provider_block(&self) -> Result<(), OnDemandMigrationConfigError> {
+        let missing = OnDemandMigrationConfigError::MissingProviderBlock;
+        let unexpected = OnDemandMigrationConfigError::UnexpectedProviderBlock;
+        let invalid = OnDemandMigrationConfigError::InvalidProviderBlock;
+
+        if self.provider != Provider::Azure && self.azure.is_some() {
+            return Err(unexpected("azure", self.provider));
+        }
+        if self.provider != Provider::GcsNative && self.gcs.is_some() {
+            return Err(unexpected("gcs", self.provider));
+        }
+
+        match self.provider {
+            Provider::Azure => {
+                let azure = self.azure.as_ref().ok_or(missing("azure", self.provider))?;
+                if azure.account.is_empty() {
+                    return Err(invalid("azure", "account must not be empty"));
+                }
+                // The account feeds a hostname when the endpoint is derived:
+                // keep it to label characters so it cannot rewrite the host.
+                if !azure.account.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') {
+                    return Err(invalid("azure", "account contains characters outside [A-Za-z0-9-]"));
+                }
+                match (azure.account_key.as_deref(), azure.sas_token.as_deref()) {
+                    (Some(_), Some(_)) => return Err(invalid("azure", "account_key and sas_token are mutually exclusive")),
+                    (None, None) => return Err(invalid("azure", "one of account_key and sas_token is required")),
+                    (Some(key), None) => {
+                        if key.is_empty() {
+                            return Err(invalid("azure", "account_key must not be empty"));
+                        }
+                        // Decoded here so a mistyped key fails at the admin
+                        // boundary instead of on the first source request.
+                        if base64_simd::STANDARD.decode_to_vec(key.as_bytes()).is_err() {
+                            return Err(invalid("azure", "account_key is not base64"));
+                        }
+                    }
+                    (None, Some(sas)) => {
+                        if sas.is_empty() {
+                            return Err(invalid("azure", "sas_token must not be empty"));
+                        }
+                        if sas.starts_with('?') {
+                            return Err(invalid("azure", "sas_token must not start with '?'"));
+                        }
+                        if sas.chars().any(char::is_whitespace) {
+                            return Err(invalid("azure", "sas_token must not contain whitespace"));
+                        }
+                    }
+                }
+            }
+            Provider::GcsNative => {
+                let gcs = self.gcs.as_ref().ok_or(missing("gcs", self.provider))?;
+                let key: serde_json::Value = serde_json::from_str(&gcs.service_account_json)
+                    .map_err(|_| invalid("gcs", "service_account_json is not valid JSON"))?;
+                let Some(object) = key.as_object() else {
+                    return Err(invalid("gcs", "service_account_json is not a JSON object"));
+                };
+                if object.get("type").and_then(serde_json::Value::as_str) != Some("service_account") {
+                    return Err(invalid("gcs", "service_account_json is not a service_account key"));
+                }
+                for field in ["client_email", "private_key"] {
+                    if object
+                        .get(field)
+                        .and_then(serde_json::Value::as_str)
+                        .is_none_or(str::is_empty)
+                    {
+                        return Err(invalid("gcs", "service_account_json is missing client_email or private_key"));
+                    }
+                }
+            }
+            Provider::S3 | Provider::Aws | Provider::Minio | Provider::Rustfs | Provider::R2 | Provider::Gcs => {}
+        }
+
+        Ok(())
+    }
 }
 
 fn validate_endpoint(endpoint: &str) -> Result<(), OnDemandMigrationConfigError> {
@@ -699,7 +895,15 @@ mod tests {
             ),
             (
                 "provider enum",
-                r#"{"source":{"provider":"azure","endpoint":"https://h","region":"r","bucket":"b"}}"#,
+                r#"{"source":{"provider":"swift","endpoint":"https://h","region":"r","bucket":"b"}}"#,
+            ),
+            (
+                "azure block",
+                r#"{"source":{"provider":"azure","region":"auto","bucket":"b","azure":{"account":"acct","account_key":"a2V5","extra":1}}}"#,
+            ),
+            (
+                "gcs block",
+                r#"{"source":{"provider":"gcs_native","region":"auto","bucket":"b","gcs":{"service_account_json":"{}","extra":1}}}"#,
             ),
         ] {
             let err = OnDemandMigrationConfig::from_json(json.as_bytes()).expect_err(label);
@@ -820,9 +1024,201 @@ mod tests {
                 "{provider}"
             );
         }
+        // The native providers never sign with a region, so "auto" is the
+        // honest value to write for them.
+        for cfg in [azure_cfg(), gcs_native_cfg()] {
+            assert_eq!(cfg.source.region, "auto");
+            cfg.validate(empty_ctx())
+                .unwrap_or_else(|err| panic!("{}: {err}", cfg.source.provider));
+        }
         assert_eq!(sample().source.effective_region(), "us-west-1");
     }
 
+    const SERVICE_ACCOUNT_JSON: &str = r#"{"type":"service_account","project_id":"p","client_email":"a@b.iam.gserviceaccount.com","private_key":"-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----\n"}"#;
+
+    fn azure_cfg() -> OnDemandMigrationConfig {
+        let mut cfg = sample();
+        cfg.source.provider = Provider::Azure;
+        cfg.source.endpoint = None;
+        cfg.source.region = "auto".to_string();
+        cfg.source.credentials = None;
+        cfg.source.azure = Some(AzureSourceConfig {
+            account: "legacyaccount".to_string(),
+            account_key: Some("c2VjcmV0LWtleQ==".to_string()),
+            sas_token: None,
+        });
+        cfg
+    }
+
+    fn gcs_native_cfg() -> OnDemandMigrationConfig {
+        let mut cfg = sample();
+        cfg.source.provider = Provider::GcsNative;
+        cfg.source.endpoint = None;
+        cfg.source.region = "auto".to_string();
+        cfg.source.credentials = None;
+        cfg.source.gcs = Some(GcsSourceConfig {
+            service_account_json: SERVICE_ACCOUNT_JSON.to_string(),
+        });
+        cfg
+    }
+
+    #[test]
+    fn native_providers_derive_their_endpoint_and_round_trip_on_the_wire() {
+        let azure = azure_cfg();
+        assert_eq!(azure.source.effective_endpoint(), "https://legacyaccount.blob.core.windows.net");
+        let gcs = gcs_native_cfg();
+        assert_eq!(gcs.source.effective_endpoint(), "https://storage.googleapis.com");
+
+        for cfg in [azure_cfg(), gcs_native_cfg()] {
+            let json = cfg.to_json().expect("config must serialize");
+            assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg);
+        }
+        // The wire labels are part of the admin contract.
+        assert!(
+            String::from_utf8(azure_cfg().to_json().expect("json"))
+                .expect("utf8")
+                .contains(r#""provider":"azure""#)
+        );
+        assert!(
+            String::from_utf8(gcs_native_cfg().to_json().expect("json"))
+                .expect("utf8")
+                .contains(r#""provider":"gcs_native""#)
+        );
+    }
+
+    #[test]
+    fn an_explicit_endpoint_overrides_the_derived_native_one() {
+        // Azurite and fake-gcs-server are addressed this way.
+        let mut cfg = azure_cfg();
+        cfg.source.endpoint = Some("http://azurite.example.com:10000".to_string());
+        cfg.validate(empty_ctx()).expect("an explicit native endpoint is allowed");
+        assert_eq!(cfg.source.effective_endpoint(), "http://azurite.example.com:10000");
+
+        cfg.source.endpoint = Some("http://azurite.example.com:10000/devstoreaccount1".to_string());
+        assert!(
+            matches!(cfg.validate(empty_ctx()), Err(OnDemandMigrationConfigError::InvalidEndpoint(_))),
+            "a native endpoint is still an origin"
+        );
+    }
+
+    #[test]
+    fn a_provider_block_belongs_to_exactly_its_own_provider() {
+        let mut cfg = sample();
+        cfg.source.azure = azure_cfg().source.azure;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("azure", Provider::S3))
+        );
+
+        let mut cfg = sample();
+        cfg.source.gcs = gcs_native_cfg().source.gcs;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("gcs", Provider::S3))
+        );
+
+        let mut cfg = azure_cfg();
+        cfg.source.azure = None;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::MissingProviderBlock("azure", Provider::Azure))
+        );
+
+        let mut cfg = gcs_native_cfg();
+        cfg.source.gcs = None;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::MissingProviderBlock("gcs", Provider::GcsNative))
+        );
+    }
+
+    #[test]
+    fn azure_block_rules() {
+        let with = |account: &str, key: Option<&str>, sas: Option<&str>| {
+            let mut cfg = azure_cfg();
+            cfg.source.azure = Some(AzureSourceConfig {
+                account: account.to_string(),
+                account_key: key.map(str::to_string),
+                sas_token: sas.map(str::to_string),
+            });
+            cfg.validate(empty_ctx())
+        };
+
+        with("legacyaccount", None, Some("sv=2021-08-06&sig=abc%3D")).expect("a SAS token is a complete credential");
+        with("legacyaccount", Some("c2VjcmV0LWtleQ=="), None).expect("an account key is a complete credential");
+
+        for (label, result) in [
+            ("empty account", with("", Some("c2VjcmV0LWtleQ=="), None)),
+            // The account becomes the first label of the derived hostname.
+            ("account with a dot", with("legacy.account", Some("c2VjcmV0LWtleQ=="), None)),
+            ("account with a slash", with("legacy/account", Some("c2VjcmV0LWtleQ=="), None)),
+            ("no credential", with("legacyaccount", None, None)),
+            ("both credentials", with("legacyaccount", Some("c2VjcmV0LWtleQ=="), Some("sv=1"))),
+            ("empty key", with("legacyaccount", Some(""), None)),
+            ("key that is not base64", with("legacyaccount", Some("not base64!"), None)),
+            ("empty sas", with("legacyaccount", None, Some(""))),
+            ("sas with a leading question mark", with("legacyaccount", None, Some("?sv=1"))),
+            ("sas with whitespace", with("legacyaccount", None, Some("sv=1 &sig=a"))),
+        ] {
+            assert!(
+                matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("azure", _))),
+                "{label}: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn gcs_native_block_requires_a_usable_service_account_key() {
+        let with = |json: &str| {
+            let mut cfg = gcs_native_cfg();
+            cfg.source.gcs = Some(GcsSourceConfig {
+                service_account_json: json.to_string(),
+            });
+            cfg.validate(empty_ctx())
+        };
+
+        with(SERVICE_ACCOUNT_JSON).expect("a service-account key is accepted");
+        for (label, json) in [
+            ("empty", ""),
+            ("not json", "not json"),
+            ("not an object", "[]"),
+            ("wrong type", r#"{"type":"authorized_user","client_email":"a@b","private_key":"k"}"#),
+            ("no private key", r#"{"type":"service_account","client_email":"a@b"}"#),
+            ("empty client email", r#"{"type":"service_account","client_email":"","private_key":"k"}"#),
+        ] {
+            let result = with(json);
+            assert!(
+                matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("gcs", _))),
+                "{label}: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn native_secrets_never_survive_redaction_or_debug() {
+        let mut azure = azure_cfg();
+        azure.source.azure.as_mut().expect("block").sas_token = Some("sv=2021-08-06&sig=top-secret".to_string());
+        azure.source.azure.as_mut().expect("block").account_key = None;
+        let gcs = gcs_native_cfg();
+
+        for rendered in [
+            format!("{:?}", azure.redacted()),
+            format!("{azure:?}"),
+            String::from_utf8(azure.redacted().to_json().expect("json")).expect("utf8"),
+        ] {
+            assert!(!rendered.contains("top-secret"), "{rendered}");
+            assert!(rendered.contains("legacyaccount"), "the account name is not a secret: {rendered}");
+        }
+        for rendered in [
+            format!("{:?}", gcs.redacted()),
+            format!("{gcs:?}"),
+            String::from_utf8(gcs.redacted().to_json().expect("json")).expect("utf8"),
+        ] {
+            assert!(!rendered.contains("BEGIN PRIVATE KEY"), "{rendered}");
+            assert!(!rendered.contains("gserviceaccount"), "{rendered}");
+        }
+    }
+
     #[test]
     fn bucket_rules() {
         let mut cfg = sample();

crates/ecstore/src/bucket/on_demand_migration/gcs.rs

@@ -0,0 +1,506 @@
+// 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.
+
+//! Native Google Cloud Storage source backend.
+//!
+//! The `gcs` provider already reaches GCS through its S3 interoperability API,
+//! which needs an HMAC key pair. This backend is the other half: it authorizes
+//! with a service-account key, the credential most GCS projects actually issue,
+//! by minting OAuth tokens through the shared `google-cloud-auth` credential
+//! machinery the tier layer already uses.
+//!
+//! Two GCS surfaces are involved, each for the half it describes best. The read
+//! path uses the XML API (`/{bucket}/{object}`), whose responses carry
+//! `x-goog-meta-*` user metadata and the `x-goog-hash` digest in one round trip.
+//! Listing uses the JSON API (`objects.list`), whose `pageToken` maps directly
+//! onto the shared page cursor and whose `prefixes` are the delimiter roll-up.
+//! Both accept the same bearer token.
+//!
+//! Every call this backend makes needs only `storage.objects.get` and
+//! `storage.objects.list`, the two permissions of the `objectViewer` role, so a
+//! key scoped to exactly the migration's needs works.
+//!
+//! `x-goog-hash` carries a base64 MD5 for every non-composite object; it is
+//! converted to hex and becomes the head's ETag, so a pulled object is checked
+//! against the digest GCS itself computed. A composite object has no MD5, and
+//! its ETag is then marked opaque rather than checked.
+
+use super::native_http::{
+    NativeHeadFields, NativeHttp, base64_md5_to_hex, header, native_source_head, parse_http_timestamp, read_text, response_body,
+};
+use super::source_client::{
+    GcsSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
+    SourceTimeouts, range_header_value,
+};
+use crate::bucket::remote_s3_client::RemoteS3ClientError;
+use crate::storage_api_contracts::range::HTTPRangeSpec;
+use google_cloud_auth::credentials::service_account::{AccessSpecifier, Builder as ServiceAccountBuilder};
+use google_cloud_auth::credentials::{CacheableResource, Credentials};
+use http::{HeaderMap, HeaderValue, Method};
+use serde::Deserialize;
+use std::collections::HashMap;
+use url::Url;
+
+/// Read-only object scope: this backend never writes to the source.
+const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only";
+const METADATA_PREFIX: &str = "x-goog-meta-";
+/// GCS reports its error code in the response body, not a header; the shared
+/// transport takes a header name, so it is given one that never matches and
+/// classification falls back to the status.
+const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code";
+/// One `objects.list` page is small; refuse an unbounded document.
+const MAX_JSON_BYTES: usize = 8 * 1024 * 1024;
+
+pub struct GcsNativeSourceBackend {
+    http: NativeHttp,
+    bucket: String,
+    credentials: Credentials,
+}
+
+impl GcsNativeSourceBackend {
+    pub fn new(
+        endpoint: &str,
+        bucket: &str,
+        spec: &GcsSourceSpec,
+        timeouts: SourceTimeouts,
+        skip_tls_verify: bool,
+        ca_cert_pem: Option<&str>,
+    ) -> Result<Self, RemoteS3ClientError> {
+        let key: serde_json::Value = serde_json::from_str(&spec.service_account_json)
+            .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not valid JSON"))?;
+        let credentials = ServiceAccountBuilder::new(key)
+            .with_access_specifier(AccessSpecifier::from_scopes([READ_ONLY_SCOPE]))
+            .build()
+            .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not usable"))?;
+        Ok(Self {
+            http: NativeHttp::new(endpoint, timeouts, skip_tls_verify, ca_cert_pem)?,
+            bucket: bucket.to_string(),
+            credentials,
+        })
+    }
+
+    /// Authorization headers for one request. A credential failure is reported
+    /// as `AccessDenied` with no message: the renderer of a credential error
+    /// has the key material in scope, and the class is what callers act on.
+    async fn auth_headers(&self) -> Result<HeaderMap, SourceError> {
+        match self.credentials.headers(http::Extensions::new()).await {
+            Ok(CacheableResource::New { data, .. }) => Ok(data),
+            // Only returned when the caller passes an entity tag, which this
+            // backend never does; an empty set is still the honest answer.
+            Ok(CacheableResource::NotModified) => Ok(HeaderMap::new()),
+            Err(_) => Err(SourceError::AccessDenied),
+        }
+    }
+
+    /// XML API URL of one object; `/` in the key stay path separators.
+    fn object_url(&self, key: &str) -> Result<Url, SourceError> {
+        self.http.url(std::iter::once(self.bucket.as_str()).chain(key.split('/')))
+    }
+
+    /// JSON API URL of the bucket's object collection.
+    fn objects_url(&self) -> Result<Url, SourceError> {
+        self.http.url(["storage", "v1", "b", self.bucket.as_str(), "o"])
+    }
+
+    async fn request(&self, method: Method, url: Url, mut headers: HeaderMap) -> Result<reqwest::Request, SourceError> {
+        for (name, value) in self.auth_headers().await? {
+            if let Some(name) = name {
+                headers.insert(name, value);
+            }
+        }
+        let mut request = reqwest::Request::new(method, url);
+        *request.headers_mut() = headers;
+        Ok(request)
+    }
+
+    /// Shared mapping for the XML API's HEAD and GET responses.
+    fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
+        if header(headers, "x-goog-encryption-key-sha256").is_some() {
+            return Err(SourceError::Unsupported(
+                "source object uses a customer-supplied encryption key; customer-key sources are not supported".to_string(),
+            ));
+        }
+        // `x-goog-hash` lists digests as `name=base64`, comma separated, and may
+        // repeat across header lines. Only the MD5 describes the whole object.
+        let md5 = headers
+            .get_all("x-goog-hash")
+            .iter()
+            .filter_map(|value| value.to_str().ok())
+            .flat_map(|value| value.split(','))
+            .filter_map(|digest| digest.trim().strip_prefix("md5="))
+            .find_map(base64_md5_to_hex);
+
+        let (etag, etag_is_opaque) = match md5 {
+            Some(md5) => (Some(md5), false),
+            // A composite object has no MD5; its ETag describes the composition
+            // rather than the bytes, so it is provenance only.
+            None => (header(headers, "etag").map(str::to_string), true),
+        };
+        native_source_head(
+            headers,
+            METADATA_PREFIX,
+            NativeHeadFields {
+                etag,
+                etag_is_opaque,
+                version_id: header(headers, "x-goog-generation").map(str::to_string),
+                storage_class: header(headers, "x-goog-storage-class").map(str::to_string),
+            },
+        )
+    }
+}
+
+#[async_trait::async_trait]
+impl SourceBackend for GcsNativeSourceBackend {
+    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
+        let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        Self::head_from_response(response.headers())
+    }
+
+    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+        let mut headers = HeaderMap::new();
+        if let Some(range) = range.map(range_header_value).transpose()? {
+            headers.insert(
+                http::header::RANGE,
+                HeaderValue::from_str(&range).map_err(|_| SourceError::Other("invalid range header".to_string()))?,
+            );
+        }
+        let request = self.request(Method::GET, self.object_url(key)?, headers).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let head = Self::head_from_response(response.headers())?;
+        let content_range = header(response.headers(), "content-range").map(str::to_string);
+        Ok(SourceGet {
+            head,
+            body: response_body(response),
+            content_range,
+        })
+    }
+
+    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+        // `objects.list` offers `startOffset`, which is inclusive, so it cannot
+        // express "resume after this key" without silently repeating it.
+        if request.start_after.is_some() {
+            return Err(SourceError::Unsupported(
+                "gcs sources cannot resume a listing from a key; use the continuation token".to_string(),
+            ));
+        }
+        let mut url = self.objects_url()?;
+        {
+            let mut query = url.query_pairs_mut();
+            if let Some(prefix) = request.prefix.filter(|prefix| !prefix.is_empty()) {
+                query.append_pair("prefix", prefix);
+            }
+            if let Some(delimiter) = request.delimiter.filter(|delimiter| !delimiter.is_empty()) {
+                query.append_pair("delimiter", delimiter);
+            }
+            if let Some(token) = request.continuation_token.filter(|token| !token.is_empty()) {
+                query.append_pair("pageToken", token);
+            }
+            if request.max_keys > 0 {
+                query.append_pair("maxResults", &request.max_keys.to_string());
+            }
+        }
+
+        let request = self.request(Method::GET, url, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let body = read_text(response, MAX_JSON_BYTES).await?;
+        parse_objects_list(&body)
+    }
+
+    /// GCS has no object tagging API; user metadata is already carried by the
+    /// head mapping. An empty map keeps `policy.copy_tags` from failing a pull
+    /// over a concept the provider does not have.
+    async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
+        Ok(HashMap::new())
+    }
+
+    /// A one-object listing, not `buckets.get`: the migration pipeline only
+    /// ever needs `storage.objects.list` and `storage.objects.get`, and a key
+    /// scoped to exactly those (the `objectViewer` role) cannot read the bucket
+    /// resource. Probing with `buckets.get` would reject a correct key.
+    async fn probe(&self) -> Result<(), SourceError> {
+        let mut url = self.objects_url()?;
+        url.query_pairs_mut().append_pair("maxResults", "1");
+        let request = self.request(Method::GET, url, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        read_text(response, MAX_JSON_BYTES)
+            .await
+            .and_then(|body| parse_objects_list(&body))?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ObjectsList {
+    #[serde(default)]
+    items: Vec<ListedObject>,
+    #[serde(default)]
+    prefixes: Vec<String>,
+    #[serde(default)]
+    next_page_token: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ListedObject {
+    name: String,
+    /// GCS renders the size as a decimal string, not a JSON number.
+    #[serde(default)]
+    size: Option<String>,
+    #[serde(default)]
+    updated: Option<String>,
+    #[serde(default)]
+    md5_hash: Option<String>,
+    #[serde(default)]
+    etag: Option<String>,
+    #[serde(default)]
+    storage_class: Option<String>,
+}
+
+fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
+    let listing: ObjectsList =
+        serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
+    let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
+    let objects = listing
+        .items
+        .into_iter()
+        .map(|item| {
+            let etag = item
+                .md5_hash
+                .as_deref()
+                .and_then(base64_md5_to_hex)
+                .or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string()))
+                .filter(|etag| !etag.is_empty());
+            SourceObject {
+                key: item.name,
+                etag,
+                size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0),
+                last_modified: item.updated.as_deref().and_then(parse_http_timestamp),
+                storage_class: item.storage_class,
+                // GCS never encodes a part count in a digest or an ETag.
+                is_multipart_etag: false,
+            }
+        })
+        .collect();
+
+    Ok(SourcePage {
+        objects,
+        common_prefixes: listing.prefixes,
+        is_truncated: next_continuation_token.is_some(),
+        next_continuation_token,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
+    use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
+    use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder;
+
+    const LIST_PAGE_ONE: &str = r#"{
+      "kind": "storage#objects",
+      "nextPageToken": "cursor-1",
+      "prefixes": ["dir/sub/"],
+      "items": [
+        {
+          "name": "dir/a.txt",
+          "size": "5",
+          "updated": "2015-10-21T07:28:00.000Z",
+          "md5Hash": "XUFAKrxLKna5cZ2REBfFkg==",
+          "etag": "CJizy9Wq0McCEAE=",
+          "storageClass": "STANDARD"
+        }
+      ]
+    }"#;
+
+    const LIST_PAGE_TWO: &str = r#"{
+      "kind": "storage#objects",
+      "items": [
+        {
+          "name": "dir/b.txt",
+          "size": "7",
+          "updated": "2015-10-21T07:28:00.000Z",
+          "etag": "\"CJizy9Wq0McCEAI=\""
+        }
+      ]
+    }"#;
+
+    fn backend(endpoint: &Url) -> GcsNativeSourceBackend {
+        GcsNativeSourceBackend {
+            http: NativeHttp::for_test(endpoint.clone()),
+            bucket: "legacy".to_string(),
+            // Anonymous credentials add no headers, so the fixture sees exactly
+            // the request this backend builds.
+            credentials: AnonymousBuilder::new().build(),
+        }
+    }
+
+    fn object_headers() -> Vec<(&'static str, String)> {
+        vec![
+            ("Content-Type", "text/plain".to_string()),
+            ("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
+            ("ETag", "\"CJizy9Wq0McCEAE=\"".to_string()),
+            ("x-goog-hash", "crc32c=AAAAAA==,md5=XUFAKrxLKna5cZ2REBfFkg==".to_string()),
+            ("x-goog-meta-owner", "alice".to_string()),
+            ("x-goog-storage-class", "STANDARD".to_string()),
+            ("x-goog-generation", "1445412480000000".to_string()),
+        ]
+    }
+
+    #[test]
+    fn objects_list_maps_items_prefixes_and_the_page_token() {
+        let page = parse_objects_list(LIST_PAGE_ONE).expect("page should parse");
+        assert_eq!(page.common_prefixes, vec!["dir/sub/"]);
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("cursor-1"));
+        assert_eq!(page.objects.len(), 1);
+        assert_eq!(page.objects[0].key, "dir/a.txt");
+        assert_eq!(page.objects[0].size, 5, "the string size is parsed");
+        assert_eq!(
+            page.objects[0].etag.as_deref(),
+            Some("5d41402abc4b2a76b9719d911017c592"),
+            "the base64 md5Hash becomes a hex ETag"
+        );
+        assert_eq!(page.objects[0].storage_class.as_deref(), Some("STANDARD"));
+        assert!(page.objects[0].last_modified.is_some(), "RFC 3339 `updated` is parsed");
+
+        let page = parse_objects_list(LIST_PAGE_TWO).expect("page should parse");
+        assert!(!page.is_truncated);
+        assert!(page.next_continuation_token.is_none());
+        assert_eq!(
+            page.objects[0].etag.as_deref(),
+            Some("CJizy9Wq0McCEAI="),
+            "without md5Hash the raw etag is carried"
+        );
+
+        assert!(parse_objects_list("not json").is_err());
+    }
+
+    #[tokio::test]
+    async fn head_prefers_the_goog_hash_md5_over_the_etag() {
+        let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, object_headers(), String::new())]).await;
+        let head = backend(&endpoint).head("dir/a b.txt").await.expect("HEAD should map");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert_eq!(recorded[0].method, "HEAD");
+        assert_eq!(recorded[0].target, "/legacy/dir/a%20b.txt", "the XML API addresses the object by path");
+        assert_eq!(
+            head.etag.as_deref(),
+            Some("5d41402abc4b2a76b9719d911017c592"),
+            "the x-goog-hash md5 is the content digest"
+        );
+        assert!(!head.etag_is_opaque, "a GCS md5 may be checked against the pulled bytes");
+        assert_eq!(head.user_metadata, HashMap::from([("owner".to_string(), "alice".to_string())]));
+        assert_eq!(head.version_id.as_deref(), Some("1445412480000000"));
+        assert_eq!(head.storage_class.as_deref(), Some("STANDARD"));
+    }
+
+    #[tokio::test]
+    async fn a_composite_object_without_an_md5_keeps_an_opaque_etag() {
+        let headers = object_headers()
+            .into_iter()
+            .map(|(name, value)| {
+                if name == "x-goog-hash" {
+                    (name, "crc32c=AAAAAA==".to_string())
+                } else {
+                    (name, value)
+                }
+            })
+            .collect();
+        let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
+        let head = backend(&endpoint).head("composed").await.expect("HEAD should map");
+        assert_eq!(head.etag.as_deref(), Some("CJizy9Wq0McCEAE="));
+        assert!(head.etag_is_opaque, "a composite ETag describes the composition, not the bytes");
+    }
+
+    #[tokio::test]
+    async fn customer_supplied_key_objects_are_refused() {
+        let mut headers = object_headers();
+        headers.push(("x-goog-encryption-key-sha256", "abc".to_string()));
+        let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
+        let err = backend(&endpoint)
+            .head("a.txt")
+            .await
+            .expect_err("CSEK objects are unsupported");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn list_and_probe_address_the_json_api() {
+        let (endpoint, recorded) = scripted_server(vec![
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
+        ])
+        .await;
+        let backend = backend(&endpoint);
+
+        backend
+            .list(&SourceListRequest {
+                prefix: Some("dir/"),
+                delimiter: Some("/"),
+                continuation_token: Some("cursor-0"),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("listing should succeed");
+        backend.probe().await.expect("probe should succeed");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert!(recorded[0].target.starts_with("/storage/v1/b/legacy/o?"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("prefix=dir%2F"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("delimiter=%2F"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("pageToken=cursor-0"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("maxResults=2"), "{}", recorded[0].target);
+        assert_eq!(
+            recorded[1].target, "/storage/v1/b/legacy/o?maxResults=1",
+            "the probe uses the listing permission the pipeline already needs"
+        );
+    }
+
+    #[tokio::test]
+    async fn gcs_native_backend_satisfies_the_shared_backend_contract() {
+        let mut ranged = object_headers();
+        ranged.push(("Content-Range", "bytes 1-3/5".to_string()));
+        // A HEAD reports the object size with no body, exactly as GCS does.
+        let mut head_only = object_headers();
+        head_only.push(("Content-Length", "5".to_string()));
+        let (endpoint, _) = scripted_server(vec![
+            ScriptedResponse::new(200, head_only, String::new()),
+            ScriptedResponse::new(200, object_headers(), "hello".to_string()),
+            ScriptedResponse::new(206, ranged, "ell".to_string()),
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_TWO.to_string()),
+            // GCS has no tagging call, so the contract's tag step issues no
+            // request; the probe is the next one on the wire.
+            ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
+            ScriptedResponse::new(404, Vec::new(), String::new()),
+            ScriptedResponse::new(403, Vec::new(), String::new()),
+        ])
+        .await;
+
+        assert_backend_contract(
+            &backend(&endpoint),
+            BackendCapabilities {
+                etag_is_opaque: false,
+                supports_start_after: false,
+                // GCS objects have no tags; the contract's tag step is skipped.
+                supports_tagging: false,
+            },
+        )
+        .await;
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -19,25 +19,36 @@
 //! client, and the per-node runtime (`sys`) that turns configs into live
 //! clients guarded by a breaker, a negative cache, singleflight and a pull
 //! concurrency limit (rustfs/backlog#2147).
+//!
+//! A source is reached through one `SourceBackend`: the S3 dialect for every
+//! S3-compatible provider, and a native backend for the providers that have no
+//! S3 API (`azure`, `gcs_native`).
 
+pub mod azure;
+#[cfg(test)]
+mod backend_contract;
 pub mod backfill;
 pub mod breaker;
 pub mod config;
+pub mod gcs;
 pub mod list_through;
+mod native_http;
 pub mod negative_cache;
 pub mod pull;
 pub mod source_client;
 pub mod stats;
 pub mod sys;
+#[cfg(test)]
+mod test_http_fixture;
 
 pub use breaker::{
     BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker,
     BreakerState, BreakerTransition, BreakerVerdict,
 };
 pub use config::{
-    ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
-    OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
-    SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+    AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
+    ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider,
+    RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
 };
 pub use list_through::{
     FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
@@ -56,5 +67,6 @@ pub use stats::{
 };
 pub use sys::{
     ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError,
-    OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec,
+    OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec,
+    source_client_spec,
 };

crates/ecstore/src/bucket/on_demand_migration/native_http.rs

@@ -0,0 +1,415 @@
+// 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.
+
+//! Shared HTTP transport for the on-demand migration source backends that do
+//! not speak S3 (Azure Blob, native GCS).
+//!
+//! The S3 backend rides the AWS SDK; these providers have no SigV4 dialect, so
+//! they talk plain HTTP through one `reqwest` client that carries the same
+//! connect/read timeouts and TLS policy the operator configured for the source.
+//! Redirects are refused: the endpoint passed the outbound policy gate once, and
+//! following a source-chosen `Location` would leave that gate behind.
+//!
+//! Errors never render the request URL. A SAS token lives in the query string,
+//! so a `reqwest` error rendered with its URL would print the credential into
+//! the log line and the admin response.
+
+use super::source_client::{SourceError, SourceHead, SourceTimeouts, USER_AGENT_SUFFIX, classify_status, is_multipart_etag};
+use crate::bucket::remote_s3_client::{RemoteS3ClientError, validate_remote_endpoint, validate_target_ca_pem};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_smithy_types::body::SdkBody;
+use futures::StreamExt;
+use http::HeaderMap;
+use std::collections::HashMap;
+use std::time::SystemTime;
+use time::OffsetDateTime;
+use time::format_description::well_known::{Rfc2822, Rfc3339};
+use url::Url;
+
+/// Origin the native backends are allowed to address, plus the HTTP client
+/// that reaches it.
+pub(super) struct NativeHttp {
+    client: reqwest::Client,
+    endpoint: Url,
+}
+
+impl NativeHttp {
+    /// `endpoint` must be a bare `scheme://host[:port]` origin; it is checked
+    /// against the outbound policy exactly like an S3 source endpoint.
+    pub(super) fn new(
+        endpoint: &str,
+        timeouts: SourceTimeouts,
+        skip_tls_verify: bool,
+        ca_cert_pem: Option<&str>,
+    ) -> Result<Self, RemoteS3ClientError> {
+        let endpoint = Url::parse(endpoint.trim()).map_err(|err| RemoteS3ClientError::InvalidEndpoint(err.to_string()))?;
+        if !matches!(endpoint.scheme(), "http" | "https") {
+            return Err(RemoteS3ClientError::InvalidEndpoint(format!(
+                "unsupported scheme {}; expected http or https",
+                endpoint.scheme()
+            )));
+        }
+        if endpoint.host_str().is_none_or(str::is_empty) {
+            return Err(RemoteS3ClientError::InvalidEndpoint("endpoint has no host".to_string()));
+        }
+        if !endpoint.username().is_empty() || endpoint.password().is_some() {
+            return Err(RemoteS3ClientError::InvalidEndpoint("endpoint must not carry userinfo".to_string()));
+        }
+        if !matches!(endpoint.path(), "" | "/") || endpoint.query().is_some() || endpoint.fragment().is_some() {
+            return Err(RemoteS3ClientError::InvalidEndpoint(
+                "endpoint must be an origin without path, query or fragment".to_string(),
+            ));
+        }
+        validate_remote_endpoint(&endpoint).map_err(RemoteS3ClientError::EndpointNotAllowed)?;
+
+        let mut builder = reqwest::Client::builder()
+            .connect_timeout(timeouts.connect)
+            .read_timeout(timeouts.read)
+            .redirect(reqwest::redirect::Policy::none())
+            .user_agent(USER_AGENT_SUFFIX);
+        if skip_tls_verify {
+            builder = builder.danger_accept_invalid_certs(true);
+        } else if let Some(pem) = ca_cert_pem.map(str::trim).filter(|pem| !pem.is_empty()) {
+            // Reject a malformed bundle the same way the S3 path does, so the
+            // operator sees "invalid CA PEM" instead of a TLS handshake failure.
+            validate_target_ca_pem(pem)?;
+            let certificate = reqwest::Certificate::from_pem(pem.as_bytes())
+                .map_err(|err| RemoteS3ClientError::InvalidCaPem(err.to_string()))?;
+            builder = builder.add_root_certificate(certificate);
+        }
+
+        let client = builder
+            .build()
+            .map_err(|err| RemoteS3ClientError::InvalidEndpoint(format!("http client cannot be built: {err}")))?;
+        Ok(Self { client, endpoint })
+    }
+
+    #[cfg(test)]
+    pub(super) fn for_test(endpoint: Url) -> Self {
+        Self {
+            client: reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .expect("test http client should build"),
+            endpoint,
+        }
+    }
+
+    /// A URL under the endpoint origin. `segments` are percent-encoded as
+    /// path segments, so a key containing `?`, `#` or a space cannot rewrite
+    /// the request target.
+    pub(super) fn url<'a>(&self, segments: impl IntoIterator<Item = &'a str>) -> Result<Url, SourceError> {
+        let mut url = self.endpoint.clone();
+        {
+            let mut path = url
+                .path_segments_mut()
+                .map_err(|_| SourceError::Other("source endpoint cannot carry a path".to_string()))?;
+            path.clear();
+            path.extend(segments);
+        }
+        Ok(url)
+    }
+
+    /// Sends the request and returns the response only for a 2xx status.
+    /// Non-2xx statuses are classified from the status and the provider's own
+    /// error-code header; response bodies are not read, so no provider message
+    /// can smuggle credentials or markup into a log line.
+    pub(super) async fn send(
+        &self,
+        request: reqwest::Request,
+        error_code_header: &str,
+    ) -> Result<reqwest::Response, SourceError> {
+        let response = self.client.execute(request).await.map_err(classify_transport_error)?;
+        let status = response.status();
+        if status.is_success() {
+            return Ok(response);
+        }
+        let code = response
+            .headers()
+            .get(error_code_header)
+            .and_then(|value| value.to_str().ok())
+            .map(str::to_string);
+        Err(classify_status(
+            status.as_u16(),
+            None,
+            match &code {
+                Some(code) => format!("source returned HTTP {status} ({code})"),
+                None => format!("source returned HTTP {status}"),
+            },
+        ))
+    }
+}
+
+/// Renders a transport failure without the request URL: a SAS token or a
+/// signed query would otherwise reach logs and admin responses.
+pub(super) fn classify_transport_error(err: reqwest::Error) -> SourceError {
+    let is_timeout = err.is_timeout();
+    let is_connect = err.is_connect();
+    let message = err.without_url().to_string();
+    if is_timeout {
+        SourceError::Timeout
+    } else if is_connect {
+        SourceError::Connect(message)
+    } else {
+        SourceError::Other(message)
+    }
+}
+
+/// Streams the response body without buffering it.
+pub(super) fn response_body(response: reqwest::Response) -> ByteStream {
+    let stream = response.bytes_stream().map(|chunk| {
+        chunk
+            .map(http_body::Frame::data)
+            .map_err(|err| std::io::Error::other(err.without_url().to_string()))
+    });
+    ByteStream::new(SdkBody::from_body_1_x(http_body_util::StreamBody::new(stream)))
+}
+
+/// Reads a bounded response body as UTF-8, for the XML and JSON listings.
+pub(super) async fn read_text(response: reqwest::Response, max_bytes: usize) -> Result<String, SourceError> {
+    let mut body = Vec::new();
+    let mut stream = response.bytes_stream();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(classify_transport_error)?;
+        if body.len().saturating_add(chunk.len()) > max_bytes {
+            return Err(SourceError::Other("source listing response exceeded the size limit".to_string()));
+        }
+        body.extend_from_slice(&chunk);
+    }
+    String::from_utf8(body).map_err(|_| SourceError::Other("source listing response is not valid UTF-8".to_string()))
+}
+
+/// Base64 digest (`Content-MD5`, `md5Hash`, `x-goog-hash`) as lowercase hex.
+/// `None` when the value is not a 16-byte digest, so a CRC32C never passes as
+/// an MD5.
+pub(super) fn base64_md5_to_hex(value: &str) -> Option<String> {
+    let raw = base64_simd::STANDARD.decode_to_vec(value.trim().as_bytes()).ok()?;
+    (raw.len() == 16).then(|| faster_hex::hex_string(&raw))
+}
+
+pub(super) fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
+    headers.get(name).and_then(|value| value.to_str().ok()).map(str::trim)
+}
+
+fn header_string(headers: &HeaderMap, name: &str) -> Option<String> {
+    header(headers, name).filter(|value| !value.is_empty()).map(str::to_string)
+}
+
+/// `Last-Modified` and friends arrive as an HTTP date; the JSON dialects use
+/// RFC 3339 for the same field, so both are accepted.
+pub(super) fn parse_http_timestamp(value: &str) -> Option<SystemTime> {
+    OffsetDateTime::parse(value, &Rfc2822)
+        .or_else(|_| OffsetDateTime::parse(value, &Rfc3339))
+        .ok()
+        .map(SystemTime::from)
+}
+
+/// Provider-specific fields the shared header mapping cannot infer.
+pub(super) struct NativeHeadFields {
+    pub(super) etag: Option<String>,
+    /// The ETag is an opaque token rather than a digest of the bytes.
+    pub(super) etag_is_opaque: bool,
+    pub(super) version_id: Option<String>,
+    pub(super) storage_class: Option<String>,
+}
+
+/// Maps a HEAD or GET response onto [`SourceHead`]. `metadata_prefix` is the
+/// provider's user-metadata header prefix (`x-ms-meta-`, `x-goog-meta-`); the
+/// stored shape drops it, matching the `x-amz-meta-` handling of the S3 path.
+pub(super) fn native_source_head(
+    headers: &HeaderMap,
+    metadata_prefix: &str,
+    fields: NativeHeadFields,
+) -> Result<SourceHead, SourceError> {
+    let size = header(headers, "content-length")
+        .and_then(|value| value.parse::<u64>().ok())
+        .ok_or_else(|| SourceError::Other("source response has no valid content-length".to_string()))?;
+
+    let mut user_metadata = HashMap::new();
+    for (name, value) in headers {
+        let name = name.as_str();
+        if let Some(key) = name.strip_prefix(metadata_prefix)
+            && !key.is_empty()
+            && let Ok(value) = value.to_str()
+        {
+            user_metadata.insert(key.to_string(), value.to_string());
+        }
+    }
+
+    let etag = fields
+        .etag
+        .map(|etag| etag.trim().trim_matches('"').to_string())
+        .filter(|etag| !etag.is_empty());
+    // An opaque ETag never encodes a part count, so the multipart flag stays
+    // false for it however the provider happens to spell the token.
+    let is_multipart_etag = !fields.etag_is_opaque && etag.as_deref().is_some_and(is_multipart_etag);
+
+    Ok(SourceHead {
+        etag,
+        size,
+        last_modified: header(headers, "last-modified").and_then(parse_http_timestamp),
+        content_type: header_string(headers, "content-type"),
+        content_encoding: header_string(headers, "content-encoding"),
+        content_disposition: header_string(headers, "content-disposition"),
+        content_language: header_string(headers, "content-language"),
+        cache_control: header_string(headers, "cache-control"),
+        expires: header_string(headers, "expires"),
+        user_metadata,
+        version_id: fields.version_id,
+        storage_class: fields.storage_class,
+        // Neither native provider hands back ciphertext: a customer-key object
+        // is refused by the backend before it reaches this mapping, and the
+        // service-managed encryption is transparent to the reader.
+        sse: None,
+        is_multipart_etag,
+        etag_is_opaque: fields.etag_is_opaque,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::HeaderValue;
+
+    fn headers(pairs: &[(&str, &str)]) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        for (name, value) in pairs {
+            headers.insert(
+                http::HeaderName::from_bytes(name.as_bytes()).expect("test header name"),
+                HeaderValue::from_str(value).expect("test header value"),
+            );
+        }
+        headers
+    }
+
+    fn fields() -> NativeHeadFields {
+        NativeHeadFields {
+            etag: None,
+            etag_is_opaque: false,
+            version_id: None,
+            storage_class: None,
+        }
+    }
+
+    #[test]
+    fn native_source_head_maps_content_headers_and_prefixed_metadata() {
+        let headers = headers(&[
+            ("content-length", "1234"),
+            ("content-type", "text/plain"),
+            ("content-encoding", "gzip"),
+            ("content-language", "en"),
+            ("content-disposition", "attachment"),
+            ("cache-control", "max-age=60"),
+            ("expires", "Thu, 01 Jan 2026 00:00:00 GMT"),
+            ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT"),
+            ("x-ms-meta-owner", "alice"),
+            ("x-goog-meta-owner", "not-mine"),
+        ]);
+        let head = native_source_head(
+            &headers,
+            "x-ms-meta-",
+            NativeHeadFields {
+                etag: Some("\"0x8DCE1D2\"".to_string()),
+                etag_is_opaque: true,
+                version_id: Some("2026-01-01T00:00:00.0000000Z".to_string()),
+                storage_class: Some("Hot".to_string()),
+            },
+        )
+        .expect("head should map");
+
+        assert_eq!(head.size, 1234);
+        assert_eq!(head.content_type.as_deref(), Some("text/plain"));
+        assert_eq!(head.content_encoding.as_deref(), Some("gzip"));
+        assert_eq!(head.content_language.as_deref(), Some("en"));
+        assert_eq!(head.content_disposition.as_deref(), Some("attachment"));
+        assert_eq!(head.cache_control.as_deref(), Some("max-age=60"));
+        assert_eq!(head.expires.as_deref(), Some("Thu, 01 Jan 2026 00:00:00 GMT"));
+        assert_eq!(
+            head.last_modified,
+            Some(SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_445_412_480)),
+            "HTTP-date Last-Modified must parse"
+        );
+        assert_eq!(
+            head.user_metadata,
+            HashMap::from([("owner".to_string(), "alice".to_string())]),
+            "only the provider's own metadata prefix is read"
+        );
+        assert_eq!(head.etag.as_deref(), Some("0x8DCE1D2"), "quotes are stripped, the token is kept");
+        assert!(head.etag_is_opaque);
+        assert!(!head.is_multipart_etag);
+        assert_eq!(head.storage_class.as_deref(), Some("Hot"));
+        assert!(head.sse.is_none());
+    }
+
+    #[test]
+    fn native_source_head_requires_a_content_length() {
+        let err = native_source_head(&headers(&[("content-type", "text/plain")]), "x-ms-meta-", fields())
+            .expect_err("a response without content-length is unusable");
+        assert!(matches!(err, SourceError::Other(_)), "{err:?}");
+    }
+
+    #[test]
+    fn opaque_etag_never_reads_as_a_multipart_etag() {
+        // A digest-shaped ETag keeps the S3 reading; the same string marked
+        // opaque must not be split into "digest-partcount".
+        for (opaque, expected) in [(false, true), (true, false)] {
+            let head = native_source_head(
+                &headers(&[("content-length", "1")]),
+                "x-ms-meta-",
+                NativeHeadFields {
+                    etag: Some("d41d8cd98f00b204e9800998ecf8427e-3".to_string()),
+                    etag_is_opaque: opaque,
+                    ..fields()
+                },
+            )
+            .expect("head should map");
+            assert_eq!(head.is_multipart_etag, expected, "opaque = {opaque}");
+        }
+    }
+
+    #[test]
+    fn base64_md5_converts_only_sixteen_byte_digests() {
+        assert_eq!(
+            base64_md5_to_hex("1B2M2Y8AsgTpgAmY7PhCfg==").as_deref(),
+            Some("d41d8cd98f00b204e9800998ecf8427e")
+        );
+        assert_eq!(base64_md5_to_hex("not base64!").as_deref(), None);
+        // A CRC32C digest is four bytes: it must not pass as an MD5.
+        assert_eq!(base64_md5_to_hex("AAAAAA==").as_deref(), None);
+    }
+
+    #[test]
+    fn native_http_rejects_endpoints_that_are_not_bare_origins() {
+        for bad in [
+            "ftp://source.example.com",
+            "https://user:pw@source.example.com",
+            "https://source.example.com/container",
+            "https://source.example.com/?x=1",
+            "not a url",
+        ] {
+            assert!(
+                NativeHttp::new(bad, SourceTimeouts::default(), false, None).is_err(),
+                "{bad} must be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn native_http_percent_encodes_every_path_segment() {
+        let http = NativeHttp::for_test(Url::parse("https://acct.blob.core.windows.net").expect("origin"));
+        let url = http.url(["container", "dir", "a b?c#d.txt"]).expect("url should build");
+        assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/container/dir/a%20b%3Fc%23d.txt");
+        assert_eq!(url.query(), None, "a key with '?' must not become a query");
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/pull.rs

@@ -1119,6 +1119,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig::default(),
             policy: PolicyConfig::default(),

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -25,6 +25,8 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.
 
+use super::azure::AzureSourceBackend;
+use super::gcs::GcsNativeSourceBackend;
 use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
     PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
@@ -65,6 +67,10 @@ pub enum SourceProvider {
     /// Generic S3-compatible service.
     #[default]
     S3,
+    /// Native Azure Blob service; not an S3 dialect.
+    Azure,
+    /// Native GCS JSON API with a service-account key; not an S3 dialect.
+    GcsNative,
 }
 
 impl SourceProvider {
@@ -76,6 +82,8 @@ impl SourceProvider {
             "minio" => Some(Self::Minio),
             "rustfs" => Some(Self::Rustfs),
             "s3" => Some(Self::S3),
+            "azure" => Some(Self::Azure),
+            "gcs_native" => Some(Self::GcsNative),
             _ => None,
         }
     }
@@ -88,6 +96,8 @@ impl SourceProvider {
             Self::Minio => "minio",
             Self::Rustfs => "rustfs",
             Self::S3 => "s3",
+            Self::Azure => "azure",
+            Self::GcsNative => "gcs_native",
         }
     }
 
@@ -159,6 +169,69 @@ pub struct SourceClientSpec {
     /// Bytes per second the pull pipeline may consume from this source;
     /// `None` means unlimited. Enforced by the consumer, not by this client.
     pub bandwidth_limit: Option<NonZeroU64>,
+    /// Which [`SourceBackend`] to build. The S3 variant reads `region`,
+    /// `path_style` and `credentials`; the native variants ignore all three
+    /// and carry their own credentials.
+    pub backend: SourceBackendSpec,
+}
+
+/// Provider-specific half of [`SourceClientSpec`].
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub enum SourceBackendSpec {
+    #[default]
+    S3,
+    Azure(AzureSourceSpec),
+    Gcs(GcsSourceSpec),
+}
+
+/// Native Azure Blob parameters. The container is [`SourceClientSpec::bucket`].
+#[derive(Clone, PartialEq, Eq)]
+pub struct AzureSourceSpec {
+    pub account: String,
+    pub auth: AzureAuth,
+}
+
+impl fmt::Debug for AzureSourceSpec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("AzureSourceSpec")
+            .field("account", &self.account)
+            .field("auth", &self.auth)
+            .finish()
+    }
+}
+
+/// How Azure requests are authorized.
+#[derive(Clone, PartialEq, Eq)]
+pub enum AzureAuth {
+    /// Base64 storage-account key, signed per request with Shared Key.
+    SharedKey(String),
+    /// SAS query string without the leading `?`, appended to every URL.
+    Sas(String),
+}
+
+impl fmt::Debug for AzureAuth {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Both variants are secrets; only the scheme may be rendered.
+        f.write_str(match self {
+            Self::SharedKey(_) => "SharedKey(REDACTED)",
+            Self::Sas(_) => "Sas(REDACTED)",
+        })
+    }
+}
+
+/// Native GCS parameters. The bucket is [`SourceClientSpec::bucket`].
+#[derive(Clone, PartialEq, Eq)]
+pub struct GcsSourceSpec {
+    /// Service-account key JSON.
+    pub service_account_json: String,
+}
+
+impl fmt::Debug for GcsSourceSpec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GcsSourceSpec")
+            .field("service_account_json", &"REDACTED")
+            .finish()
+    }
 }
 
 impl SourceClientSpec {
@@ -272,7 +345,7 @@ const ACCESS_DENIED_CODES: &[&str] = &[
     "InvalidToken",
 ];
 
-fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError {
+pub(super) fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError {
     if let Some(code) = code {
         if THROTTLE_CODES.contains(&code) {
             return SourceError::Throttled;
@@ -345,6 +418,11 @@ pub struct SourceHead {
     pub storage_class: Option<String>,
     pub sse: Option<SourceSse>,
     pub is_multipart_etag: bool,
+    /// The provider's ETag is not derived from the object bytes (Azure
+    /// stamps an opaque concurrency token). Such an ETag is recorded for
+    /// provenance but must never be read as a content digest, so the
+    /// write-back path refuses to use it as the expected MD5.
+    pub etag_is_opaque: bool,
 }
 
 /// Per-operation fields shared by HEAD and GET outputs.
@@ -366,7 +444,7 @@ struct HeadParts {
     sse_customer_algorithm: Option<String>,
 }
 
-fn normalize_etag(etag: Option<String>) -> Option<String> {
+pub(super) fn normalize_etag(etag: Option<String>) -> Option<String> {
     etag.map(|etag| etag.trim().trim_matches('"').to_string())
         .filter(|etag| !etag.is_empty())
 }
@@ -415,6 +493,7 @@ fn source_head(parts: HeadParts) -> Result<SourceHead, SourceError> {
         storage_class: parts.storage_class,
         sse,
         is_multipart_etag,
+        etag_is_opaque: false,
     })
 }
 
@@ -625,9 +704,48 @@ impl fmt::Debug for SourceClient {
 
 impl SourceClient {
     pub async fn new(spec: &SourceClientSpec) -> Result<Self, RemoteS3ClientError> {
-        let endpoint = spec.endpoint_spec()?;
-        let config = build_remote_s3_config(&endpoint).await?;
-        Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
+        match &spec.backend {
+            SourceBackendSpec::S3 => {
+                let endpoint = spec.endpoint_spec()?;
+                let config = build_remote_s3_config(&endpoint).await?;
+                Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
+            }
+            SourceBackendSpec::Azure(azure) => {
+                let backend = AzureSourceBackend::new(
+                    &spec.endpoint,
+                    &spec.bucket,
+                    azure,
+                    spec.timeouts,
+                    spec.skip_tls_verify,
+                    spec.ca_cert_pem.as_deref(),
+                )?;
+                Ok(Self::from_backend(Box::new(backend), spec))
+            }
+            SourceBackendSpec::Gcs(gcs) => {
+                let backend = GcsNativeSourceBackend::new(
+                    &spec.endpoint,
+                    &spec.bucket,
+                    gcs,
+                    spec.timeouts,
+                    spec.skip_tls_verify,
+                    spec.ca_cert_pem.as_deref(),
+                )?;
+                Ok(Self::from_backend(Box::new(backend), spec))
+            }
+        }
+    }
+
+    /// Wraps a ready backend in the prefix-mapping client. The endpoint is
+    /// kept only for `Debug` and admin status.
+    fn from_backend(backend: Box<dyn SourceBackend>, spec: &SourceClientSpec) -> Self {
+        Self {
+            backend,
+            endpoint: spec.endpoint.clone(),
+            bucket: spec.bucket.clone(),
+            source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
+            timeouts: spec.timeouts,
+            bandwidth_limit: spec.bandwidth_limit,
+        }
     }
 
     /// `config` must come from [`SourceClientSpec::endpoint_spec`], which is
@@ -866,6 +984,7 @@ fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract};
     use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn};
     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::result::ConnectorError;
@@ -983,6 +1102,7 @@ mod tests {
             retry: RemoteS3RetryPolicy::Disabled,
             timeouts: SourceTimeouts::default(),
             bandwidth_limit: NonZeroU64::new(1_000_000),
+            backend: SourceBackendSpec::S3,
         }
     }
 
@@ -1586,7 +1706,101 @@ mod tests {
         assert_eq!(resolve_path_style(PathStyle::VirtualHost, Minio, "10.0.0.1"), PathStyle::VirtualHost);
         assert_eq!(resolve_path_style(PathStyle::Path, Aws, "s3.amazonaws.com"), PathStyle::Path);
         assert_eq!(SourceProvider::from_label(" AWS "), Some(Aws));
-        assert_eq!(SourceProvider::from_label("azure"), None);
+        assert_eq!(SourceProvider::from_label(" Azure "), Some(Azure));
+        assert_eq!(SourceProvider::from_label("gcs_native"), Some(GcsNative));
+        assert_eq!(SourceProvider::from_label("swift"), None);
+    }
+
+    const CONTRACT_LIST_PAGE_ONE: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Name>source-bucket</Name>
+  <IsTruncated>true</IsTruncated>
+  <NextContinuationToken>cursor-1</NextContinuationToken>
+  <Contents>
+    <Key>dir/a.txt</Key>
+    <LastModified>2015-10-21T07:28:00.000Z</LastModified>
+    <ETag>&quot;5d41402abc4b2a76b9719d911017c592&quot;</ETag>
+    <Size>5</Size>
+    <StorageClass>STANDARD</StorageClass>
+  </Contents>
+  <CommonPrefixes><Prefix>dir/sub/</Prefix></CommonPrefixes>
+</ListBucketResult>"#;
+
+    const CONTRACT_LIST_PAGE_TWO: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Name>source-bucket</Name>
+  <IsTruncated>false</IsTruncated>
+  <Contents>
+    <Key>dir/b.txt</Key>
+    <LastModified>2015-10-21T07:28:00.000Z</LastModified>
+    <ETag>&quot;7d41402abc4b2a76b9719d911017c592&quot;</ETag>
+    <Size>7</Size>
+  </Contents>
+</ListBucketResult>"#;
+
+    const CONTRACT_TAGGING: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><TagSet>
+  <Tag><Key>env</Key><Value>prod</Value></Tag>
+</TagSet></Tagging>"#;
+
+    fn contract_object_headers(content_length: u64) -> Vec<(&'static str, String)> {
+        vec![
+            ("etag", format!("\"{OBJECT_MD5}\"")),
+            ("content-length", content_length.to_string()),
+            ("content-type", "text/plain".to_string()),
+            ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
+            ("x-amz-meta-owner", "alice".to_string()),
+            ("x-amz-storage-class", "STANDARD".to_string()),
+        ]
+    }
+
+    /// The S3 backend behind the scripted connector, without the prefix-mapping
+    /// client on top: the contract is a property of the backend itself.
+    async fn scripted_s3_backend(responses: Vec<Scripted>) -> S3SourceBackend {
+        let spec = spec(None);
+        let connector = SharedHttpConnector::new(ScriptedConnector {
+            requests: Arc::new(Mutex::new(Vec::new())),
+            responses: Arc::new(Mutex::new(responses.into_iter().collect())),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+        let endpoint = spec.endpoint_spec().expect("test spec endpoint should parse");
+        let config = build_remote_s3_config(&endpoint)
+            .await
+            .expect("test spec should build")
+            .http_client(http_client)
+            .interceptor(SourceProxyMarkerInterceptor::new());
+        S3SourceBackend {
+            client: S3Client::from_conf(config.build()),
+            bucket: spec.bucket.clone(),
+        }
+    }
+
+    #[tokio::test]
+    async fn s3_backend_satisfies_the_shared_backend_contract() {
+        let mut ranged = contract_object_headers(3);
+        ranged.push(("content-range", "bytes 1-3/5".to_string()));
+        let backend = scripted_s3_backend(vec![
+            ok(contract_object_headers(5), ""),
+            ok(contract_object_headers(5), "hello"),
+            ok(ranged, "ell"),
+            ok(Vec::new(), CONTRACT_LIST_PAGE_ONE),
+            ok(Vec::new(), CONTRACT_LIST_PAGE_TWO),
+            ok(Vec::new(), CONTRACT_TAGGING),
+            ok(Vec::new(), ""),
+            status(404, ""),
+            status(403, ACCESS_DENIED_BODY),
+        ])
+        .await;
+
+        assert_backend_contract(
+            &backend,
+            BackendCapabilities {
+                etag_is_opaque: false,
+                supports_start_after: true,
+                supports_tagging: true,
+            },
+        )
+        .await;
     }
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {

crates/ecstore/src/bucket/on_demand_migration/sys.rs

@@ -47,7 +47,10 @@ use super::config::{
 use super::list_through::{SOURCE_LIST_RATE_PER_SEC, SourceListRateLimiter};
 use super::negative_cache::NegativeCache;
 use super::pull::{OdmWriteBack, PullQueue};
-use super::source_client::{SourceClient, SourceClientSpec, SourceError, SourceProvider, SourceTimeouts};
+use super::source_client::{
+    AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError, SourceProvider,
+    SourceTimeouts,
+};
 use super::stats::{GaugeGuard, OdmStats, OdmStatsSnapshot, PullFailureReason};
 use crate::bucket::remote_s3_client::{
     PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
@@ -619,6 +622,7 @@ pub fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClientSpec
         // load on a source that is already failing.
         retry: RemoteS3RetryPolicy::Disabled,
         bandwidth_limit: policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
+        backend: source_backend_spec(source),
     }
 }
 
@@ -630,6 +634,31 @@ fn source_provider(provider: Provider) -> SourceProvider {
         Provider::Rustfs => SourceProvider::Rustfs,
         Provider::R2 => SourceProvider::R2,
         Provider::Gcs => SourceProvider::Gcs,
+        Provider::Azure => SourceProvider::Azure,
+        Provider::GcsNative => SourceProvider::GcsNative,
+    }
+}
+
+/// Which backend the client builds. A native provider whose block is missing
+/// falls back to the S3 spec, where the builder reports the missing
+/// credentials: the config layer already refuses to store that shape, so this
+/// only covers a config written by an older or hand-edited build.
+pub fn source_backend_spec(source: &SourceConfig) -> SourceBackendSpec {
+    match (source.provider, source.azure.as_ref(), source.gcs.as_ref()) {
+        (Provider::Azure, Some(azure), _) => SourceBackendSpec::Azure(AzureSourceSpec {
+            account: azure.account.clone(),
+            auth: match (&azure.account_key, &azure.sas_token) {
+                (Some(key), _) => AzureAuth::SharedKey(key.clone()),
+                (None, Some(sas)) => AzureAuth::Sas(sas.clone()),
+                // Refused by `SourceConfig::validate`; an empty shared key
+                // fails closed at the builder rather than signing with none.
+                (None, None) => AzureAuth::SharedKey(String::new()),
+            },
+        }),
+        (Provider::GcsNative, _, Some(gcs)) => SourceBackendSpec::Gcs(GcsSourceSpec {
+            service_account_json: gcs.service_account_json.clone(),
+        }),
+        _ => SourceBackendSpec::S3,
     }
 }
 
@@ -929,6 +958,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: prefix.map(str::to_string),

crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs

@@ -0,0 +1,120 @@
+// 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.
+
+//! Scripted HTTP server for the native source backends' tests.
+//!
+//! The S3 backend can be driven through the SDK's own connector; the native
+//! backends talk to a real socket, so their tests need a server that answers a
+//! fixed script and records what it was asked. Every response closes its
+//! connection, which keeps one request on one socket and makes the script order
+//! exactly the request order.
+
+use std::sync::{Arc, Mutex};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use url::Url;
+
+pub(super) struct ScriptedResponse {
+    status: u16,
+    headers: Vec<(&'static str, String)>,
+    body: String,
+}
+
+impl ScriptedResponse {
+    pub(super) fn new(status: u16, headers: Vec<(&'static str, String)>, body: String) -> Self {
+        Self { status, headers, body }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct RecordedRequest {
+    pub(super) method: String,
+    /// Request target as it appeared on the wire: path plus query.
+    pub(super) target: String,
+    pub(super) headers: Vec<(String, String)>,
+}
+
+impl RecordedRequest {
+    pub(super) fn header(&self, name: &str) -> Option<&str> {
+        self.headers
+            .iter()
+            .find(|(key, _)| key.eq_ignore_ascii_case(name))
+            .map(|(_, value)| value.as_str())
+    }
+}
+
+pub(super) type Recorder = Arc<Mutex<Vec<RecordedRequest>>>;
+
+/// Binds a loopback listener that answers `responses` in order and returns its
+/// origin plus the recorder. The task ends once the script is exhausted.
+pub(super) async fn scripted_server(responses: Vec<ScriptedResponse>) -> (Url, Recorder) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind");
+    let port = listener.local_addr().expect("fixture address").port();
+    let recorder: Recorder = Arc::new(Mutex::new(Vec::new()));
+    let sink = Arc::clone(&recorder);
+
+    tokio::spawn(async move {
+        for response in responses {
+            let Ok((mut stream, _)) = listener.accept().await else {
+                return;
+            };
+            let mut request = Vec::new();
+            let mut buffer = [0_u8; 2048];
+            while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+                match stream.read(&mut buffer).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(read) => request.extend_from_slice(&buffer[..read]),
+                }
+            }
+            let text = String::from_utf8_lossy(&request).into_owned();
+            let mut lines = text.lines();
+            let start = lines.next().unwrap_or_default().to_string();
+            let mut parts = start.split_whitespace();
+            sink.lock().expect("recorder lock").push(RecordedRequest {
+                method: parts.next().unwrap_or_default().to_string(),
+                target: parts.next().unwrap_or_default().to_string(),
+                headers: lines
+                    .take_while(|line| !line.is_empty())
+                    .filter_map(|line| line.split_once(':'))
+                    .map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
+                    .collect(),
+            });
+
+            // A scripted HEAD declares the object size in its own headers while
+            // carrying no body, so an explicit `Content-Length` wins over the
+            // body length.
+            let declares_length = response
+                .headers
+                .iter()
+                .any(|(name, _)| name.eq_ignore_ascii_case("content-length"));
+            let mut rendered = match declares_length {
+                true => format!("HTTP/1.1 {} Scripted\r\nConnection: close\r\n", response.status),
+                false => format!(
+                    "HTTP/1.1 {} Scripted\r\nContent-Length: {}\r\nConnection: close\r\n",
+                    response.status,
+                    response.body.len()
+                ),
+            };
+            for (name, value) in response.headers {
+                rendered.push_str(&format!("{name}: {value}\r\n"));
+            }
+            rendered.push_str("\r\n");
+            rendered.push_str(&response.body);
+            let _ = stream.write_all(rendered.as_bytes()).await;
+            let _ = stream.flush().await;
+        }
+    });
+
+    (Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorder)
+}

crates/madmin/fixtures/on_demand_migration/get_response.json

@@ -1 +1 @@
-{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}
+{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}

crates/madmin/fixtures/on_demand_migration/set_request.json

@@ -1 +1 @@
-{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}
+{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}

crates/madmin/fixtures/on_demand_migration/set_response.json

@@ -1 +1 @@
-{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}
+{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}

crates/madmin/src/on_demand_migration.rs

@@ -78,10 +78,18 @@ pub struct OnDemandMigrationSource {
     #[serde(default)]
     pub path_style: OnDemandMigrationPathStyle,
     /// `None` means anonymous access to a public source bucket.
+    /// `None` means anonymous access to a public source bucket. The native
+    /// providers carry their credentials in `azure` / `gcs` instead.
     #[serde(default)]
     pub credentials: Option<OnDemandMigrationCredentials>,
     #[serde(default)]
     pub tls: OnDemandMigrationTls,
+    /// Required for `azure` and rejected for every other provider.
+    #[serde(default)]
+    pub azure: Option<OnDemandMigrationAzure>,
+    /// Required for `gcs_native` and rejected for every other provider.
+    #[serde(default)]
+    pub gcs: Option<OnDemandMigrationGcs>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -92,7 +100,49 @@ pub enum OnDemandMigrationProvider {
     Minio,
     Rustfs,
     R2,
+    /// GCS XML interoperability API with HMAC keys.
     Gcs,
+    /// Native Azure Blob service.
+    Azure,
+    /// Native GCS JSON API with a service-account key.
+    #[serde(rename = "gcs_native")]
+    GcsNative,
+}
+
+/// Native Azure Blob parameters. The container is `source.bucket`; exactly one
+/// of `account_key` and `sas_token` is set. Responses carry both as `REDACTED`.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationAzure {
+    pub account: String,
+    #[serde(default)]
+    pub account_key: Option<String>,
+    #[serde(default)]
+    pub sas_token: Option<String>,
+}
+
+impl fmt::Debug for OnDemandMigrationAzure {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("OnDemandMigrationAzure")
+            .field("account", &self.account)
+            .field("account_key", &self.account_key.as_ref().map(|_| "REDACTED"))
+            .field("sas_token", &self.sas_token.as_ref().map(|_| "REDACTED"))
+            .finish()
+    }
+}
+
+/// Native GCS parameters. The bucket is `source.bucket`; the key JSON embeds a
+/// private key, so responses carry it as `REDACTED`.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationGcs {
+    pub service_account_json: String,
+}
+
+impl fmt::Debug for OnDemandMigrationGcs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("OnDemandMigrationGcs")
+            .field("service_account_json", &"REDACTED")
+            .finish()
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
@@ -806,6 +856,8 @@ mod tests {
                 session_token: None,
             }),
             tls: OnDemandMigrationTls::default(),
+            azure: None,
+            gcs: None,
         });
         let mut expected: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).expect("fixture");
         expected.filter.source_prefix = None;
@@ -821,6 +873,42 @@ mod tests {
         assert!(minimal.source.credentials.is_none());
     }
 
+    #[test]
+    fn native_provider_documents_round_trip_and_hide_their_secrets() {
+        for (label, json) in [
+            (
+                "azure",
+                r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"},"gcs":null}"#,
+            ),
+            (
+                "gcs_native",
+                r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#,
+            ),
+        ] {
+            let source: OnDemandMigrationSource = serde_json::from_str(json).unwrap_or_else(|err| panic!("{label}: {err}"));
+            assert_eq!(
+                serde_json::to_string(&source).expect("re-encodes"),
+                json,
+                "{label} must reproduce the server wire shape byte for byte"
+            );
+        }
+
+        let azure = OnDemandMigrationAzure {
+            account: "legacyaccount".to_string(),
+            account_key: Some("c2VjcmV0".to_string()),
+            sas_token: Some("sig=topsecret".to_string()),
+        };
+        let rendered = format!("{azure:?}");
+        assert!(rendered.contains("legacyaccount"));
+        assert!(!rendered.contains("c2VjcmV0"), "{rendered}");
+        assert!(!rendered.contains("topsecret"), "{rendered}");
+
+        let gcs = OnDemandMigrationGcs {
+            service_account_json: r#"{"private_key":"-----BEGIN PRIVATE KEY-----"}"#.to_string(),
+        };
+        assert!(!format!("{gcs:?}").contains("PRIVATE KEY"), "{gcs:?}");
+    }
+
     #[test]
     fn credentials_debug_never_prints_secrets() {
         let credentials = OnDemandMigrationCredentials {

docs/operations/on-demand-migration.md

@@ -91,14 +91,20 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
 |---|---|---|---|
 | `version` | integer | `1` | Must be `1` |
 | `enabled` | bool | `true` | `false` keeps the config but stops all source traffic |
-| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` | — (required) | Drives endpoint and addressing defaults |
-| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required for every provider except `aws`, where it is derived from `region` |
-| `source.region` | string | — (required) | Non-empty. `auto` is accepted only for `r2`, `minio`, `rustfs` and is signed as `us-east-1` |
-| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace |
+| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` \| `azure` \| `gcs_native` | — (required) | Drives endpoint and addressing defaults, and which backend the client builds: every value but `azure` and `gcs_native` speaks S3 |
+| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required except for `aws` (derived from `region`), `azure` (derived as `https://<account>.blob.core.windows.net`) and `gcs_native` (`https://storage.googleapis.com`). Set it explicitly to point at Azurite or fake-gcs-server, subject to the same outbound policy as any other source endpoint |
+| `source.region` | string | — (required) | Non-empty. `auto` is accepted for `r2`, `minio`, `rustfs` and for the native providers, and is signed as `us-east-1`. `azure` and `gcs_native` never sign with a region, so `auto` is the honest value there |
+| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace. For `azure` this is the container name, for `gcs_native` the bucket name; the provider block never repeats it |
 | `source.path_style` | `auto` \| `path` \| `virtual` | `auto` | `auto` resolves to path-style for IP-literal or `localhost` endpoints and for `s3`/`minio`/`rustfs`; virtual-host for `aws`/`gcs`/`r2` |
-| `source.credentials` | object \| null | `null` | `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present |
+| `source.credentials` | object \| null | `null` | Read only by the S3 providers; `azure` and `gcs_native` must leave it `null` and carry their credentials in their own block. `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present |
 | `source.tls.skip_verify` | bool | `false` | Disables certificate verification for the source connection |
 | `source.tls.ca_cert_pem` | string \| null | `null` | Must contain `-----BEGIN CERTIFICATE-----` |
+| `source.azure` | object \| null | `null` | Required for `provider = "azure"` and rejected for every other provider |
+| `source.azure.account` | string | — (required) | Storage account name; `[A-Za-z0-9-]` only, because it becomes the first label of the derived host |
+| `source.azure.account_key` | string \| null | `null` | Base64 storage-account key, signed per request with Shared Key. Mutually exclusive with `sas_token`; exactly one of the two is required |
+| `source.azure.sas_token` | string \| null | `null` | SAS query string without the leading `?` and without whitespace, appended to every request URL |
+| `source.gcs` | object \| null | `null` | Required for `provider = "gcs_native"` and rejected for every other provider |
+| `source.gcs.service_account_json` | string | — (required) | Service-account key JSON; must parse and carry `type: service_account`, `client_email` and `private_key`. Tokens are minted read-only (`devstorage.read_only`) |
 | `filter.prefix` | string \| null | `null` | Null or non-empty. Only local keys with this prefix consult the source |
 | `filter.source_prefix` | string \| null | `null` | Null or non-empty. Prepended to the local key to form the source key |
 | `policy.head` | `proxy` \| `local_only` | `proxy` | `local_only` answers a HEAD miss with 404 and no source traffic |
@@ -107,7 +113,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
 | `policy.list_through` | bool | `false` | Merges the source listing into `ListObjectsV2` so clients see the whole namespace during the migration. Off by default: it puts the source in the path of every listing |
 | `policy.respect_local_delete_marker` | bool | `true` | A local delete marker is the final answer; only a versioned bucket can produce one |
 | `policy.preserve_etag` | bool | `true` | Keeps the source ETag on the stored object unless the bucket encrypts by default |
-| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull |
+| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull. `azure` reads blob tags instead; `gcs_native` has no tags and always finds none |
 | `policy.emit_events` | bool | `true` | Whether a write-back emits `ObjectCreated` notifications |
 | `policy.negative_cache_ttl_secs` | integer | `30` | `0..=3600`; `0` disables the negative cache |
 | `policy.inline_max_bytes` | integer | `16777216` (16 MiB) | `0..=268435456` (256 MiB). At or below this size a GET miss is teed inline; above it the response streams through and a background pull stores the object |
@@ -133,8 +139,14 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket
 | `rustfs` | Required | Path-style | `auto` allowed | A RustFS source answers the migration request locally thanks to the anti-loop marker | `real_source_test.rs` in the `e2e-nightly` lane |
 | `r2` | `https://<account-id>.r2.cloudflarestorage.com` | Virtual-host | `auto` allowed (signed as `us-east-1`) | | `cloud-source (r2)`, only while `ODM_INTEROP_R2_*` are configured; no difference recorded yet |
 | `gcs` | `https://storage.googleapis.com` | Virtual-host | Real region required | Uses the GCS XML interoperability API with an HMAC key pair, not a service-account JSON key | `cloud-source (gcs)`, only while `ODM_INTEROP_GCS_HMAC_*` are configured; no difference recorded yet |
+| `azure` | Optional; derived as `https://<account>.blob.core.windows.net` | Native Blob REST, not S3 | Unused; write `auto` | Needs `source.azure`; the container is `source.bucket`. Reads need `Read` on the blob and `List` on the container, plus `Tags` when `policy.copy_tags` is on | None yet: no interop job covers Azure |
+| `gcs_native` | Optional; derived as `https://storage.googleapis.com` | Native GCS API, not S3 | Unused; write `auto` | Needs `source.gcs`. Reads use the XML API for objects and `objects.list` for listings, both with an OAuth token minted from the service-account key; the key needs `storage.objects.get` and `storage.objects.list` | None yet: no interop job covers native GCS |
 
-Azure Blob has no preset; a native provider is deferred (rustfs/backlog#2166).
+Every backend answers the same trait contract, pinned by `backend_contract.rs` in `crates/ecstore/src/bucket/on_demand_migration/`, and the three differences that contract allows are the ones documented here.
+
+`azure` differs in two of them. Its ETag is a concurrency token rather than a digest of the bytes, so it is stored as `odm-source-etag` provenance and never used as the expected MD5 of a pulled object — the write-back integrity check falls back to the local digest. And its listing paginates only with an opaque marker: there is no "start after this key" form, so a caller that asks for one gets `Unsupported` instead of a listing that silently starts over.
+
+`gcs_native` differs in the other two. Its listing also has no exclusive "start after" form (`startOffset` is inclusive), so it refuses one the same way. And GCS has no object tagging at all: `policy.copy_tags` finds no tags rather than failing the pull, because GCS custom metadata is already carried by the head mapping. Its ETag is normally usable: the `x-goog-hash` MD5 is converted to hex and checked against the pulled bytes, except on a composite object, which has no MD5 and whose ETag is then treated as opaque.
 
 The "Interop evidence" column names the job in `.github/workflows/on-demand-migration-interop.yml` (rustfs/backlog#2167) that last exercised the preset against a real implementation, and is where a provider difference belongs once the lane finds one. That lane is report-only and scheduled: it runs `crates/e2e_test/src/on_demand_migration/interop_test.rs` — the same case bodies as the merge-gate suite, with the source injected through `RUSTFS_ODM_INTEROP_*` — against a pinned MinIO container, and against each cloud provider whose repository secrets are configured. A provider without secrets is skipped with a note in the run summary rather than failing, so "no difference recorded yet" means exactly that and not "verified clean"; see [ci-gates.md](../testing/ci-gates.md) for the row.
 

rustfs/src/admin/handlers/on_demand_migration.rs

@@ -46,6 +46,7 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{
 };
 use crate::admin::storage_api::bucket::on_demand_migration::{
     OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext,
+    source_backend_spec,
 };
 use crate::admin::storage_api::bucket::remote_s3_client::{
     PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
@@ -585,6 +586,8 @@ fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider {
         Provider::Rustfs => SourceProvider::Rustfs,
         Provider::R2 => SourceProvider::R2,
         Provider::Gcs => SourceProvider::Gcs,
+        Provider::Azure => SourceProvider::Azure,
+        Provider::GcsNative => SourceProvider::GcsNative,
     }
 }
 
@@ -621,6 +624,9 @@ pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClie
         // a flapping source behind a success and triple the probe's cost.
         retry: RemoteS3RetryPolicy::Disabled,
         bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
+        // One mapping serves the probe and the runtime, so an admin probe
+        // always exercises the backend the runtime will build.
+        backend: source_backend_spec(source),
     }
 }
 

rustfs/src/admin/storage_api.rs

@@ -292,6 +292,7 @@ pub(crate) mod on_demand_migration {
     pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle;
     pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider;
     pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>;
+    pub(crate) use super::ecstore_bucket::on_demand_migration::source_backend_spec;
 
     pub(crate) mod backfill {
         pub(crate) type BackfillCheckpoint = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillCheckpoint;

rustfs/src/app/object/get.rs

@@ -4821,6 +4821,8 @@ mod on_demand_migration_tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: None,

rustfs/src/app/object/head.rs

@@ -665,6 +665,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: None,
@@ -743,6 +745,7 @@ mod tests {
                 },
             ),
             is_multipart_etag: true,
+            etag_is_opaque: false,
         }
     }
 

rustfs/src/app/object/on_demand_migration_put.rs

@@ -115,6 +115,12 @@ pub(super) fn expected_md5_hex(head: &SourceHead) -> Option<String> {
     if head.sse.is_some() {
         return None;
     }
+    // Azure stamps an opaque concurrency token in the ETag slot. It is
+    // recorded as provenance, but reading it as a digest would compare the
+    // pulled bytes against a value that never described them.
+    if head.etag_is_opaque {
+        return None;
+    }
     let etag = head.etag.as_deref()?;
     if etag.len() != 32 || is_multipart_etag(etag) || !etag.bytes().all(|byte| byte.is_ascii_hexdigit()) {
         return None;
@@ -874,6 +880,12 @@ mod tests {
         head.sse = None;
         head.etag = None;
         assert_eq!(expected_md5_hex(&head), None);
+
+        // An Azure ETag can be any string the service chooses; even one that
+        // happens to look like an MD5 must not be checked against the bytes.
+        let mut head = source_head(b"abc");
+        head.etag_is_opaque = true;
+        assert_eq!(expected_md5_hex(&head), None, "opaque provider ETag");
     }
 
     #[test]

rustfs/src/app/object/shared.rs

@@ -309,7 +309,7 @@ fn classify_bucket_default_sse_lookup(
 ) -> S3Result<Option<(ServerSideEncryptionConfiguration, OffsetDateTime)>> {
     match lookup {
         Ok(config) => Ok(Some(config)),
-        Err(err) if err == StorageError::ConfigNotFound => Ok(None),
+        Err(StorageError::ConfigNotFound) => Ok(None),
         Err(err) => {
             let api_error = ApiError::from(err);
             error!(