You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/10/13 05:14:07 UTC
(opendal) branch 6560-reqsign-core-v1-s3 created (now d6e5da507)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch 6560-reqsign-core-v1-s3
in repository https://gitbox.apache.org/repos/asf/opendal.git
at d6e5da507 Address assume role support
This branch includes the following new commits:
new 7b9dab80f Save work
new d6e5da507 Address assume role support
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/02: Save work
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch 6560-reqsign-core-v1-s3
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 7b9dab80f528542d8c8cb28002ea7e13cde0557e
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Oct 13 13:03:32 2025 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/Cargo.lock | 111 +++++++++++++----
core/Cargo.toml | 12 +-
core/src/services/s3/backend.rs | 264 +++++++++++++++++-----------------------
core/src/services/s3/core.rs | 160 ++++++++++--------------
core/src/services/s3/writer.rs | 12 +-
5 files changed, 272 insertions(+), 287 deletions(-)
diff --git a/core/Cargo.lock b/core/Cargo.lock
index 7d1d09700..c88d54cb6 100644
--- a/core/Cargo.lock
+++ b/core/Cargo.lock
@@ -4440,7 +4440,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6a793df0d7afeac54f95b471d3af7f0d4fb975699f972341a4b76988d49cdf0c"
dependencies = [
"cfg-if",
- "windows-targets 0.53.0",
+ "windows-targets 0.53.3",
]
[[package]]
@@ -5360,11 +5360,15 @@ dependencies = [
"prometheus 0.14.0",
"prometheus-client",
"prost 0.13.5",
- "quick-xml 0.38.3",
+ "quick-xml",
"rand 0.8.5",
"redb",
"redis",
"reqsign",
+ "reqsign-aws-v4",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "reqsign-http-send-reqwest",
"reqwest",
"rocksdb",
"rustls-native-certs 0.8.1",
@@ -6513,16 +6517,6 @@ dependencies = [
"winapi",
]
-[[package]]
-name = "quick-xml"
-version = "0.37.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb"
-dependencies = [
- "memchr",
- "serde",
-]
-
[[package]]
name = "quick-xml"
version = "0.38.3"
@@ -6941,18 +6935,86 @@ dependencies = [
"log",
"once_cell",
"percent-encoding",
- "quick-xml 0.37.5",
"rand 0.8.5",
"reqwest",
"rsa",
+ "serde",
+ "serde_json",
+ "sha1",
+ "sha2",
+]
+
+[[package]]
+name = "reqsign-aws-v4"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c50993dfb45a89b82dba66b2251984baad70e1b3c502db980f077f095615a26e"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "bytes",
+ "form_urlencoded",
+ "http 1.3.1",
+ "log",
+ "percent-encoding",
+ "quick-xml",
+ "reqsign-core",
"rust-ini",
"serde",
"serde_json",
+ "serde_urlencoded",
+ "sha1",
+]
+
+[[package]]
+name = "reqsign-core"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f2f07d63648c81c8dbccc19e8e10ef8d57daafb8174e4c2a75f14f33fe8c5ec"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "base64 0.22.1",
+ "bytes",
+ "form_urlencoded",
+ "hex",
+ "hmac",
+ "http 1.3.1",
+ "jiff",
+ "log",
+ "percent-encoding",
"sha1",
"sha2",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "reqsign-file-read-tokio"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "262eb485bb6e8213b13ef10e86ef8613539fb03daa2123b57d96675f784b15b6"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "reqsign-core",
"tokio",
]
+[[package]]
+name = "reqsign-http-send-reqwest"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ff9bb6507b23175dbda8a91ae1a0ad2317471f6ee117e500d1cf6b9ed1eeb0b"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "bytes",
+ "http 1.3.1",
+ "http-body-util",
+ "reqsign-core",
+ "reqwest",
+]
+
[[package]]
name = "reqwest"
version = "0.12.23"
@@ -9702,7 +9764,7 @@ dependencies = [
"windows-collections",
"windows-core 0.61.2",
"windows-future",
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
"windows-numerics",
]
@@ -9735,7 +9797,7 @@ checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
dependencies = [
"windows-implement 0.60.0",
"windows-interface 0.59.1",
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
"windows-result 0.3.4",
"windows-strings",
]
@@ -9747,7 +9809,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
dependencies = [
"windows-core 0.61.2",
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
"windows-threading",
]
@@ -9797,9 +9859,9 @@ dependencies = [
[[package]]
name = "windows-link"
-version = "0.1.1"
+version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76840935b766e1b0a05c0066835fb9ec80071d4c09a16f6bd5f7e655e3c14c38"
+checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
[[package]]
name = "windows-link"
@@ -9814,7 +9876,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1"
dependencies = [
"windows-core 0.61.2",
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
]
[[package]]
@@ -9832,7 +9894,7 @@ version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
dependencies = [
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
]
[[package]]
@@ -9841,7 +9903,7 @@ version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
dependencies = [
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
]
[[package]]
@@ -9913,10 +9975,11 @@ dependencies = [
[[package]]
name = "windows-targets"
-version = "0.53.0"
+version = "0.53.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1e4c7e8ceaaf9cb7d7507c974735728ab453b67ef8f18febdd7c11fe59dca8b"
+checksum = "d5fe6031c4041849d7c496a8ded650796e7b6ecc19df1a431c1a363342e5dc91"
dependencies = [
+ "windows-link 0.1.3",
"windows_aarch64_gnullvm 0.53.0",
"windows_aarch64_msvc 0.53.0",
"windows_i686_gnu 0.53.0",
@@ -9933,7 +9996,7 @@ version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6"
dependencies = [
- "windows-link 0.1.1",
+ "windows-link 0.1.3",
]
[[package]]
diff --git a/core/Cargo.toml b/core/Cargo.toml
index cede6ede5..8f4a4cafa 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -199,9 +199,10 @@ services-redis = ["dep:redis", "dep:bb8", "redis?/tokio-rustls-comp"]
services-redis-native-tls = ["services-redis", "redis?/tokio-native-tls-comp"]
services-rocksdb = ["dep:rocksdb", "internal-tokio-rt"]
services-s3 = [
- "dep:reqsign",
- "reqsign?/services-aws",
- "reqsign?/reqwest_request",
+ "dep:reqsign-core",
+ "dep:reqsign-aws-v4",
+ "dep:reqsign-file-read-tokio",
+ "dep:reqsign-http-send-reqwest",
"dep:crc32c",
]
services-seafile = []
@@ -279,6 +280,11 @@ sqlx = { version = "0.8.0", features = [
# For http based services.
reqsign = { version = "0.16.5", default-features = false, optional = true }
+# For S3 service migration to v1
+reqsign-core = { version = "2.0", default-features = false, optional = true }
+reqsign-aws-v4 = { version = "2.0", default-features = false, optional = true }
+reqsign-file-read-tokio = { version = "2.0", default-features = false, optional = true }
+reqsign-http-send-reqwest = { version = "2.0", default-features = false, optional = true }
# for self-referencing structs
ouroboros = { version = "0.18.4", optional = true }
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 4570ce6b0..d0cb7a695 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -22,7 +22,6 @@ use std::fmt::Write;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::LazyLock;
-use std::sync::atomic::AtomicBool;
use base64::Engine;
use base64::prelude::BASE64_STANDARD;
@@ -36,11 +35,16 @@ use log::warn;
use md5::Digest;
use md5::Md5;
use percent_encoding::percent_decode_str;
-use reqsign::AwsAssumeRoleLoader;
-use reqsign::AwsConfig;
-use reqsign::AwsCredentialLoad;
-use reqsign::AwsDefaultLoader;
-use reqsign::AwsV4Signer;
+use reqsign_aws_v4::Credential;
+use reqsign_aws_v4::DefaultCredentialProvider;
+use reqsign_aws_v4::RequestSigner as AwsV4Signer;
+use reqsign_aws_v4::StaticCredentialProvider;
+use reqsign_core::Context;
+use reqsign_core::ProvideCredential;
+use reqsign_core::ProvideCredentialChain;
+use reqsign_core::Signer;
+use reqsign_file_read_tokio::TokioFileRead;
+use reqsign_http_send_reqwest::ReqwestHttpSend;
use reqwest::Url;
use super::S3_SCHEME;
@@ -102,9 +106,9 @@ impl Configurator for S3Config {
fn into_builder(self) -> Self::Builder {
S3Builder {
config: self,
- customized_credential_load: None,
http_client: None,
+ credential_providers: None,
}
}
}
@@ -117,10 +121,9 @@ impl Configurator for S3Config {
pub struct S3Builder {
config: S3Config,
- customized_credential_load: Option<Box<dyn AwsCredentialLoad>>,
-
#[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
http_client: Option<HttpClient>,
+ credential_providers: Option<ProvideCredentialChain<Credential>>,
}
impl Debug for S3Builder {
@@ -492,15 +495,6 @@ impl S3Builder {
self
}
- /// Adding a customized credential load for service.
- ///
- /// If customized_credential_load has been set, we will ignore all other
- /// credential load methods.
- pub fn customized_credential_load(mut self, cred: Box<dyn AwsCredentialLoad>) -> Self {
- self.customized_credential_load = Some(cred);
- self
- }
-
/// Specify the http client that used by this service.
///
/// # Notes
@@ -521,31 +515,37 @@ impl S3Builder {
self
}
- /// Check if `bucket` is valid
+ /// Replace the credential providers with a custom chain.
+ pub fn credential_provider_chain(mut self, chain: ProvideCredentialChain<Credential>) -> Self {
+ self.credential_providers = Some(chain);
+ self
+ }
+
+ /// Check if `bucket` is valid.
/// `bucket` must be not empty and if `enable_virtual_host_style` is true
- /// it couldn't contain dot(.) character
- fn is_bucket_valid(&self) -> bool {
- if self.config.bucket.is_empty() {
+ /// it could not contain dot (.) character.
+ fn is_bucket_valid(config: &S3Config) -> bool {
+ if config.bucket.is_empty() {
return false;
}
// If enable virtual host style, `bucket` will reside in domain part,
// for example `https://bucket_name.s3.us-east-1.amazonaws.com`,
// so `bucket` with dot can't be recognized correctly for this format.
- if self.config.enable_virtual_host_style && self.config.bucket.contains('.') {
+ if config.enable_virtual_host_style && config.bucket.contains('.') {
return false;
}
true
}
/// Build endpoint with given region.
- fn build_endpoint(&self, region: &str) -> String {
+ fn build_endpoint(config: &S3Config, region: &str) -> String {
let bucket = {
- debug_assert!(self.is_bucket_valid(), "bucket must be valid");
+ debug_assert!(Self::is_bucket_valid(config), "bucket must be valid");
- self.config.bucket.as_str()
+ config.bucket.as_str()
};
- let mut endpoint = match &self.config.endpoint {
+ let mut endpoint = match &config.endpoint {
Some(endpoint) => {
if endpoint.starts_with("http") {
endpoint.to_string()
@@ -576,7 +576,7 @@ impl S3Builder {
};
// Apply virtual host style.
- if self.config.enable_virtual_host_style {
+ if config.enable_virtual_host_style {
endpoint = endpoint.replace("//", &format!("//{bucket}."))
} else {
write!(endpoint, "/{bucket}").expect("write into string must succeed");
@@ -745,15 +745,22 @@ impl S3Builder {
impl Builder for S3Builder {
type Config = S3Config;
- fn build(mut self) -> Result<impl Access> {
+ fn build(self) -> Result<impl Access> {
debug!("backend build started: {:?}", &self);
- let root = normalize_root(&self.config.root.clone().unwrap_or_default());
+ #[allow(deprecated)]
+ let S3Builder {
+ config,
+ http_client,
+ credential_providers,
+ } = self;
+
+ let root = normalize_root(&config.root.clone().unwrap_or_default());
debug!("backend use root {}", &root);
// Handle bucket name.
- let bucket = if self.is_bucket_valid() {
- Ok(&self.config.bucket)
+ let bucket = if Self::is_bucket_valid(&config) {
+ Ok(&config.bucket)
} else {
Err(
Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
@@ -762,14 +769,14 @@ impl Builder for S3Builder {
}?;
debug!("backend use bucket {}", &bucket);
- let default_storage_class = match &self.config.default_storage_class {
+ let default_storage_class = match &config.default_storage_class {
None => None,
Some(v) => Some(
build_header_value(v).map_err(|err| err.with_context("key", "storage_class"))?,
),
};
- let server_side_encryption = match &self.config.server_side_encryption {
+ let server_side_encryption = match &config.server_side_encryption {
None => None,
Some(v) => Some(
build_header_value(v)
@@ -778,7 +785,7 @@ impl Builder for S3Builder {
};
let server_side_encryption_aws_kms_key_id =
- match &self.config.server_side_encryption_aws_kms_key_id {
+ match &config.server_side_encryption_aws_kms_key_id {
None => None,
Some(v) => Some(build_header_value(v).map_err(|err| {
err.with_context("key", "server_side_encryption_aws_kms_key_id")
@@ -786,7 +793,7 @@ impl Builder for S3Builder {
};
let server_side_encryption_customer_algorithm =
- match &self.config.server_side_encryption_customer_algorithm {
+ match &config.server_side_encryption_customer_algorithm {
None => None,
Some(v) => Some(build_header_value(v).map_err(|err| {
err.with_context("key", "server_side_encryption_customer_algorithm")
@@ -794,7 +801,7 @@ impl Builder for S3Builder {
};
let server_side_encryption_customer_key =
- match &self.config.server_side_encryption_customer_key {
+ match &config.server_side_encryption_customer_key {
None => None,
Some(v) => Some(build_header_value(v).map_err(|err| {
err.with_context("key", "server_side_encryption_customer_key")
@@ -802,14 +809,14 @@ impl Builder for S3Builder {
};
let server_side_encryption_customer_key_md5 =
- match &self.config.server_side_encryption_customer_key_md5 {
+ match &config.server_side_encryption_customer_key_md5 {
None => None,
Some(v) => Some(build_header_value(v).map_err(|err| {
err.with_context("key", "server_side_encryption_customer_key_md5")
})?),
};
- let checksum_algorithm = match self.config.checksum_algorithm.as_deref() {
+ let checksum_algorithm = match config.checksum_algorithm.as_deref() {
Some("crc32c") => Some(ChecksumAlgorithm::Crc32c),
None => None,
v => {
@@ -820,109 +827,65 @@ impl Builder for S3Builder {
}
};
- // This is our current config.
- let mut cfg = AwsConfig::default();
- if !self.config.disable_config_load {
- #[cfg(not(target_arch = "wasm32"))]
- {
- cfg = cfg.from_profile();
- cfg = cfg.from_env();
- }
- }
-
- if let Some(ref v) = self.config.region {
- cfg.region = Some(v.to_string());
- }
-
- if cfg.region.is_none() {
- return Err(Error::new(
- ErrorKind::ConfigInvalid,
- "region is missing. Please find it by S3::detect_region() or set them in env.",
- )
- .with_operation("Builder::build")
- .with_context("service", Scheme::S3));
- }
-
- let region = cfg.region.to_owned().unwrap();
+ // Determine the region
+ let region = if let Some(ref v) = config.region {
+ v.to_string()
+ } else {
+ // Try to get region from environment
+ std::env::var("AWS_REGION")
+ .or_else(|_| std::env::var("AWS_DEFAULT_REGION"))
+ .map_err(|_| {
+ Error::new(
+ ErrorKind::ConfigInvalid,
+ "region is missing. Please find it by S3::detect_region() or set them in env.",
+ )
+ .with_operation("Builder::build")
+ .with_context("service", Scheme::S3)
+ })?
+ };
debug!("backend use region: {region}");
- // Retain the user's endpoint if it exists; otherwise, try loading it from the environment.
- self.config.endpoint = self.config.endpoint.or_else(|| cfg.endpoint_url.clone());
-
// Building endpoint.
- let endpoint = self.build_endpoint(®ion);
+ let endpoint = Self::build_endpoint(&config, ®ion);
debug!("backend use endpoint: {endpoint}");
- // Setting all value from user input if available.
- if let Some(v) = self.config.access_key_id {
- cfg.access_key_id = Some(v)
- }
- if let Some(v) = self.config.secret_access_key {
- cfg.secret_access_key = Some(v)
- }
- if let Some(v) = self.config.session_token {
- cfg.session_token = Some(v)
- }
+ // Create the context for reqsign-core
+ let ctx = Context::new()
+ .with_file_read(TokioFileRead)
+ .with_http_send(ReqwestHttpSend::new(GLOBAL_REQWEST_CLIENT.clone()));
- let mut loader: Option<Box<dyn AwsCredentialLoad>> = None;
- // If customized_credential_load is set, we will use it.
- if let Some(v) = self.customized_credential_load {
- loader = Some(v);
- }
+ let mut provider = if let Some(chain) = credential_providers {
+ chain
+ } else {
+ let mut builder = DefaultCredentialProvider::builder();
- // If role_arn is set, we must use AssumeRoleLoad.
- if let Some(role_arn) = self.config.role_arn {
- // use current env as source credential loader.
- let default_loader =
- AwsDefaultLoader::new(GLOBAL_REQWEST_CLIENT.clone().clone(), cfg.clone());
-
- // Build the config for assume role.
- let mut assume_role_cfg = AwsConfig {
- region: Some(region.clone()),
- role_arn: Some(role_arn),
- external_id: self.config.external_id.clone(),
- sts_regional_endpoints: "regional".to_string(),
- ..Default::default()
- };
+ if config.disable_config_load {
+ builder = builder.disable_env(true).disable_profile(true);
+ }
- // override default role_session_name if set
- if let Some(name) = self.config.role_session_name {
- assume_role_cfg.role_session_name = name;
+ if config.disable_ec2_metadata {
+ builder = builder.disable_imds(true);
}
- let assume_role_loader = AwsAssumeRoleLoader::new(
- GLOBAL_REQWEST_CLIENT.clone().clone(),
- assume_role_cfg,
- Box::new(default_loader),
- )
- .map_err(|err| {
- Error::new(
- ErrorKind::ConfigInvalid,
- "The assume_role_loader is misconfigured",
- )
- .with_context("service", Scheme::S3)
- .set_source(err)
- })?;
- loader = Some(Box::new(assume_role_loader));
+ ProvideCredentialChain::new().push(builder.build())
+ };
+
+ if let (Some(ak), Some(sk)) = (&config.access_key_id, &config.secret_access_key) {
+ let static_provider = if let Some(token) = config.session_token.as_deref() {
+ StaticCredentialProvider::new(ak, sk).with_session_token(token)
+ } else {
+ StaticCredentialProvider::new(ak, sk)
+ };
+ provider = provider.push_front(static_provider);
}
- // If loader is not set, we will use default loader.
- let loader = match loader {
- Some(v) => v,
- None => {
- let mut default_loader =
- AwsDefaultLoader::new(GLOBAL_REQWEST_CLIENT.clone().clone(), cfg);
- if self.config.disable_ec2_metadata {
- default_loader = default_loader.with_disable_ec2_metadata();
- }
- Box::new(default_loader)
- }
- };
+ // Create request signer for S3
+ let request_signer = AwsV4Signer::new("s3", ®ion);
- let signer = AwsV4Signer::new("s3", ®ion);
+ // Create the signer
+ let signer = Signer::new(ctx, provider, request_signer);
- let delete_max_size = self
- .config
+ let delete_max_size = config
.delete_max_size
.unwrap_or(DEFAULT_BATCH_MAX_OPERATIONS);
@@ -939,16 +902,11 @@ impl Builder for S3Builder {
stat_with_if_none_match: true,
stat_with_if_modified_since: true,
stat_with_if_unmodified_since: true,
- stat_with_override_cache_control: !self
- .config
- .disable_stat_with_override,
- stat_with_override_content_disposition: !self
- .config
- .disable_stat_with_override,
- stat_with_override_content_type: !self
- .config
+ stat_with_override_cache_control: !config.disable_stat_with_override,
+ stat_with_override_content_disposition: !config
.disable_stat_with_override,
- stat_with_version: self.config.enable_versioning,
+ stat_with_override_content_type: !config.disable_stat_with_override,
+ stat_with_version: config.enable_versioning,
read: true,
read_with_if_match: true,
@@ -958,17 +916,17 @@ impl Builder for S3Builder {
read_with_override_cache_control: true,
read_with_override_content_disposition: true,
read_with_override_content_type: true,
- read_with_version: self.config.enable_versioning,
+ read_with_version: config.enable_versioning,
write: true,
write_can_empty: true,
write_can_multi: true,
- write_can_append: self.config.enable_write_with_append,
+ write_can_append: config.enable_write_with_append,
write_with_cache_control: true,
write_with_content_type: true,
write_with_content_encoding: true,
- write_with_if_match: !self.config.disable_write_with_if_match,
+ write_with_if_match: !config.disable_write_with_if_match,
write_with_if_not_exists: true,
write_with_user_metadata: true,
@@ -987,7 +945,7 @@ impl Builder for S3Builder {
delete: true,
delete_max_size: Some(delete_max_size),
- delete_with_version: self.config.enable_versioning,
+ delete_with_version: config.enable_versioning,
copy: true,
@@ -995,8 +953,8 @@ impl Builder for S3Builder {
list_with_limit: true,
list_with_start_after: true,
list_with_recursive: true,
- list_with_versions: self.config.enable_versioning,
- list_with_deleted: self.config.enable_versioning,
+ list_with_versions: config.enable_versioning,
+ list_with_deleted: config.enable_versioning,
presign: true,
presign_stat: true,
@@ -1010,7 +968,7 @@ impl Builder for S3Builder {
// allow deprecated api here for compatibility
#[allow(deprecated)]
- if let Some(client) = self.http_client {
+ if let Some(client) = http_client {
am.update_http_client(|_| client);
}
@@ -1025,12 +983,10 @@ impl Builder for S3Builder {
server_side_encryption_customer_key,
server_side_encryption_customer_key_md5,
default_storage_class,
- allow_anonymous: self.config.allow_anonymous,
- disable_list_objects_v2: self.config.disable_list_objects_v2,
- enable_request_payer: self.config.enable_request_payer,
+ allow_anonymous: config.allow_anonymous,
+ disable_list_objects_v2: config.disable_list_objects_v2,
+ enable_request_payer: config.enable_request_payer,
signer,
- loader,
- credential_loaded: AtomicBool::new(false),
checksum_algorithm,
}),
})
@@ -1170,9 +1126,9 @@ impl Access for S3Backend {
"operation is not supported",
)),
};
- let mut req = req?;
+ let req = req?;
- self.core.sign_query(&mut req, expire).await?;
+ let req = self.core.sign_query(req, expire).await?;
// We don't need this request anymore, consume it directly.
let (parts, _) = req.into_parts();
@@ -1206,7 +1162,7 @@ mod tests {
if enable_virtual_host_style {
b = b.enable_virtual_host_style();
}
- assert_eq!(b.is_bucket_valid(), expected)
+ assert_eq!(S3Builder::is_bucket_valid(&b.config), expected)
}
}
@@ -1227,7 +1183,7 @@ mod tests {
b = b.endpoint(endpoint);
}
- let endpoint = b.build_endpoint("us-east-2");
+ let endpoint = S3Builder::build_endpoint(&b.config, "us-east-2");
assert_eq!(endpoint, "https://s3.us-east-2.amazonaws.com/test");
}
@@ -1239,7 +1195,7 @@ mod tests {
b = b.endpoint(endpoint);
}
- let endpoint = b.build_endpoint("us-east-2");
+ let endpoint = S3Builder::build_endpoint(&b.config, "us-east-2");
assert_eq!(endpoint, "https://test.s3.us-east-2.amazonaws.com");
}
}
diff --git a/core/src/services/s3/core.rs b/core/src/services/s3/core.rs
index 22e7d1f57..d063ce921 100644
--- a/core/src/services/s3/core.rs
+++ b/core/src/services/s3/core.rs
@@ -21,8 +21,6 @@ use std::fmt::Display;
use std::fmt::Formatter;
use std::fmt::Write;
use std::sync::Arc;
-use std::sync::atomic;
-use std::sync::atomic::AtomicBool;
use std::time::Duration;
use base64::Engine;
@@ -43,9 +41,8 @@ use http::header::IF_MATCH;
use http::header::IF_MODIFIED_SINCE;
use http::header::IF_NONE_MATCH;
use http::header::IF_UNMODIFIED_SINCE;
-use reqsign::AwsCredential;
-use reqsign::AwsCredentialLoad;
-use reqsign::AwsV4Signer;
+use reqsign_aws_v4::Credential;
+use reqsign_core::Signer;
use serde::Deserialize;
use serde::Serialize;
@@ -104,9 +101,7 @@ pub struct S3Core {
pub disable_list_objects_v2: bool,
pub enable_request_payer: bool,
- pub signer: AwsV4Signer,
- pub loader: Box<dyn AwsCredentialLoad>,
- pub credential_loaded: AtomicBool,
+ pub signer: Signer<Credential>,
pub checksum_algorithm: Option<ChecksumAlgorithm>,
}
@@ -121,52 +116,43 @@ impl Debug for S3Core {
}
impl S3Core {
- /// If credential is not found, we will not sign the request.
- async fn load_credential(&self) -> Result<Option<AwsCredential>> {
- let cred = self
- .loader
- .load_credential(GLOBAL_REQWEST_CLIENT.clone())
+ pub async fn sign_query<T>(&self, req: Request<T>, duration: Duration) -> Result<Request<T>> {
+ // Skip signing for anonymous access
+ if self.allow_anonymous {
+ return Ok(req);
+ }
+
+ // Sign the request with presigned URL
+ let (mut parts, body) = req.into_parts();
+
+ self.signer
+ .sign(&mut parts, Some(duration))
.await
- .map_err(new_request_credential_error)?;
+ .map_err(|e| new_request_sign_error(e.into()))?;
- if let Some(cred) = cred {
- // Update credential_loaded to true if we have load credential successfully.
- self.credential_loaded
- .store(true, atomic::Ordering::Relaxed);
- return Ok(Some(cred));
- }
+ // Always remove host header, let users' client to set it based on HTTP
+ // version.
+ //
+ // As discussed in <https://github.com/seanmonstar/reqwest/issues/1809>,
+ // google server could send RST_STREAM of PROTOCOL_ERROR if our request
+ // contains host header.
+ parts.headers.remove(HOST);
- // If we have load credential before but failed to load this time, we should
- // return error instead.
- if self.credential_loaded.load(atomic::Ordering::Relaxed) {
- return Err(Error::new(
- ErrorKind::PermissionDenied,
- "credential was previously loaded successfully but has failed this time",
- )
- .set_temporary());
- }
+ Ok(Request::from_parts(parts, body))
+ }
- // Credential is empty and users allow anonymous access, we will not sign the request.
+ pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
+ // Skip signing for anonymous access
if self.allow_anonymous {
- return Ok(None);
+ return self.info.http_client().send(req).await;
}
- Err(Error::new(
- ErrorKind::PermissionDenied,
- "no valid credential found and anonymous access is not allowed",
- ))
- }
-
- pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
- let cred = if let Some(cred) = self.load_credential().await? {
- cred
- } else {
- return Ok(());
- };
+ let (mut parts, body) = req.into_parts();
self.signer
- .sign(req, &cred)
- .map_err(new_request_sign_error)?;
+ .sign(&mut parts, None)
+ .await
+ .map_err(|e| new_request_sign_error(e.into()))?;
// Always remove host header, let users' client to set it based on HTTP
// version.
@@ -174,21 +160,26 @@ impl S3Core {
// As discussed in <https://github.com/seanmonstar/reqwest/issues/1809>,
// google server could send RST_STREAM of PROTOCOL_ERROR if our request
// contains host header.
- req.headers_mut().remove(HOST);
+ parts.headers.remove(HOST);
- Ok(())
+ self.info
+ .http_client()
+ .send(Request::from_parts(parts, body))
+ .await
}
- pub async fn sign_query<T>(&self, req: &mut Request<T>, duration: Duration) -> Result<()> {
- let cred = if let Some(cred) = self.load_credential().await? {
- cred
- } else {
- return Ok(());
- };
+ pub async fn fetch(&self, req: Request<Buffer>) -> Result<Response<HttpBody>> {
+ // Skip signing for anonymous access
+ if self.allow_anonymous {
+ return self.info.http_client().fetch(req).await;
+ }
+
+ let (mut parts, body) = req.into_parts();
self.signer
- .sign_query(req, duration, &cred)
- .map_err(new_request_sign_error)?;
+ .sign(&mut parts, None)
+ .await
+ .map_err(|e| new_request_sign_error(e.into()))?;
// Always remove host header, let users' client to set it based on HTTP
// version.
@@ -196,14 +187,12 @@ impl S3Core {
// As discussed in <https://github.com/seanmonstar/reqwest/issues/1809>,
// google server could send RST_STREAM of PROTOCOL_ERROR if our request
// contains host header.
- req.headers_mut().remove(HOST);
+ parts.headers.remove(HOST);
- Ok(())
- }
-
- #[inline]
- pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
- self.info.http_client().send(req).await
+ self.info
+ .http_client()
+ .fetch(Request::from_parts(parts, body))
+ .await
}
/// # Note
@@ -529,11 +518,8 @@ impl S3Core {
range: BytesRange,
args: &OpRead,
) -> Result<Response<HttpBody>> {
- let mut req = self.s3_get_object_request(path, range, args)?;
-
- self.sign(&mut req).await?;
-
- self.info.http_client().fetch(req).await
+ let req = self.s3_get_object_request(path, range, args)?;
+ self.fetch(req).await
}
pub fn s3_put_object_request(
@@ -610,10 +596,7 @@ impl S3Core {
}
pub async fn s3_head_object(&self, path: &str, args: OpStat) -> Result<Response<Buffer>> {
- let mut req = self.s3_head_object_request(path, args)?;
-
- self.sign(&mut req).await?;
-
+ let req = self.s3_head_object_request(path, args)?;
self.send(req).await
}
@@ -641,14 +624,12 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::Delete)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -703,15 +684,13 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::Copy)
.header(constants::X_AMZ_COPY_SOURCE, &source)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -744,14 +723,12 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::List)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -796,14 +773,12 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::List)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -854,9 +829,7 @@ impl S3Core {
// Inject operation to the request.
req = req.extension(Operation::Write);
- let mut req = req.body(Buffer::new()).map_err(new_request_build_error)?;
-
- self.sign(&mut req).await?;
+ let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
self.send(req).await
}
@@ -937,12 +910,10 @@ impl S3Core {
// Inject operation to the request.
req = req.extension(Operation::Write);
- let mut req = req
+ let req = req
.body(Buffer::from(Bytes::from(content)))
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -966,13 +937,12 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::Write)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
self.send(req).await
}
@@ -1008,12 +978,10 @@ impl S3Core {
// Inject operation to the request.
req = req.extension(Operation::Delete);
- let mut req = req
+ let req = req
.body(Buffer::from(Bytes::from(content)))
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
@@ -1057,14 +1025,12 @@ impl S3Core {
// Set request payer header if enabled.
req = self.insert_request_payer_header(req);
- let mut req = req
+ let req = req
// Inject operation to the request.
.extension(Operation::List)
.body(Buffer::new())
.map_err(new_request_build_error)?;
- self.sign(&mut req).await?;
-
self.send(req).await
}
}
diff --git a/core/src/services/s3/writer.rs b/core/src/services/s3/writer.rs
index 54a1b6e3f..82932cae8 100644
--- a/core/src/services/s3/writer.rs
+++ b/core/src/services/s3/writer.rs
@@ -66,12 +66,10 @@ impl S3Writer {
impl oio::MultipartWrite for S3Writer {
async fn write_once(&self, size: u64, body: Buffer) -> Result<Metadata> {
- let mut req = self
+ let req = self
.core
.s3_put_object_request(&self.path, Some(size), &self.op, body)?;
- self.core.sign(&mut req).await?;
-
let resp = self.core.send(req).await?;
let status = resp.status();
@@ -117,7 +115,7 @@ impl oio::MultipartWrite for S3Writer {
let checksum = self.core.calculate_checksum(&body);
- let mut req = self.core.s3_upload_part_request(
+ let req = self.core.s3_upload_part_request(
&self.path,
upload_id,
part_number,
@@ -126,8 +124,6 @@ impl oio::MultipartWrite for S3Writer {
checksum.clone(),
)?;
- self.core.sign(&mut req).await?;
-
let resp = self.core.send(req).await?;
let status = resp.status();
@@ -242,12 +238,10 @@ impl oio::AppendWrite for S3Writer {
}
async fn append(&self, offset: u64, size: u64, body: Buffer) -> Result<Metadata> {
- let mut req = self
+ let req = self
.core
.s3_append_object_request(&self.path, offset, size, &self.op, body)?;
- self.core.sign(&mut req).await?;
-
let resp = self.core.send(req).await?;
let status = resp.status();
(opendal) 02/02: Address assume role support
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch 6560-reqsign-core-v1-s3
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit d6e5da5078feb6ee9dc671955399300be612f9c6
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Oct 13 13:13:47 2025 +0800
Address assume role support
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/services/s3/backend.rs | 47 ++++++++++++++++++++++++++++++++++++++---
1 file changed, 44 insertions(+), 3 deletions(-)
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index d0cb7a695..5c319b2c7 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -35,6 +35,7 @@ use log::warn;
use md5::Digest;
use md5::Md5;
use percent_encoding::percent_decode_str;
+use reqsign_aws_v4::AssumeRoleCredentialProvider;
use reqsign_aws_v4::Credential;
use reqsign_aws_v4::DefaultCredentialProvider;
use reqsign_aws_v4::RequestSigner as AwsV4Signer;
@@ -515,6 +516,22 @@ impl S3Builder {
self
}
+ /// Append a custom credential provider that will be tried before the default chain.
+ ///
+ /// Providers are evaluated in the order they are added.
+ pub fn credential_provider(
+ mut self,
+ provider: impl ProvideCredential<Credential = Credential>,
+ ) -> Self {
+ let chain = self
+ .credential_providers
+ .take()
+ .unwrap_or_default()
+ .push(provider);
+ self.credential_providers = Some(chain);
+ self
+ }
+
/// Replace the credential providers with a custom chain.
pub fn credential_provider_chain(mut self, chain: ProvideCredentialChain<Credential>) -> Self {
self.credential_providers = Some(chain);
@@ -854,9 +871,7 @@ impl Builder for S3Builder {
.with_file_read(TokioFileRead)
.with_http_send(ReqwestHttpSend::new(GLOBAL_REQWEST_CLIENT.clone()));
- let mut provider = if let Some(chain) = credential_providers {
- chain
- } else {
+ let mut provider = {
let mut builder = DefaultCredentialProvider::builder();
if config.disable_config_load {
@@ -870,6 +885,7 @@ impl Builder for S3Builder {
ProvideCredentialChain::new().push(builder.build())
};
+ // Insert static key if user provided.
if let (Some(ak), Some(sk)) = (&config.access_key_id, &config.secret_access_key) {
let static_provider = if let Some(token) = config.session_token.as_deref() {
StaticCredentialProvider::new(ak, sk).with_session_token(token)
@@ -879,6 +895,31 @@ impl Builder for S3Builder {
provider = provider.push_front(static_provider);
}
+ // Insert assume role provider if user provided.
+ if let Some(role_arn) = &config.role_arn {
+ let sts_ctx = ctx.clone();
+ let sts_request_signer = AwsV4Signer::new("sts", ®ion);
+ let sts_signer = Signer::new(sts_ctx, provider, sts_request_signer);
+ let mut assume_role_provider =
+ AssumeRoleCredentialProvider::new(role_arn.clone(), sts_signer);
+
+ if let Some(external_id) = &config.external_id {
+ assume_role_provider = assume_role_provider.with_external_id(external_id.clone());
+ }
+ if let Some(role_session_name) = &config.role_session_name {
+ assume_role_provider =
+ assume_role_provider.with_role_session_name(role_session_name.clone());
+ }
+ provider = ProvideCredentialChain::new().push(assume_role_provider);
+ }
+
+ // Replace provider if user provide their own.
+ let provider = if let Some(credential_providers) = credential_providers {
+ credential_providers
+ } else {
+ provider
+ };
+
// Create request signer for S3
let request_signer = AwsV4Signer::new("s3", ®ion);