diff --git a/Cargo.lock b/Cargo.lock index f4cc41aae..dc72ece63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9792,6 +9792,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 d36c9d85b..dd2063129 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/api/mod.rs b/crates/ecstore/src/api/mod.rs index fd66c897a..e752713ff 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -153,12 +153,13 @@ pub mod bucket { LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason, PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS, - SourceLatencySnapshot, source_client_spec, + SourceLatencySnapshot, source_backend_spec, source_client_spec, }; pub use crate::bucket::on_demand_migration::{ - ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION, - OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, - SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, + AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, + ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, + Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, + ValidationContext, }; pub use crate::bucket::on_demand_migration::{ EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS, @@ -185,9 +186,9 @@ pub mod bucket { } pub mod source_client { pub use crate::bucket::on_demand_migration::source_client::{ - SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, - SourceProbe, SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, - resolve_path_style, + AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError, + SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse, + SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style, }; } } 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..08e687797 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/azure.rs @@ -0,0 +1,1046 @@ +// 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) => { + let key = base64_simd::STANDARD + .decode_to_vec(key.as_bytes()) + .map_err(|_| RemoteS3ClientError::Credentials("azure account key is not base64"))?; + // HMAC accepts a zero-length key, so an absent one would sign + // every request with nothing rather than fail here. + if key.is_empty() { + return Err(RemoteS3ClientError::Credentials("azure account key is empty")); + } + Credential::SharedKey(key) + } + 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::backend_contract::{BackendCapabilities, assert_backend_contract}; + use crate::bucket::on_demand_migration::source_client::SourceError; + use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; + + 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, + } + } + + 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()), + ] + } + + #[test] + fn an_absent_or_malformed_account_key_is_refused_before_any_request() { + for key in ["", "not base64!"] { + let spec = AzureSourceSpec { + account: "acct".to_string(), + auth: AzureAuth::SharedKey(key.to_string()), + }; + let built = AzureSourceBackend::new( + "https://acct.blob.core.windows.net", + "legacy", + &spec, + SourceTimeouts::default(), + false, + None, + ); + assert!( + matches!(built, Err(RemoteS3ClientError::Credentials(_))), + "{key:?} must not build a client" + ); + } + let spec = AzureSourceSpec { + account: "acct".to_string(), + auth: AzureAuth::Sas(String::new()), + }; + assert!( + AzureSourceBackend::new( + "https://acct.blob.core.windows.net", + "legacy", + &spec, + SourceTimeouts::default(), + false, + None + ) + .is_err(), + "an empty SAS token carries no parameters" + ); + } + + #[tokio::test] + async fn head_signs_the_request_and_maps_azure_metadata() { + let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(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![ScriptedResponse::new(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![ScriptedResponse::new(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![ScriptedResponse::new(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![ + ScriptedResponse::new(200, Vec::new(), LIST_PAGE.to_string()), + ScriptedResponse::new(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![ + ScriptedResponse::new(200, Vec::new(), TAGS.to_string()), + ScriptedResponse::new(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![ScriptedResponse::new(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:?}"); + } + } + + const CONTRACT_LIST_PAGE_ONE: &str = r#" + + + + dir/a.txt + + Wed, 21 Oct 2015 07:28:00 GMT + 0x8D2F1B0A1B2C3D4 + 5 + Hot + + + dir/sub/ + + cursor-1 +"#; + + const CONTRACT_LIST_PAGE_TWO: &str = r#" + + + + dir/b.txt + + Wed, 21 Oct 2015 07:28:00 GMT + 7 + + + + +"#; + + const CONTRACT_TAGS: &str = r#" +envprod"#; + + fn contract_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", "text/plain".to_string()), + ("x-ms-meta-owner", "alice".to_string()), + ("x-ms-access-tier", "Hot".to_string()), + ("x-ms-blob-type", "BlockBlob".to_string()), + ] + } + + #[tokio::test] + async fn azure_backend_satisfies_the_shared_backend_contract() { + let mut ranged = contract_blob_headers(); + ranged.push(("Content-Range", "bytes 1-3/5".to_string())); + // A HEAD reports the object size with no body, exactly as Azure does. + let mut head_only = contract_blob_headers(); + head_only.push(("Content-Length", "5".to_string())); + let (endpoint, _) = scripted_server(vec![ + ScriptedResponse::new(200, head_only, String::new()), + ScriptedResponse::new(200, contract_blob_headers(), "hello".to_string()), + ScriptedResponse::new(206, ranged, "ell".to_string()), + ScriptedResponse::new(200, Vec::new(), CONTRACT_LIST_PAGE_ONE.to_string()), + ScriptedResponse::new(200, Vec::new(), CONTRACT_LIST_PAGE_TWO.to_string()), + ScriptedResponse::new(200, Vec::new(), CONTRACT_TAGS.to_string()), + ScriptedResponse::new(200, Vec::new(), String::new()), + ScriptedResponse::new(404, vec![(HEADER_ERROR_CODE, "BlobNotFound".to_string())], String::new()), + ScriptedResponse::new( + 403, + vec![(HEADER_ERROR_CODE, "AuthorizationPermissionMismatch".to_string())], + String::new(), + ), + ]) + .await; + let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32])); + + assert_backend_contract( + &backend, + BackendCapabilities { + // Azure's ETag is a concurrency token; the contract requires it + // to be carried but never read as a digest. + etag_is_opaque: true, + // Azure paginates only with an opaque marker. + supports_start_after: false, + supports_tagging: true, + }, + ) + .await; + } + + #[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/backend_contract.rs b/crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs new file mode 100644 index 000000000..a8e1b1337 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs @@ -0,0 +1,172 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! One contract every [`SourceBackend`] implementation must satisfy. +//! +//! The migration pipeline talks to a source only through the trait, so a new +//! provider is correct exactly when it answers the same questions the same way: +//! the same head fields, the same range semantics, the same page shape, the +//! same error classes. Each backend supplies a fixture that answers this fixed +//! corpus in its own dialect and then runs [`assert_backend_contract`], so a +//! provider-specific mapping bug shows up as a contract failure rather than as +//! a surprise in the pull pipeline. +//! +//! Backends differ in two documented ways, declared through +//! [`BackendCapabilities`]: whether the provider's ETag is a content digest, +//! and whether the provider can resume a listing from a key. + +use super::source_client::{SourceBackend, SourceError, SourceListRequest}; +use crate::storage_api_contracts::range::HTTPRangeSpec; +use std::collections::HashMap; + +/// The single object every fixture serves. +pub(super) const OBJECT_KEY: &str = "dir/a.txt"; +pub(super) const OBJECT_BODY: &[u8] = b"hello"; +/// MD5 of [`OBJECT_BODY`]; the ETag of the object on a digest provider. +pub(super) const OBJECT_MD5: &str = "5d41402abc4b2a76b9719d911017c592"; +/// The second key the fixture's listing returns, on its second page. +pub(super) const SECOND_KEY: &str = "dir/b.txt"; +pub(super) const COMMON_PREFIX: &str = "dir/sub/"; +pub(super) const LIST_CURSOR: &str = "cursor-1"; +/// A key the fixture answers with the provider's "no such object". +pub(super) const MISSING_KEY: &str = "missing"; +/// A key the fixture answers with the provider's "not authorized". +pub(super) const FORBIDDEN_KEY: &str = "secret"; + +/// Where backends are allowed to differ. +#[derive(Clone, Copy, Debug)] +pub(super) struct BackendCapabilities { + /// The provider's ETag is an opaque token, not a digest of the bytes. + pub(super) etag_is_opaque: bool, + /// The provider can resume a listing from a key rather than only from an + /// opaque cursor. + pub(super) supports_start_after: bool, + /// The provider has an object-tagging concept at all. GCS does not, and + /// answers with an empty map instead of failing a pull. + pub(super) supports_tagging: bool, +} + +/// Drives `backend` through the shared corpus. Fixtures are scripted in +/// request order, so the call order here is part of the contract. +pub(super) async fn assert_backend_contract(backend: &dyn SourceBackend, caps: BackendCapabilities) { + // 1. HEAD maps the object's shared fields. + let head = backend.head(OBJECT_KEY).await.expect("HEAD of the fixture object"); + assert_eq!(head.size, OBJECT_BODY.len() as u64, "HEAD reports the object size"); + assert_eq!(head.content_type.as_deref(), Some("text/plain")); + assert_eq!( + head.user_metadata, + HashMap::from([("owner".to_string(), "alice".to_string())]), + "user metadata is keyed without the provider prefix" + ); + assert!(head.storage_class.is_some(), "the provider's tier is recorded"); + assert!(head.last_modified.is_some(), "the provider's timestamp is parsed"); + assert!(head.sse.is_none(), "the fixture object is not server-side encrypted"); + assert!(!head.is_multipart_etag); + assert_eq!(head.etag_is_opaque, caps.etag_is_opaque); + match caps.etag_is_opaque { + false => assert_eq!(head.etag.as_deref(), Some(OBJECT_MD5), "a digest ETag is mapped verbatim"), + true => assert!(head.etag.is_some(), "an opaque ETag is still recorded"), + } + + // 2. An unranged GET streams the whole object and reports no range. + let got = backend.get(OBJECT_KEY, None).await.expect("unranged GET"); + assert_eq!(got.head.size, OBJECT_BODY.len() as u64); + assert!(got.content_range.is_none(), "an unranged GET has no content-range"); + assert_eq!(got.head.etag_is_opaque, caps.etag_is_opaque, "GET and HEAD agree about the ETag"); + let body = got.body.collect().await.expect("body streams").into_bytes(); + assert_eq!(body.as_ref(), OBJECT_BODY); + + // 3. A ranged GET returns exactly the requested interval, and `size` is + // the length of the returned bytes rather than of the object. + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 1, + end: 3, + }; + let got = backend.get(OBJECT_KEY, Some(&range)).await.expect("ranged GET"); + assert_eq!(got.head.size, 3, "a ranged GET reports the range length"); + assert_eq!(got.content_range.as_deref(), Some("bytes 1-3/5")); + let body = got.body.collect().await.expect("body streams").into_bytes(); + assert_eq!(body.as_ref(), &OBJECT_BODY[1..=3]); + + // 4. A delimiter listing rolls prefixes up and hands back a cursor. + let page = backend + .list(&SourceListRequest { + prefix: Some("dir/"), + delimiter: Some("/"), + max_keys: 2, + ..Default::default() + }) + .await + .expect("first listing page"); + assert_eq!(page.objects.len(), 1, "the first page holds one object"); + assert_eq!(page.objects[0].key, OBJECT_KEY, "listing keys are in the source namespace"); + assert_eq!(page.objects[0].size, OBJECT_BODY.len() as u64); + assert!(page.objects[0].last_modified.is_some()); + assert_eq!(page.common_prefixes, vec![COMMON_PREFIX.to_string()]); + assert!(page.is_truncated); + assert_eq!(page.next_continuation_token.as_deref(), Some(LIST_CURSOR)); + + // 5. The cursor is passed back verbatim and the last page ends the walk. + let page = backend + .list(&SourceListRequest { + prefix: Some("dir/"), + delimiter: Some("/"), + continuation_token: Some(LIST_CURSOR), + max_keys: 2, + ..Default::default() + }) + .await + .expect("second listing page"); + assert_eq!(page.objects.len(), 1); + assert_eq!(page.objects[0].key, SECOND_KEY); + assert!(!page.is_truncated); + assert!(page.next_continuation_token.is_none(), "a complete listing carries no cursor"); + + // 6. Tags come back as a flat map, empty on a provider without tags. + let tags = backend.tagging(OBJECT_KEY).await.expect("object tags"); + match caps.supports_tagging { + true => assert_eq!(tags, HashMap::from([("env".to_string(), "prod".to_string())])), + false => assert!(tags.is_empty(), "a provider without tags reports none: {tags:?}"), + } + + // 7. The probe confirms the bucket or container answers. + backend.probe().await.expect("probe of the fixture bucket"); + + // 8. A missing object is `NotFound`, and never retried. + let err = backend.head(MISSING_KEY).await.expect_err("a missing object must fail"); + assert!(matches!(err, SourceError::NotFound), "{err:?}"); + assert_eq!(err.class_label(), "not_found"); + assert!(!err.is_retryable()); + + // 9. A denied object is `AccessDenied`, and never retried. + let err = backend.head(FORBIDDEN_KEY).await.expect_err("a denied object must fail"); + assert!(matches!(err, SourceError::AccessDenied), "{err:?}"); + assert_eq!(err.class_label(), "access_denied"); + assert!(!err.is_retryable()); + + // 10. A provider without a key cursor must refuse one instead of listing + // from the wrong position. This issues no request either way. + if !caps.supports_start_after { + let err = backend + .list(&SourceListRequest { + start_after: Some(OBJECT_KEY), + max_keys: 1, + ..Default::default() + }) + .await + .expect_err("a backend without a key cursor must refuse start_after"); + assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}"); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/config.rs b/crates/ecstore/src/bucket/on_demand_migration/config.rs index 759762d26..d00f08c2c 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); @@ -820,9 +1024,201 @@ mod tests { "{provider}" ); } + // The native providers never sign with a region, so "auto" is the + // honest value to write for them. + for cfg in [azure_cfg(), gcs_native_cfg()] { + assert_eq!(cfg.source.region, "auto"); + cfg.validate(empty_ctx()) + .unwrap_or_else(|err| panic!("{}: {err}", cfg.source.provider)); + } assert_eq!(sample().source.effective_region(), "us-west-1"); } + const SERVICE_ACCOUNT_JSON: &str = r#"{"type":"service_account","project_id":"p","client_email":"a@b.iam.gserviceaccount.com","private_key":"-----BEGIN PRIVATE KEY-----\nsecret\n-----END PRIVATE KEY-----"}"#; + + fn azure_cfg() -> OnDemandMigrationConfig { + let mut cfg = sample(); + cfg.source.provider = Provider::Azure; + cfg.source.endpoint = None; + cfg.source.region = "auto".to_string(); + cfg.source.credentials = None; + cfg.source.azure = Some(AzureSourceConfig { + account: "legacyaccount".to_string(), + account_key: Some("c2VjcmV0LWtleQ==".to_string()), + sas_token: None, + }); + cfg + } + + fn gcs_native_cfg() -> OnDemandMigrationConfig { + let mut cfg = sample(); + cfg.source.provider = Provider::GcsNative; + cfg.source.endpoint = None; + cfg.source.region = "auto".to_string(); + cfg.source.credentials = None; + cfg.source.gcs = Some(GcsSourceConfig { + service_account_json: SERVICE_ACCOUNT_JSON.to_string(), + }); + cfg + } + + #[test] + fn native_providers_derive_their_endpoint_and_round_trip_on_the_wire() { + let azure = azure_cfg(); + assert_eq!(azure.source.effective_endpoint(), "https://legacyaccount.blob.core.windows.net"); + let gcs = gcs_native_cfg(); + assert_eq!(gcs.source.effective_endpoint(), "https://storage.googleapis.com"); + + for cfg in [azure_cfg(), gcs_native_cfg()] { + let json = cfg.to_json().expect("config must serialize"); + assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg); + } + // The wire labels are part of the admin contract. + assert!( + String::from_utf8(azure_cfg().to_json().expect("json")) + .expect("utf8") + .contains(r#""provider":"azure""#) + ); + assert!( + String::from_utf8(gcs_native_cfg().to_json().expect("json")) + .expect("utf8") + .contains(r#""provider":"gcs_native""#) + ); + } + + #[test] + fn an_explicit_endpoint_overrides_the_derived_native_one() { + // Azurite and fake-gcs-server are addressed this way. + let mut cfg = azure_cfg(); + cfg.source.endpoint = Some("http://azurite.example.com:10000".to_string()); + cfg.validate(empty_ctx()).expect("an explicit native endpoint is allowed"); + assert_eq!(cfg.source.effective_endpoint(), "http://azurite.example.com:10000"); + + cfg.source.endpoint = Some("http://azurite.example.com:10000/devstoreaccount1".to_string()); + assert!( + matches!(cfg.validate(empty_ctx()), Err(OnDemandMigrationConfigError::InvalidEndpoint(_))), + "a native endpoint is still an origin" + ); + } + + #[test] + fn a_provider_block_belongs_to_exactly_its_own_provider() { + let mut cfg = sample(); + cfg.source.azure = azure_cfg().source.azure; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("azure", Provider::S3)) + ); + + let mut cfg = sample(); + cfg.source.gcs = gcs_native_cfg().source.gcs; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("gcs", Provider::S3)) + ); + + let mut cfg = azure_cfg(); + cfg.source.azure = None; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::MissingProviderBlock("azure", Provider::Azure)) + ); + + let mut cfg = gcs_native_cfg(); + cfg.source.gcs = None; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::MissingProviderBlock("gcs", Provider::GcsNative)) + ); + } + + #[test] + fn azure_block_rules() { + let with = |account: &str, key: Option<&str>, sas: Option<&str>| { + let mut cfg = azure_cfg(); + cfg.source.azure = Some(AzureSourceConfig { + account: account.to_string(), + account_key: key.map(str::to_string), + sas_token: sas.map(str::to_string), + }); + cfg.validate(empty_ctx()) + }; + + with("legacyaccount", None, Some("sv=2021-08-06&sig=abc%3D")).expect("a SAS token is a complete credential"); + with("legacyaccount", Some("c2VjcmV0LWtleQ=="), None).expect("an account key is a complete credential"); + + for (label, result) in [ + ("empty account", with("", Some("c2VjcmV0LWtleQ=="), None)), + // The account becomes the first label of the derived hostname. + ("account with a dot", with("legacy.account", Some("c2VjcmV0LWtleQ=="), None)), + ("account with a slash", with("legacy/account", Some("c2VjcmV0LWtleQ=="), None)), + ("no credential", with("legacyaccount", None, None)), + ("both credentials", with("legacyaccount", Some("c2VjcmV0LWtleQ=="), Some("sv=1"))), + ("empty key", with("legacyaccount", Some(""), None)), + ("key that is not base64", with("legacyaccount", Some("not base64!"), None)), + ("empty sas", with("legacyaccount", None, Some(""))), + ("sas with a leading question mark", with("legacyaccount", None, Some("?sv=1"))), + ("sas with whitespace", with("legacyaccount", None, Some("sv=1 &sig=a"))), + ] { + assert!( + matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("azure", _))), + "{label}: {result:?}" + ); + } + } + + #[test] + fn gcs_native_block_requires_a_usable_service_account_key() { + let with = |json: &str| { + let mut cfg = gcs_native_cfg(); + cfg.source.gcs = Some(GcsSourceConfig { + service_account_json: json.to_string(), + }); + cfg.validate(empty_ctx()) + }; + + with(SERVICE_ACCOUNT_JSON).expect("a service-account key is accepted"); + for (label, json) in [ + ("empty", ""), + ("not json", "not json"), + ("not an object", "[]"), + ("wrong type", r#"{"type":"authorized_user","client_email":"a@b","private_key":"k"}"#), + ("no private key", r#"{"type":"service_account","client_email":"a@b"}"#), + ("empty client email", r#"{"type":"service_account","client_email":"","private_key":"k"}"#), + ] { + let result = with(json); + assert!( + matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("gcs", _))), + "{label}: {result:?}" + ); + } + } + + #[test] + fn native_secrets_never_survive_redaction_or_debug() { + let mut azure = azure_cfg(); + azure.source.azure.as_mut().expect("block").sas_token = Some("sv=2021-08-06&sig=top-secret".to_string()); + azure.source.azure.as_mut().expect("block").account_key = None; + let gcs = gcs_native_cfg(); + + for rendered in [ + format!("{:?}", azure.redacted()), + format!("{azure:?}"), + String::from_utf8(azure.redacted().to_json().expect("json")).expect("utf8"), + ] { + assert!(!rendered.contains("top-secret"), "{rendered}"); + assert!(rendered.contains("legacyaccount"), "the account name is not a secret: {rendered}"); + } + for rendered in [ + format!("{:?}", gcs.redacted()), + format!("{gcs:?}"), + String::from_utf8(gcs.redacted().to_json().expect("json")).expect("utf8"), + ] { + assert!(!rendered.contains("PRIVATE KEY-----"), "{rendered}"); + assert!(!rendered.contains("gserviceaccount"), "{rendered}"); + } + } + #[test] fn bucket_rules() { let mut cfg = sample(); diff --git a/crates/ecstore/src/bucket/on_demand_migration/gcs.rs b/crates/ecstore/src/bucket/on_demand_migration/gcs.rs new file mode 100644 index 000000000..707e1cc82 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/gcs.rs @@ -0,0 +1,506 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Native Google Cloud Storage source backend. +//! +//! The `gcs` provider already reaches GCS through its S3 interoperability API, +//! which needs an HMAC key pair. This backend is the other half: it authorizes +//! with a service-account key, the credential most GCS projects actually issue, +//! by minting OAuth tokens through the shared `google-cloud-auth` credential +//! machinery the tier layer already uses. +//! +//! Two GCS surfaces are involved, each for the half it describes best. The read +//! path uses the XML API (`/{bucket}/{object}`), whose responses carry +//! `x-goog-meta-*` user metadata and the `x-goog-hash` digest in one round trip. +//! Listing uses the JSON API (`objects.list`), whose `pageToken` maps directly +//! onto the shared page cursor and whose `prefixes` are the delimiter roll-up. +//! Both accept the same bearer token. +//! +//! Every call this backend makes needs only `storage.objects.get` and +//! `storage.objects.list`, the two permissions of the `objectViewer` role, so a +//! key scoped to exactly the migration's needs works. +//! +//! `x-goog-hash` carries a base64 MD5 for every non-composite object; it is +//! converted to hex and becomes the head's ETag, so a pulled object is checked +//! against the digest GCS itself computed. A composite object has no MD5, and +//! its ETag is then marked opaque rather than checked. + +use super::native_http::{ + NativeHeadFields, NativeHttp, base64_md5_to_hex, header, native_source_head, parse_http_timestamp, read_text, response_body, +}; +use super::source_client::{ + GcsSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, + SourceTimeouts, range_header_value, +}; +use crate::bucket::remote_s3_client::RemoteS3ClientError; +use crate::storage_api_contracts::range::HTTPRangeSpec; +use google_cloud_auth::credentials::service_account::{AccessSpecifier, Builder as ServiceAccountBuilder}; +use google_cloud_auth::credentials::{CacheableResource, Credentials}; +use http::{HeaderMap, HeaderValue, Method}; +use serde::Deserialize; +use std::collections::HashMap; +use url::Url; + +/// Read-only object scope: this backend never writes to the source. +const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only"; +const METADATA_PREFIX: &str = "x-goog-meta-"; +/// GCS reports its error code in the response body, not a header; the shared +/// transport takes a header name, so it is given one that never matches and +/// classification falls back to the status. +const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code"; +/// One `objects.list` page is small; refuse an unbounded document. +const MAX_JSON_BYTES: usize = 8 * 1024 * 1024; + +pub struct GcsNativeSourceBackend { + http: NativeHttp, + bucket: String, + credentials: Credentials, +} + +impl GcsNativeSourceBackend { + pub fn new( + endpoint: &str, + bucket: &str, + spec: &GcsSourceSpec, + timeouts: SourceTimeouts, + skip_tls_verify: bool, + ca_cert_pem: Option<&str>, + ) -> Result { + let key: serde_json::Value = serde_json::from_str(&spec.service_account_json) + .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not valid JSON"))?; + let credentials = ServiceAccountBuilder::new(key) + .with_access_specifier(AccessSpecifier::from_scopes([READ_ONLY_SCOPE])) + .build() + .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not usable"))?; + Ok(Self { + http: NativeHttp::new(endpoint, timeouts, skip_tls_verify, ca_cert_pem)?, + bucket: bucket.to_string(), + credentials, + }) + } + + /// Authorization headers for one request. A credential failure is reported + /// as `AccessDenied` with no message: the renderer of a credential error + /// has the key material in scope, and the class is what callers act on. + async fn auth_headers(&self) -> Result { + match self.credentials.headers(http::Extensions::new()).await { + Ok(CacheableResource::New { data, .. }) => Ok(data), + // Only returned when the caller passes an entity tag, which this + // backend never does; an empty set is still the honest answer. + Ok(CacheableResource::NotModified) => Ok(HeaderMap::new()), + Err(_) => Err(SourceError::AccessDenied), + } + } + + /// XML API URL of one object; `/` in the key stay path separators. + fn object_url(&self, key: &str) -> Result { + self.http.url(std::iter::once(self.bucket.as_str()).chain(key.split('/'))) + } + + /// JSON API URL of the bucket's object collection. + fn objects_url(&self) -> Result { + self.http.url(["storage", "v1", "b", self.bucket.as_str(), "o"]) + } + + async fn request(&self, method: Method, url: Url, mut headers: HeaderMap) -> Result { + for (name, value) in self.auth_headers().await? { + if let Some(name) = name { + headers.insert(name, value); + } + } + let mut request = reqwest::Request::new(method, url); + *request.headers_mut() = headers; + Ok(request) + } + + /// Shared mapping for the XML API's HEAD and GET responses. + fn head_from_response(headers: &HeaderMap) -> Result { + if header(headers, "x-goog-encryption-key-sha256").is_some() { + return Err(SourceError::Unsupported( + "source object uses a customer-supplied encryption key; customer-key sources are not supported".to_string(), + )); + } + // `x-goog-hash` lists digests as `name=base64`, comma separated, and may + // repeat across header lines. Only the MD5 describes the whole object. + let md5 = headers + .get_all("x-goog-hash") + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .filter_map(|digest| digest.trim().strip_prefix("md5=")) + .find_map(base64_md5_to_hex); + + let (etag, etag_is_opaque) = match md5 { + Some(md5) => (Some(md5), false), + // A composite object has no MD5; its ETag describes the composition + // rather than the bytes, so it is provenance only. + None => (header(headers, "etag").map(str::to_string), true), + }; + native_source_head( + headers, + METADATA_PREFIX, + NativeHeadFields { + etag, + etag_is_opaque, + version_id: header(headers, "x-goog-generation").map(str::to_string), + storage_class: header(headers, "x-goog-storage-class").map(str::to_string), + }, + ) + } +} + +#[async_trait::async_trait] +impl SourceBackend for GcsNativeSourceBackend { + async fn head(&self, key: &str) -> Result { + let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?; + let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + Self::head_from_response(response.headers()) + } + + async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result { + let mut headers = HeaderMap::new(); + if let Some(range) = range.map(range_header_value).transpose()? { + headers.insert( + http::header::RANGE, + HeaderValue::from_str(&range).map_err(|_| SourceError::Other("invalid range header".to_string()))?, + ); + } + let request = self.request(Method::GET, self.object_url(key)?, headers).await?; + let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + let head = Self::head_from_response(response.headers())?; + let content_range = header(response.headers(), "content-range").map(str::to_string); + Ok(SourceGet { + head, + body: response_body(response), + content_range, + }) + } + + async fn list(&self, request: &SourceListRequest<'_>) -> Result { + // `objects.list` offers `startOffset`, which is inclusive, so it cannot + // express "resume after this key" without silently repeating it. + if request.start_after.is_some() { + return Err(SourceError::Unsupported( + "gcs sources cannot resume a listing from a key; use the continuation token".to_string(), + )); + } + let mut url = self.objects_url()?; + { + let mut query = url.query_pairs_mut(); + if let Some(prefix) = request.prefix.filter(|prefix| !prefix.is_empty()) { + query.append_pair("prefix", prefix); + } + if let Some(delimiter) = request.delimiter.filter(|delimiter| !delimiter.is_empty()) { + query.append_pair("delimiter", delimiter); + } + if let Some(token) = request.continuation_token.filter(|token| !token.is_empty()) { + query.append_pair("pageToken", token); + } + if request.max_keys > 0 { + query.append_pair("maxResults", &request.max_keys.to_string()); + } + } + + let request = self.request(Method::GET, url, HeaderMap::new()).await?; + let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + let body = read_text(response, MAX_JSON_BYTES).await?; + parse_objects_list(&body) + } + + /// GCS has no object tagging API; user metadata is already carried by the + /// head mapping. An empty map keeps `policy.copy_tags` from failing a pull + /// over a concept the provider does not have. + async fn tagging(&self, _key: &str) -> Result, SourceError> { + Ok(HashMap::new()) + } + + /// A one-object listing, not `buckets.get`: the migration pipeline only + /// ever needs `storage.objects.list` and `storage.objects.get`, and a key + /// scoped to exactly those (the `objectViewer` role) cannot read the bucket + /// resource. Probing with `buckets.get` would reject a correct key. + async fn probe(&self) -> Result<(), SourceError> { + let mut url = self.objects_url()?; + url.query_pairs_mut().append_pair("maxResults", "1"); + let request = self.request(Method::GET, url, HeaderMap::new()).await?; + let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?; + read_text(response, MAX_JSON_BYTES) + .await + .and_then(|body| parse_objects_list(&body))?; + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ObjectsList { + #[serde(default)] + items: Vec, + #[serde(default)] + prefixes: Vec, + #[serde(default)] + next_page_token: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct ListedObject { + name: String, + /// GCS renders the size as a decimal string, not a JSON number. + #[serde(default)] + size: Option, + #[serde(default)] + updated: Option, + #[serde(default)] + md5_hash: Option, + #[serde(default)] + etag: Option, + #[serde(default)] + storage_class: Option, +} + +fn parse_objects_list(body: &str) -> Result { + let listing: ObjectsList = + serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?; + let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty()); + let objects = listing + .items + .into_iter() + .map(|item| { + let etag = item + .md5_hash + .as_deref() + .and_then(base64_md5_to_hex) + .or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string())) + .filter(|etag| !etag.is_empty()); + SourceObject { + key: item.name, + etag, + size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0), + last_modified: item.updated.as_deref().and_then(parse_http_timestamp), + storage_class: item.storage_class, + // GCS never encodes a part count in a digest or an ETag. + is_multipart_etag: false, + } + }) + .collect(); + + Ok(SourcePage { + objects, + common_prefixes: listing.prefixes, + is_truncated: next_continuation_token.is_some(), + next_continuation_token, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract}; + use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; + use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder; + + const LIST_PAGE_ONE: &str = r#"{ + "kind": "storage#objects", + "nextPageToken": "cursor-1", + "prefixes": ["dir/sub/"], + "items": [ + { + "name": "dir/a.txt", + "size": "5", + "updated": "2015-10-21T07:28:00.000Z", + "md5Hash": "XUFAKrxLKna5cZ2REBfFkg==", + "etag": "CJizy9Wq0McCEAE=", + "storageClass": "STANDARD" + } + ] + }"#; + + const LIST_PAGE_TWO: &str = r#"{ + "kind": "storage#objects", + "items": [ + { + "name": "dir/b.txt", + "size": "7", + "updated": "2015-10-21T07:28:00.000Z", + "etag": "\"CJizy9Wq0McCEAI=\"" + } + ] + }"#; + + fn backend(endpoint: &Url) -> GcsNativeSourceBackend { + GcsNativeSourceBackend { + http: NativeHttp::for_test(endpoint.clone()), + bucket: "legacy".to_string(), + // Anonymous credentials add no headers, so the fixture sees exactly + // the request this backend builds. + credentials: AnonymousBuilder::new().build(), + } + } + + fn object_headers() -> Vec<(&'static str, String)> { + vec![ + ("Content-Type", "text/plain".to_string()), + ("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()), + ("ETag", "\"CJizy9Wq0McCEAE=\"".to_string()), + ("x-goog-hash", "crc32c=AAAAAA==,md5=XUFAKrxLKna5cZ2REBfFkg==".to_string()), + ("x-goog-meta-owner", "alice".to_string()), + ("x-goog-storage-class", "STANDARD".to_string()), + ("x-goog-generation", "1445412480000000".to_string()), + ] + } + + #[test] + fn objects_list_maps_items_prefixes_and_the_page_token() { + let page = parse_objects_list(LIST_PAGE_ONE).expect("page should parse"); + assert_eq!(page.common_prefixes, vec!["dir/sub/"]); + assert!(page.is_truncated); + assert_eq!(page.next_continuation_token.as_deref(), Some("cursor-1")); + assert_eq!(page.objects.len(), 1); + assert_eq!(page.objects[0].key, "dir/a.txt"); + assert_eq!(page.objects[0].size, 5, "the string size is parsed"); + assert_eq!( + page.objects[0].etag.as_deref(), + Some("5d41402abc4b2a76b9719d911017c592"), + "the base64 md5Hash becomes a hex ETag" + ); + assert_eq!(page.objects[0].storage_class.as_deref(), Some("STANDARD")); + assert!(page.objects[0].last_modified.is_some(), "RFC 3339 `updated` is parsed"); + + let page = parse_objects_list(LIST_PAGE_TWO).expect("page should parse"); + assert!(!page.is_truncated); + assert!(page.next_continuation_token.is_none()); + assert_eq!( + page.objects[0].etag.as_deref(), + Some("CJizy9Wq0McCEAI="), + "without md5Hash the raw etag is carried" + ); + + assert!(parse_objects_list("not json").is_err()); + } + + #[tokio::test] + async fn head_prefers_the_goog_hash_md5_over_the_etag() { + let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, object_headers(), String::new())]).await; + let head = backend(&endpoint).head("dir/a b.txt").await.expect("HEAD should map"); + + let recorded = recorded.lock().expect("recorder lock").clone(); + assert_eq!(recorded[0].method, "HEAD"); + assert_eq!(recorded[0].target, "/legacy/dir/a%20b.txt", "the XML API addresses the object by path"); + assert_eq!( + head.etag.as_deref(), + Some("5d41402abc4b2a76b9719d911017c592"), + "the x-goog-hash md5 is the content digest" + ); + assert!(!head.etag_is_opaque, "a GCS md5 may be checked against the pulled bytes"); + assert_eq!(head.user_metadata, HashMap::from([("owner".to_string(), "alice".to_string())])); + assert_eq!(head.version_id.as_deref(), Some("1445412480000000")); + assert_eq!(head.storage_class.as_deref(), Some("STANDARD")); + } + + #[tokio::test] + async fn a_composite_object_without_an_md5_keeps_an_opaque_etag() { + let headers = object_headers() + .into_iter() + .map(|(name, value)| { + if name == "x-goog-hash" { + (name, "crc32c=AAAAAA==".to_string()) + } else { + (name, value) + } + }) + .collect(); + let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await; + let head = backend(&endpoint).head("composed").await.expect("HEAD should map"); + assert_eq!(head.etag.as_deref(), Some("CJizy9Wq0McCEAE=")); + assert!(head.etag_is_opaque, "a composite ETag describes the composition, not the bytes"); + } + + #[tokio::test] + async fn customer_supplied_key_objects_are_refused() { + let mut headers = object_headers(); + headers.push(("x-goog-encryption-key-sha256", "abc".to_string())); + let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await; + let err = backend(&endpoint) + .head("a.txt") + .await + .expect_err("CSEK objects are unsupported"); + assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}"); + } + + #[tokio::test] + async fn list_and_probe_address_the_json_api() { + let (endpoint, recorded) = scripted_server(vec![ + ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()), + ScriptedResponse::new(200, Vec::new(), "{}".to_string()), + ]) + .await; + let backend = backend(&endpoint); + + backend + .list(&SourceListRequest { + prefix: Some("dir/"), + delimiter: Some("/"), + continuation_token: Some("cursor-0"), + max_keys: 2, + ..Default::default() + }) + .await + .expect("listing should succeed"); + backend.probe().await.expect("probe should succeed"); + + let recorded = recorded.lock().expect("recorder lock").clone(); + assert!(recorded[0].target.starts_with("/storage/v1/b/legacy/o?"), "{}", recorded[0].target); + assert!(recorded[0].target.contains("prefix=dir%2F"), "{}", recorded[0].target); + assert!(recorded[0].target.contains("delimiter=%2F"), "{}", recorded[0].target); + assert!(recorded[0].target.contains("pageToken=cursor-0"), "{}", recorded[0].target); + assert!(recorded[0].target.contains("maxResults=2"), "{}", recorded[0].target); + assert_eq!( + recorded[1].target, "/storage/v1/b/legacy/o?maxResults=1", + "the probe uses the listing permission the pipeline already needs" + ); + } + + #[tokio::test] + async fn gcs_native_backend_satisfies_the_shared_backend_contract() { + let mut ranged = object_headers(); + ranged.push(("Content-Range", "bytes 1-3/5".to_string())); + // A HEAD reports the object size with no body, exactly as GCS does. + let mut head_only = object_headers(); + head_only.push(("Content-Length", "5".to_string())); + let (endpoint, _) = scripted_server(vec![ + ScriptedResponse::new(200, head_only, String::new()), + ScriptedResponse::new(200, object_headers(), "hello".to_string()), + ScriptedResponse::new(206, ranged, "ell".to_string()), + ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()), + ScriptedResponse::new(200, Vec::new(), LIST_PAGE_TWO.to_string()), + // GCS has no tagging call, so the contract's tag step issues no + // request; the probe is the next one on the wire. + ScriptedResponse::new(200, Vec::new(), "{}".to_string()), + ScriptedResponse::new(404, Vec::new(), String::new()), + ScriptedResponse::new(403, Vec::new(), String::new()), + ]) + .await; + + assert_backend_contract( + &backend(&endpoint), + BackendCapabilities { + etag_is_opaque: false, + supports_start_after: false, + // GCS objects have no tags; the contract's tag step is skipped. + supports_tagging: false, + }, + ) + .await; + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/mod.rs b/crates/ecstore/src/bucket/on_demand_migration/mod.rs index 0ca4a29ce..6147f1f94 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/mod.rs @@ -19,25 +19,36 @@ //! client, and the per-node runtime (`sys`) that turns configs into live //! clients guarded by a breaker, a negative cache, singleflight and a pull //! concurrency limit (rustfs/backlog#2147). +//! +//! A source is reached through one `SourceBackend`: the S3 dialect for every +//! S3-compatible provider, and a native backend for the providers that have no +//! S3 API (`azure`, `gcs_native`). +pub mod azure; +#[cfg(test)] +mod backend_contract; pub mod backfill; pub mod breaker; pub mod config; +pub mod gcs; pub mod list_through; +mod native_http; pub mod negative_cache; pub mod pull; pub mod source_client; pub mod stats; pub mod sys; +#[cfg(test)] +mod test_http_fixture; pub use breaker::{ BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker, BreakerState, BreakerTransition, BreakerVerdict, }; pub use config::{ - ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION, - OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig, - SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, + AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, + ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, + RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, }; pub use list_through::{ FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, @@ -57,5 +68,6 @@ pub use stats::{ }; pub use sys::{ ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError, - OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec, + OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec, + source_client_spec, }; 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..881db697d --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs @@ -0,0 +1,415 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared HTTP transport for the on-demand migration source backends that do +//! not speak S3 (Azure Blob, native GCS). +//! +//! The S3 backend rides the AWS SDK; these providers have no SigV4 dialect, so +//! they talk plain HTTP through one `reqwest` client that carries the same +//! connect/read timeouts and TLS policy the operator configured for the source. +//! Redirects are refused: the endpoint passed the outbound policy gate once, and +//! following a source-chosen `Location` would leave that gate behind. +//! +//! Errors never render the request URL. A SAS token lives in the query string, +//! so a `reqwest` error rendered with its URL would print the credential into +//! the log line and the admin response. + +use super::source_client::{SourceError, SourceHead, SourceTimeouts, USER_AGENT_SUFFIX, classify_status, is_multipart_etag}; +use crate::bucket::remote_s3_client::{RemoteS3ClientError, validate_remote_endpoint, validate_target_ca_pem}; +use aws_sdk_s3::primitives::ByteStream; +use aws_smithy_types::body::SdkBody; +use futures::StreamExt; +use http::HeaderMap; +use std::collections::HashMap; +use std::time::SystemTime; +use time::OffsetDateTime; +use time::format_description::well_known::{Rfc2822, Rfc3339}; +use url::Url; + +/// Origin the native backends are allowed to address, plus the HTTP client +/// that reaches it. +pub(super) struct NativeHttp { + client: reqwest::Client, + endpoint: Url, +} + +impl NativeHttp { + /// `endpoint` must be a bare `scheme://host[:port]` origin; it is checked + /// against the outbound policy exactly like an S3 source endpoint. + pub(super) fn new( + endpoint: &str, + timeouts: SourceTimeouts, + skip_tls_verify: bool, + ca_cert_pem: Option<&str>, + ) -> Result { + 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())) +} + +/// Base64 digest (`Content-MD5`, `md5Hash`, `x-goog-hash`) as lowercase hex. +/// `None` when the value is not a 16-byte digest, so a CRC32C never passes as +/// an MD5. +pub(super) fn base64_md5_to_hex(value: &str) -> Option { + let raw = base64_simd::STANDARD.decode_to_vec(value.trim().as_bytes()).ok()?; + (raw.len() == 16).then(|| faster_hex::hex_string(&raw)) +} + +pub(super) fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name).and_then(|value| value.to_str().ok()).map(str::trim) +} + +fn header_string(headers: &HeaderMap, name: &str) -> Option { + 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 base64_md5_converts_only_sixteen_byte_digests() { + assert_eq!( + base64_md5_to_hex("1B2M2Y8AsgTpgAmY7PhCfg==").as_deref(), + Some("d41d8cd98f00b204e9800998ecf8427e") + ); + assert_eq!(base64_md5_to_hex("not base64!").as_deref(), None); + // A CRC32C digest is four bytes: it must not pass as an MD5. + assert_eq!(base64_md5_to_hex("AAAAAA==").as_deref(), None); + } + + #[test] + fn native_http_rejects_endpoints_that_are_not_bare_origins() { + for bad in [ + "ftp://source.example.com", + "https://user:pw@source.example.com", + "https://source.example.com/container", + "https://source.example.com/?x=1", + "not a url", + ] { + assert!( + NativeHttp::new(bad, SourceTimeouts::default(), false, None).is_err(), + "{bad} must be rejected" + ); + } + } + + #[test] + fn native_http_percent_encodes_every_path_segment() { + let http = NativeHttp::for_test(Url::parse("https://acct.blob.core.windows.net").expect("origin")); + let url = http.url(["container", "dir", "a b?c#d.txt"]).expect("url should build"); + assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/container/dir/a%20b%3Fc%23d.txt"); + assert_eq!(url.query(), None, "a key with '?' must not become a query"); + } +} diff --git a/crates/ecstore/src/bucket/on_demand_migration/pull.rs b/crates/ecstore/src/bucket/on_demand_migration/pull.rs index df3dd6462..8d18bdbdd 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/pull.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/pull.rs @@ -1118,6 +1118,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 a97cf5a0a..618fff00e 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,8 @@ //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never //! forwarded: v1 rejects SSE-C source objects outright. +use super::azure::AzureSourceBackend; +use super::gcs::GcsNativeSourceBackend; use super::list_through::{ListPageError, validate_list_page}; use crate::bucket::remote_s3_client::{ PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config, @@ -65,6 +67,10 @@ pub enum SourceProvider { /// Generic S3-compatible service. #[default] S3, + /// Native Azure Blob service; not an S3 dialect. + Azure, + /// Native GCS JSON API with a service-account key; not an S3 dialect. + GcsNative, } impl SourceProvider { @@ -76,6 +82,8 @@ impl SourceProvider { "minio" => Some(Self::Minio), "rustfs" => Some(Self::Rustfs), "s3" => Some(Self::S3), + "azure" => Some(Self::Azure), + "gcs_native" => Some(Self::GcsNative), _ => None, } } @@ -88,6 +96,8 @@ impl SourceProvider { Self::Minio => "minio", Self::Rustfs => "rustfs", Self::S3 => "s3", + Self::Azure => "azure", + Self::GcsNative => "gcs_native", } } @@ -159,6 +169,69 @@ pub struct SourceClientSpec { /// Bytes per second the pull pipeline may consume from this source; /// `None` means unlimited. Enforced by the consumer, not by this client. pub bandwidth_limit: Option, + /// Which [`SourceBackend`] to build. The S3 variant reads `region`, + /// `path_style` and `credentials`; the native variants ignore all three + /// and carry their own credentials. + pub backend: SourceBackendSpec, +} + +/// Provider-specific half of [`SourceClientSpec`]. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub enum SourceBackendSpec { + #[default] + S3, + Azure(AzureSourceSpec), + Gcs(GcsSourceSpec), +} + +/// Native Azure Blob parameters. The container is [`SourceClientSpec::bucket`]. +#[derive(Clone, PartialEq, Eq)] +pub struct AzureSourceSpec { + pub account: String, + pub auth: AzureAuth, +} + +impl fmt::Debug for AzureSourceSpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AzureSourceSpec") + .field("account", &self.account) + .field("auth", &self.auth) + .finish() + } +} + +/// How Azure requests are authorized. +#[derive(Clone, PartialEq, Eq)] +pub enum AzureAuth { + /// Base64 storage-account key, signed per request with Shared Key. + SharedKey(String), + /// SAS query string without the leading `?`, appended to every URL. + Sas(String), +} + +impl fmt::Debug for AzureAuth { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Both variants are secrets; only the scheme may be rendered. + f.write_str(match self { + Self::SharedKey(_) => "SharedKey(REDACTED)", + Self::Sas(_) => "Sas(REDACTED)", + }) + } +} + +/// Native GCS parameters. The bucket is [`SourceClientSpec::bucket`]. +#[derive(Clone, PartialEq, Eq)] +pub struct GcsSourceSpec { + /// Service-account key JSON. + pub service_account_json: String, +} + +impl fmt::Debug for GcsSourceSpec { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("GcsSourceSpec") + .field("service_account_json", &"REDACTED") + .finish() + } } impl SourceClientSpec { @@ -272,7 +345,7 @@ const ACCESS_DENIED_CODES: &[&str] = &[ "InvalidToken", ]; -fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError { +pub(super) fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError { if let Some(code) = code { if THROTTLE_CODES.contains(&code) { return SourceError::Throttled; @@ -344,6 +417,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. @@ -365,7 +443,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()) } @@ -414,6 +492,7 @@ fn source_head(parts: HeadParts) -> Result { storage_class: parts.storage_class, sse, is_multipart_etag, + etag_is_opaque: false, }) } @@ -624,9 +703,48 @@ 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(gcs) => { + let backend = GcsNativeSourceBackend::new( + &spec.endpoint, + &spec.bucket, + gcs, + spec.timeouts, + spec.skip_tls_verify, + spec.ca_cert_pem.as_deref(), + )?; + Ok(Self::from_backend(Box::new(backend), spec)) + } + } + } + + /// Wraps a ready backend in the prefix-mapping client. The endpoint is + /// kept only for `Debug` and admin status. + fn from_backend(backend: Box, 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 @@ -871,6 +989,7 @@ fn s3_source_object(object: SdkObject) -> Result { #[cfg(test)] mod tests { use super::*; + use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract}; use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn}; use aws_smithy_runtime_api::client::orchestrator::HttpRequest; use aws_smithy_runtime_api::client::result::ConnectorError; @@ -988,6 +1107,7 @@ mod tests { retry: RemoteS3RetryPolicy::Disabled, timeouts: SourceTimeouts::default(), bandwidth_limit: NonZeroU64::new(1_000_000), + backend: SourceBackendSpec::S3, } } @@ -1615,7 +1735,101 @@ mod tests { assert_eq!(resolve_path_style(PathStyle::VirtualHost, Minio, "10.0.0.1"), PathStyle::VirtualHost); assert_eq!(resolve_path_style(PathStyle::Path, Aws, "s3.amazonaws.com"), PathStyle::Path); assert_eq!(SourceProvider::from_label(" AWS "), Some(Aws)); - assert_eq!(SourceProvider::from_label("azure"), None); + assert_eq!(SourceProvider::from_label(" Azure "), Some(Azure)); + assert_eq!(SourceProvider::from_label("gcs_native"), Some(GcsNative)); + assert_eq!(SourceProvider::from_label("swift"), None); + } + + const CONTRACT_LIST_PAGE_ONE: &str = r#" + + source-bucket + true + cursor-1 + + dir/a.txt + 2015-10-21T07:28:00.000Z + "5d41402abc4b2a76b9719d911017c592" + 5 + STANDARD + + dir/sub/ +"#; + + const CONTRACT_LIST_PAGE_TWO: &str = r#" + + source-bucket + false + + dir/b.txt + 2015-10-21T07:28:00.000Z + "7d41402abc4b2a76b9719d911017c592" + 7 + +"#; + + const CONTRACT_TAGGING: &str = r#" + + envprod +"#; + + fn contract_object_headers(content_length: u64) -> Vec<(&'static str, String)> { + vec![ + ("etag", format!("\"{OBJECT_MD5}\"")), + ("content-length", content_length.to_string()), + ("content-type", "text/plain".to_string()), + ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()), + ("x-amz-meta-owner", "alice".to_string()), + ("x-amz-storage-class", "STANDARD".to_string()), + ] + } + + /// The S3 backend behind the scripted connector, without the prefix-mapping + /// client on top: the contract is a property of the backend itself. + async fn scripted_s3_backend(responses: Vec) -> S3SourceBackend { + let spec = spec(None); + let connector = SharedHttpConnector::new(ScriptedConnector { + requests: Arc::new(Mutex::new(Vec::new())), + responses: Arc::new(Mutex::new(responses.into_iter().collect())), + }); + let http_client = http_client_fn(move |_settings, _components| connector.clone()); + let endpoint = spec.endpoint_spec().expect("test spec endpoint should parse"); + let config = build_remote_s3_config(&endpoint) + .await + .expect("test spec should build") + .http_client(http_client) + .interceptor(SourceProxyMarkerInterceptor::new()); + S3SourceBackend { + client: S3Client::from_conf(config.build()), + bucket: spec.bucket.clone(), + } + } + + #[tokio::test] + async fn s3_backend_satisfies_the_shared_backend_contract() { + let mut ranged = contract_object_headers(3); + ranged.push(("content-range", "bytes 1-3/5".to_string())); + let backend = scripted_s3_backend(vec![ + ok(contract_object_headers(5), ""), + ok(contract_object_headers(5), "hello"), + ok(ranged, "ell"), + ok(Vec::new(), CONTRACT_LIST_PAGE_ONE), + ok(Vec::new(), CONTRACT_LIST_PAGE_TWO), + ok(Vec::new(), CONTRACT_TAGGING), + ok(Vec::new(), ""), + status(404, ""), + status(403, ACCESS_DENIED_BODY), + ]) + .await; + + assert_backend_contract( + &backend, + BackendCapabilities { + etag_is_opaque: false, + supports_start_after: true, + supports_tagging: true, + }, + ) + .await; } fn prefix_client(prefix: Option) -> 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..6c348749e 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 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), diff --git a/crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs b/crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs new file mode 100644 index 000000000..eb6dda014 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs @@ -0,0 +1,120 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Scripted HTTP server for the native source backends' tests. +//! +//! The S3 backend can be driven through the SDK's own connector; the native +//! backends talk to a real socket, so their tests need a server that answers a +//! fixed script and records what it was asked. Every response closes its +//! connection, which keeps one request on one socket and makes the script order +//! exactly the request order. + +use std::sync::{Arc, Mutex}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use url::Url; + +pub(super) struct ScriptedResponse { + status: u16, + headers: Vec<(&'static str, String)>, + body: String, +} + +impl ScriptedResponse { + pub(super) fn new(status: u16, headers: Vec<(&'static str, String)>, body: String) -> Self { + Self { status, headers, body } + } +} + +#[derive(Clone, Debug)] +pub(super) struct RecordedRequest { + pub(super) method: String, + /// Request target as it appeared on the wire: path plus query. + pub(super) target: String, + pub(super) headers: Vec<(String, String)>, +} + +impl RecordedRequest { + pub(super) fn header(&self, name: &str) -> Option<&str> { + self.headers + .iter() + .find(|(key, _)| key.eq_ignore_ascii_case(name)) + .map(|(_, value)| value.as_str()) + } +} + +pub(super) type Recorder = Arc>>; + +/// Binds a loopback listener that answers `responses` in order and returns its +/// origin plus the recorder. The task ends once the script is exhausted. +pub(super) async fn scripted_server(responses: Vec) -> (Url, Recorder) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind"); + let port = listener.local_addr().expect("fixture address").port(); + let recorder: Recorder = Arc::new(Mutex::new(Vec::new())); + let sink = Arc::clone(&recorder); + + tokio::spawn(async move { + for response in responses { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let mut request = Vec::new(); + let mut buffer = [0_u8; 2048]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + match stream.read(&mut buffer).await { + Ok(0) | Err(_) => break, + Ok(read) => request.extend_from_slice(&buffer[..read]), + } + } + let text = String::from_utf8_lossy(&request).into_owned(); + let mut lines = text.lines(); + let start = lines.next().unwrap_or_default().to_string(); + let mut parts = start.split_whitespace(); + sink.lock().expect("recorder lock").push(RecordedRequest { + method: parts.next().unwrap_or_default().to_string(), + target: parts.next().unwrap_or_default().to_string(), + headers: lines + .take_while(|line| !line.is_empty()) + .filter_map(|line| line.split_once(':')) + .map(|(name, value)| (name.trim().to_string(), value.trim().to_string())) + .collect(), + }); + + // A scripted HEAD declares the object size in its own headers while + // carrying no body, so an explicit `Content-Length` wins over the + // body length. + let declares_length = response + .headers + .iter() + .any(|(name, _)| name.eq_ignore_ascii_case("content-length")); + let mut rendered = match declares_length { + true => format!("HTTP/1.1 {} Scripted\r\nConnection: close\r\n", response.status), + false => format!( + "HTTP/1.1 {} Scripted\r\nContent-Length: {}\r\nConnection: close\r\n", + response.status, + response.body.len() + ), + }; + for (name, value) in response.headers { + rendered.push_str(&format!("{name}: {value}\r\n")); + } + rendered.push_str("\r\n"); + rendered.push_str(&response.body); + let _ = stream.write_all(rendered.as_bytes()).await; + let _ = stream.flush().await; + } + }); + + (Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorder) +} diff --git a/crates/madmin/fixtures/on_demand_migration/get_response.json b/crates/madmin/fixtures/on_demand_migration/get_response.json index aff3a808b..4101403c3 100644 --- a/crates/madmin/fixtures/on_demand_migration/get_response.json +++ b/crates/madmin/fixtures/on_demand_migration/get_response.json @@ -1 +1 @@ -{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"} +{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"} diff --git a/crates/madmin/fixtures/on_demand_migration/set_request.json b/crates/madmin/fixtures/on_demand_migration/set_request.json index e5b7fb03a..a67d359da 100644 --- a/crates/madmin/fixtures/on_demand_migration/set_request.json +++ b/crates/madmin/fixtures/on_demand_migration/set_request.json @@ -1 +1 @@ -{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}} +{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}} diff --git a/crates/madmin/fixtures/on_demand_migration/set_response.json b/crates/madmin/fixtures/on_demand_migration/set_response.json index 81bf69669..2ea089aa3 100644 --- a/crates/madmin/fixtures/on_demand_migration/set_response.json +++ b/crates/madmin/fixtures/on_demand_migration/set_response.json @@ -1 +1 @@ -{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}} +{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}} diff --git a/crates/madmin/src/on_demand_migration.rs b/crates/madmin/src/on_demand_migration.rs index 941ab0233..0fce8af6c 100644 --- a/crates/madmin/src/on_demand_migration.rs +++ b/crates/madmin/src/on_demand_migration.rs @@ -78,10 +78,18 @@ pub struct OnDemandMigrationSource { #[serde(default)] pub path_style: OnDemandMigrationPathStyle, /// `None` means anonymous access to a public source bucket. + /// `None` means anonymous access to a public source bucket. The native + /// providers carry their credentials in `azure` / `gcs` instead. #[serde(default)] pub credentials: Option, #[serde(default)] pub tls: OnDemandMigrationTls, + /// Required for `azure` and rejected for every other provider. + #[serde(default)] + pub azure: Option, + /// Required for `gcs_native` and rejected for every other provider. + #[serde(default)] + pub gcs: Option, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -92,7 +100,49 @@ pub enum OnDemandMigrationProvider { Minio, Rustfs, R2, + /// GCS XML interoperability API with HMAC keys. Gcs, + /// Native Azure Blob service. + Azure, + /// Native GCS JSON API with a service-account key. + #[serde(rename = "gcs_native")] + GcsNative, +} + +/// Native Azure Blob parameters. The container is `source.bucket`; exactly one +/// of `account_key` and `sas_token` is set. Responses carry both as `REDACTED`. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationAzure { + pub account: String, + #[serde(default)] + pub account_key: Option, + #[serde(default)] + pub sas_token: Option, +} + +impl fmt::Debug for OnDemandMigrationAzure { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OnDemandMigrationAzure") + .field("account", &self.account) + .field("account_key", &self.account_key.as_ref().map(|_| "REDACTED")) + .field("sas_token", &self.sas_token.as_ref().map(|_| "REDACTED")) + .finish() + } +} + +/// Native GCS parameters. The bucket is `source.bucket`; the key JSON embeds a +/// private key, so responses carry it as `REDACTED`. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OnDemandMigrationGcs { + pub service_account_json: String, +} + +impl fmt::Debug for OnDemandMigrationGcs { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OnDemandMigrationGcs") + .field("service_account_json", &"REDACTED") + .finish() + } } #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] @@ -806,6 +856,8 @@ mod tests { session_token: None, }), tls: OnDemandMigrationTls::default(), + azure: None, + gcs: None, }); let mut expected: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).expect("fixture"); expected.filter.source_prefix = None; @@ -821,6 +873,42 @@ mod tests { assert!(minimal.source.credentials.is_none()); } + #[test] + fn native_provider_documents_round_trip_and_hide_their_secrets() { + for (label, json) in [ + ( + "azure", + r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"},"gcs":null}"#, + ), + ( + "gcs_native", + r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#, + ), + ] { + let source: OnDemandMigrationSource = serde_json::from_str(json).unwrap_or_else(|err| panic!("{label}: {err}")); + assert_eq!( + serde_json::to_string(&source).expect("re-encodes"), + json, + "{label} must reproduce the server wire shape byte for byte" + ); + } + + let azure = OnDemandMigrationAzure { + account: "legacyaccount".to_string(), + account_key: Some("c2VjcmV0".to_string()), + sas_token: Some("sig=topsecret".to_string()), + }; + let rendered = format!("{azure:?}"); + assert!(rendered.contains("legacyaccount")); + assert!(!rendered.contains("c2VjcmV0"), "{rendered}"); + assert!(!rendered.contains("topsecret"), "{rendered}"); + + let gcs = OnDemandMigrationGcs { + service_account_json: r#"{"private_key":"-----BEGIN PRIVATE KEY-----"}"#.to_string(), + }; + assert!(!format!("{gcs:?}").contains("PRIVATE KEY"), "{gcs:?}"); + } + #[test] fn credentials_debug_never_prints_secrets() { let credentials = OnDemandMigrationCredentials { diff --git a/docs/operations/on-demand-migration.md b/docs/operations/on-demand-migration.md index 8155c6f11..d253e33a2 100644 --- a/docs/operations/on-demand-migration.md +++ b/docs/operations/on-demand-migration.md @@ -103,14 +103,20 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno |---|---|---|---| | `version` | integer | `1` | Must be `1` | | `enabled` | bool | `true` | `false` keeps the config but stops all source traffic | -| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` | — (required) | Drives endpoint and addressing defaults | -| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required for every provider except `aws`, where it is derived from `region` | -| `source.region` | string | — (required) | Non-empty. `auto` is accepted only for `r2`, `minio`, `rustfs` and is signed as `us-east-1` | -| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace | +| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` \| `azure` \| `gcs_native` | — (required) | Drives endpoint and addressing defaults, and which backend the client builds: every value but `azure` and `gcs_native` speaks S3 | +| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required except for `aws` (derived from `region`), `azure` (derived as `https://.blob.core.windows.net`) and `gcs_native` (`https://storage.googleapis.com`). Set it explicitly to point at Azurite or fake-gcs-server, subject to the same outbound policy as any other source endpoint | +| `source.region` | string | — (required) | Non-empty. `auto` is accepted for `r2`, `minio`, `rustfs` and for the native providers, and is signed as `us-east-1`. `azure` and `gcs_native` never sign with a region, so `auto` is the honest value there | +| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace. For `azure` this is the container name, for `gcs_native` the bucket name; the provider block never repeats it | | `source.path_style` | `auto` \| `path` \| `virtual` | `auto` | `auto` resolves to path-style for IP-literal or `localhost` endpoints and for `s3`/`minio`/`rustfs`; virtual-host for `aws`/`gcs`/`r2` | -| `source.credentials` | object \| null | `null` | `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present | +| `source.credentials` | object \| null | `null` | Read only by the S3 providers; `azure` and `gcs_native` must leave it `null` and carry their credentials in their own block. `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present | | `source.tls.skip_verify` | bool | `false` | Disables certificate verification for the source connection | | `source.tls.ca_cert_pem` | string \| null | `null` | Must contain `-----BEGIN CERTIFICATE-----` | +| `source.azure` | object \| null | `null` | Required for `provider = "azure"` and rejected for every other provider | +| `source.azure.account` | string | — (required) | Storage account name; `[A-Za-z0-9-]` only, because it becomes the first label of the derived host | +| `source.azure.account_key` | string \| null | `null` | Base64 storage-account key, signed per request with Shared Key. Mutually exclusive with `sas_token`; exactly one of the two is required | +| `source.azure.sas_token` | string \| null | `null` | SAS query string without the leading `?` and without whitespace, appended to every request URL | +| `source.gcs` | object \| null | `null` | Required for `provider = "gcs_native"` and rejected for every other provider | +| `source.gcs.service_account_json` | string | — (required) | Service-account key JSON; must parse and carry `type: service_account`, `client_email` and `private_key`. Tokens are minted read-only (`devstorage.read_only`) | | `filter.prefix` | string \| null | `null` | Null or non-empty. Only local keys with this prefix consult the source | | `filter.source_prefix` | string \| null | `null` | Null or non-empty. Prepended to the local key to form the source key | | `policy.head` | `proxy` \| `local_only` | `proxy` | `local_only` answers a HEAD miss with 404 and no source traffic | @@ -119,7 +125,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno | `policy.list_through` | bool | `false` | Merges the source listing into `ListObjectsV2` so clients see the whole namespace during the migration. Off by default: it puts the source in the path of every listing | | `policy.respect_local_delete_marker` | bool | `true` | A local delete marker is the final answer; only a versioned bucket can produce one | | `policy.preserve_etag` | bool | `true` | Keeps the source ETag on the stored object unless the bucket encrypts by default | -| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull | +| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull. `azure` reads blob tags instead; `gcs_native` has no tags and always finds none | | `policy.emit_events` | bool | `true` | Whether a write-back emits `ObjectCreated` notifications | | `policy.negative_cache_ttl_secs` | integer | `30` | `0..=3600`; `0` disables the negative cache | | `policy.inline_max_bytes` | integer | `16777216` (16 MiB) | `0..=268435456` (256 MiB). At or below this size a GET miss is teed inline; above it the response streams through and a background pull stores the object | @@ -145,8 +151,14 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket | `rustfs` | Required | Path-style | `auto` allowed | A RustFS source answers the migration request locally thanks to the anti-loop marker | `real_source_test.rs` in the `e2e-nightly` lane | | `r2` | `https://.r2.cloudflarestorage.com` | Virtual-host | `auto` allowed (signed as `us-east-1`) | | `cloud-source (r2)`, only while `ODM_INTEROP_R2_*` are configured; no difference recorded yet | | `gcs` | `https://storage.googleapis.com` | Virtual-host | Real region required | Uses the GCS XML interoperability API with an HMAC key pair, not a service-account JSON key | `cloud-source (gcs)`, only while `ODM_INTEROP_GCS_HMAC_*` are configured; no difference recorded yet | +| `azure` | Optional; derived as `https://.blob.core.windows.net` | Native Blob REST, not S3 | Unused; write `auto` | Needs `source.azure`; the container is `source.bucket`. Reads need `Read` on the blob and `List` on the container, plus `Tags` when `policy.copy_tags` is on | None yet: no interop job covers Azure | +| `gcs_native` | Optional; derived as `https://storage.googleapis.com` | Native GCS API, not S3 | Unused; write `auto` | Needs `source.gcs`. Reads use the XML API for objects and `objects.list` for listings, both with an OAuth token minted from the service-account key; the key needs `storage.objects.get` and `storage.objects.list` | None yet: no interop job covers native GCS | -Azure Blob has no preset; a native provider is deferred (rustfs/backlog#2166). +Every backend answers the same trait contract, pinned by `backend_contract.rs` in `crates/ecstore/src/bucket/on_demand_migration/`, and the three differences that contract allows are the ones documented here. + +`azure` differs in two of them. Its ETag is a concurrency token rather than a digest of the bytes, so it is stored as `odm-source-etag` provenance and never used as the expected MD5 of a pulled object — the write-back integrity check falls back to the local digest. And its listing paginates only with an opaque marker: there is no "start after this key" form, so a caller that asks for one gets `Unsupported` instead of a listing that silently starts over. + +`gcs_native` differs in the other two. Its listing also has no exclusive "start after" form (`startOffset` is inclusive), so it refuses one the same way. And GCS has no object tagging at all: `policy.copy_tags` finds no tags rather than failing the pull, because GCS custom metadata is already carried by the head mapping. Its ETag is normally usable: the `x-goog-hash` MD5 is converted to hex and checked against the pulled bytes, except on a composite object, which has no MD5 and whose ETag is then treated as opaque. The "Interop evidence" column names the job in `.github/workflows/on-demand-migration-interop.yml` (rustfs/backlog#2167) that last exercised the preset against a real implementation, and is where a provider difference belongs once the lane finds one. That lane is report-only and scheduled: it runs `crates/e2e_test/src/on_demand_migration/interop_test.rs` — the same case bodies as the merge-gate suite, with the source injected through `RUSTFS_ODM_INTEROP_*` — against a pinned MinIO container, and against each cloud provider whose repository secrets are configured. A provider without secrets is skipped with a note in the run summary rather than failing, so "no difference recorded yet" means exactly that and not "verified clean"; see [ci-gates.md](../testing/ci-gates.md) for the row. diff --git a/rustfs/src/admin/handlers/on_demand_migration.rs b/rustfs/src/admin/handlers/on_demand_migration.rs index 5fdf3b2f4..103e6ee8d 100644 --- a/rustfs/src/admin/handlers/on_demand_migration.rs +++ b/rustfs/src/admin/handlers/on_demand_migration.rs @@ -46,6 +46,7 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{ }; use crate::admin::storage_api::bucket::on_demand_migration::{ OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext, + source_backend_spec, }; use crate::admin::storage_api::bucket::remote_s3_client::{ PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy, @@ -585,6 +586,8 @@ fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider { Provider::Rustfs => SourceProvider::Rustfs, Provider::R2 => SourceProvider::R2, Provider::Gcs => SourceProvider::Gcs, + Provider::Azure => SourceProvider::Azure, + Provider::GcsNative => SourceProvider::GcsNative, } } @@ -621,6 +624,9 @@ pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClie // a flapping source behind a success and triple the probe's cost. retry: RemoteS3RetryPolicy::Disabled, bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new), + // One mapping serves the probe and the runtime, so an admin probe + // always exercises the backend the runtime will build. + backend: source_backend_spec(source), } } diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 03adac373..ea7ae56b5 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -292,6 +292,7 @@ pub(crate) mod on_demand_migration { pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle; pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider; pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>; + pub(crate) use super::ecstore_bucket::on_demand_migration::source_backend_spec; pub(crate) mod backfill { pub(crate) type BackfillCheckpoint = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillCheckpoint; diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index c1511ebba..65039350c 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -4832,6 +4832,8 @@ mod on_demand_migration_tests { session_token: None, }), tls: TlsConfig::default(), + azure: None, + gcs: None, }, filter: FilterConfig { prefix: None, diff --git a/rustfs/src/app/object/head.rs b/rustfs/src/app/object/head.rs index 73fd09b39..6d877af8a 100644 --- a/rustfs/src/app/object/head.rs +++ b/rustfs/src/app/object/head.rs @@ -665,6 +665,8 @@ mod tests { session_token: None, }), tls: TlsConfig::default(), + azure: None, + gcs: None, }, filter: FilterConfig { prefix: None, @@ -743,6 +745,7 @@ mod tests { }, ), is_multipart_etag: true, + etag_is_opaque: false, } } diff --git a/rustfs/src/app/object/on_demand_migration_put.rs b/rustfs/src/app/object/on_demand_migration_put.rs index 685e60b38..7edea71f8 100644 --- a/rustfs/src/app/object/on_demand_migration_put.rs +++ b/rustfs/src/app/object/on_demand_migration_put.rs @@ -124,6 +124,12 @@ pub(super) fn expected_md5_hex(head: &SourceHead) -> Option { if head.sse.is_some() { return None; } + // Azure stamps an opaque concurrency token in the ETag slot. It is + // recorded as provenance, but reading it as a digest would compare the + // pulled bytes against a value that never described them. + if head.etag_is_opaque { + return None; + } let etag = head.etag.as_deref()?; if etag.len() != 32 || is_multipart_etag(etag) || !etag.bytes().all(|byte| byte.is_ascii_hexdigit()) { return None; @@ -1043,6 +1049,12 @@ mod tests { head.sse = None; head.etag = None; assert_eq!(expected_md5_hex(&head), None); + + // An Azure ETag can be any string the service chooses; even one that + // happens to look like an MD5 must not be checked against the bytes. + let mut head = source_head(b"abc"); + head.etag_is_opaque = true; + assert_eq!(expected_md5_hex(&head), None, "opaque provider ETag"); } #[test]