add traits for custom subprocess implementations to aws-types and aws-config

.changelog/1775189986.md

@@ -0,0 +1,15 @@
+---
+applies_to:
+- aws-sdk-rust
+authors:
+- codeshaunted
+references:
+- smithy-rs#4590
+breaking: false
+new_feature: true
+bug_fix: false
+---
+
+Adds a public `ProvideProcess` trait and `Process` shim to `aws_types::os_shim_internal`, following the same pattern as `ProvideEnv`/`Env` and `ProvideFs`/`Fs`. This enables custom implementations of process execution for `credential_process`, allowing users to provide custom backends (e.g., sandboxed environments, remote execution, WASM) and enabling tests to mock process execution without spawning real subprocesses.
+
+Also updates `ProviderConfig` and `ConfigLoader` in `aws_config` to accept a `Process`, and updates `CredentialProcessProvider` to use the `Process` abstraction. The default tokio-based process provider uses `tokio::process`, pulled in via the existing `credentials-process` feature. Users can provide their own `Process` implementation via `ConfigLoader::process()` or `ProviderConfig::with_process()`.

aws/rust-runtime/aws-config/src/credential_process.rs

@@ -14,12 +14,52 @@ use aws_credential_types::credential_feature::AwsCredentialFeature;
 use aws_credential_types::provider::{self, error::CredentialsError, future, ProvideCredentials};
 use aws_credential_types::Credentials;
 use aws_smithy_json::deserialize::Token;
+use aws_types::os_shim_internal::Process;
 use std::borrow::Cow;
-use std::process::Command;
 use std::time::SystemTime;
 use time::format_description::well_known::Rfc3339;
 use time::OffsetDateTime;
 
+fn default_process() -> Process {
+    use aws_types::os_shim_internal::{CommandOutput, ExitStatus, ProvideProcess};
+    use std::future::Future;
+    use std::pin::Pin;
+
+    /// Process provider that uses `tokio::process::Command` for async execution.
+    #[derive(Debug)]
+    struct TokioProcessProvider;
+
+    impl ProvideProcess for TokioProcessProvider {
+        fn execute(
+            &self,
+            command: &str,
+        ) -> Pin<Box<dyn Future<Output = std::io::Result<CommandOutput>> + Send + '_>> {
+            let command = command.to_string();
+            Box::pin(async move {
+                let std_command = if cfg!(windows) {
+                    let mut cmd = std::process::Command::new("cmd.exe");
+                    cmd.args(["/C", &command]);
+                    cmd
+                } else {
+                    let mut cmd = std::process::Command::new("sh");
+                    cmd.args(["-c", &command]);
+                    cmd
+                };
+                let output = tokio::process::Command::from(std_command)
+                    .output()
+                    .await?;
+                Ok(CommandOutput {
+                    status: ExitStatus::from(output.status),
+                    stdout: output.stdout,
+                    stderr: output.stderr,
+                })
+            })
+        }
+    }
+
+    Process::from_custom(TokioProcessProvider)
+}
+
 /// External process credentials provider
 ///
 /// This credentials provider runs a configured external process and parses
