feat: exec, to run a command with credentials and the proxy endpoint

source-coop exec -- aws s3 ls s3://acct/prod/ runs any S3 tool as the
signed-in user, as aws-vault exec does: the role's current credentials
(cached, or minted as creds would) go into the command's environment,
with the proxy as AWS_ENDPOINT_URL(_S3) and S3_ENDPOINT_URL, a region if
none is set, and AWS_PROFILE removed. On Unix the CLI becomes the
command, so its exit code and signals reach the shell directly.

creds and exec share current_creds(), which mint() now takes a role and
proxy for rather than CredsArgs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: alukach

README.md

@@ -132,6 +132,21 @@ eval $(source-coop creds --format env)
 
 This sets `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, and `AWS_SESSION_TOKEN` in your current shell.
 
+### Running a command with your credentials (`exec`)
+
+`source-coop exec` runs any S3 tool with your credentials and the data proxy as its endpoint, without a profile or exported variables, as `aws-vault exec` does for AWS:
+
+```bash
+source-coop exec -- aws s3 ls s3://my-org/my-product/
+source-coop exec -- aws s3 sync ./data s3://my-org/my-product/data/
+source-coop exec -- s5cmd cp 's3://my-org/my-product/*.parquet' ./local/
+( source .env.staging && source-coop exec -- aws s3 ls s3://my-org/ )   # staging
+```
+
+The command gets `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN` and `AWS_CREDENTIAL_EXPIRATION`, and the proxy URL as `AWS_ENDPOINT_URL_S3` (the AWS CLI v2.13+ and SDKs, for S3 only, so calls to other AWS services are unaffected) and `S3_ENDPOINT_URL` (s5cmd). `AWS_REGION` is set to `us-east-1` only if no region is set already (an empty one counts as unset). `AWS_PROFILE` and `AWS_DEFAULT_PROFILE` are removed so a profile's own credentials can't be mistaken for these, and `AWS_IGNORE_CONFIGURED_ENDPOINT_URLS` is removed so the command can't send them to AWS instead of the proxy. `--role-arn` and `--proxy-url` (or `SOURCE_ROLE_ARN` and `SOURCE_PROXY_URL`) pick the role and environment, as for `creds`. Everything after `--` is the command, flags included.
+
+The command's exit code is passed on. Credentials are fetched once, when it starts: cached ones are used only with at least 15 minutes left, otherwise new ones are fetched. A command still running when they expire (an hour after they're fetched, unless you logged in with `--duration`) needs running again. If you're not signed in, `exec` exits with 4 without running the command.
+
 ### Multiple roles
 
 > [!WARNING]

src/cache.rs

@@ -366,19 +366,16 @@ pub(crate) fn write_file(path: &Path, json: &str) -> Result<String, String> {
     Ok(path.display().to_string())
 }
 
-/// Check if credentials are expired or will expire within a 60-second buffer.
-pub fn is_expired(creds: &Credentials) -> Result<bool, String> {
+/// Whether `creds` run out within `left` from now (or already have).
+pub fn expires_within(creds: &Credentials, left: chrono::Duration) -> Result<bool, String> {
     let expiration = chrono::DateTime::parse_from_rfc3339(&creds.expiration).map_err(|e| {
         format!(
             "Failed to parse expiration timestamp '{}': {e}",
             creds.expiration
         )
     })?;
 
-    let now = Utc::now();
-    let buffer = chrono::Duration::seconds(60);
-
-    Ok(expiration <= now + buffer)
+    Ok(expiration <= Utc::now() + left)
 }
 
 #[cfg(test)]
@@ -497,28 +494,37 @@ mod tests {
     fn expired_future_date() {
         let future = (Utc::now() + chrono::Duration::hours(1)).to_rfc3339();
         let creds = sample_creds(&future);
-        assert!(!is_expired(&creds).unwrap());
+        assert!(!expires_within(&creds, chrono::Duration::seconds(60)).unwrap());
     }
 
     #[test]
     fn expired_past_date() {
         let past = (Utc::now() - chrono::Duration::hours(1)).to_rfc3339();
         let creds = sample_creds(&past);
-        assert!(is_expired(&creds).unwrap());
+        assert!(expires_within(&creds, chrono::Duration::seconds(60)).unwrap());
     }
 
     #[test]
     fn expired_within_buffer() {
         // 30 seconds from now is within the 60s buffer
         let near_future = (Utc::now() + chrono::Duration::seconds(30)).to_rfc3339();
         let creds = sample_creds(&near_future);
-        assert!(is_expired(&creds).unwrap());
+        assert!(expires_within(&creds, chrono::Duration::seconds(60)).unwrap());
     }
 
     #[test]
     fn expired_invalid_timestamp() {
         let creds = sample_creds("not-a-timestamp");
-        assert!(is_expired(&creds).is_err());
+        assert!(expires_within(&creds, chrono::Duration::seconds(60)).is_err());
+    }
+
+    #[test]
+    fn expires_within_a_longer_margin() {
+        // Fine for `creds`, too short for `exec`'s 15 minutes.
+        let soon = (Utc::now() + chrono::Duration::minutes(10)).to_rfc3339();
+        let creds = sample_creds(&soon);
+        assert!(!expires_within(&creds, chrono::Duration::seconds(60)).unwrap());
+        assert!(expires_within(&creds, chrono::Duration::minutes(15)).unwrap());
     }
 
     #[test]

src/exec.rs

@@ -0,0 +1,113 @@
+//! `source-coop exec`: run any S3 tool (the AWS CLI, s5cmd, rclone, an SDK
+//! script) as the signed-in user, with credentials and the data proxy's
+//! endpoint in its environment, as `aws-vault exec` does for AWS.
+
+use crate::sts::Credentials;
+use std::process::Command;
+
+/// The environment `command` runs with: the credentials, where they expire,
+/// and the proxy as the S3 endpoint under the names the common tools read.
+/// A region is only given when none is set, as the proxy ignores it but
+/// signing needs one.
+pub fn vars(creds: &Credentials, proxy_url: &str, region_set: bool) -> Vec<(&'static str, String)> {
+    let endpoint = proxy_url.trim_end_matches('/').to_string();
+    let mut vars = vec![
+        ("AWS_ACCESS_KEY_ID", creds.access_key_id.clone()),
+        ("AWS_SECRET_ACCESS_KEY", creds.secret_access_key.clone()),
+        ("AWS_SESSION_TOKEN", creds.session_token.clone()),
+        ("AWS_CREDENTIAL_EXPIRATION", creds.expiration.clone()),
+        // The AWS CLI (v2.13+) and SDKs, for S3 alone: other services (STS,
+        // KMS) a script calls keep their own endpoints.
+        ("AWS_ENDPOINT_URL_S3", endpoint.clone()),
+        // s5cmd.
+        ("S3_ENDPOINT_URL", endpoint),
+    ];
+    if !region_set {
+        vars.push(("AWS_REGION", "us-east-1".into()));
+        vars.push(("AWS_DEFAULT_REGION", "us-east-1".into()));
+    }
+    vars
+}
+
+/// Run `command` with `vars` added to the environment, and leave with its
+/// exit code. On Unix this process becomes the command, so signals (Ctrl+C)
+/// and the exit status reach the shell as if it had been run directly.
+pub fn run(command: &[String], vars: Vec<(&'static str, String)>) -> Result<(), String> {
+    let (program, args) = command.split_first().ok_or("Name a command to run.")?;
+    let mut cmd = Command::new(program);
+    cmd.args(args).envs(vars);
+    // A profile could hold other credentials or an endpoint; the env vars
+    // already win, but naming one invites confusion over which applied. And
+    // tools told to ignore configured endpoints would send the credentials
+    // to AWS itself rather than the proxy.
+    for name in [
+        "AWS_PROFILE",
+        "AWS_DEFAULT_PROFILE",
+        "AWS_IGNORE_CONFIGURED_ENDPOINT_URLS",
+    ] {
+        cmd.env_remove(name);
+    }
+    let failed = |e: std::io::Error| format!("Couldn't run {program}: {e}");
+
+    #[cfg(unix)]
+    {
+        use std::os::unix::process::CommandExt;
+        // Only returns if the command couldn't be started.
+        Err(failed(cmd.exec()))
+    }
+    #[cfg(not(unix))]
+    {
+        let status = cmd.status().map_err(failed)?;
+        std::process::exit(status.code().unwrap_or(1));
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn creds() -> Credentials {
+        Credentials {
+            access_key_id: "AKID".into(),
+            secret_access_key: "SECRET".into(),
+            session_token: "TOKEN".into(),
+            expiration: "2026-10-14T13:00:00Z".into(),
+        }
+    }
+
+    #[test]
+    fn the_command_gets_credentials_and_the_proxy_as_its_endpoint() {
+        let vars = vars(&creds(), "https://data.staging.source.coop/", false);
+        let get = |k: &str| {
+            vars.iter()
+                .find(|(name, _)| *name == k)
+                .map(|(_, v)| v.as_str())
+        };
+        assert_eq!(get("AWS_ACCESS_KEY_ID"), Some("AKID"));
+        assert_eq!(get("AWS_SESSION_TOKEN"), Some("TOKEN"));
+        assert_eq!(
+            get("AWS_CREDENTIAL_EXPIRATION"),
+            Some("2026-10-14T13:00:00Z")
+        );
+        for name in ["AWS_ENDPOINT_URL_S3", "S3_ENDPOINT_URL"] {
+            assert_eq!(
+                get(name),
+                Some("https://data.staging.source.coop"),
+                "{name}"
+            );
+        }
+        assert_eq!(get("AWS_ENDPOINT_URL"), None, "only S3 goes to the proxy");
+        assert_eq!(get("AWS_REGION"), Some("us-east-1"));
+    }
+
+    #[test]
+    fn a_region_already_set_is_kept() {
+        let vars = vars(&creds(), "https://data.source.coop", true);
+        assert!(!vars.iter().any(|(name, _)| name.contains("REGION")));
+    }
+
+    #[test]
+    fn a_missing_command_is_an_error() {
+        assert_eq!(run(&[], vec![]), Err("Name a command to run.".to_string()));
+    }
+}

src/main.rs

@@ -1,5 +1,6 @@
 mod api_cmd;
 mod cache;
+mod exec;
 mod failure;
 mod oidc;
 mod output;
@@ -88,6 +89,26 @@ enum Commands {
     Login(LoginArgs),
     /// Make any request to the source.coop API, signed in as you
     Api(ApiCommandArgs),
+    /// Run a command (the AWS CLI, s5cmd, an SDK script) with credentials and
+    /// the data proxy as its S3 endpoint in its environment
+    /// (e.g. `source-coop exec -- aws s3 ls s3://my-org/my-product/`)
+    Exec(ExecArgs),
+}
+
+#[derive(Parser)]
+struct ExecArgs {
+    /// Role ARN whose credentials to use
+    #[arg(long, env = "SOURCE_ROLE_ARN", default_value = defaults::ROLE_ARN)]
+    role_arn: String,
+
+    /// S3 proxy URL: the environment whose credentials to use, and the
+    /// endpoint the command is pointed at
+    #[arg(long, env = "SOURCE_PROXY_URL", default_value = defaults::PROXY_URL)]
+    proxy_url: String,
+
+    /// The command to run, with its arguments, after `--`
+    #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)]
+    command: Vec<String>,
 }
 
 #[derive(Parser)]
