feat(ecstore): add a native azure blob odm source backend

Cargo.lock

@@ -9703,6 +9703,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/bucket/on_demand_migration/azure.rs

@@ -0,0 +1,982 @@
+// 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 Azure Blob source backend.
+//!
+//! Azure has no S3 API, so this backend speaks the Blob REST service directly:
+//! Get Blob / Get Blob Properties for the read path, List Blobs for the listing,
+//! Get Blob Tags for the tags and Get Container Properties for the probe. The
+//! local key is the blob name inside the container named by `source.bucket`.
+//!
+//! Two authorization schemes are supported, matching the two forms an operator
+//! can hold: a storage-account key signed per request with Shared Key, and a SAS
+//! token appended to the request query.
+//!
+//! Azure's ETag is a concurrency token, not a digest of the bytes, so every head
+//! this backend produces is marked [`SourceHead::etag_is_opaque`]: the write-back
+//! path records the value for provenance and refuses to check content against it.
+//! `Content-MD5` is the only Azure digest, it is optional per blob, and it is not
+//! mapped onto the ETag slot precisely so that the two never get confused.
+//!
+//! The anti-loop `source-proxy-request` markers the S3 backend sends are omitted:
+//! they mean something only to a RustFS or MinIO source, and Azure would have to
+//! carry them through Shared Key canonicalization for no gain.
+
+use super::native_http::{
+    NativeHeadFields, NativeHttp, header, native_source_head, parse_http_timestamp, read_text, response_body,
+};
+use super::source_client::{
+    AzureAuth, AzureSourceSpec, 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 hmac::{Hmac, Mac, digest::KeyInit};
+use http::{HeaderMap, HeaderValue, Method};
+use quick_xml::Reader;
+use quick_xml::events::Event;
+use sha2::Sha256;
+use std::collections::{BTreeMap, HashMap};
+use url::Url;
+
+type HmacSha256 = Hmac<Sha256>;
+
+/// Blob REST version this backend pins. Every response field it reads exists
+/// from this version on, including blob tags and blob versioning.
+const API_VERSION: &str = "2021-08-06";
+const HEADER_VERSION: &str = "x-ms-version";
+const HEADER_DATE: &str = "x-ms-date";
+const HEADER_ERROR_CODE: &str = "x-ms-error-code";
+const METADATA_PREFIX: &str = "x-ms-meta-";
+/// A List Blobs or Get Blob Tags response is small; refuse a source that
+/// streams an unbounded document at us instead of buffering it.
+const MAX_XML_BYTES: usize = 8 * 1024 * 1024;
+
+/// `Sun, 06 Nov 1994 08:49:37 GMT`, the only `x-ms-date` form Azure accepts.
+const HTTP_DATE: &[time::format_description::BorrowedFormatItem<'static>] =
+    time::macros::format_description!("[weekday repr:short], [day] [month repr:short] [year] [hour]:[minute]:[second] GMT");
+
+enum Credential {
+    /// Decoded storage-account key.
+    SharedKey(Vec<u8>),
+    /// SAS parameters, decoded once so re-encoding cannot double-escape them.
+    Sas(Vec<(String, String)>),
+}
+
+pub struct AzureSourceBackend {
+    http: NativeHttp,
+    account: String,
+    container: String,
+    credential: Credential,
+}
+
+impl AzureSourceBackend {
+    pub fn new(
+        endpoint: &str,
+        container: &str,
+        spec: &AzureSourceSpec,
+        timeouts: SourceTimeouts,
+        skip_tls_verify: bool,
+        ca_cert_pem: Option<&str>,
+    ) -> Result<Self, RemoteS3ClientError> {
+        let credential = match &spec.auth {
+            AzureAuth::SharedKey(key) => Credential::SharedKey(
+                base64_simd::STANDARD
+                    .decode_to_vec(key.as_bytes())
+                    .map_err(|_| RemoteS3ClientError::Credentials("azure account key is not base64"))?,
+            ),
+            AzureAuth::Sas(sas) => {
+                let pairs: Vec<(String, String)> = url::form_urlencoded::parse(sas.trim_start_matches('?').as_bytes())
+                    .into_owned()
+                    .collect();
+                if pairs.is_empty() {
+                    return Err(RemoteS3ClientError::Credentials("azure sas token has no parameters"));
+                }
+                Credential::Sas(pairs)
+            }
+        };
+        Ok(Self {
+            http: NativeHttp::new(endpoint, timeouts, skip_tls_verify, ca_cert_pem)?,
+            account: spec.account.clone(),
+            container: container.to_string(),
+            credential,
+        })
+    }
+
+    /// URL of one blob in the container. The key is split so its `/` stay path
+    /// separators while every other character is percent-encoded.
+    fn blob_url(&self, key: &str) -> Result<Url, SourceError> {
+        self.http.url(std::iter::once(self.container.as_str()).chain(key.split('/')))
+    }
+
+    fn container_url(&self) -> Result<Url, SourceError> {
+        self.http.url(std::iter::once(self.container.as_str()))
+    }
+
+    /// Builds a signed (or SAS-carrying) request. `headers` holds the
+    /// operation's own headers; the service headers and authorization are
+    /// added here so every request is authorized the same way.
+    fn request(&self, method: Method, mut url: Url, mut headers: HeaderMap) -> Result<reqwest::Request, SourceError> {
+        headers.insert(HEADER_VERSION, HeaderValue::from_static(API_VERSION));
+        let now = time::OffsetDateTime::now_utc()
+            .format(HTTP_DATE)
+            .map_err(|err| SourceError::Other(format!("cannot render the request date: {err}")))?;
+        headers.insert(
+            HEADER_DATE,
+            HeaderValue::from_str(&now).map_err(|_| SourceError::Other("cannot render the request date".to_string()))?,
+        );
+
+        match &self.credential {
+            Credential::SharedKey(key) => {
+                let signature = shared_key_signature(key, &self.account, method.as_str(), &url, &headers)?;
+                headers.insert(
+                    http::header::AUTHORIZATION,
+                    HeaderValue::from_str(&format!("SharedKey {}:{signature}", self.account))
+                        .map_err(|_| SourceError::Other("cannot render the authorization header".to_string()))?,
+                );
+            }
+            Credential::Sas(pairs) => {
+                url.query_pairs_mut().extend_pairs(pairs.iter().map(|(k, v)| (k, v)));
+            }
+        }
+
+        let mut request = reqwest::Request::new(method, url);
+        *request.headers_mut() = headers;
+        Ok(request)
+    }
+
+    /// Shared mapping for Get Blob and Get Blob Properties.
+    fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
+        // A customer-provided key means the service holds ciphertext it cannot
+        // decrypt for us; the same rule the S3 path applies to SSE-C.
+        if header(headers, "x-ms-encryption-key-sha256").is_some() {
+            return Err(SourceError::Unsupported(
+                "source blob uses a customer-provided encryption key; customer-key sources are not supported".to_string(),
+            ));
+        }
+        native_source_head(
+            headers,
+            METADATA_PREFIX,
+            NativeHeadFields {
+                etag: header(headers, "etag").map(str::to_string),
+                etag_is_opaque: true,
+                version_id: header(headers, "x-ms-version-id").map(str::to_string),
+                storage_class: header(headers, "x-ms-access-tier").map(str::to_string),
+            },
+        )
+    }
+}
+
+#[async_trait::async_trait]
+impl SourceBackend for AzureSourceBackend {
+    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
+        let request = self.request(Method::HEAD, self.blob_url(key)?, HeaderMap::new())?;
+        let response = self.http.send(request, HEADER_ERROR_CODE).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.blob_url(key)?, headers)?;
+        let response = self.http.send(request, HEADER_ERROR_CODE).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> {
+        // Azure paginates with an opaque marker and has no "start after this
+        // key" form. Refuse rather than silently listing from the beginning.
+        if request.start_after.is_some() {
+            return Err(SourceError::Unsupported(
+                "azure sources cannot resume a listing from a key; use the continuation token".to_string(),
+            ));
+        }
+        let mut url = self.container_url()?;
+        {
+            let mut query = url.query_pairs_mut();
+            query.append_pair("restype", "container");
+            query.append_pair("comp", "list");
+            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(marker) = request.continuation_token.filter(|marker| !marker.is_empty()) {
+                query.append_pair("marker", marker);
+            }
+            if request.max_keys > 0 {
+                query.append_pair("maxresults", &request.max_keys.to_string());
+            }
+        }
+
+        let request = self.request(Method::GET, url, HeaderMap::new())?;
+        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let body = read_text(response, MAX_XML_BYTES).await?;
+        let listing = parse_list_blobs(&body)?;
+
+        Ok(SourcePage {
+            objects: listing.objects,
+            common_prefixes: listing.prefixes,
+            is_truncated: listing.next_marker.is_some(),
+            next_continuation_token: listing.next_marker,
+        })
+    }
+
+    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+        let mut url = self.blob_url(key)?;
+        url.query_pairs_mut().append_pair("comp", "tags");
+        let request = self.request(Method::GET, url, HeaderMap::new())?;
+        let response = self.http.send(request, HEADER_ERROR_CODE).await?;
+        let body = read_text(response, MAX_XML_BYTES).await?;
+        parse_blob_tags(&body)
+    }
+
+    async fn probe(&self) -> Result<(), SourceError> {
+        let mut url = self.container_url()?;
+        url.query_pairs_mut().append_pair("restype", "container");
+        let request = self.request(Method::HEAD, url, HeaderMap::new())?;
+        self.http.send(request, HEADER_ERROR_CODE).await?;
+        Ok(())
+    }
+}
+
+/// Shared Key signature over the canonical request. Only the fields this
+/// backend ever sets are non-empty: `Range`, the `x-ms-*` headers and the
+/// canonicalized resource. GET and HEAD carry no body, so every `Content-*`
+/// slot stays empty.
+fn shared_key_signature(key: &[u8], account: &str, method: &str, url: &Url, headers: &HeaderMap) -> Result<String, SourceError> {
+    let mut string_to_sign = String::with_capacity(256);
+    string_to_sign.push_str(method);
+    string_to_sign.push('\n');
+    // Content-Encoding, Content-Language, Content-Length, Content-MD5,
+    // Content-Type, Date, If-Modified-Since, If-Match, If-None-Match,
+    // If-Unmodified-Since: all empty. `Date` stays empty because `x-ms-date`
+    // carries the timestamp and Azure then ignores this slot.
+    for _ in 0..10 {
+        string_to_sign.push('\n');
+    }
+    string_to_sign.push_str(header(headers, "range").unwrap_or_default());
+    string_to_sign.push('\n');
+
+    // Canonicalized headers: every `x-ms-*` header, lowercased and sorted.
+    let mut canonical_headers = BTreeMap::new();
+    for (name, value) in headers {
+        let name = name.as_str();
+        if let Some(rest) = name.strip_prefix("x-ms-")
+            && !rest.is_empty()
+            && let Ok(value) = value.to_str()
+        {
+            canonical_headers.insert(name.to_string(), value.trim().to_string());
+        }
+    }
+    for (name, value) in &canonical_headers {
+        string_to_sign.push_str(name);
+        string_to_sign.push(':');
+        string_to_sign.push_str(value);
+        string_to_sign.push('\n');
+    }
+
+    // Canonicalized resource: the account, the encoded path, then every query
+    // parameter lowercased and sorted, with repeated values joined by commas.
+    string_to_sign.push('/');
+    string_to_sign.push_str(account);
+    string_to_sign.push_str(url.path());
+    let mut canonical_query: BTreeMap<String, Vec<String>> = BTreeMap::new();
+    for (name, value) in url.query_pairs() {
+        canonical_query
+            .entry(name.to_ascii_lowercase())
+            .or_default()
+            .push(value.into_owned());
+    }
+    for (name, mut values) in canonical_query {
+        values.sort();
+        string_to_sign.push('\n');
+        string_to_sign.push_str(&name);
+        string_to_sign.push(':');
+        string_to_sign.push_str(&values.join(","));
+    }
+
+    let mut mac = HmacSha256::new_from_slice(key)
+        .map_err(|_| SourceError::Other("azure account key has an unusable length".to_string()))?;
+    mac.update(string_to_sign.as_bytes());
+    Ok(base64_simd::STANDARD.encode_to_string(mac.finalize().into_bytes()))
+}
+
+#[derive(Debug)]
+struct AzureListing {
+    objects: Vec<SourceObject>,
+    prefixes: Vec<String>,
+    /// `None` when the listing is complete; Azure marks the end with an empty
+    /// `NextMarker`.
+    next_marker: Option<String>,
+}
+
+#[derive(Default)]
+struct BlobEntry {
+    name: String,
+    etag: Option<String>,
+    size: u64,
+    last_modified: Option<std::time::SystemTime>,
+    access_tier: Option<String>,
+}
+
+/// Parses one `List Blobs` page.
+fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
+    let mut reader = xml_reader(xml);
+    let mut objects = Vec::new();
+    let mut prefixes = Vec::new();
+    let mut next_marker = None;
+    let mut blob: Option<BlobEntry> = None;
+    let mut in_blob_prefix = false;
+    // Open container elements. quick-xml reports a truncated document as a
+    // plain end of input, so a non-zero depth at EOF is the only signal that
+    // the page was cut short and must not be read as a complete listing.
+    let mut depth = 0_usize;
+
+    loop {
+        match reader.read_event() {
+            Ok(Event::Start(start)) => {
+                let name = local_name(start.name().as_ref());
+                match name.as_str() {
+                    "blob" => {
+                        depth += 1;
+                        blob = Some(BlobEntry::default());
+                    }
+                    "blobprefix" => {
+                        depth += 1;
+                        in_blob_prefix = true;
+                    }
+                    "properties" | "blobs" | "enumerationresults" => depth += 1,
+                    _ => {
+                        let end = start.to_end().into_owned();
+                        let text = leaf_text(&mut reader, end.name())?;
+                        apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
+                    }
+                }
+            }
+            Ok(Event::Empty(empty)) => {
+                let name = local_name(empty.name().as_ref());
+                apply_list_field(&name, String::new(), &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
+            }
+            Ok(Event::End(end)) => match local_name(end.name().as_ref()).as_str() {
+                "blob" => {
+                    depth = depth.saturating_sub(1);
+                    if let Some(entry) = blob.take() {
+                        objects.push(SourceObject {
+                            key: entry.name,
+                            etag: entry.etag,
+                            size: entry.size,
+                            last_modified: entry.last_modified,
+                            storage_class: entry.access_tier,
+                            // Azure ETags carry no part count; the listing
+                            // never describes a composed object.
+                            is_multipart_etag: false,
+                        });
+                    }
+                }
+                "blobprefix" => {
+                    depth = depth.saturating_sub(1);
+                    in_blob_prefix = false;
+                }
+                "properties" | "blobs" | "enumerationresults" => depth = depth.saturating_sub(1),
+                _ => {}
+            },
+            Ok(Event::Eof) => break,
+            Ok(_) => {}
+            Err(err) => return Err(SourceError::Other(format!("source listing is not valid XML: {err}"))),
+        }
+    }
+    if depth != 0 {
+        return Err(SourceError::Other("source listing ended before every element was closed".to_string()));
+    }
+
+    Ok(AzureListing {
+        objects,
+        prefixes,
+        next_marker: next_marker.filter(|marker| !marker.is_empty()),
+    })
+}
+
+fn apply_list_field(
+    name: &str,
+    text: String,
+    blob: &mut Option<BlobEntry>,
+    prefixes: &mut Vec<String>,
+    next_marker: &mut Option<String>,
+    in_blob_prefix: bool,
+) {
+    match name {
+        "name" => {
+            if in_blob_prefix {
+                prefixes.push(text);
+            } else if let Some(entry) = blob.as_mut() {
+                entry.name = text;
+            }
+        }
+        "nextmarker" => *next_marker = Some(text),
+        "etag" => {
+            if let Some(entry) = blob.as_mut() {
+                entry.etag = Some(text.trim().trim_matches('"').to_string()).filter(|etag| !etag.is_empty());
+            }
+        }
+        "content-length" => {
+            if let Some(entry) = blob.as_mut() {
+                entry.size = text.trim().parse().unwrap_or(0);
+            }
+        }
+        "last-modified" => {
+            if let Some(entry) = blob.as_mut() {
+                entry.last_modified = parse_http_timestamp(text.trim());
+            }
+        }
+        "accesstier" => {
+            if let Some(entry) = blob.as_mut() {
+                entry.access_tier = Some(text).filter(|tier| !tier.is_empty());
+            }
+        }
+        _ => {}
+    }
+}
+
+/// Parses a `Get Blob Tags` response.
+fn parse_blob_tags(xml: &str) -> Result<HashMap<String, String>, SourceError> {
+    let mut reader = xml_reader(xml);
+    let mut tags = HashMap::new();
+    let mut key = None;
+    let mut value = None;
+    let mut depth = 0_usize;
+
+    loop {
+        match reader.read_event() {
+            Ok(Event::Start(start)) => {
+                let name = local_name(start.name().as_ref());
+                match name.as_str() {
+                    "tags" | "tagset" | "tag" => depth += 1,
+                    _ => {
+                        let end = start.to_end().into_owned();
+                        let text = leaf_text(&mut reader, end.name())?;
+                        match name.as_str() {
+                            "key" => key = Some(text),
+                            "value" => value = Some(text),
+                            _ => {}
+                        }
+                    }
+                }
+            }
+            Ok(Event::Empty(empty)) => match local_name(empty.name().as_ref()).as_str() {
+                "key" => key = Some(String::new()),
+                "value" => value = Some(String::new()),
+                _ => {}
+            },
+            Ok(Event::End(end)) => {
+                let name = local_name(end.name().as_ref());
+                if matches!(name.as_str(), "tags" | "tagset" | "tag") {
+                    depth = depth.saturating_sub(1);
+                }
+                if name == "tag"
+                    && let (Some(key), Some(value)) = (key.take(), value.take())
+                {
+                    tags.insert(key, value);
+                }
+            }
+            Ok(Event::Eof) => break,
+            Ok(_) => {}
+            Err(err) => return Err(SourceError::Other(format!("source tags are not valid XML: {err}"))),
+        }
+    }
+    if depth != 0 {
+        return Err(SourceError::Other("source tags ended before every element was closed".to_string()));
+    }
+
+    Ok(tags)
+}
+
+fn xml_reader(xml: &str) -> Reader<&[u8]> {
+    let mut reader = Reader::from_str(xml);
+    let config = reader.config_mut();
+    config.trim_text_start = true;
+    config.trim_text_end = true;
+    reader
+}
+
+/// Lowercased element name without its namespace prefix.
+fn local_name(raw: &str) -> String {
+    raw.rsplit(':').next().unwrap_or(raw).to_ascii_lowercase()
+}
+
+/// Text of a leaf element, consuming through its end tag.
+fn leaf_text(reader: &mut Reader<&[u8]>, end: quick_xml::name::QName<'_>) -> Result<String, SourceError> {
+    let raw = reader
+        .read_text(end)
+        .map_err(|err| format!("source response is not valid XML: {err}"))
+        .and_then(|text| {
+            quick_xml::escape::unescape(text.as_ref())
+                .map(|text| text.into_owned())
+                .map_err(|err| format!("source response has invalid XML escapes: {err}"))
+        });
+    raw.map_err(SourceError::Other)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bucket::on_demand_migration::source_client::SourceError;
+    use std::sync::{Arc, Mutex};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    const LIST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
+<EnumerationResults ServiceEndpoint="https://acct.blob.core.windows.net/" ContainerName="legacy">
+  <Prefix>photos/</Prefix>
+  <Delimiter>/</Delimiter>
+  <MaxResults>2</MaxResults>
+  <Blobs>
+    <Blob>
+      <Name>photos/a &amp; b.jpg</Name>
+      <Properties>
+        <Last-Modified>Wed, 21 Oct 2015 07:28:00 GMT</Last-Modified>
+        <Etag>0x8D2F1B0A1B2C3D4</Etag>
+        <Content-Length>42</Content-Length>
+        <Content-MD5>1B2M2Y8AsgTpgAmY7PhCfg==</Content-MD5>
+        <BlobType>BlockBlob</BlobType>
+        <AccessTier>Hot</AccessTier>
+      </Properties>
+    </Blob>
+    <Blob>
+      <Name>photos/b.jpg</Name>
+      <Properties>
+        <Content-Length>7</Content-Length>
+      </Properties>
+    </Blob>
+    <BlobPrefix>
+      <Name>photos/raw/</Name>
+    </BlobPrefix>
+  </Blobs>
+  <NextMarker>2!76!MDAwMDI0</NextMarker>
+</EnumerationResults>"#;
+
+    const LAST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
+<EnumerationResults><Blobs><Blob><Name>only.txt</Name><Properties><Content-Length>1</Content-Length></Properties></Blob></Blobs><NextMarker /></EnumerationResults>"#;
+
+    const TAGS: &str = r#"<?xml version="1.0" encoding="utf-8"?>
+<Tags><TagSet>
+  <Tag><Key>env</Key><Value>prod</Value></Tag>
+  <Tag><Key>team</Key><Value>storage &amp; co</Value></Tag>
+</TagSet></Tags>"#;
+
+    #[test]
+    fn list_blobs_maps_entries_prefixes_and_the_marker() {
+        let listing = parse_list_blobs(LIST_PAGE).expect("page should parse");
+        assert_eq!(listing.prefixes, vec!["photos/raw/"]);
+        assert_eq!(listing.next_marker.as_deref(), Some("2!76!MDAwMDI0"));
+        assert_eq!(listing.objects.len(), 2);
+        let first = &listing.objects[0];
+        assert_eq!(first.key, "photos/a & b.jpg", "XML entities in a blob name are decoded");
+        assert_eq!(first.etag.as_deref(), Some("0x8D2F1B0A1B2C3D4"));
+        assert_eq!(first.size, 42);
+        assert_eq!(
+            first.last_modified,
+            Some(std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_445_412_480))
+        );
+        assert_eq!(first.storage_class.as_deref(), Some("Hot"));
+        assert!(!first.is_multipart_etag);
+        assert_eq!(listing.objects[1].key, "photos/b.jpg");
+        assert_eq!(listing.objects[1].size, 7);
+        assert!(listing.objects[1].etag.is_none());
+    }
+
+    #[test]
+    fn an_empty_next_marker_ends_the_listing() {
+        let listing = parse_list_blobs(LAST_PAGE).expect("page should parse");
+        assert_eq!(listing.objects.len(), 1);
+        assert!(listing.next_marker.is_none(), "an empty NextMarker is not a cursor");
+    }
+
+    #[test]
+    fn malformed_listing_xml_is_an_error() {
+        for bad in [
+            "<EnumerationResults><Blobs>",
+            "<EnumerationResults><Blobs><Blob><Name>a</Name></Blobs></EnumerationResults>",
+            "not xml at <all",
+        ] {
+            let err = parse_list_blobs(bad).expect_err("{bad} must fail");
+            assert!(matches!(err, SourceError::Other(_)), "{bad}: {err:?}");
+        }
+        assert!(parse_blob_tags("<Tags><TagSet>").is_err(), "a truncated tag set must fail");
+    }
+
+    #[test]
+    fn blob_tags_parse_into_the_shared_tag_map() {
+        let tags = parse_blob_tags(TAGS).expect("tags should parse");
+        assert_eq!(
+            tags,
+            HashMap::from([
+                ("env".to_string(), "prod".to_string()),
+                ("team".to_string(), "storage & co".to_string())
+            ])
+        );
+        assert!(parse_blob_tags("<Tags><TagSet /></Tags>").expect("empty tag set").is_empty());
+    }
+
+    /// Signature fixture from a request this backend really builds: it pins the
+    /// canonical form so a change to the header set or the query canonicalization
+    /// cannot silently start producing signatures Azure rejects.
+    #[test]
+    fn shared_key_signs_the_canonical_request() {
+        let key = base64_simd::STANDARD.decode_to_vec(b"c2VjcmV0LWtleQ==").expect("test key");
+        let mut headers = HeaderMap::new();
+        headers.insert(HEADER_VERSION, HeaderValue::from_static(API_VERSION));
+        headers.insert(HEADER_DATE, HeaderValue::from_static("Sun, 06 Nov 1994 08:49:37 GMT"));
+        headers.insert(http::header::RANGE, HeaderValue::from_static("bytes=10-14"));
+        let url = Url::parse("https://acct.blob.core.windows.net/legacy/photos/a.jpg").expect("url");
+
+        let signature = shared_key_signature(&key, "acct", "GET", &url, &headers).expect("signature");
+        let expected = {
+            let string_to_sign = concat!(
+                "GET\n\n\n\n\n\n\n\n\n\n\n",
+                "bytes=10-14\n",
+                "x-ms-date:Sun, 06 Nov 1994 08:49:37 GMT\n",
+                "x-ms-version:2021-08-06\n",
+                "/acct/legacy/photos/a.jpg"
+            );
+            let mut mac = HmacSha256::new_from_slice(&key).expect("hmac");
+            mac.update(string_to_sign.as_bytes());
+            base64_simd::STANDARD.encode_to_string(mac.finalize().into_bytes())
+        };
+        assert_eq!(signature, expected);
+    }
+
+    #[test]
+    fn shared_key_canonicalizes_query_parameters() {
+        let key = vec![1_u8; 32];
+        let mut headers = HeaderMap::new();
+        headers.insert(HEADER_VERSION, HeaderValue::from_static(API_VERSION));
+        // Query order must not change the signature: Azure canonicalizes by
+        // lowercased parameter name.
+        let a = Url::parse("https://acct.blob.core.windows.net/legacy?restype=container&comp=list&prefix=p%2F").expect("url");
+        let b = Url::parse("https://acct.blob.core.windows.net/legacy?prefix=p%2F&COMP=list&restype=container").expect("url");
+        assert_eq!(
+            shared_key_signature(&key, "acct", "GET", &a, &headers).expect("a"),
+            shared_key_signature(&key, "acct", "GET", &b, &headers).expect("b")
+        );
+    }
+
+    fn backend(endpoint: &Url, credential: Credential) -> AzureSourceBackend {
+        AzureSourceBackend {
+            http: NativeHttp::for_test(endpoint.clone()),
+            account: "acct".to_string(),
+            container: "legacy".to_string(),
+            credential,
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    struct Recorded {
+        method: String,
+        target: String,
+        headers: Vec<(String, String)>,
+    }
+
+    impl Recorded {
+        fn header(&self, name: &str) -> Option<&str> {
+            self.headers
+                .iter()
+                .find(|(key, _)| key.eq_ignore_ascii_case(name))
+                .map(|(_, value)| value.as_str())
+        }
+    }
+
+    /// One canned HTTP/1.1 response per connection. `Connection: close` keeps
+    /// every request on its own socket so the script order is deterministic.
+    async fn scripted_server(responses: Vec<(u16, Vec<(&'static str, String)>, String)>) -> (Url, Arc<Mutex<Vec<Recorded>>>) {
+        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 recorded: Arc<Mutex<Vec<Recorded>>> = Arc::new(Mutex::new(Vec::new()));
+        let sink = Arc::clone(&recorded);
+
+        tokio::spawn(async move {
+            for (status, headers, body) 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();
+                let recorded = Recorded {
+                    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(),
+                };
+                sink.lock().expect("recorder lock").push(recorded);
+
+                let mut response = format!("HTTP/1.1 {status} X\r\nContent-Length: {}\r\nConnection: close\r\n", body.len());
+                for (name, value) in headers {
+                    response.push_str(&format!("{name}: {value}\r\n"));
+                }
+                response.push_str("\r\n");
+                response.push_str(&body);
+                let _ = stream.write_all(response.as_bytes()).await;
+                let _ = stream.flush().await;
+            }
+        });
+
+        (Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorded)
+    }
+
+    fn blob_headers() -> Vec<(&'static str, String)> {
+        vec![
+            ("ETag", "\"0x8D2F1B0A1B2C3D4\"".to_string()),
+            ("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
+            ("Content-Type", "image/jpeg".to_string()),
+            ("Content-MD5", "1B2M2Y8AsgTpgAmY7PhCfg==".to_string()),
+            ("x-ms-meta-owner", "alice".to_string()),
+            ("x-ms-access-tier", "Cool".to_string()),
+            ("x-ms-version-id", "2026-01-01T00:00:00.0000000Z".to_string()),
+            ("x-ms-blob-type", "BlockBlob".to_string()),
+        ]
+    }
+
+    #[tokio::test]
+    async fn head_signs_the_request_and_maps_azure_metadata() {
+        let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await;
+        let backend = backend(&endpoint, Credential::SharedKey(b"0123456789abcdef0123456789abcdef".to_vec()));
+
+        let head = backend.head("photos/a b.jpg").await.expect("HEAD should map");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert_eq!(recorded.len(), 1);
+        assert_eq!(recorded[0].method, "HEAD");
+        assert_eq!(recorded[0].target, "/legacy/photos/a%20b.jpg", "the blob name is path-encoded");
+        assert_eq!(recorded[0].header("x-ms-version"), Some(API_VERSION));
+        assert!(recorded[0].header("x-ms-date").is_some(), "a signed request must carry x-ms-date");
+        assert!(
+            recorded[0]
+                .header("authorization")
+                .is_some_and(|value| value.starts_with("SharedKey acct:")),
+            "{:?}",
+            recorded[0].header("authorization")
+        );
+
+        assert_eq!(head.etag.as_deref(), Some("0x8D2F1B0A1B2C3D4"));
+        assert!(head.etag_is_opaque, "an Azure ETag is never a content digest");
+        assert!(!head.is_multipart_etag);
+        assert_eq!(head.size, 0);
+        assert_eq!(head.content_type.as_deref(), Some("image/jpeg"));
+        assert_eq!(head.storage_class.as_deref(), Some("Cool"));
+        assert_eq!(head.version_id.as_deref(), Some("2026-01-01T00:00:00.0000000Z"));
+        assert_eq!(head.user_metadata, HashMap::from([("owner".to_string(), "alice".to_string())]));
+        assert!(head.sse.is_none());
+    }
+
+    #[tokio::test]
+    async fn sas_credentials_travel_in_the_query_and_never_sign() {
+        let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await;
+        let backend = backend(
+            &endpoint,
+            Credential::Sas(vec![
+                ("sv".to_string(), "2021-08-06".to_string()),
+                ("sig".to_string(), "a+b/c=".to_string()),
+            ]),
+        );
+
+        backend.head("a.txt").await.expect("HEAD should map");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert!(recorded[0].header("authorization").is_none(), "a SAS request must not be signed");
+        assert!(recorded[0].target.contains("sv=2021-08-06"), "{}", recorded[0].target);
+        assert!(
+            recorded[0].target.contains("sig=a%2Bb%2Fc%3D"),
+            "the SAS signature must be re-encoded exactly once: {}",
+            recorded[0].target
+        );
+    }
+
+    #[tokio::test]
+    async fn get_passes_the_range_through_and_streams_the_body() {
+        let mut headers = blob_headers();
+        headers.push(("Content-Range", "bytes 10-14/100".to_string()));
+        let (endpoint, recorded) = scripted_server(vec![(206, headers, "hello".to_string())]).await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+
+        let range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 10,
+            end: 14,
+        };
+        let got = backend.get("a.txt", Some(&range)).await.expect("ranged GET should succeed");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert_eq!(recorded[0].method, "GET");
+        assert_eq!(recorded[0].header("range"), Some("bytes=10-14"));
+        assert_eq!(got.content_range.as_deref(), Some("bytes 10-14/100"));
+        assert_eq!(got.head.size, 5);
+        let body = got.body.collect().await.expect("body should stream").into_bytes();
+        assert_eq!(body.as_ref(), b"hello");
+    }
+
+    #[tokio::test]
+    async fn customer_key_blobs_are_refused() {
+        let mut headers = blob_headers();
+        headers.push(("x-ms-encryption-key-sha256", "abc".to_string()));
+        let (endpoint, _) = scripted_server(vec![(200, headers, String::new())]).await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+
+        let err = backend.head("a.txt").await.expect_err("customer-key blobs are unsupported");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+        assert_eq!(err.class_label(), "unsupported");
+        assert!(!err.is_retryable());
+    }
+
+    #[tokio::test]
+    async fn list_requests_the_container_and_pages_with_the_marker() {
+        let (endpoint, recorded) = scripted_server(vec![
+            (200, Vec::new(), LIST_PAGE.to_string()),
+            (200, Vec::new(), LAST_PAGE.to_string()),
+        ])
+        .await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+
+        let page = backend
+            .list(&SourceListRequest {
+                prefix: Some("photos/"),
+                delimiter: Some("/"),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("first page should list");
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("2!76!MDAwMDI0"));
+        assert_eq!(page.common_prefixes, vec!["photos/raw/"]);
+
+        let page = backend
+            .list(&SourceListRequest {
+                prefix: Some("photos/"),
+                continuation_token: page.next_continuation_token.as_deref(),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("second page should list");
+        assert!(!page.is_truncated);
+        assert!(page.next_continuation_token.is_none());
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        for request in &recorded {
+            assert!(request.target.starts_with("/legacy?"), "{}", request.target);
+            assert!(request.target.contains("restype=container"), "{}", request.target);
+            assert!(request.target.contains("comp=list"), "{}", request.target);
+            assert!(request.target.contains("prefix=photos%2F"), "{}", request.target);
+            assert!(request.target.contains("maxresults=2"), "{}", request.target);
+        }
+        assert!(!recorded[0].target.contains("marker="), "{}", recorded[0].target);
+        assert!(recorded[1].target.contains("marker=2%2176%21MDAwMDI0"), "{}", recorded[1].target);
+    }
+
+    #[tokio::test]
+    async fn list_refuses_a_start_after_cursor_before_sending() {
+        let (endpoint, recorded) = scripted_server(Vec::new()).await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+
+        let err = backend
+            .list(&SourceListRequest {
+                start_after: Some("a"),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect_err("azure has no start-after form");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+        assert!(
+            recorded.lock().expect("recorder lock").is_empty(),
+            "an unsupported request must never reach the source"
+        );
+    }
+
+    #[tokio::test]
+    async fn tagging_and_probe_address_the_right_resources() {
+        let (endpoint, recorded) =
+            scripted_server(vec![(200, Vec::new(), TAGS.to_string()), (200, Vec::new(), String::new())]).await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+
+        let tags = backend.tagging("a.txt").await.expect("tags should parse");
+        assert_eq!(tags.get("env").map(String::as_str), Some("prod"));
+        backend.probe().await.expect("probe should succeed");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert_eq!(recorded[0].target, "/legacy/a.txt?comp=tags");
+        assert_eq!(recorded[1].method, "HEAD");
+        assert_eq!(recorded[1].target, "/legacy?restype=container");
+    }
+
+    #[tokio::test]
+    async fn azure_statuses_map_onto_the_shared_error_classes() {
+        for (status, code, expected, retryable) in [
+            (404_u16, Some("BlobNotFound"), "not_found", false),
+            (403, Some("AuthorizationPermissionMismatch"), "access_denied", false),
+            (401, None, "access_denied", false),
+            (429, None, "throttled", true),
+            (503, Some("ServerBusy"), "throttled", true),
+            (500, None, "server_error", true),
+        ] {
+            let headers = code
+                .map(|code| vec![(HEADER_ERROR_CODE, code.to_string())])
+                .unwrap_or_default();
+            let (endpoint, _) = scripted_server(vec![(status, headers, String::new())]).await;
+            let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+            let err = backend.head("a.txt").await.expect_err("{status} must fail");
+            assert_eq!(err.class_label(), expected, "status {status} -> {err:?}");
+            assert_eq!(err.is_retryable(), retryable, "status {status} -> {err:?}");
+        }
+    }
+
+    #[tokio::test]
+    async fn transport_failures_never_render_the_request_url() {
+        // Nothing is listening on the reserved port, so the connect fails and
+        // the error must not carry the SAS-bearing URL.
+        let backend = backend(
+            &Url::parse("http://127.0.0.1:1").expect("endpoint"),
+            Credential::Sas(vec![("sig".to_string(), "top-secret-signature".to_string())]),
+        );
+        let err = backend.head("a.txt").await.expect_err("a closed port must fail");
+        let rendered = err.to_string();
+        assert!(!rendered.contains("top-secret-signature"), "{rendered}");
+        assert!(!rendered.contains("127.0.0.1"), "{rendered}");
+    }
+}

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

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

@@ -19,11 +19,17 @@
 //! 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`).
 
+pub mod azure;
 pub mod backfill;
 pub mod breaker;
 pub mod config;
 pub mod list_through;
+mod native_http;
 pub mod negative_cache;
 pub mod pull;
 pub mod source_client;

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

@@ -0,0 +1,396 @@
+// 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()))
+}
+
+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 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,7 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.
 
+use super::azure::AzureSourceBackend;
 use crate::bucket::remote_s3_client::{
     PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
 };
@@ -64,6 +65,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 {
@@ -75,6 +80,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,
         }
     }
@@ -87,6 +94,8 @@ impl SourceProvider {
             Self::Minio => "minio",
             Self::Rustfs => "rustfs",
             Self::S3 => "s3",
+            Self::Azure => "azure",
+            Self::GcsNative => "gcs_native",
         }
     }
 
@@ -158,6 +167,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 {
@@ -268,7 +340,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;
@@ -341,6 +413,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.
@@ -362,7 +439,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())
 }
@@ -411,6 +488,7 @@ fn source_head(parts: HeadParts) -> Result<SourceHead, SourceError> {
         storage_class: parts.storage_class,
         sse,
         is_multipart_etag,
+        etag_is_opaque: false,
     })
 }
 
@@ -621,9 +699,40 @@ impl fmt::Debug for SourceClient {
 
 impl SourceClient {
     pub async fn new(spec: &SourceClientSpec) -> Result<Self, RemoteS3ClientError> {
-        let endpoint = spec.endpoint_spec()?;
+        match &spec.backend {
-        let config = build_remote_s3_config(&endpoint).await?;
+            SourceBackendSpec::S3 => {
-        Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
+                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(_) => {
+                Err(RemoteS3ClientError::Credentials("the native gcs source backend is not implemented yet"))
+            }
+        }
+    }
+
+    /// 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
@@ -983,6 +1092,7 @@ mod tests {
             retry: RemoteS3RetryPolicy::Disabled,
             timeouts: SourceTimeouts::default(),
             bandwidth_limit: NonZeroU64::new(1_000_000),
+            backend: SourceBackendSpec::S3,
         }
     }
 
@@ -1487,7 +1597,9 @@ 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);
     }
 
     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(crate) 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),