feat(aws): auto-refresh credentials via reqsign Signer

Replace manual AwsCredential passing with reqsign's Signer<Credential>
which handles credential caching and lazy refresh internally. This
enables production deployments (EC2/ECS/EKS) where temporary credentials
from IMDS/STS expire periodically.

- Add context.rs with TokioFileRead, TokioCommandExecute, ReqwestHttpSend
  implementations for reqsign Context (hand-written async, no async-trait dep)
- Replace S3Options.credential with S3Options.signer (reqsign Signer)
- Simplify sign.rs to delegate to signer.sign() instead of manual RequestSigner
- Add AmazonS3Builder::default_credential_provider() for auto-detect chain
  (env, profile, SSO, IRSA, ECS, IMDS), feature-gated to tokio-http/web-http
- Update fusio-manifest s3::Builder with matching default_credential_provider()
- Rewrite tests to use AmazonS3Builder instead of direct S3Options construction

Author: ethe

fusio-manifest/src/s3.rs

@@ -105,6 +105,7 @@ where
     region: Option<String>,
     endpoint: Option<String>,
     credential: Option<AwsCredential>,
+    use_default_credential_provider: bool,
     sign_payload: bool,
     checksum: bool,
     opts: Arc<ManifestContext<R, E>>,
@@ -128,6 +129,7 @@ impl Builder<DefaultRetention, DefaultExecutor> {
             region: None,
             endpoint: None,
             credential: None,
+            use_default_credential_provider: false,
             sign_payload: false,
             checksum: false,
             opts: Arc::new(opts),
@@ -180,6 +182,15 @@ where
         self
     }
 