@@ -57,6 +97,7 @@ use time::OffsetDateTime;
 pub struct CredentialProcessProvider {
     command: CommandWithSensitiveArgs<String>,
     profile_account_id: Option<AccountId>,
+    process: Process,
 }
 
 impl ProvideCredentials for CredentialProcessProvider {
@@ -74,6 +115,7 @@ impl CredentialProcessProvider {
         Self {
             command: CommandWithSensitiveArgs::new(command),
             profile_account_id: None,
+            process: default_process(),
         }
     }
 
@@ -85,17 +127,9 @@ impl CredentialProcessProvider {
         // Security: command arguments must be redacted at debug level
         tracing::debug!(command = %self.command, "loading credentials from external process");
 
-        let command = if cfg!(windows) {
-            let mut command = Command::new("cmd.exe");
-            command.args(["/C", self.command.unredacted()]);
-            command
-        } else {
-            let mut command = Command::new("sh");
-            command.args(["-c", self.command.unredacted()]);
-            command
-        };
-        let output = tokio::process::Command::from(command)
-            .output()
+        let output = self
+            .process
+            .execute(self.command.unredacted())
             .await
             .map_err(|e| {
                 CredentialsError::provider_error(format!(
@@ -140,6 +174,7 @@ impl CredentialProcessProvider {
 pub(crate) struct Builder {
     command: Option<CommandWithSensitiveArgs<String>>,
     profile_account_id: Option<AccountId>,
+    process: Option<Process>,
 }
 
 impl Builder {
@@ -158,10 +193,16 @@ impl Builder {
         self.profile_account_id = account_id;
     }
 
+    pub(crate) fn process(mut self, process: Process) -> Self {
+        self.process = Some(process);
+        self
+    }
+
     pub(crate) fn build(self) -> CredentialProcessProvider {
         CredentialProcessProvider {
             command: self.command.expect("should be set"),
             profile_account_id: self.profile_account_id,
+            process: self.process.unwrap_or_else(default_process),
         }
     }
 }
@@ -269,10 +310,9 @@ mod test {
     use crate::sensitive_command::CommandWithSensitiveArgs;
     use aws_credential_types::credential_feature::AwsCredentialFeature;
     use aws_credential_types::provider::ProvideCredentials;
-    use std::time::{Duration, SystemTime};
+    use std::time::SystemTime;
     use time::format_description::well_known::Rfc3339;
     use time::OffsetDateTime;
-    use tokio::time::timeout;
 
     // TODO(https://github.com/awslabs/aws-sdk-rust/issues/1117) This test is ignored on Windows because it uses Unix-style paths
     #[tokio::test]
@@ -312,8 +352,26 @@ mod test {
         assert_eq!(creds.expiry(), None);
     }
 
+    #[tokio::test]
+    async fn credentials_process_io_error() {
+        use aws_types::os_shim_internal::Process;
+        let process = Process::from_slice(&[]);
+        let provider = CredentialProcessProvider::builder()
+            .command(CommandWithSensitiveArgs::new(String::from("missing")))
+            .process(process)
+            .build();
+        let err = provider.provide_credentials().await.expect_err("should fail");
+        let msg = format!("{err:?}");
+        assert!(
+            msg.contains("not found"),
+            "error should contain cause: {msg}"
+        );
+    }
+
     #[tokio::test]
     async fn credentials_process_timeouts() {
+        use std::time::Duration;
+        use tokio::time::timeout;
         let provider = CredentialProcessProvider::new(String::from("sleep 1000"));
         let _creds = timeout(Duration::from_millis(1), provider.provide_credentials())
             .await

aws/rust-runtime/aws-config/src/lib.rs

@@ -240,7 +240,7 @@ mod loader {
     use aws_types::docs_for;
     use aws_types::endpoint_config::AccountIdEndpointMode;
     use aws_types::origin::Origin;
-    use aws_types::os_shim_internal::{Env, Fs};
+    use aws_types::os_shim_internal::{Env, Fs, Process};
     use aws_types::region::SigningRegionSet;
     use aws_types::sdk_config::SharedHttpClient;
     use aws_types::SdkConfig;
@@ -300,6 +300,7 @@ mod loader {
         stalled_stream_protection_config: Option<StalledStreamProtectionConfig>,
         env: Option<Env>,
         fs: Option<Fs>,
+        process: Option<Process>,
         behavior_version: Option<BehaviorVersion>,
         request_checksum_calculation: Option<RequestChecksumCalculation>,
         response_checksum_validation: Option<ResponseChecksumValidation>,
@@ -567,6 +568,15 @@ mod loader {
             self
         }
 
+        /// Override the process execution abstraction used during config resolution.
+        ///
+        /// This can be used with [`Process::from_custom`] to provide a custom process
+        /// execution backend (e.g., sandboxed environments, remote execution).
+        pub fn process(mut self, process: Process) -> Self {
+            self.process = Some(process);
+            self
+        }
+
         /// Override the access token provider used to build [`SdkConfig`].
         ///
         /// # Examples
@@ -836,7 +846,8 @@ mod loader {
                 .unwrap_or_else(|| {
                     let mut config = ProviderConfig::init(time_source.clone(), sleep_impl.clone())
                         .with_fs(self.fs.unwrap_or_default())
-                        .with_env(self.env.unwrap_or_default());
+                        .with_env(self.env.unwrap_or_default())
+                        .with_process(self.process.unwrap_or_default());
                     if let Some(http_client) = self.http_client.clone() {
                         config = config.with_http_client(http_client);
                     }

aws/rust-runtime/aws-config/src/profile/credentials/exec.rs

@@ -99,7 +99,8 @@ impl ProviderChain {
                 {
                     Arc::new({
                         let mut builder = CredentialProcessProvider::builder()
-                            .command(command_with_sensitive_args.to_owned_string());
+                            .command(command_with_sensitive_args.to_owned_string())
+                            .process(provider_config.process());
                         builder.set_account_id(
                             account_id.map(aws_credential_types::attributes::AccountId::from),
                         );

aws/rust-runtime/aws-config/src/provider_config.rs

@@ -16,7 +16,7 @@ use aws_smithy_runtime_api::client::http::HttpClient;
 use aws_smithy_runtime_api::shared::IntoShared;
 use aws_smithy_types::error::display::DisplayErrorContext;
 use aws_smithy_types::retry::RetryConfig;
-use aws_types::os_shim_internal::{Env, Fs};
+use aws_types::os_shim_internal::{Env, Fs, Process};
 use aws_types::region::Region;
 use aws_types::sdk_config::SharedHttpClient;
 use aws_types::SdkConfig;
@@ -37,6 +37,7 @@ use tokio::sync::OnceCell;
 pub struct ProviderConfig {
     env: Env,
     fs: Fs,
+    process: Process,
     time_source: SharedTimeSource,
     http_client: Option<SharedHttpClient>,
     retry_config: Option<RetryConfig>,
@@ -58,6 +59,7 @@ impl Debug for ProviderConfig {
         f.debug_struct("ProviderConfig")
             .field("env", &self.env)
             .field("fs", &self.fs)
+            .field("process", &self.process)
             .field("time_source", &self.time_source)
             .field("http_client", &self.http_client)
             .field("retry_config", &self.retry_config)
@@ -75,6 +77,7 @@ impl Default for ProviderConfig {
         Self {
             env: Env::default(),
             fs: Fs::default(),
+            process: Process::default(),
             time_source: SharedTimeSource::default(),
             http_client: None,
             retry_config: None,
@@ -108,6 +111,7 @@ impl ProviderConfig {
             profile_files: ProfileFiles::default(),
             env,
             fs,
+            process: Process::default(),
             time_source: SharedTimeSource::new(StaticTimeSource::new(UNIX_EPOCH)),
             http_client: None,
             retry_config: None,
@@ -151,6 +155,7 @@ impl ProviderConfig {
         ProviderConfig {
             env: Env::default(),
             fs: Fs::default(),
+            process: Process::default(),
             time_source: SharedTimeSource::default(),
             http_client: None,
             retry_config: None,
@@ -176,6 +181,7 @@ impl ProviderConfig {
             profile_files: ProfileFiles::default(),
             env: Env::default(),
             fs: Fs::default(),
+            process: Process::default(),
             time_source,
             http_client: None,
             retry_config: None,
@@ -245,6 +251,11 @@ impl ProviderConfig {
         self.fs.clone()
     }
 
+    #[allow(dead_code)]
+    pub(crate) fn process(&self) -> Process {
+        self.process.clone()
+    }
+
     #[allow(dead_code)]
     pub(crate) fn time_source(&self) -> SharedTimeSource {
         self.time_source.clone()
@@ -392,6 +403,10 @@ impl ProviderConfig {
         }
     }
 
+    pub(crate) fn with_process(self, process: Process) -> Self {
+        ProviderConfig { process, ..self }
+    }
+
     /// Override the time source for this configuration
     pub fn with_time_source(self, time_source: impl TimeSource + 'static) -> Self {
         ProviderConfig {