From b95f4a328b14354321fdbfd1a9addbd1f525c504 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 5 Sep 2026 16:01:34 +0800 Subject: [PATCH] feat(ecstore): add a native gcs odm source backend and one backend contract --- crates/ecstore/src/api/mod.rs | 15 +- .../src/bucket/on_demand_migration/azure.rs | 176 ++++--- .../on_demand_migration/backend_contract.rs | 172 ++++++ .../src/bucket/on_demand_migration/config.rs | 192 +++++++ .../src/bucket/on_demand_migration/gcs.rs | 494 ++++++++++++++++++ .../src/bucket/on_demand_migration/mod.rs | 16 +- .../bucket/on_demand_migration/native_http.rs | 19 + .../on_demand_migration/source_client.rs | 106 +++- .../src/bucket/on_demand_migration/sys.rs | 2 +- .../on_demand_migration/test_http_fixture.rs | 120 +++++ .../on_demand_migration/get_response.json | 2 +- .../on_demand_migration/set_request.json | 2 +- .../on_demand_migration/set_response.json | 2 +- crates/madmin/src/on_demand_migration.rs | 88 ++++ docs/operations/on-demand-migration.md | 24 +- .../src/admin/handlers/on_demand_migration.rs | 6 + rustfs/src/admin/storage_api.rs | 1 + rustfs/src/app/object/get.rs | 2 + rustfs/src/app/object/head.rs | 3 + .../src/app/object/on_demand_migration_put.rs | 12 + 20 files changed, 1352 insertions(+), 102 deletions(-) create mode 100644 crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs create mode 100644 crates/ecstore/src/bucket/on_demand_migration/gcs.rs create mode 100644 crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 34e784288..c2c5b1574 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, @@ -184,9 +185,9 @@ pub mod bucket { } pub mod source_client { pub use crate::bucket::on_demand_migration::source_client::{ - SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, - SourceProbe, SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, - resolve_path_style, + AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError, + SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse, + SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style, }; } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/azure.rs b/crates/ecstore/src/bucket/on_demand_migration/azure.rs index b5fab6fa4..f2a6b7f60 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/azure.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/azure.rs @@ -543,10 +543,9 @@ fn leaf_text(reader: &mut Reader<&[u8]>, end: quick_xml::name::QName<'_>) -> Res #[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 std::sync::{Arc, Mutex}; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; + use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server}; const LIST_PAGE: &str = r#" @@ -693,72 +692,6 @@ mod tests { } } - #[derive(Clone, Debug)] - struct Recorded { - method: String, - target: String, - headers: Vec<(String, String)>, - } - - impl Recorded { - fn header(&self, name: &str) -> Option<&str> { - self.headers - .iter() - .find(|(key, _)| key.eq_ignore_ascii_case(name)) - .map(|(_, value)| value.as_str()) - } - } - - /// One canned HTTP/1.1 response per connection. `Connection: close` keeps - /// every request on its own socket so the script order is deterministic. - async fn scripted_server(responses: Vec<(u16, Vec<(&'static str, String)>, String)>) -> (Url, Arc>>) { - let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind"); - let port = listener.local_addr().expect("fixture address").port(); - let recorded: Arc>> = Arc::new(Mutex::new(Vec::new())); - let sink = Arc::clone(&recorded); - - tokio::spawn(async move { - for (status, headers, body) in responses { - let Ok((mut stream, _)) = listener.accept().await else { - return; - }; - let mut request = Vec::new(); - let mut buffer = [0_u8; 2048]; - while !request.windows(4).any(|window| window == b"\r\n\r\n") { - match stream.read(&mut buffer).await { - Ok(0) | Err(_) => break, - Ok(read) => request.extend_from_slice(&buffer[..read]), - } - } - let text = String::from_utf8_lossy(&request).into_owned(); - let mut lines = text.lines(); - let start = lines.next().unwrap_or_default().to_string(); - let mut parts = start.split_whitespace(); - let recorded = Recorded { - method: parts.next().unwrap_or_default().to_string(), - target: parts.next().unwrap_or_default().to_string(), - headers: lines - .take_while(|line| !line.is_empty()) - .filter_map(|line| line.split_once(':')) - .map(|(name, value)| (name.trim().to_string(), value.trim().to_string())) - .collect(), - }; - sink.lock().expect("recorder lock").push(recorded); - - let mut response = format!("HTTP/1.1 {status} X\r\nContent-Length: {}\r\nConnection: close\r\n", body.len()); - for (name, value) in headers { - response.push_str(&format!("{name}: {value}\r\n")); - } - response.push_str("\r\n"); - response.push_str(&body); - let _ = stream.write_all(response.as_bytes()).await; - let _ = stream.flush().await; - } - }); - - (Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorded) - } - fn blob_headers() -> Vec<(&'static str, String)> { vec![ ("ETag", "\"0x8D2F1B0A1B2C3D4\"".to_string()), @@ -774,7 +707,7 @@ mod tests { #[tokio::test] async fn head_signs_the_request_and_maps_azure_metadata() { - let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await; + let (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"); @@ -806,7 +739,7 @@ mod tests { #[tokio::test] async fn sas_credentials_travel_in_the_query_and_never_sign() { - let (endpoint, recorded) = scripted_server(vec![(200, blob_headers(), String::new())]).await; + let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, blob_headers(), String::new())]).await; let backend = backend( &endpoint, Credential::Sas(vec![ @@ -831,7 +764,7 @@ mod tests { async fn get_passes_the_range_through_and_streams_the_body() { let mut headers = blob_headers(); headers.push(("Content-Range", "bytes 10-14/100".to_string())); - let (endpoint, recorded) = scripted_server(vec![(206, headers, "hello".to_string())]).await; + let (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 { @@ -854,7 +787,7 @@ mod tests { async fn customer_key_blobs_are_refused() { let mut headers = blob_headers(); headers.push(("x-ms-encryption-key-sha256", "abc".to_string())); - let (endpoint, _) = scripted_server(vec![(200, headers, String::new())]).await; + let (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"); @@ -866,8 +799,8 @@ mod tests { #[tokio::test] async fn list_requests_the_container_and_pages_with_the_marker() { let (endpoint, recorded) = scripted_server(vec![ - (200, Vec::new(), LIST_PAGE.to_string()), - (200, Vec::new(), LAST_PAGE.to_string()), + 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])); @@ -931,8 +864,11 @@ mod tests { #[tokio::test] async fn tagging_and_probe_address_the_right_resources() { - let (endpoint, recorded) = - scripted_server(vec![(200, Vec::new(), TAGS.to_string()), (200, Vec::new(), String::new())]).await; + let (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"); @@ -958,7 +894,7 @@ mod tests { let headers = code .map(|code| vec![(HEADER_ERROR_CODE, code.to_string())]) .unwrap_or_default(); - let (endpoint, _) = scripted_server(vec![(status, headers, String::new())]).await; + let (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:?}"); @@ -966,6 +902,90 @@ mod tests { } } + 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 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 a7deabf34..0d0d98ade 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/config.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/config.rs @@ -1024,9 +1024,201 @@ mod tests { "{provider}" ); } + // The native providers never sign with a region, so "auto" is the + // honest value to write for them. + for cfg in [azure_cfg(), gcs_native_cfg()] { + assert_eq!(cfg.source.region, "auto"); + cfg.validate(empty_ctx()) + .unwrap_or_else(|err| panic!("{}: {err}", cfg.source.provider)); + } assert_eq!(sample().source.effective_region(), "us-west-1"); } + const SERVICE_ACCOUNT_JSON: &str = r#"{"type":"service_account","project_id":"p","client_email":"a@b.iam.gserviceaccount.com","private_key":"-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----\n"}"#; + + fn azure_cfg() -> OnDemandMigrationConfig { + let mut cfg = sample(); + cfg.source.provider = Provider::Azure; + cfg.source.endpoint = None; + cfg.source.region = "auto".to_string(); + cfg.source.credentials = None; + cfg.source.azure = Some(AzureSourceConfig { + account: "legacyaccount".to_string(), + account_key: Some("c2VjcmV0LWtleQ==".to_string()), + sas_token: None, + }); + cfg + } + + fn gcs_native_cfg() -> OnDemandMigrationConfig { + let mut cfg = sample(); + cfg.source.provider = Provider::GcsNative; + cfg.source.endpoint = None; + cfg.source.region = "auto".to_string(); + cfg.source.credentials = None; + cfg.source.gcs = Some(GcsSourceConfig { + service_account_json: SERVICE_ACCOUNT_JSON.to_string(), + }); + cfg + } + + #[test] + fn native_providers_derive_their_endpoint_and_round_trip_on_the_wire() { + let azure = azure_cfg(); + assert_eq!(azure.source.effective_endpoint(), "https://legacyaccount.blob.core.windows.net"); + let gcs = gcs_native_cfg(); + assert_eq!(gcs.source.effective_endpoint(), "https://storage.googleapis.com"); + + for cfg in [azure_cfg(), gcs_native_cfg()] { + let json = cfg.to_json().expect("config must serialize"); + assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg); + } + // The wire labels are part of the admin contract. + assert!( + String::from_utf8(azure_cfg().to_json().expect("json")) + .expect("utf8") + .contains(r#""provider":"azure""#) + ); + assert!( + String::from_utf8(gcs_native_cfg().to_json().expect("json")) + .expect("utf8") + .contains(r#""provider":"gcs_native""#) + ); + } + + #[test] + fn an_explicit_endpoint_overrides_the_derived_native_one() { + // Azurite and fake-gcs-server are addressed this way. + let mut cfg = azure_cfg(); + cfg.source.endpoint = Some("http://azurite.example.com:10000".to_string()); + cfg.validate(empty_ctx()).expect("an explicit native endpoint is allowed"); + assert_eq!(cfg.source.effective_endpoint(), "http://azurite.example.com:10000"); + + cfg.source.endpoint = Some("http://azurite.example.com:10000/devstoreaccount1".to_string()); + assert!( + matches!(cfg.validate(empty_ctx()), Err(OnDemandMigrationConfigError::InvalidEndpoint(_))), + "a native endpoint is still an origin" + ); + } + + #[test] + fn a_provider_block_belongs_to_exactly_its_own_provider() { + let mut cfg = sample(); + cfg.source.azure = azure_cfg().source.azure; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("azure", Provider::S3)) + ); + + let mut cfg = sample(); + cfg.source.gcs = gcs_native_cfg().source.gcs; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("gcs", Provider::S3)) + ); + + let mut cfg = azure_cfg(); + cfg.source.azure = None; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::MissingProviderBlock("azure", Provider::Azure)) + ); + + let mut cfg = gcs_native_cfg(); + cfg.source.gcs = None; + assert_eq!( + cfg.validate(empty_ctx()), + Err(OnDemandMigrationConfigError::MissingProviderBlock("gcs", Provider::GcsNative)) + ); + } + + #[test] + fn azure_block_rules() { + let with = |account: &str, key: Option<&str>, sas: Option<&str>| { + let mut cfg = azure_cfg(); + cfg.source.azure = Some(AzureSourceConfig { + account: account.to_string(), + account_key: key.map(str::to_string), + sas_token: sas.map(str::to_string), + }); + cfg.validate(empty_ctx()) + }; + + with("legacyaccount", None, Some("sv=2021-08-06&sig=abc%3D")).expect("a SAS token is a complete credential"); + with("legacyaccount", Some("c2VjcmV0LWtleQ=="), None).expect("an account key is a complete credential"); + + for (label, result) in [ + ("empty account", with("", Some("c2VjcmV0LWtleQ=="), None)), + // The account becomes the first label of the derived hostname. + ("account with a dot", with("legacy.account", Some("c2VjcmV0LWtleQ=="), None)), + ("account with a slash", with("legacy/account", Some("c2VjcmV0LWtleQ=="), None)), + ("no credential", with("legacyaccount", None, None)), + ("both credentials", with("legacyaccount", Some("c2VjcmV0LWtleQ=="), Some("sv=1"))), + ("empty key", with("legacyaccount", Some(""), None)), + ("key that is not base64", with("legacyaccount", Some("not base64!"), None)), + ("empty sas", with("legacyaccount", None, Some(""))), + ("sas with a leading question mark", with("legacyaccount", None, Some("?sv=1"))), + ("sas with whitespace", with("legacyaccount", None, Some("sv=1 &sig=a"))), + ] { + assert!( + matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("azure", _))), + "{label}: {result:?}" + ); + } + } + + #[test] + fn gcs_native_block_requires_a_usable_service_account_key() { + let with = |json: &str| { + let mut cfg = gcs_native_cfg(); + cfg.source.gcs = Some(GcsSourceConfig { + service_account_json: json.to_string(), + }); + cfg.validate(empty_ctx()) + }; + + with(SERVICE_ACCOUNT_JSON).expect("a service-account key is accepted"); + for (label, json) in [ + ("empty", ""), + ("not json", "not json"), + ("not an object", "[]"), + ("wrong type", r#"{"type":"authorized_user","client_email":"a@b","private_key":"k"}"#), + ("no private key", r#"{"type":"service_account","client_email":"a@b"}"#), + ("empty client email", r#"{"type":"service_account","client_email":"","private_key":"k"}"#), + ] { + let result = with(json); + assert!( + matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("gcs", _))), + "{label}: {result:?}" + ); + } + } + + #[test] + fn native_secrets_never_survive_redaction_or_debug() { + let mut azure = azure_cfg(); + azure.source.azure.as_mut().expect("block").sas_token = Some("sv=2021-08-06&sig=top-secret".to_string()); + azure.source.azure.as_mut().expect("block").account_key = None; + let gcs = gcs_native_cfg(); + + for rendered in [ + format!("{:?}", azure.redacted()), + format!("{azure:?}"), + String::from_utf8(azure.redacted().to_json().expect("json")).expect("utf8"), + ] { + assert!(!rendered.contains("top-secret"), "{rendered}"); + assert!(rendered.contains("legacyaccount"), "the account name is not a secret: {rendered}"); + } + for rendered in [ + format!("{:?}", gcs.redacted()), + format!("{gcs:?}"), + String::from_utf8(gcs.redacted().to_json().expect("json")).expect("utf8"), + ] { + assert!(!rendered.contains("BEGIN PRIVATE KEY"), "{rendered}"); + assert!(!rendered.contains("gserviceaccount"), "{rendered}"); + } + } + #[test] fn bucket_rules() { let mut cfg = sample(); 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..d52ea5160 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/gcs.rs @@ -0,0 +1,494 @@ +// 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. +//! +//! `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"]) + } + + fn bucket_url(&self) -> Result { + self.http.url(["storage", "v1", "b", self.bucket.as_str()]) + } + + 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()) + } + + async fn probe(&self) -> Result<(), SourceError> { + let request = self.request(Method::GET, self.bucket_url()?, HeaderMap::new()).await?; + self.http.send(request, NO_ERROR_CODE_HEADER).await?; + 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"); + } + + #[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 5f6acf9c2..30aa6e96d 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/mod.rs @@ -22,12 +22,15 @@ //! //! 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`). +//! 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; @@ -35,15 +38,17 @@ pub mod pull; pub mod source_client; pub mod stats; pub mod sys; +#[cfg(test)] +mod test_http_fixture; pub use breaker::{ BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker, BreakerState, BreakerTransition, BreakerVerdict, }; pub use config::{ - ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION, - OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig, - SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, + AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, + ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, + RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext, }; pub use list_through::{ FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, @@ -62,5 +67,6 @@ pub use stats::{ }; pub use sys::{ ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError, - OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec, + OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec, + source_client_spec, }; diff --git a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs index 9f64860ee..881db697d 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/native_http.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/native_http.rs @@ -190,6 +190,14 @@ pub(super) async fn read_text(response: reqwest::Response, max_bytes: usize) -> 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) } @@ -370,6 +378,17 @@ mod tests { } } + #[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 [ 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 bad312b6b..69e7b5695 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs @@ -26,6 +26,7 @@ //! forwarded: v1 rejects SSE-C source objects outright. use super::azure::AzureSourceBackend; +use super::gcs::GcsNativeSourceBackend; use crate::bucket::remote_s3_client::{ PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config, }; @@ -716,8 +717,16 @@ impl SourceClient { )?; Ok(Self::from_backend(Box::new(backend), spec)) } - SourceBackendSpec::Gcs(_) => { - Err(RemoteS3ClientError::Credentials("the native gcs source backend is not implemented yet")) + 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)) } } } @@ -975,6 +984,7 @@ fn s3_source_object(object: SdkObject) -> Option { #[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; @@ -1602,6 +1612,98 @@ mod tests { 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 { SourceClient { backend: Box::new(S3SourceBackend { diff --git a/crates/ecstore/src/bucket/on_demand_migration/sys.rs b/crates/ecstore/src/bucket/on_demand_migration/sys.rs index 2a1bac332..6c348749e 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/sys.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/sys.rs @@ -643,7 +643,7 @@ fn source_provider(provider: Provider) -> SourceProvider { /// falls back to the S3 spec, where the builder reports the missing /// credentials: the config layer already refuses to store that shape, so this /// only covers a config written by an older or hand-edited build. -pub(crate) fn source_backend_spec(source: &SourceConfig) -> SourceBackendSpec { +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(), 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 7a29db72c..b353bba42 100644 --- a/docs/operations/on-demand-migration.md +++ b/docs/operations/on-demand-migration.md @@ -91,14 +91,20 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno |---|---|---|---| | `version` | integer | `1` | Must be `1` | | `enabled` | bool | `true` | `false` keeps the config but stops all source traffic | -| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` | — (required) | Drives endpoint and addressing defaults | -| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required for every provider except `aws`, where it is derived from `region` | -| `source.region` | string | — (required) | Non-empty. `auto` is accepted only for `r2`, `minio`, `rustfs` and is signed as `us-east-1` | -| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace | +| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` \| `azure` \| `gcs_native` | — (required) | Drives endpoint and addressing defaults, and which backend the client builds: every value but `azure` and `gcs_native` speaks S3 | +| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required except for `aws` (derived from `region`), `azure` (derived as `https://.blob.core.windows.net`) and `gcs_native` (`https://storage.googleapis.com`). Set it explicitly to point at Azurite or fake-gcs-server | +| `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.tls.skip_verify` | bool | `false` | Disables certificate verification for the source connection | | `source.tls.ca_cert_pem` | string \| null | `null` | Must contain `-----BEGIN CERTIFICATE-----` | +| `source.azure` | object \| null | `null` | Required for `provider = "azure"` and rejected for every other provider | +| `source.azure.account` | string | — (required) | Storage account name; `[A-Za-z0-9-]` only, because it becomes the first label of the derived host | +| `source.azure.account_key` | string \| null | `null` | Base64 storage-account key, signed per request with Shared Key. Mutually exclusive with `sas_token`; exactly one of the two is required | +| `source.azure.sas_token` | string \| null | `null` | SAS query string without the leading `?` and without whitespace, appended to every request URL | +| `source.gcs` | object \| null | `null` | Required for `provider = "gcs_native"` and rejected for every other provider | +| `source.gcs.service_account_json` | string | — (required) | Service-account key JSON; must parse and carry `type: service_account`, `client_email` and `private_key`. Tokens are minted read-only (`devstorage.read_only`) | | `filter.prefix` | string \| null | `null` | Null or non-empty. Only local keys with this prefix consult the source | | `filter.source_prefix` | string \| null | `null` | Null or non-empty. Prepended to the local key to form the source key | | `policy.head` | `proxy` \| `local_only` | `proxy` | `local_only` answers a HEAD miss with 404 and no source traffic | @@ -107,7 +113,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno | `policy.list_through` | bool | `false` | Merges the source listing into `ListObjectsV2` so clients see the whole namespace during the migration. Off by default: it puts the source in the path of every listing | | `policy.respect_local_delete_marker` | bool | `true` | A local delete marker is the final answer; only a versioned bucket can produce one | | `policy.preserve_etag` | bool | `true` | Keeps the source ETag on the stored object unless the bucket encrypts by default | -| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull | +| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull. `azure` reads blob tags instead; `gcs_native` has no tags and always finds none | | `policy.emit_events` | bool | `true` | Whether a write-back emits `ObjectCreated` notifications | | `policy.negative_cache_ttl_secs` | integer | `30` | `0..=3600`; `0` disables the negative cache | | `policy.inline_max_bytes` | integer | `16777216` (16 MiB) | `0..=268435456` (256 MiB). At or below this size a GET miss is teed inline; above it the response streams through and a background pull stores the object | @@ -133,8 +139,14 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket | `rustfs` | Required | Path-style | `auto` allowed | A RustFS source answers the migration request locally thanks to the anti-loop marker | `real_source_test.rs` in the `e2e-nightly` lane | | `r2` | `https://.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 b9158afda..b777b5767 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -4821,6 +4821,8 @@ mod on_demand_migration_tests { session_token: None, }), tls: TlsConfig::default(), + azure: None, + gcs: None, }, filter: FilterConfig { prefix: None, 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 fc81009a1..281467db5 100644 --- a/rustfs/src/app/object/on_demand_migration_put.rs +++ b/rustfs/src/app/object/on_demand_migration_put.rs @@ -115,6 +115,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; @@ -874,6 +880,12 @@ mod tests { head.sse = None; head.etag = None; assert_eq!(expected_md5_hex(&head), None); + + // An Azure ETag can be any string the service chooses; even one that + // happens to look like an MD5 must not be checked against the bytes. + let mut head = source_head(b"abc"); + head.etag_is_opaque = true; + assert_eq!(expected_md5_hex(&head), None, "opaque provider ETag"); } #[test]