+    /// Use the default credential provider chain (env, profile, SSO, IRSA, ECS, IMDS).
+    ///
+    /// Available with the `tokio` feature or with `web` on `wasm32`.
+    #[cfg(any(feature = "tokio", all(feature = "web", target_arch = "wasm32")))]
+    pub fn default_credential_provider(mut self) -> Self {
+        self.use_default_credential_provider = true;
+        self
+    }
+
     /// Enable SigV4 payload signing (default: false). Recommended for AWS/MinIO.
     pub fn sign_payload(mut self, yes: bool) -> Self {
         self.sign_payload = yes;
@@ -212,6 +223,7 @@ where
             region: self.region,
             endpoint: self.endpoint,
             credential: self.credential,
+            use_default_credential_provider: self.use_default_credential_provider,
             sign_payload: self.sign_payload,
             checksum: self.checksum,
             opts: opts.into(),
@@ -224,6 +236,10 @@ where
         if let Some(cred) = self.credential {
             b = b.credential(cred);
         }
+        #[cfg(any(feature = "tokio", all(feature = "web", target_arch = "wasm32")))]
+        if self.use_default_credential_provider {
+            b = b.default_credential_provider();
+        }
         if let Some(ep) = self.endpoint {
             b = b.endpoint(ep);
         }
@@ -256,6 +272,7 @@ where
             region: self.region.clone(),
             endpoint: self.endpoint.clone(),
             credential: self.credential.clone(),
+            use_default_credential_provider: self.use_default_credential_provider,
             sign_payload: self.sign_payload,
             checksum: self.checksum,
             opts: self.opts.clone(),

fusio-manifest/src/session.rs

@@ -1121,7 +1121,8 @@ mod tests {
             assert_eq!(keys_equal_to_v1, 2);
 
             session.put(key, "v2".to_owned());
-            return session.commit().await.map(|_| ()); // This should fail for one of the transactions
+            return session.commit().await.map(|_| ()); // This should fail for one of the
+                                                       // transactions
         }
 
         // 2. Run both transactions concurrently.

fusio/Cargo.toml

@@ -24,6 +24,7 @@ aws = [
     "serde",
     "serde_json",
     "serde_urlencoded",
+    "tokio?/process",
 ]
 bytes = ["dep:bytes", "fusio-core/bytes"]
 completion-based = ["fusio-core/completion-based"]

fusio/src/fs/mod.rs

@@ -110,10 +110,7 @@ mod tests {
             fs::{Fs, OpenOptions},
             impls::disk::tokio::fs::TokioFs,
             path::Path,
-            remotes::{
-                aws::{credential::AwsCredential, fs::AmazonS3, options::S3Options, s3::S3File},
-                http::tokio::TokioClient,
-            },
+            remotes::aws::{credential::AwsCredential, fs::AmazonS3Builder, s3::S3File},
             DynFs, Read, Write,
         };
 
@@ -122,25 +119,18 @@ mod tests {
             .map_err(|err| Error::Path(Box::new(err)))?;
         let s3_path: Path = "s3_copy_test.file".into();
 
-        let key_id = "user".to_string();
-        let secret_key = "password".to_string();
-
-        let client = TokioClient::new();
-        let region = "ap-southeast-1";
-        let options = S3Options {
-            endpoint: "http://localhost:9000/data".into(),
-            bucket: "data".to_string(),
-            credential: Some(AwsCredential {
-                key_id,
-                secret_key,
-                token: None,
-            }),
-            region: region.into(),
-            sign_payload: true,
-            checksum: false,
-        };
-
-        let s3_fs = Arc::new(AmazonS3::new(Box::new(client), options));
+        let s3_fs = Arc::new(
+            AmazonS3Builder::new("data".to_string())
+                .endpoint("http://localhost:9000".to_string())
+                .region("ap-southeast-1".to_string())
+                .credential(AwsCredential {
+                    key_id: "user".to_string(),
+                    secret_key: "password".to_string(),
+                    token: None,
+                })
+                .sign_payload(true)
+                .build(),
+        );
         let local_fs = Arc::new(TokioFs);
 
         {

fusio/src/impls/remotes/aws/context.rs

@@ -0,0 +1,149 @@
+use reqsign_core::Context;
+
+cfg_if::cfg_if! {
+    if #[cfg(all(feature = "tokio", not(target_arch = "wasm32")))] {
+        #[derive(Debug)]
+        pub(crate) struct TokioFileRead;
+
+        impl reqsign_core::FileRead for TokioFileRead {
+            fn file_read<'a, 'b, 'c>(
+                &'a self,
+                path: &'b str,
+            ) -> std::pin::Pin<
+                Box<dyn std::future::Future<Output = reqsign_core::Result<Vec<u8>>> + Send + 'c>,
+            >
+            where
+                'a: 'c,
+                'b: 'c,
+                Self: 'c,
+            {
+                Box::pin(async move { tokio::fs::read(path).await.map_err(Into::into) })
+            }
+        }
+
+        #[derive(Debug)]
+        pub(crate) struct TokioCommandExecute;
+
+        impl reqsign_core::CommandExecute for TokioCommandExecute {
+            fn command_execute<'a, 'b, 'c, 'd, 'fut>(
+                &'a self,
+                program: &'b str,
+                args: &'c [&'d str],
+            ) -> std::pin::Pin<
+                Box<
+                    dyn std::future::Future<Output = reqsign_core::Result<reqsign_core::CommandOutput>>
+                        + Send
+                        + 'fut,
+                >,
+            >
+            where
+                'a: 'fut,
+                'b: 'fut,
+                'c: 'fut,
+                'd: 'fut,
+                Self: 'fut,
+            {
+                let program = program.to_owned();
+                let args: Vec<String> = args.iter().map(|s| s.to_string()).collect();
+                Box::pin(async move {
+                    let output = tokio::process::Command::new(&program)
+                        .args(&args)
+                        .output()
+                        .await
+                        .map_err(|e| {
+                            reqsign_core::Error::unexpected(format!(
+                                "execute command {program}: {e}"
+                            ))
+                        })?;
+                    Ok(reqsign_core::CommandOutput {
+                        status: output.status.code().unwrap_or(-1),
+                        stdout: output.stdout,
+                        stderr: output.stderr,
+                    })
+                })
+            }
+        }
+    }
+}
+
+cfg_if::cfg_if! {
+    if #[cfg(any(feature = "tokio-http", feature = "web-http"))] {
+        #[derive(Debug)]
+        pub(crate) struct ReqwestHttpSend {
+            client: reqwest::Client,
+        }
+
+        impl Default for ReqwestHttpSend {
+            fn default() -> Self {
+                Self {
+                    client: reqwest::Client::new(),
+                }
+            }
+        }
+
+        impl reqsign_core::HttpSend for ReqwestHttpSend {
+            fn http_send<'a, 'b>(
+                &'a self,
+                req: http::Request<bytes::Bytes>,
+            ) -> std::pin::Pin<
+                Box<
+                    dyn std::future::Future<
+                            Output = reqsign_core::Result<http::Response<bytes::Bytes>>,
+                        > + Send
+                        + 'b,
+                >,
+            >
+            where
+                'a: 'b,
+                Self: 'b,
+            {
+                Box::pin(async move {
+                    let (parts, body) = req.into_parts();
+                    let resp = self
+                        .client
+                        .request(parts.method, parts.uri.to_string())
+                        .headers(parts.headers)
+                        .body(body)
+                        .send()
+                        .await
+                        .map_err(|e| {
+                            reqsign_core::Error::unexpected(format!("http request failed: {e}"))
+                        })?;
+
+                    let status = resp.status();
+                    let headers = resp.headers().clone();
+                    let body = resp.bytes().await.map_err(|e| {
+                        reqsign_core::Error::unexpected(format!("read response body: {e}"))
+                    })?;
+
+                    let mut builder = http::Response::builder().status(status);
+                    *builder.headers_mut().unwrap() = headers;
+                    builder.body(body).map_err(|e| {
+                        reqsign_core::Error::unexpected(format!("build http response: {e}"))
+                    })
+                })
+            }
+        }
+    }
+}
+
+pub(crate) fn default_context() -> Context {
+    #[allow(unused_mut)]
+    let mut ctx = Context::new();
+
+    cfg_if::cfg_if! {
+        if #[cfg(all(feature = "tokio", not(target_arch = "wasm32")))] {
+            ctx = ctx
+                .with_file_read(TokioFileRead)
+                .with_command_execute(TokioCommandExecute);
+        }
+    }
+
+    cfg_if::cfg_if! {
+        if #[cfg(any(feature = "tokio-http", feature = "web-http"))] {
+            ctx = ctx.with_http_send(ReqwestHttpSend::default());
+        }
+    }
+
+    ctx.with_env(reqsign_core::OsEnv)
+}