@@ -293,6 +314,11 @@ async fn main() {
                 fail(e, false);
             }
         }
+        Commands::Exec(args) => {
+            if let Err(e) = run_exec(args, verbose).await {
+                fail(e, false);
+            }
+        }
     }
 }
 
@@ -779,24 +805,36 @@ async fn run_login(args: LoginArgs, verbose: bool) -> Result<(), String> {
     Ok(())
 }
 
-async fn run_creds(args: CredsArgs, verbose: bool) -> Result<(), String> {
-    let env = cache::env(&args.proxy_url)?;
-    let env = env.as_deref();
-    let cached = cache::read_credentials(env, &args.role_arn)?;
-    let entry = match cached {
-        Some(entry) if !cache::is_expired(&entry.creds)? => entry,
-        _ => {
-            // Re-check under the lock: another process may have minted already.
-            let _lock = cache::lock(env, &args.role_arn)?;
-            match cache::read_credentials(env, &args.role_arn)? {
-                Some(entry) if !cache::is_expired(&entry.creds)? => entry,
-                stale => mint(&args, env, stale, verbose).await.map_err(|e| {
-                    format!("Couldn't get new credentials ({e}). Run 'source-coop auth login'.")
-                })?,
-            }
-        }
-    };
+/// Run the command with the role's credentials. They're fetched once, so
+/// cached ones are only reused with `EXEC_MIN_LIFETIME` left; a command
+/// running past their expiry (an hour unless `login --duration` said
+/// otherwise) needs running again.
+async fn run_exec(args: ExecArgs, verbose: bool) -> Result<(), String> {
+    let entry = current_creds(&args.role_arn, &args.proxy_url, EXEC_MIN_LIFETIME, verbose).await?;
+    // Set but empty is unset, as signing would fail on an empty region.
+    let region_set = ["AWS_REGION", "AWS_DEFAULT_REGION"]
+        .iter()
+        .any(|v| std::env::var_os(v).is_some_and(|r| !r.is_empty()));
+    let vars = exec::vars(&entry.creds, &args.proxy_url, region_set);
+    if verbose {
+        eprintln!(
+            "[verbose] Running {:?} against {} with credentials expiring {}",
+            args.command, args.proxy_url, entry.creds.expiration
+        );
+    }
+    exec::run(&args.command, vars)
+}
+
+/// How long cached credentials must still last for `exec` to hand them to a
+/// command, which can't ask for new ones as `credential_process` can.
+const EXEC_MIN_LIFETIME: chrono::Duration = chrono::Duration::minutes(15);
+
+/// How long cached credentials must still last for `creds`: its callers ask
+/// again when they run out.
+const CREDS_MIN_LIFETIME: chrono::Duration = chrono::Duration::seconds(60);
 
+async fn run_creds(args: CredsArgs, verbose: bool) -> Result<(), String> {
+    let entry = current_creds(&args.role_arn, &args.proxy_url, CREDS_MIN_LIFETIME, verbose).await?;
     match args.format {
         OutputFormat::CredentialProcess => output::print_credential_process(&entry.creds),
         OutputFormat::Env => output::print_env(&entry.creds),
@@ -805,36 +843,66 @@ async fn run_creds(args: CredsArgs, verbose: bool) -> Result<(), String> {
     Ok(())
 }
 
+/// The role's credentials at `proxy_url`'s environment: the cached ones while
+/// they have `min_left` to run, otherwise new ones, minted once however many
+/// ask at a time.
+async fn current_creds(
+    role_arn: &str,
+    proxy_url: &str,
+    min_left: chrono::Duration,
+    verbose: bool,
+) -> Result<cache::CacheEntry, String> {
+    let env = cache::env(proxy_url)?;
+    let env = env.as_deref();
+    let fresh = |entry: &cache::CacheEntry| -> Result<bool, String> {
+        Ok(!cache::expires_within(&entry.creds, min_left)?)
+    };
+    match cache::read_credentials(env, role_arn)? {
+        Some(entry) if fresh(&entry)? => return Ok(entry),
+        _ => {}
+    }
+    // Re-check under the lock: another process may have minted already.
+    let _lock = cache::lock(env, role_arn)?;
+    match cache::read_credentials(env, role_arn)? {
+        Some(entry) if fresh(&entry)? => Ok(entry),
+        stale => mint(role_arn, proxy_url, env, stale, verbose)
+            .await
+            .map_err(|e| match e.as_str() {
+                NOT_LOGGED_IN => "Not signed in. Run 'source-coop auth login'.".to_string(),
+                _ => format!("Couldn't get new credentials ({e}). Run 'source-coop auth login'."),
+            }),
+    }
+}
+
+/// `mint`'s error when there's no login session to mint with.
+const NOT_LOGGED_IN: &str = "not logged in";
+
 /// New credentials for the role, cached before they're returned. An entry
 /// from before the session was shared refreshes with its own token; anything
 /// else exchanges the session's ID token, refreshing the session if need be.
 async fn mint(
-    args: &CredsArgs,
+    role_arn: &str,
+    proxy_url: &str,
     env: Option<&str>,
     stale: Option<cache::CacheEntry>,
     verbose: bool,
 ) -> Result<cache::CacheEntry, String> {
-    let save = |e: &cache::CacheEntry| cache::write_credentials(env, &args.role_arn, e).map(drop);
+    let save = |e: &cache::CacheEntry| cache::write_credentials(env, role_arn, e).map(drop);
     match stale {
         Some(entry) if entry.sts.is_none() && entry.refresh.is_some() => {
-            refresh(&args.role_arn, entry, verbose, save).await
+            refresh(role_arn, entry, verbose, save).await
         }
         stale => {
             let sts = stale.and_then(|e| e.sts).unwrap_or(cache::StsSettings {
-                proxy_url: args.proxy_url.clone(),
+                proxy_url: proxy_url.to_string(),
                 duration: None,
             });
             let id_token = session::token(env, session::Want::Id, verbose)
                 .await?
-                .ok_or("not logged in")?;
-            let creds = sts::assume_role(
-                &sts.proxy_url,
-                &args.role_arn,
-                &id_token,
-                sts.duration,
-                verbose,
-            )
-            .await?;
+                .ok_or(NOT_LOGGED_IN)?;
+            let creds =
+                sts::assume_role(&sts.proxy_url, role_arn, &id_token, sts.duration, verbose)
+                    .await?;
             let entry = cache::CacheEntry {
                 creds,
                 sts: Some(sts),
@@ -1436,7 +1504,7 @@ mod tests {
         let entry = result.unwrap();
 
         assert_eq!(entry.creds.access_key_id, "NEWKEY");
-        assert!(!cache::is_expired(&entry.creds).unwrap());
+        assert!(!cache::expires_within(&entry.creds, chrono::Duration::seconds(60)).unwrap());
         assert_eq!(entry.refresh.unwrap().refresh_token, "new-rt");
         // Rotated token persisted first (with old creds), then the new creds.
         assert_eq!(