diff --git a/Cargo.lock b/Cargo.lock index e9e7d9fe5..880262cff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9703,6 +9703,7 @@ dependencies = [ "path-absolutize", "pin-project-lite", "proptest", + "quick-xml", "rand 0.10.2", "ratelimit", "rcgen", diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 50aa7c1d3..95e368ddb 100644 --- a/crates/ecstore/Cargo.toml +++ b/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"] } diff --git a/crates/ecstore/src/bucket/on_demand_migration/azure.rs b/crates/ecstore/src/bucket/on_demand_migration/azure.rs new file mode 100644 index 000000000..b5fab6fa4 --- /dev/null +++ b/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; + +/// 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), + /// 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 { + 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 { + self.http.url(std::iter::once(self.container.as_str()).chain(key.split('/'))) + } + + fn container_url(&self) -> Result { + 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 { + 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 { + // 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 { + 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 { + 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 { + // 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, 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 { + 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> = 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, + prefixes: Vec, + /// `None` when the listing is complete; Azure marks the end with an empty + /// `NextMarker`. + next_marker: Option, +} + +#[derive(Default)] +struct BlobEntry { + name: String, + etag: Option, + size: u64, + last_modified: Option, + access_tier: Option, +} + +/// Parses one `List Blobs` page. +fn parse_list_blobs(xml: &str) -> Result { + 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 = 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, + prefixes: &mut Vec, + next_marker: &mut Option, + 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, 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 { + 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#" + + photos/ + / + 2 + + + photos/a & b.jpg + + Wed, 21 Oct 2015 07:28:00 GMT + 0x8D2F1B0A1B2C3D4 + 42 + 1B2M2Y8AsgTpgAmY7PhCfg== + BlockBlob + Hot + + + + photos/b.jpg + + 7 + + + + photos/raw/ + + + 2!76!MDAwMDI0 +"#; + + const LAST_PAGE: &str = r#" +only.txt1"#; + + const TAGS: &str = r#" + + envprod + teamstorage & co +"#; + + #[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 [ + "", + "a", + "not xml at ").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("").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>>) { + 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>> = 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}"); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/config.rs b/crates/ecstore/src/bucket/on_demand_migration/config.rs index 759762d26..a7deabf34 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/config.rs +++ b/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, #[serde(default)] pub tls: TlsConfig, + /// Required for [`Provider::Azure`] and rejected for every other + /// provider. + #[serde(default)] + pub azure: Option, + /// 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, } -/// 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, + /// SAS query string without the leading `?`. + #[serde(default)] + pub sas_token: Option, +} + +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); diff --git a/crates/ecstore/src/bucket/on_demand_migration/mod.rs b/crates/ecstore/src/bucket/on_demand_migration/mod.rs index 85a4d5a99..5f6acf9c2 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/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; diff --git a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs new file mode 100644 index 000000000..9f64860ee --- /dev/null +++ b/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 { + 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) -> Result { + 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 { + 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 { + 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 { + 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 { + 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, + /// The ETag is an opaque token rather than a digest of the bytes. + pub(super) etag_is_opaque: bool, + pub(super) version_id: Option, + pub(super) storage_class: Option, +} + +/// 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 { + let size = header(headers, "content-length") + .and_then(|value| value.parse::().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"); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/pull.rs b/crates/ecstore/src/bucket/on_demand_migration/pull.rs index 60f7145a2..b0b5dc36d 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/pull.rs +++ b/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(), diff --git a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs index 4782e05f0..bad312b6b 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/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, + /// 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, pub sse: Option, 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, } -fn normalize_etag(etag: Option) -> Option { +pub(super) fn normalize_etag(etag: Option) -> Option { etag.map(|etag| etag.trim().trim_matches('"').to_string()) .filter(|etag| !etag.is_empty()) } @@ -411,6 +488,7 @@ fn source_head(parts: HeadParts) -> Result { 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 { - 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(_) => { + 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, 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) -> SourceClient { diff --git a/crates/ecstore/src/bucket/on_demand_migration/sys.rs b/crates/ecstore/src/bucket/on_demand_migration/sys.rs index 16cb8ea65..2a1bac332 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/sys.rs +++ b/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),