From da325a6f1e22c485c2d899ef619445d8db5919ac Mon Sep 17 00:00:00 2001 From: hoophalab <200652805+hoophalab@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:10:27 -0400 Subject: [PATCH 1/5] fix(clp-tdl-package): Resolve AWS credentials via the SDK's default provider chain for `default` authentication. --- Cargo.lock | 1 + .../src/clp_config/s3_config.rs | 2 + components/clp-rust-utils/src/s3/client.rs | 2 +- components/clp-rust-utils/src/sqs/client.rs | 2 +- .../clp-rust-utils/tests/clp_config_test.rs | 1 + components/clp-tdl-package/Cargo.toml | 1 + .../src/task/compression/compress.rs | 92 ++++++++++++++----- .../log-ingestor/tests/test_ingestion_job.rs | 3 + components/log-ingestor/tests/test_scan.rs | 1 + 9 files changed, 81 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 03b01ad06e..78cc5491d2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -973,6 +973,7 @@ name = "clp-tdl-package" version = "0.13.1-dev" dependencies = [ "anyhow", + "aws-config", "aws-sdk-s3", "clp-rust-utils", "non-empty-string", diff --git a/components/clp-rust-utils/src/clp_config/s3_config.rs b/components/clp-rust-utils/src/clp_config/s3_config.rs index 50792094a1..123e16dc42 100644 --- a/components/clp-rust-utils/src/clp_config/s3_config.rs +++ b/components/clp-rust-utils/src/clp_config/s3_config.rs @@ -29,4 +29,6 @@ pub enum AwsAuthentication { pub struct AwsCredentials { pub access_key_id: String, pub secret_access_key: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_token: Option, } diff --git a/components/clp-rust-utils/src/s3/client.rs b/components/clp-rust-utils/src/s3/client.rs index 134b1645a9..3ababdbf54 100644 --- a/components/clp-rust-utils/src/s3/client.rs +++ b/components/clp-rust-utils/src/s3/client.rs @@ -33,7 +33,7 @@ pub async fn create_new_client( config_defaults = config_defaults.credentials_provider(Credentials::new( credentials.access_key_id.as_str(), credentials.secret_access_key.as_str(), - None, + credentials.session_token.clone(), None, "clp-credentials-provider", )); diff --git a/components/clp-rust-utils/src/sqs/client.rs b/components/clp-rust-utils/src/sqs/client.rs index 2d5be46171..40799fa046 100644 --- a/components/clp-rust-utils/src/sqs/client.rs +++ b/components/clp-rust-utils/src/sqs/client.rs @@ -30,7 +30,7 @@ pub async fn create_new_client( config_defaults = config_defaults.credentials_provider(Credentials::new( credentials.access_key_id.as_str(), credentials.secret_access_key.as_str(), - None, + credentials.session_token.clone(), None, "clp-credentials-provider", )); diff --git a/components/clp-rust-utils/tests/clp_config_test.rs b/components/clp-rust-utils/tests/clp_config_test.rs index adbfab2b21..8ce03c31ce 100644 --- a/components/clp-rust-utils/tests/clp_config_test.rs +++ b/components/clp-rust-utils/tests/clp_config_test.rs @@ -21,6 +21,7 @@ fn test_clp_io_config_serialization() { credentials: AwsCredentials { access_key_id: "ACCESS_KEY_ID".into(), secret_access_key: "SECRET_ACCESS_KEY".into(), + session_token: None, }, }, }; diff --git a/components/clp-tdl-package/Cargo.toml b/components/clp-tdl-package/Cargo.toml index 8db1a8b52e..c78112ac91 100644 --- a/components/clp-tdl-package/Cargo.toml +++ b/components/clp-tdl-package/Cargo.toml @@ -8,6 +8,7 @@ crate-type = ["cdylib", "rlib"] [dependencies] anyhow = "1.0.100" +aws-config = "1.8.12" aws-sdk-s3 = "1.121.0" clp-rust-utils = { path = "../clp-rust-utils" } non-empty-string = { version = "0.2.6", features = ["serde"] } diff --git a/components/clp-tdl-package/src/task/compression/compress.rs b/components/clp-tdl-package/src/task/compression/compress.rs index af2eb94ff0..71329ce021 100644 --- a/components/clp-tdl-package/src/task/compression/compress.rs +++ b/components/clp-tdl-package/src/task/compression/compress.rs @@ -10,6 +10,9 @@ use std::process::Command; use std::process::Stdio; use anyhow::Context; +use aws_config::BehaviorVersion; +use aws_sdk_s3::config::Credentials; +use aws_sdk_s3::config::ProvideCredentials; use clp_rust_utils::aws::AWS_DEFAULT_REGION; use clp_rust_utils::clp_config::AwsAuthentication; use clp_rust_utils::clp_config::S3Config; @@ -46,6 +49,7 @@ use crate::common::runtime; /// * A spawned archive finisher panics. /// * Forwards [`build_s3_logs_list`]'s return values on failure. /// * Forwards [`std::fs::write`]'s return values on failure. +/// * Forwards [`s3_credential_env`]'s return values on failure. /// * Forwards [`extract_s3_output_config`]'s return values on failure. /// * Forwards [`prepare_clp_s_input`]'s return values on failure. /// * Forwards [`run_clp_s`]'s return values on failure. @@ -84,10 +88,16 @@ pub(super) fn compress( })?; tmp_file_deleter.add(list_path.clone()); + let runtime = runtime(); let S3InputSource { - aws_authentication, .. + region_code, + aws_authentication, + .. } = input_source; - let credential_env = s3_credential_env(&aws_authentication); + let input_region = region_code + .as_ref() + .map_or(AWS_DEFAULT_REGION, NonEmptyString::as_str); + let credential_env = s3_credential_env(&runtime, input_region, &aws_authentication)?; let s3_config = extract_s3_output_config(config)?; let archive_dir = config @@ -107,7 +117,6 @@ pub(super) fn compress( &mut tmp_file_deleter, )?; - let runtime = runtime(); let client = build_s3_client(&runtime, s3_config); let bucket = s3_config.bucket.to_string(); @@ -341,26 +350,66 @@ fn build_s3_logs_list(input_source: &S3InputSource) -> anyhow::Result { /// /// # Returns /// -/// * The env-var name-value pairs for [`AwsAuthentication::Credentials`]. -/// * An empty vector for [`AwsAuthentication::Default`] (which assumes credentials are already in -/// the ambient env). -fn s3_credential_env(auth: &AwsAuthentication) -> Vec<(&'static str, String)> { +/// The env-var name-value pairs containing the resolved access key, secret key, and optional +/// session token. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * The default AWS SDK credential provider chain has no provider. +/// * The default AWS SDK credential provider chain fails to resolve credentials. +fn s3_credential_env( + runtime: &tokio::runtime::Handle, + region: &str, + auth: &AwsAuthentication, +) -> anyhow::Result> { /// The env var holding the AWS access key ID. const AWS_ACCESS_KEY_ID_ENV_VAR: &str = "AWS_ACCESS_KEY_ID"; /// The env var holding the AWS secret access key. const AWS_SECRET_ACCESS_KEY_ENV_VAR: &str = "AWS_SECRET_ACCESS_KEY"; - match auth { - AwsAuthentication::Credentials { credentials } => vec![ - (AWS_ACCESS_KEY_ID_ENV_VAR, credentials.access_key_id.clone()), - ( - AWS_SECRET_ACCESS_KEY_ENV_VAR, - credentials.secret_access_key.clone(), - ), - ], - AwsAuthentication::Default => Vec::new(), + /// The env var holding the AWS session token. + const AWS_SESSION_TOKEN_ENV_VAR: &str = "AWS_SESSION_TOKEN"; + + let credentials = match auth { + AwsAuthentication::Credentials { credentials } => Credentials::new( + credentials.access_key_id.clone(), + credentials.secret_access_key.clone(), + credentials.session_token.clone(), + None, + "clp-credentials-provider", + ), + AwsAuthentication::Default => { + let sdk_config = runtime.block_on( + aws_config::defaults(BehaviorVersion::latest()) + .region(aws_sdk_s3::config::Region::new(region.to_string())) + .load(), + ); + let provider = sdk_config + .credentials_provider() + .context("default AWS SDK credential provider is unavailable")?; + runtime + .block_on(provider.provide_credentials()) + .context("failed to resolve credentials from the default AWS SDK provider chain")? + } + }; + + let mut env = vec![ + ( + AWS_ACCESS_KEY_ID_ENV_VAR, + credentials.access_key_id().to_string(), + ), + ( + AWS_SECRET_ACCESS_KEY_ENV_VAR, + credentials.secret_access_key().to_string(), + ), + ]; + if let Some(session_token) = credentials.session_token() { + env.push((AWS_SESSION_TOKEN_ENV_VAR, session_token.to_string())); } + Ok(env) } /// Parses a single clp-s `--print-archive-stats` stdout line into an [`ArchiveMetadata`]. @@ -897,25 +946,24 @@ mod tests { Ok(()) } - #[test] - fn s3_credential_env_default() { - assert_eq!(s3_credential_env(&AwsAuthentication::Default), Vec::new()); - } - #[test] fn s3_credential_env_credentials() { + let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime"); let auth = AwsAuthentication::Credentials { credentials: AwsCredentials { access_key_id: "the-access-key".to_string(), secret_access_key: "the-secret-key".to_string(), + session_token: Some("the-session-token".to_string()), }, }; assert_eq!( - s3_credential_env(&auth), + s3_credential_env(runtime.handle(), "us-east-1", &auth) + .expect("failed to resolve credentials"), vec![ ("AWS_ACCESS_KEY_ID", "the-access-key".to_string()), ("AWS_SECRET_ACCESS_KEY", "the-secret-key".to_string()), + ("AWS_SESSION_TOKEN", "the-session-token".to_string()), ] ); } diff --git a/components/log-ingestor/tests/test_ingestion_job.rs b/components/log-ingestor/tests/test_ingestion_job.rs index f9ad1a327e..86ba913865 100644 --- a/components/log-ingestor/tests/test_ingestion_job.rs +++ b/components/log-ingestor/tests/test_ingestion_job.rs @@ -175,6 +175,7 @@ async fn run_sqs_listener_test( credentials: AwsCredentials { access_key_id: aws_config.access_key_id.clone(), secret_access_key: aws_config.secret_access_key.clone(), + session_token: None, }, }; let sqs_client = clp_rust_utils::sqs::create_new_client( @@ -290,6 +291,7 @@ async fn test_s3_scanner() -> Result<()> { credentials: AwsCredentials { access_key_id: aws_config.access_key_id.clone(), secret_access_key: aws_config.secret_access_key.clone(), + session_token: None, }, }; let s3_client = clp_rust_utils::s3::create_new_client( @@ -368,6 +370,7 @@ async fn test_s3_scanner_single_object() -> Result<()> { credentials: AwsCredentials { access_key_id: aws_config.access_key_id.clone(), secret_access_key: aws_config.secret_access_key.clone(), + session_token: None, }, }; let s3_client = clp_rust_utils::s3::create_new_client( diff --git a/components/log-ingestor/tests/test_scan.rs b/components/log-ingestor/tests/test_scan.rs index 39f5900b66..a49e0affba 100644 --- a/components/log-ingestor/tests/test_scan.rs +++ b/components/log-ingestor/tests/test_scan.rs @@ -24,6 +24,7 @@ async fn create_test_s3_client(aws_config: &AwsConfig) -> aws_sdk_s3::Client { credentials: AwsCredentials { access_key_id: aws_config.access_key_id.clone(), secret_access_key: aws_config.secret_access_key.clone(), + session_token: None, }, }; From 8c4d1d7b5e13f31dbbf7351ac8961a2ba5cc093a Mon Sep 17 00:00:00 2001 From: ChenXing Yang <20001020ycx@gmail.com> Date: Thu, 30 Jul 2026 18:01:11 -0400 Subject: [PATCH 2/5] test(clp-tdl-package): Cover the `default` credential-resolution path in `s3_credential_env`. --- .../src/task/compression/compress.rs | 23 +++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/components/clp-tdl-package/src/task/compression/compress.rs b/components/clp-tdl-package/src/task/compression/compress.rs index 71329ce021..6d58ee6990 100644 --- a/components/clp-tdl-package/src/task/compression/compress.rs +++ b/components/clp-tdl-package/src/task/compression/compress.rs @@ -968,6 +968,29 @@ mod tests { ); } + #[test] + fn s3_credential_env_default() { + let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime"); + // SAFETY: No other test in this binary reads or writes these env vars, and the env + // provider is the first source in the SDK's default chain, so the test is deterministic + // and network-free. + unsafe { + std::env::set_var("AWS_ACCESS_KEY_ID", "the-env-access-key"); + std::env::set_var("AWS_SECRET_ACCESS_KEY", "the-env-secret-key"); + std::env::set_var("AWS_SESSION_TOKEN", "the-env-session-token"); + } + + assert_eq!( + s3_credential_env(runtime.handle(), "us-east-1", &AwsAuthentication::Default) + .expect("failed to resolve credentials"), + vec![ + ("AWS_ACCESS_KEY_ID", "the-env-access-key".to_string()), + ("AWS_SECRET_ACCESS_KEY", "the-env-secret-key".to_string()), + ("AWS_SESSION_TOKEN", "the-env-session-token".to_string()), + ] + ); + } + #[test] fn parse_archive_stats_ignores_extra_keys() { let line = concat!( From 20283bfb029dd3461f21342b95384297368efc71 Mon Sep 17 00:00:00 2001 From: ChenXing Yang <60459812+20001020ycx@users.noreply.github.com> Date: Fri, 31 Jul 2026 13:57:48 -0400 Subject: [PATCH 3/5] Apply suggestions from code review Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com> --- .../src/task/compression/compress.rs | 40 ++++++++++--------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/components/clp-tdl-package/src/task/compression/compress.rs b/components/clp-tdl-package/src/task/compression/compress.rs index 6d58ee6990..dadd96939e 100644 --- a/components/clp-tdl-package/src/task/compression/compress.rs +++ b/components/clp-tdl-package/src/task/compression/compress.rs @@ -350,15 +350,18 @@ fn build_s3_logs_list(input_source: &S3InputSource) -> anyhow::Result { /// /// # Returns /// -/// The env-var name-value pairs containing the resolved access key, secret key, and optional -/// session token. +/// The env-var name-value pairs with the following environment variables set: +/// +/// * AWS_ACCESS_KEY_ID +/// * AWS_SECRET_ACCESS_KEY +/// * AWS_SESSION_TOKEN (if any) /// /// # Errors /// /// Returns an error if: /// /// * The default AWS SDK credential provider chain has no provider. -/// * The default AWS SDK credential provider chain fails to resolve credentials. +/// * Forwards [`ProvideCredentials::provide_credentials`]'s return values on failure. fn s3_credential_env( runtime: &tokio::runtime::Handle, region: &str, @@ -373,13 +376,11 @@ fn s3_credential_env( /// The env var holding the AWS session token. const AWS_SESSION_TOKEN_ENV_VAR: &str = "AWS_SESSION_TOKEN"; - let credentials = match auth { - AwsAuthentication::Credentials { credentials } => Credentials::new( + let (access_key_id, secret_access_key, session_token) = match auth { + AwsAuthentication::Credentials { credentials } => ( credentials.access_key_id.clone(), credentials.secret_access_key.clone(), credentials.session_token.clone(), - None, - "clp-credentials-provider", ), AwsAuthentication::Default => { let sdk_config = runtime.block_on( @@ -390,24 +391,25 @@ fn s3_credential_env( let provider = sdk_config .credentials_provider() .context("default AWS SDK credential provider is unavailable")?; - runtime + let credentials = runtime .block_on(provider.provide_credentials()) - .context("failed to resolve credentials from the default AWS SDK provider chain")? + .context("failed to resolve credentials from the default AWS SDK provider chain")?; + ( + credentials.access_key_id().to_string(), + credentials.secret_access_key().to_string(), + credentials + .session_token() + .map(std::string::ToString::to_string), + ) } }; let mut env = vec![ - ( - AWS_ACCESS_KEY_ID_ENV_VAR, - credentials.access_key_id().to_string(), - ), - ( - AWS_SECRET_ACCESS_KEY_ENV_VAR, - credentials.secret_access_key().to_string(), - ), + (AWS_ACCESS_KEY_ID_ENV_VAR, access_key_id), + (AWS_SECRET_ACCESS_KEY_ENV_VAR, secret_access_key), ]; - if let Some(session_token) = credentials.session_token() { - env.push((AWS_SESSION_TOKEN_ENV_VAR, session_token.to_string())); + if let Some(session_token) = session_token { + env.push((AWS_SESSION_TOKEN_ENV_VAR, session_token)); } Ok(env) } From 33abcf1b30c32b96eea8bf46e7df7b2afd6f7483 Mon Sep 17 00:00:00 2001 From: ChenXing Yang <20001020ycx@gmail.com> Date: Fri, 31 Jul 2026 14:11:25 -0400 Subject: [PATCH 4/5] test(clp-tdl-package): Remove the env-mutating `default`-path unit test; drop the now-unused `Credentials` import. --- .../src/task/compression/compress.rs | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/components/clp-tdl-package/src/task/compression/compress.rs b/components/clp-tdl-package/src/task/compression/compress.rs index dadd96939e..9e715ca302 100644 --- a/components/clp-tdl-package/src/task/compression/compress.rs +++ b/components/clp-tdl-package/src/task/compression/compress.rs @@ -11,7 +11,6 @@ use std::process::Stdio; use anyhow::Context; use aws_config::BehaviorVersion; -use aws_sdk_s3::config::Credentials; use aws_sdk_s3::config::ProvideCredentials; use clp_rust_utils::aws::AWS_DEFAULT_REGION; use clp_rust_utils::clp_config::AwsAuthentication; @@ -970,29 +969,6 @@ mod tests { ); } - #[test] - fn s3_credential_env_default() { - let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime"); - // SAFETY: No other test in this binary reads or writes these env vars, and the env - // provider is the first source in the SDK's default chain, so the test is deterministic - // and network-free. - unsafe { - std::env::set_var("AWS_ACCESS_KEY_ID", "the-env-access-key"); - std::env::set_var("AWS_SECRET_ACCESS_KEY", "the-env-secret-key"); - std::env::set_var("AWS_SESSION_TOKEN", "the-env-session-token"); - } - - assert_eq!( - s3_credential_env(runtime.handle(), "us-east-1", &AwsAuthentication::Default) - .expect("failed to resolve credentials"), - vec![ - ("AWS_ACCESS_KEY_ID", "the-env-access-key".to_string()), - ("AWS_SECRET_ACCESS_KEY", "the-env-secret-key".to_string()), - ("AWS_SESSION_TOKEN", "the-env-session-token".to_string()), - ] - ); - } - #[test] fn parse_archive_stats_ignores_extra_keys() { let line = concat!( From 605288740911ae286204a4c690b0e679ef493953 Mon Sep 17 00:00:00 2001 From: ChenXing Yang <20001020ycx@gmail.com> Date: Fri, 31 Jul 2026 14:20:25 -0400 Subject: [PATCH 5/5] docs(clp-tdl-package): Add missing backticks in `s3_credential_env`'s docstring. --- components/clp-tdl-package/src/task/compression/compress.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/clp-tdl-package/src/task/compression/compress.rs b/components/clp-tdl-package/src/task/compression/compress.rs index 9e715ca302..eda4c04158 100644 --- a/components/clp-tdl-package/src/task/compression/compress.rs +++ b/components/clp-tdl-package/src/task/compression/compress.rs @@ -351,9 +351,9 @@ fn build_s3_logs_list(input_source: &S3InputSource) -> anyhow::Result { /// /// The env-var name-value pairs with the following environment variables set: /// -/// * AWS_ACCESS_KEY_ID -/// * AWS_SECRET_ACCESS_KEY -/// * AWS_SESSION_TOKEN (if any) +/// * `AWS_ACCESS_KEY_ID` +/// * `AWS_SECRET_ACCESS_KEY` +/// * `AWS_SESSION_TOKEN` (if any) /// /// # Errors ///