From 6c654f7707272a20a60e302e6d6d51deefdc8f0c Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 3 Oct 2026 01:30:53 +0800 Subject: [PATCH 1/5] feat(aws): expose shared configuration and region discovery --- services/aws-core/Cargo.toml | 2 + services/aws-core/README.md | 65 +++ services/aws-core/src/config.rs | 248 +++++++++ services/aws-core/src/imds.rs | 139 +++++ services/aws-core/src/lib.rs | 6 + .../aws-core/src/provide_credential/imds.rs | 97 +--- .../src/provide_credential/process.rs | 40 +- .../src/provide_credential/profile.rs | 96 +--- .../aws-core/src/provide_credential/sso.rs | 40 +- services/aws-core/src/region.rs | 109 ++++ services/aws-core/tests/shared_config.rs | 482 ++++++++++++++++++ services/aws-v4/src/lib.rs | 1 + services/aws-v4a/src/lib.rs | 1 + 13 files changed, 1095 insertions(+), 231 deletions(-) create mode 100644 services/aws-core/src/config.rs create mode 100644 services/aws-core/src/imds.rs create mode 100644 services/aws-core/src/region.rs create mode 100644 services/aws-core/tests/shared_config.rs diff --git a/services/aws-core/Cargo.toml b/services/aws-core/Cargo.toml index 4ce24227..15a081db 100644 --- a/services/aws-core/Cargo.toml +++ b/services/aws-core/Cargo.toml @@ -28,6 +28,8 @@ rust-version.workspace = true [dependencies] bytes = { workspace = true } +futures = { workspace = true } +futures-timer = { version = "3", features = ["wasm-bindgen"] } form_urlencoded = { workspace = true } hex = { workspace = true } http = { workspace = true } diff --git a/services/aws-core/README.md b/services/aws-core/README.md index 04ab37dc..017bfe2c 100644 --- a/services/aws-core/README.md +++ b/services/aws-core/README.md @@ -7,3 +7,68 @@ This crate contains the AWS credential type, algorithm-independent credential providers, and canonical request primitives shared by SigV4 and SigV4a. Most users should depend on `reqsign-aws-v4` or `reqsign-aws-v4a`, which re-export the relevant public types. + +## Shared configuration and region discovery + +`SharedConfig` reads the selected AWS profile without acquiring credentials, +executing `credential_process`, or contacting metadata or SSO services. It is +also re-exported by both signing crates and `reqsign::aws`. + +```no_run +use reqsign_aws_core::{IMDSv2RegionProvider, SharedConfig}; +use reqsign_core::{Context, OsEnv}; + +# async fn example(ctx: Context) -> reqsign_core::Result<()> { +// Supply FileRead and HttpSend adapters through Context as needed. +let ctx = ctx.with_env(OsEnv); +let config = SharedConfig::new().with_profile("example"); +let profile = config.load(&ctx).await?; +let endpoint = profile.endpoint_url(); +let configured_region = profile.region(); + +// Metadata fallback is opt-in and independent of authentication. +let region = config + .with_imds(IMDSv2RegionProvider::new()) + .resolve_region(&ctx) + .await?; +# Ok(()) +# } +``` + +Profile and file selection use explicit options first, then `AWS_PROFILE`, +`AWS_CONFIG_FILE`, and `AWS_SHARED_CREDENTIALS_FILE`, then `default`, +`~/.aws/config`, and `~/.aws/credentials`. The Context supplies the home directory, +file reader, and environment. Config files use `[default]` and `[profile name]`; +credentials files use `[default]` and `[name]`. Named profiles do not inherit +`default`. During shared configuration loading, credentials-file properties +replace config-file properties with the same name, preserving other properties. +The static credential provider retains its existing complete-credentials-file +precedence and config-file fallback. + +Region precedence is `with_region`, `AWS_REGION`, `AWS_DEFAULT_REGION`, then the +merged profile's `region`. Global endpoint precedence is `with_endpoint_url`, +`AWS_ENDPOINT_URL`, then the merged profile's `endpoint_url`. `Profile::get` +returns the raw merged file property, before these overrides. Service-specific +endpoint variables, `[services ...]` sections, endpoint rules, and +`AWS_IGNORE_CONFIGURED_ENDPOINT_URLS` are not interpreted. The endpoint is a +setting for the caller to use, not an automatically selected signing endpoint. + +Unreadable files and unavailable home directories are treated as absent. +Malformed readable files return an error without including their contents. +Missing values remain `None`; applications choose their own fallback regions. +Calling `load` never queries metadata, even after `with_imds`. `resolve_region` +skips file reads when an explicit or environment region is available, and only +queries metadata when no configured region is found and discovery is enabled. + +`IMDSv2RegionProvider` performs a token request and an authenticated +`/latest/meta-data/placement/region` request, without requiring an IAM role. It +honors `AWS_EC2_METADATA_DISABLED`, `AWS_EC2_METADATA_SERVICE_ENDPOINT`, and +`AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE=IPv6`. An explicit endpoint takes +precedence. The whole lookup has a configurable two-second deadline and no +retries or IMDSv1 fallback. Disabled discovery returns `None`; transport, +protocol, and timeout failures return errors. The Context HTTP adapter must +cooperate with future cancellation. The timer works independently of the +application's async runtime (using browser timers on wasm32). + +See the AWS documentation for [shared configuration file format](https://docs.aws.amazon.com/sdkref/latest/guide/file-format.html) +and [EC2 metadata categories](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/instancedata-data-categories.html). diff --git a/services/aws-core/src/config.rs b/services/aws-core/src/config.rs new file mode 100644 index 00000000..2edcd3f9 --- /dev/null +++ b/services/aws-core/src/config.rs @@ -0,0 +1,248 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::IMDSv2RegionProvider; +use ini::{Ini, ParseOption}; +use reqsign_core::{Context, Error, Result}; +use std::collections::HashMap; +use std::fmt; + +/// Loads AWS shared configuration without acquiring credentials or making HTTP requests. +/// +/// Selection precedence is explicit profile/path, the corresponding `AWS_PROFILE`, +/// `AWS_CONFIG_FILE`, or `AWS_SHARED_CREDENTIALS_FILE` environment variable, then +/// `default`, `~/.aws/config`, or `~/.aws/credentials`. Named config sections use +/// `[profile name]`; credentials sections use `[name]`. A named profile does not +/// inherit `[default]`. Credentials-file properties override config-file properties +/// individually, while properties present in only one file are retained. +/// +/// Unreadable files (including missing files) and unavailable home directories are +/// treated as absent, as in the credential providers. Malformed readable files are +/// errors. No credential processes, SSO sessions, or role chains are executed. +#[derive(Clone, Debug, Default)] +pub struct SharedConfig { + profile: Option, + config_file: Option, + credentials_file: Option, + region: Option, + endpoint_url: Option, + imds: Option, +} + +impl SharedConfig { + /// Create a loader. Metadata discovery is disabled by default. + pub fn new() -> Self { + Self::default() + } + + /// Select a profile, overriding `AWS_PROFILE`. + pub fn with_profile(mut self, profile: impl Into) -> Self { + self.profile = Some(profile.into()); + self + } + + /// Override `AWS_CONFIG_FILE` and the default config path. + pub fn with_config_file(mut self, path: impl Into) -> Self { + self.config_file = Some(path.into()); + self + } + + /// Override `AWS_SHARED_CREDENTIALS_FILE` and the default credentials path. + pub fn with_credentials_file(mut self, path: impl Into) -> Self { + self.credentials_file = Some(path.into()); + self + } + + /// Override region environment variables and profile settings. + pub fn with_region(mut self, region: impl Into) -> Self { + self.region = Some(region.into()); + self + } + + /// Override `AWS_ENDPOINT_URL` and the profile's global `endpoint_url`. + pub fn with_endpoint_url(mut self, endpoint: impl Into) -> Self { + self.endpoint_url = Some(endpoint.into()); + self + } + + /// Enable metadata fallback for [`Self::resolve_region`] only. + pub fn with_imds(mut self, provider: IMDSv2RegionProvider) -> Self { + self.imds = Some(provider); + self + } + + pub(crate) fn profile_name(&self, ctx: &Context) -> String { + self.profile + .clone() + .or_else(|| ctx.env_var("AWS_PROFILE")) + .unwrap_or_else(|| "default".into()) + } + + fn configured_region(&self, ctx: &Context) -> Option { + self.region + .clone() + .or_else(|| ctx.env_var("AWS_REGION")) + .or_else(|| ctx.env_var("AWS_DEFAULT_REGION")) + } + + pub(crate) async fn load_config_file(&self, ctx: &Context) -> Result { + read_ini( + ctx, + self.config_file.as_deref(), + "AWS_CONFIG_FILE", + "~/.aws/config", + ) + .await + } + + pub(crate) async fn load_credentials_file(&self, ctx: &Context) -> Result { + read_ini( + ctx, + self.credentials_file.as_deref(), + "AWS_SHARED_CREDENTIALS_FILE", + "~/.aws/credentials", + ) + .await + } + + /// Read and merge the selected profile. This never contacts IMDS, even when enabled. + /// + /// Region precedence: explicit override, `AWS_REGION`, `AWS_DEFAULT_REGION`, + /// merged profile. Endpoint precedence: explicit override, `AWS_ENDPOINT_URL`, + /// merged profile. Missing values remain `None`; no application defaults are added. + /// Service-specific endpoint settings and endpoint rules are not resolved. + pub async fn load(&self, ctx: &Context) -> Result { + let name = self.profile_name(ctx); + let config = self.load_config_file(ctx).await?; + let credentials = self.load_credentials_file(ctx).await?; + let mut properties = HashMap::new(); + for section in [ + config.section(Some(config_section(&name))), + credentials.section(Some(&name)), + ] + .into_iter() + .flatten() + { + properties.extend(section.iter().map(|(k, v)| (k.to_owned(), v.to_owned()))); + } + let region = self + .configured_region(ctx) + .or_else(|| properties.get("region").cloned()); + let endpoint_url = self + .endpoint_url + .clone() + .or_else(|| ctx.env_var("AWS_ENDPOINT_URL")) + .or_else(|| properties.get("endpoint_url").cloned()); + Ok(Profile { + name, + properties, + region, + endpoint_url, + }) + } + + /// Resolve region independently of credentials, optionally falling back to IMDSv2. + /// + /// Explicit/environment regions skip file reads. A profile region skips metadata. + /// Disabled discovery or an absent region returns `None`; configuration, metadata, + /// and timeout errors are returned to the caller without an application fallback. + pub async fn resolve_region(&self, ctx: &Context) -> Result> { + if let Some(region) = self.configured_region(ctx) { + return Ok(Some(region)); + } + if let Some(region) = self.load(ctx).await?.region { + return Ok(Some(region)); + } + match &self.imds { + Some(imds) => imds.resolve_region(ctx).await, + None => Ok(None), + } + } +} + +/// A selected AWS profile with resolved region and global endpoint settings. +/// +/// Raw properties may contain secrets. Debug output omits all property values. +#[derive(Clone)] +pub struct Profile { + name: String, + properties: HashMap, + region: Option, + endpoint_url: Option, +} + +impl fmt::Debug for Profile { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Profile") + .field("name", &self.name) + .finish_non_exhaustive() + } +} + +impl Profile { + /// Selected profile name. + pub fn name(&self) -> &str { + &self.name + } + /// Merged file property, before explicit or environment overrides. + pub fn get(&self, key: &str) -> Option<&str> { + self.properties.get(key).map(String::as_str) + } + /// Region after explicit and environment overrides, without metadata discovery. + pub fn region(&self) -> Option<&str> { + self.region.as_deref() + } + /// Global endpoint after explicit and environment overrides. No service rules are applied. + pub fn endpoint_url(&self) -> Option<&str> { + self.endpoint_url.as_deref() + } +} + +pub(crate) fn config_section(profile: &str) -> String { + if profile == "default" { + "default".into() + } else { + format!("profile {profile}") + } +} + +async fn read_ini(ctx: &Context, explicit: Option<&str>, env: &str, default: &str) -> Result { + let path = explicit + .map(str::to_owned) + .or_else(|| ctx.env_var(env)) + .unwrap_or_else(|| default.into()); + let Some(path) = ctx.expand_home_dir(&path) else { + return Ok(Ini::new()); + }; + let Ok(content) = ctx.file_read(&path).await else { + return Ok(Ini::new()); + }; + let content = std::str::from_utf8(&content) + .map_err(|_| Error::config_invalid("AWS shared configuration is not UTF-8"))?; + // AWS values, especially credential_process commands, contain literal quotes and backslashes. + // Keep nested service settings distinct from top-level profile properties. + Ini::load_from_str_opt( + content, + ParseOption { + enabled_quote: false, + enabled_escape: false, + enabled_indented_mutiline_value: true, + ..Default::default() + }, + ) + .map_err(|_| Error::config_invalid("failed to parse AWS shared configuration")) +} diff --git a/services/aws-core/src/imds.rs b/services/aws-core/src/imds.rs new file mode 100644 index 00000000..13f4039f --- /dev/null +++ b/services/aws-core/src/imds.rs @@ -0,0 +1,139 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::provide_credential::utils::parse_imds_error; +use bytes::Bytes; +use http::Method; +use http::header::CONTENT_LENGTH; +use reqsign_core::time::Timestamp; +use reqsign_core::{Context, Error, Result}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +#[derive(Clone)] +pub(crate) struct ImdsClient { + endpoint: Option, + token: Arc>, +} + +impl Default for ImdsClient { + fn default() -> Self { + Self { + endpoint: None, + token: Arc::new(Mutex::new(( + String::new(), + String::new(), + Timestamp::default(), + ))), + } + } +} + +impl ImdsClient { + /// Create a new `ImdsClient` instance. + pub fn new() -> Self { + Self::default() + } + + /// Set the endpoint for the metadata service. + pub fn with_endpoint(mut self, endpoint: impl Into) -> Self { + self.endpoint = Some(endpoint.into()); + self + } +} + +impl ImdsClient { + pub(crate) fn get_endpoint(&self, ctx: &Context) -> String { + // First check configured endpoint, then environment, then default + self.endpoint.clone().unwrap_or_else(|| { + ctx.env_var("AWS_EC2_METADATA_SERVICE_ENDPOINT") + .unwrap_or_else(|| { + if ctx + .env_var("AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE") + .is_some_and(|v| v.eq_ignore_ascii_case("IPv6")) + { + "http://[fd00:ec2::254]".into() + } else { + "http://169.254.169.254".into() + } + }) + }) + } + + pub(crate) async fn load_ec2_metadata_token(&self, ctx: &Context) -> Result { + let endpoint = self.get_endpoint(ctx); + { + let (cached_endpoint, token, expires_in) = + self.token.lock().expect("lock poisoned").clone(); + if cached_endpoint == endpoint && expires_in > Timestamp::now() { + return Ok(token); + } + } + + let url = format!("{}/latest/api/token", endpoint.trim_end_matches('/')); + let req = http::Request::builder() + .uri(&url) + .method(Method::PUT) + .header(CONTENT_LENGTH, "0") + // 21600s (6h) is recommended by AWS. + .header("x-aws-ec2-metadata-token-ttl-seconds", "21600") + .body(Bytes::new()) + .map_err(|e| { + Error::request_invalid("failed to build IMDS token request") + .with_source(e) + .with_context(format!("url: {url}")) + })?; + + let resp = ctx.http_send_as_string(req).await.map_err(|e| { + Error::unexpected("failed to connect to IMDS") + .with_source(e) + .with_context(format!("endpoint: {endpoint}")) + .with_context("hint: check if running on EC2 instance") + .set_retryable(true) + })?; + + if resp.status() != http::StatusCode::OK { + return Err(parse_imds_error( + "fetch_imds_token", + resp.status(), + resp.body(), + )); + } + let ec2_token = resp.into_body(); + // Refresh ten minutes before the token expires. + let expires_in = Timestamp::now() + Duration::from_secs(21600) - Duration::from_secs(600); + + { + *self.token.lock().expect("lock poisoned") = (endpoint, ec2_token.clone(), expires_in); + } + + Ok(ec2_token) + } +} + +impl std::fmt::Debug for ImdsClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ImdsClient") + .field("endpoint", &self.endpoint) + .finish_non_exhaustive() + } +} + +pub(crate) fn disabled(ctx: &Context) -> bool { + ctx.env_var("AWS_EC2_METADATA_DISABLED") + .is_some_and(|v| v.eq_ignore_ascii_case("true")) +} diff --git a/services/aws-core/src/lib.rs b/services/aws-core/src/lib.rs index 498744e8..ec6db1bf 100644 --- a/services/aws-core/src/lib.rs +++ b/services/aws-core/src/lib.rs @@ -19,6 +19,12 @@ pub mod constants; +mod config; +pub use config::{Profile, SharedConfig}; +mod imds; +mod region; +pub use region::IMDSv2RegionProvider; + #[doc(hidden)] pub mod assume_role; pub use assume_role::AssumeRoleGrant; diff --git a/services/aws-core/src/provide_credential/imds.rs b/services/aws-core/src/provide_credential/imds.rs index 31fd03d9..989b991b 100644 --- a/services/aws-core/src/provide_credential/imds.rs +++ b/services/aws-core/src/provide_credential/imds.rs @@ -16,29 +16,16 @@ // under the License. use crate::Credential; +use crate::imds::ImdsClient; use crate::provide_credential::utils::parse_imds_error; use bytes::Bytes; use http::Method; -use http::header::CONTENT_LENGTH; -use reqsign_core::time::Timestamp; use reqsign_core::{Context, Error, ProvideCredential, Result}; use serde::Deserialize; -use std::sync::{Arc, Mutex}; -use std::time::Duration; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct IMDSv2CredentialProvider { - endpoint: Option, - token: Arc>, -} - -impl Default for IMDSv2CredentialProvider { - fn default() -> Self { - Self { - endpoint: None, - token: Arc::new(Mutex::new((String::new(), Timestamp::default()))), - } - } + client: ImdsClient, } impl IMDSv2CredentialProvider { @@ -46,93 +33,25 @@ impl IMDSv2CredentialProvider { pub fn new() -> Self { Self::default() } - /// Set the endpoint for the metadata service. pub fn with_endpoint(mut self, endpoint: impl Into) -> Self { - self.endpoint = Some(endpoint.into()); + self.client = self.client.with_endpoint(endpoint); self } } -impl IMDSv2CredentialProvider { - fn get_endpoint(&self, ctx: &Context) -> String { - // First check configured endpoint, then environment, then default - self.endpoint.clone().unwrap_or_else(|| { - ctx.env_vars() - .get("AWS_EC2_METADATA_SERVICE_ENDPOINT") - .cloned() - .unwrap_or_else(|| "http://169.254.169.254".into()) - }) - } - - async fn load_ec2_metadata_token(&self, ctx: &Context) -> Result { - { - let (token, expires_in) = self.token.lock().expect("lock poisoned").clone(); - if expires_in > Timestamp::now() { - return Ok(token); - } - } - - let endpoint = self.get_endpoint(ctx); - let url = format!("{endpoint}/latest/api/token"); - let req = http::Request::builder() - .uri(&url) - .method(Method::PUT) - .header(CONTENT_LENGTH, "0") - // 21600s (6h) is recommended by AWS. - .header("x-aws-ec2-metadata-token-ttl-seconds", "21600") - .body(Bytes::new()) - .map_err(|e| { - Error::request_invalid("failed to build IMDS token request") - .with_source(e) - .with_context(format!("url: {url}")) - })?; - - let resp = ctx.http_send_as_string(req).await.map_err(|e| { - Error::unexpected("failed to connect to IMDS") - .with_source(e) - .with_context("endpoint: {endpoint}") - .with_context("hint: check if running on EC2 instance") - .set_retryable(true) - })?; - - if resp.status() != http::StatusCode::OK { - return Err(parse_imds_error( - "fetch_imds_token", - resp.status(), - resp.body(), - )); - } - let ec2_token = resp.into_body(); - // Set expires_in to 10 minutes to enforce re-read. - let expires_in = Timestamp::now() + Duration::from_secs(21600) - Duration::from_secs(600); - - { - *self.token.lock().expect("lock poisoned") = (ec2_token.clone(), expires_in); - } - - Ok(ec2_token) - } -} impl ProvideCredential for IMDSv2CredentialProvider { type Credential = Credential; async fn provide_credential(&self, ctx: &Context) -> Result> { - // Check if disabled via environment - let disabled_env = ctx - .env_vars() - .get("AWS_EC2_METADATA_DISABLED") - .map(|v| v == "true") - .unwrap_or(false); - - if disabled_env { + if crate::imds::disabled(ctx) { return Ok(None); } - let token = self.load_ec2_metadata_token(ctx).await?; + let token = self.client.load_ec2_metadata_token(ctx).await?; // List all credentials that node has. - let endpoint = self.get_endpoint(ctx); + let endpoint = self.client.get_endpoint(ctx); let url = format!("{endpoint}/latest/meta-data/iam/security-credentials/"); let req = http::Request::builder() .uri(&url) @@ -171,7 +90,7 @@ impl ProvideCredential for IMDSv2CredentialProvider { } // Get the credentials via role_name. - let endpoint = self.get_endpoint(ctx); + let endpoint = self.client.get_endpoint(ctx); let url = format!("{endpoint}/latest/meta-data/iam/security-credentials/{profile_name}"); let req = http::Request::builder() .uri(url) diff --git a/services/aws-core/src/provide_credential/process.rs b/services/aws-core/src/provide_credential/process.rs index 2fe482f7..29506357 100644 --- a/services/aws-core/src/provide_credential/process.rs +++ b/services/aws-core/src/provide_credential/process.rs @@ -16,7 +16,6 @@ // under the License. use crate::Credential; -use ini::Ini; use log::debug; use reqsign_core::{Context, Error, ProvideCredential, Result}; use serde::Deserialize; @@ -84,42 +83,17 @@ impl ProcessCredentialProvider { } // Otherwise, load from config file - // Priority: 1. self.profile, 2. AWS_PROFILE env var, 3. "default" - let profile_name = self - .profile - .clone() - .or_else(|| ctx.env_var("AWS_PROFILE")) - .unwrap_or_else(|| "default".to_string()); + let mut shared = crate::SharedConfig::new(); + if let Some(profile) = &self.profile { + shared = shared.with_profile(profile); + } + let profile_name = shared.profile_name(ctx); self.load_command_from_config(ctx, &profile_name).await } async fn load_command_from_config(&self, ctx: &Context, profile: &str) -> Result { - // Load AWS config file - let config_path = ctx - .env_var("AWS_CONFIG_FILE") - .unwrap_or_else(|| "~/.aws/config".to_string()); - - let expanded_path = if config_path.starts_with("~/") { - match ctx.expand_home_dir(&config_path) { - Some(expanded) => expanded, - None => return Err(Error::config_invalid("failed to expand home directory")), - } - } else { - config_path - }; - - let content = ctx.file_read(&expanded_path).await.map_err(|_| { - Error::config_invalid(format!("failed to read config file: {expanded_path}")) - })?; - - let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)) - .map_err(|e| Error::config_invalid(format!("failed to parse config file: {e}")))?; - - let profile_section = if profile == "default" { - profile.to_string() - } else { - format!("profile {profile}") - }; + let conf = crate::SharedConfig::new().load_config_file(ctx).await?; + let profile_section = crate::config::config_section(profile); let section = conf.section(Some(profile_section)).ok_or_else(|| { Error::config_invalid(format!("profile '{profile}' not found in config")) diff --git a/services/aws-core/src/provide_credential/profile.rs b/services/aws-core/src/provide_credential/profile.rs index 5221a43a..4943182c 100644 --- a/services/aws-core/src/provide_credential/profile.rs +++ b/services/aws-core/src/provide_credential/profile.rs @@ -16,14 +16,10 @@ // under the License. use crate::Credential; -#[cfg(not(target_arch = "wasm32"))] +#[cfg(test)] use crate::constants::*; #[cfg(not(target_arch = "wasm32"))] -use ini::Ini; -#[cfg(not(target_arch = "wasm32"))] use log::debug; -#[cfg(not(target_arch = "wasm32"))] -use reqsign_core::Error; use reqsign_core::{Context, ProvideCredential, Result}; /// ProfileCredentialProvider loads AWS credentials from configuration files. @@ -77,43 +73,28 @@ impl ProfileCredentialProvider { self } + #[cfg(not(target_arch = "wasm32"))] + fn shared_config(&self) -> crate::SharedConfig { + let mut config = crate::SharedConfig::new(); + if let Some(v) = &self.profile { + config = config.with_profile(v); + } + if let Some(v) = &self.config_file { + config = config.with_config_file(v); + } + if let Some(v) = &self.credentials_file { + config = config.with_credentials_file(v); + } + config + } + #[cfg(not(target_arch = "wasm32"))] async fn load_from_credentials_file( &self, ctx: &Context, profile: &str, ) -> Result> { - let path = if let Some(path) = &self.credentials_file { - path.clone() - } else if let Some(path) = ctx.env_var(AWS_SHARED_CREDENTIALS_FILE) { - path - } else { - "~/.aws/credentials".to_string() - }; - - let expanded_path = if path.starts_with("~/") { - match ctx.expand_home_dir(&path) { - Some(expanded) => expanded, - None => { - debug!("failed to expand homedir for path: {path}"); - return Ok(None); - } - } - } else { - path.clone() - }; - - let content = match ctx.file_read(&expanded_path).await { - Ok(content) => content, - Err(err) => { - debug!("failed to read credentials file {expanded_path}: {err:?}"); - return Ok(None); - } - }; - - let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)).map_err(|e| { - Error::config_invalid("failed to parse credentials file").with_source(e) - })?; + let conf = self.shared_config().load_credentials_file(ctx).await?; let props = match conf.section(Some(profile)) { Some(props) => props, @@ -143,41 +124,8 @@ impl ProfileCredentialProvider { ctx: &Context, profile: &str, ) -> Result> { - let path = if let Some(path) = &self.config_file { - path.clone() - } else if let Some(path) = ctx.env_var(AWS_CONFIG_FILE) { - path - } else { - "~/.aws/config".to_string() - }; - - let expanded_path = if path.starts_with("~/") { - match ctx.expand_home_dir(&path) { - Some(expanded) => expanded, - None => { - debug!("failed to expand homedir for path: {path}"); - return Ok(None); - } - } - } else { - path.clone() - }; - - let content = match ctx.file_read(&expanded_path).await { - Ok(content) => content, - Err(err) => { - debug!("failed to read config file {expanded_path}: {err:?}"); - return Ok(None); - } - }; - - let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)) - .map_err(|e| Error::config_invalid("failed to parse config file").with_source(e))?; - - let section = match profile { - "default" => "default".to_string(), - x => format!("profile {x}"), - }; + let conf = self.shared_config().load_config_file(ctx).await?; + let section = crate::config::config_section(profile); let props = match conf.section(Some(§ion)) { Some(props) => props, @@ -213,11 +161,7 @@ impl ProvideCredential for ProfileCredentialProvider { #[cfg(not(target_arch = "wasm32"))] { - let profile = self - .profile - .clone() - .or_else(|| ctx.env_var(AWS_PROFILE)) - .unwrap_or_else(|| "default".to_string()); + let profile = self.shared_config().profile_name(ctx); // Try credentials file first if let Some(cred) = self.load_from_credentials_file(ctx, &profile).await? { diff --git a/services/aws-core/src/provide_credential/sso.rs b/services/aws-core/src/provide_credential/sso.rs index 9fb6f287..da9fceb0 100644 --- a/services/aws-core/src/provide_credential/sso.rs +++ b/services/aws-core/src/provide_credential/sso.rs @@ -17,7 +17,6 @@ use crate::Credential; use http::{Method, Request, StatusCode}; -use ini::Ini; use log::{debug, warn}; use reqsign_core::time::Timestamp; use reqsign_core::{Context, Error, ProvideCredential, Result}; @@ -126,42 +125,17 @@ impl SSOCredentialProvider { } // Otherwise, load from config file - // Priority: 1. self.profile, 2. AWS_PROFILE env var, 3. "default" - let profile_name = self - .profile - .clone() - .or_else(|| ctx.env_var("AWS_PROFILE")) - .unwrap_or_else(|| "default".to_string()); + let mut shared = crate::SharedConfig::new(); + if let Some(profile) = &self.profile { + shared = shared.with_profile(profile); + } + let profile_name = shared.profile_name(ctx); self.load_from_config_file(ctx, &profile_name).await } async fn load_from_config_file(&self, ctx: &Context, profile: &str) -> Result { - // Load AWS config file - let config_path = ctx - .env_var("AWS_CONFIG_FILE") - .unwrap_or_else(|| "~/.aws/config".to_string()); - - let expanded_path = if config_path.starts_with("~/") { - match ctx.expand_home_dir(&config_path) { - Some(expanded) => expanded, - None => return Err(Error::config_invalid("failed to expand home directory")), - } - } else { - config_path - }; - - let content = ctx.file_read(&expanded_path).await.map_err(|_| { - Error::config_invalid(format!("failed to read config file: {expanded_path}")) - })?; - - let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)) - .map_err(|e| Error::config_invalid(format!("failed to parse config file: {e}")))?; - - let profile_section = if profile == "default" { - profile.to_string() - } else { - format!("profile {profile}") - }; + let conf = crate::SharedConfig::new().load_config_file(ctx).await?; + let profile_section = crate::config::config_section(profile); let section = conf.section(Some(profile_section)).ok_or_else(|| { Error::config_invalid(format!("profile '{profile}' not found in config")) diff --git a/services/aws-core/src/region.rs b/services/aws-core/src/region.rs new file mode 100644 index 00000000..f2fea13e --- /dev/null +++ b/services/aws-core/src/region.rs @@ -0,0 +1,109 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::imds::ImdsClient; +use bytes::Bytes; +use futures::future::{Either, select}; +use reqsign_core::{Context, Error, Result}; +use std::time::Duration; + +/// Discovers an EC2 region without obtaining IAM credentials. +/// +/// Uses IMDSv2 only: one token request followed by `meta-data/placement/region`. +/// `AWS_EC2_METADATA_DISABLED=true` disables discovery. Endpoint precedence is +/// explicit endpoint, `AWS_EC2_METADATA_SERVICE_ENDPOINT`, then the IPv4 metadata +/// endpoint (or IPv6 when `AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE=IPv6`). +/// +/// The entire discovery has a default two-second deadline with no retries. HTTP +/// requests use the supplied Context. Timeout cancels the in-flight HTTP future; +/// the HTTP adapter must cooperate with cancellation and must not block polling. +#[derive(Debug, Clone)] +pub struct IMDSv2RegionProvider { + client: ImdsClient, + timeout: Duration, +} + +impl Default for IMDSv2RegionProvider { + fn default() -> Self { + Self { + client: ImdsClient::new(), + timeout: Duration::from_secs(2), + } + } +} + +impl IMDSv2RegionProvider { + /// Create a region provider with a two-second deadline. + pub fn new() -> Self { + Self::default() + } + /// Override the metadata service endpoint. + pub fn with_endpoint(mut self, endpoint: impl Into) -> Self { + self.client = self.client.with_endpoint(endpoint); + self + } + /// Set the deadline for the whole discovery, including the token request. + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + /// Return the EC2 region, or `None` when metadata is disabled. + /// + /// Transport errors, non-success responses, empty/invalid region responses, and + /// deadline expiry return errors. No credentials or instance role are required. + pub async fn resolve_region(&self, ctx: &Context) -> Result> { + if crate::imds::disabled(ctx) { + return Ok(None); + } + let discovery = Box::pin(self.discover(ctx)); + match select(discovery, futures_timer::Delay::new(self.timeout)).await { + Either::Left((result, _)) => result.map(Some), + Either::Right(_) => { + Err(Error::unexpected("IMDS region discovery timed out").set_retryable(true)) + } + } + } + + async fn discover(&self, ctx: &Context) -> Result { + let token = self.client.load_ec2_metadata_token(ctx).await?; + let endpoint = self.client.get_endpoint(ctx); + let request = http::Request::builder() + .uri(format!( + "{}/latest/meta-data/placement/region", + endpoint.trim_end_matches('/') + )) + .header("x-aws-ec2-metadata-token", token) + .body(Bytes::new()) + .map_err(|_| Error::request_invalid("failed to build IMDS region request"))?; + let response = ctx.http_send(request).await?; + if response.status() != http::StatusCode::OK { + return Err(Error::unexpected("failed to fetch IMDS region") + .with_context(format!("status: {}", response.status()))); + } + let region = std::str::from_utf8(response.body()) + .map_err(|_| Error::unexpected("invalid IMDS region response"))? + .trim(); + if region.is_empty() + || !region + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-') + { + return Err(Error::unexpected("invalid IMDS region response")); + } + Ok(region.to_owned()) + } +} diff --git a/services/aws-core/tests/shared_config.rs b/services/aws-core/tests/shared_config.rs new file mode 100644 index 00000000..fefd090e --- /dev/null +++ b/services/aws-core/tests/shared_config.rs @@ -0,0 +1,482 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use bytes::Bytes; +use reqsign_aws_core::{IMDSv2RegionProvider, SharedConfig}; +use reqsign_core::{Context, Error, ErrorKind, FileRead, HttpSend, Result, StaticEnv}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +#[derive(Debug, Clone, Default)] +struct Files(HashMap>); +impl FileRead for Files { + async fn file_read(&self, path: &str) -> Result> { + self.0 + .get(path) + .cloned() + .ok_or_else(|| Error::unexpected("file absent")) + } +} + +#[derive(Debug)] +struct NoIo; +impl HttpSend for NoIo { + async fn http_send(&self, _: http::Request) -> Result> { + panic!("unexpected network discovery") + } +} +impl FileRead for NoIo { + async fn file_read(&self, _: &str) -> Result> { + panic!("unexpected file read") + } +} +impl reqsign_core::CommandExecute for NoIo { + async fn command_execute(&self, _: &str, _: &[&str]) -> Result { + panic!("unexpected credential process") + } +} + +fn context(files: &[(&str, &str)], env: &[(&str, &str)]) -> Context { + Context::new() + .with_file_read(Files( + files + .iter() + .map(|(k, v)| (k.to_string(), v.as_bytes().to_vec())) + .collect(), + )) + .with_env(StaticEnv { + home_dir: Some("/test-home".into()), + envs: env + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + }) + .with_http_send(NoIo) + .with_command_execute(NoIo) +} + +#[tokio::test] +async fn selection_merge_and_no_side_effects() -> Result<()> { + let ctx = context( + &[ + ( + "/test-home/.aws/config", + "[default]\nregion = default-region\n[profile example]\nregion = config-region\nendpoint_url = https://s3.example.invalid\ncredential_process = forbidden\nsso_session = forbidden\nrole_arn = forbidden\naws_secret_access_key = secret-value\n", + ), + ( + "/test-home/.aws/credentials", + "[example]\nregion = credentials-region\naws_access_key_id = key\n", + ), + ], + &[("AWS_PROFILE", "example")], + ); + let config = SharedConfig::new().with_imds(IMDSv2RegionProvider::new()); + let profile = config.load(&ctx).await?; + assert_eq!(profile.name(), "example"); + assert_eq!(profile.region(), Some("credentials-region")); + assert_eq!(profile.endpoint_url(), Some("https://s3.example.invalid")); + assert_eq!(profile.get("aws_access_key_id"), Some("key")); + assert_eq!(profile.get("aws_secret_access_key"), Some("secret-value")); + assert!(!format!("{profile:?}").contains("secret-value")); + assert_eq!( + config.resolve_region(&ctx).await?.as_deref(), + Some("credentials-region") + ); + let explicit = SharedConfig::new() + .with_profile("default") + .load(&ctx) + .await?; + assert_eq!(explicit.region(), Some("default-region")); + let missing = SharedConfig::new() + .with_profile("missing") + .load(&ctx) + .await?; + assert_eq!(missing.region(), None); + assert_eq!(missing.endpoint_url(), None); + Ok(()) +} + +#[tokio::test] +async fn default_profile_and_alternate_paths() -> Result<()> { + let ctx = context( + &[ + ( + "/config", + "[default]\nregion = config-region\nendpoint_url = https://config.invalid\n", + ), + ("/credentials", "[default]\nregion = credentials-region\n"), + ("/explicit-config", "[default]\nregion = explicit-region\n"), + ( + "/explicit-credentials", + "[default]\nendpoint_url = https://explicit.invalid\n", + ), + ], + &[ + ("AWS_CONFIG_FILE", "/config"), + ("AWS_SHARED_CREDENTIALS_FILE", "/credentials"), + ], + ); + let p = SharedConfig::new().load(&ctx).await?; + assert_eq!(p.name(), "default"); + assert_eq!(p.region(), Some("credentials-region")); + assert_eq!(p.endpoint_url(), Some("https://config.invalid")); + let p = SharedConfig::new() + .with_config_file("/explicit-config") + .with_credentials_file("/explicit-credentials") + .load(&ctx) + .await?; + assert_eq!(p.region(), Some("explicit-region")); + assert_eq!(p.endpoint_url(), Some("https://explicit.invalid")); + Ok(()) +} + +#[tokio::test] +async fn override_and_environment_precedence() -> Result<()> { + let ctx = context( + &[( + "/test-home/.aws/config", + "[default]\nregion = profile-region\nendpoint_url = https://profile.invalid\n", + )], + &[ + ("AWS_REGION", "env-region"), + ("AWS_DEFAULT_REGION", "env-default-region"), + ("AWS_ENDPOINT_URL", "https://env.invalid"), + ], + ); + let p = SharedConfig::new().load(&ctx).await?; + assert_eq!(p.region(), Some("env-region")); + assert_eq!(p.get("region"), Some("profile-region")); + assert_eq!(p.endpoint_url(), Some("https://env.invalid")); + let p = SharedConfig::new() + .with_region("explicit-region") + .with_endpoint_url("https://explicit.invalid") + .load(&ctx) + .await?; + assert_eq!(p.region(), Some("explicit-region")); + assert_eq!(p.endpoint_url(), Some("https://explicit.invalid")); + let ctx = ctx.with_file_read(NoIo); + assert_eq!( + SharedConfig::new().resolve_region(&ctx).await?.as_deref(), + Some("env-region") + ); + assert_eq!( + SharedConfig::new() + .with_region("explicit") + .resolve_region(&ctx) + .await? + .as_deref(), + Some("explicit") + ); + let ctx = context(&[], &[("AWS_DEFAULT_REGION", "fallback")]).with_file_read(NoIo); + assert_eq!( + SharedConfig::new().resolve_region(&ctx).await?.as_deref(), + Some("fallback") + ); + Ok(()) +} + +#[tokio::test] +async fn missing_malformed_and_literal_values() -> Result<()> { + let ctx = context(&[], &[]); + assert!(SharedConfig::new().load(&ctx).await?.region().is_none()); + assert!(SharedConfig::new().resolve_region(&ctx).await?.is_none()); + let ctx = ctx + .with_env(StaticEnv { + home_dir: None, + envs: HashMap::new(), + }) + .with_file_read(NoIo); + assert!(SharedConfig::new().load(&ctx).await?.region().is_none()); + for path in ["/test-home/.aws/config", "/test-home/.aws/credentials"] { + let ctx = context(&[(path, "[unterminated-secret-value")], &[]); + let err = SharedConfig::new().load(&ctx).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::ConfigInvalid); + assert!(!format!("{err:?}").contains("secret-value")); + } + let ctx = context( + &[( + "/test-home/.aws/config", + r#"[default] +credential_process = "C:\Program Files\helper.exe" "argument with spaces" +s3 = + endpoint_url = https://nested.invalid +region = us-east-1 +"#, + )], + &[], + ); + let p = SharedConfig::new().load(&ctx).await?; + assert_eq!( + p.get("credential_process"), + Some(r#""C:\Program Files\helper.exe" "argument with spaces""#) + ); + assert_eq!(p.region(), Some("us-east-1")); + assert_eq!(p.endpoint_url(), None); + Ok(()) +} + +#[derive(Debug, Clone)] +struct Metadata { + requests: Arc>>, + fail_at: Option<&'static str>, + pending_at: Option<&'static str>, + region: &'static str, +} +impl Default for Metadata { + fn default() -> Self { + Self { + requests: Default::default(), + fail_at: None, + pending_at: None, + region: "us-east-2\n", + } + } +} +impl HttpSend for Metadata { + async fn http_send(&self, req: http::Request) -> Result> { + let path = req.uri().path(); + self.requests.lock().unwrap().push(req.uri().to_string()); + if self.pending_at == Some(path) { + return std::future::pending().await; + } + if self.fail_at == Some(path) { + return Ok(http::Response::builder() + .status(503) + .body(Bytes::from_static(b"unavailable")) + .unwrap()); + } + let body = match path { + "/latest/api/token" => { + assert_eq!(req.method(), http::Method::PUT); + assert_eq!( + req.headers()["x-aws-ec2-metadata-token-ttl-seconds"], + "21600" + ); + "secret-imds-token" + } + "/latest/meta-data/placement/region" => { + assert_eq!(req.method(), http::Method::GET); + assert_eq!( + req.headers()["x-aws-ec2-metadata-token"], + "secret-imds-token" + ); + self.region + } + "/latest/meta-data/iam/security-credentials/" => "test-role", + "/latest/meta-data/iam/security-credentials/test-role" => { + assert_eq!( + req.headers()["x-aws-ec2-metadata-token"], + "secret-imds-token" + ); + r#"{"Code":"Success","AccessKeyId":"key","SecretAccessKey":"secret","Token":"session","Expiration":"2099-01-01T00:00:00Z"}"# + } + _ => panic!("unexpected metadata lookup: {path}"), + }; + Ok(http::Response::new(Bytes::from(body))) + } +} + +#[tokio::test] +async fn metadata_discovery_and_endpoint_selection() -> Result<()> { + for (explicit, env, host) in [ + (None, vec![], "http://169.254.169.254"), + ( + None, + vec![("AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE", "IPv6")], + "http://[fd00:ec2::254]", + ), + ( + None, + vec![("AWS_EC2_METADATA_SERVICE_ENDPOINT", "http://env.invalid")], + "http://env.invalid", + ), + ( + Some("http://explicit.invalid/"), + vec![("AWS_EC2_METADATA_SERVICE_ENDPOINT", "http://env.invalid")], + "http://explicit.invalid", + ), + ] { + let http = Metadata::default(); + let ctx = context(&[], &env).with_http_send(http.clone()); + let mut provider = IMDSv2RegionProvider::new(); + if let Some(endpoint) = explicit { + provider = provider.with_endpoint(endpoint); + } + let config = SharedConfig::new().with_imds(provider.clone()); + assert!(config.load(&ctx).await?.region().is_none()); + assert!(http.requests.lock().unwrap().is_empty()); + assert_eq!( + config.resolve_region(&ctx).await?.as_deref(), + Some("us-east-2") + ); + assert_eq!( + *http.requests.lock().unwrap(), + vec![ + format!("{host}/latest/api/token"), + format!("{host}/latest/meta-data/placement/region") + ] + ); + assert!(!format!("{provider:?}").contains("secret-imds-token")); + } + Ok(()) +} + +#[tokio::test] +async fn disabled_metadata_and_missing_region() -> Result<()> { + let ctx = context(&[], &[("AWS_EC2_METADATA_DISABLED", "TRUE")]); + let config = SharedConfig::new().with_imds(IMDSv2RegionProvider::new()); + assert!(config.resolve_region(&ctx).await?.is_none()); + Ok(()) +} + +#[tokio::test] +async fn metadata_failures_and_deadline_cover_both_requests() { + for path in ["/latest/api/token", "/latest/meta-data/placement/region"] { + let http = Metadata { + fail_at: Some(path), + ..Default::default() + }; + let ctx = context(&[], &[]).with_http_send(http.clone()); + assert!( + IMDSv2RegionProvider::new() + .resolve_region(&ctx) + .await + .is_err() + ); + assert!( + http.requests + .lock() + .unwrap() + .last() + .unwrap() + .ends_with(path) + ); + let http = Metadata { + pending_at: Some(path), + ..Default::default() + }; + let ctx = context(&[], &[]).with_http_send(http.clone()); + let err = IMDSv2RegionProvider::new() + .with_timeout(Duration::from_millis(10)) + .resolve_region(&ctx) + .await + .unwrap_err(); + assert!(err.to_string().contains("timed out")); + assert!( + http.requests + .lock() + .unwrap() + .last() + .unwrap() + .ends_with(path) + ); + } + for region in ["", "\n", "invalid region"] { + let ctx = context(&[], &[]).with_http_send(Metadata { + region, + ..Default::default() + }); + let err = IMDSv2RegionProvider::new() + .resolve_region(&ctx) + .await + .unwrap_err(); + assert!(err.to_string().contains("invalid IMDS region")); + } +} + +#[cfg(not(target_arch = "wasm32"))] +#[tokio::test] +async fn credential_provider_uses_the_same_profile_and_files() -> Result<()> { + use reqsign_core::ProvideCredential; + let ctx = context( + &[ + ( + "/config", + "[profile selected]\nregion = eu-west-1\naws_access_key_id = config-key\naws_secret_access_key = config-secret\n", + ), + ( + "/credentials", + "[selected]\naws_access_key_id = credentials-key\naws_secret_access_key = credentials-secret\n", + ), + ], + &[ + ("AWS_PROFILE", "selected"), + ("AWS_CONFIG_FILE", "/config"), + ("AWS_SHARED_CREDENTIALS_FILE", "/credentials"), + ], + ); + let profile = SharedConfig::new().load(&ctx).await?; + let credential = reqsign_aws_core::ProfileCredentialProvider::new() + .provide_credential(&ctx) + .await? + .unwrap(); + assert_eq!(profile.region(), Some("eu-west-1")); + assert_eq!( + profile.get("aws_access_key_id"), + Some(credential.access_key_id.as_str()) + ); + assert_eq!( + profile.get("aws_secret_access_key"), + Some(credential.secret_access_key.as_str()) + ); + Ok(()) +} + +#[tokio::test] +async fn metadata_transport_failure_is_an_error() { + let ctx = context(&[], &[]).with_http_send(reqsign_core::NoopHttpSend); + let err = IMDSv2RegionProvider::new() + .resolve_region(&ctx) + .await + .unwrap_err(); + assert!(err.to_string().contains("failed to connect to IMDS")); +} + +#[tokio::test] +async fn credential_metadata_still_uses_shared_token_protocol() -> Result<()> { + use reqsign_core::ProvideCredential; + let http = Metadata::default(); + let ctx = context(&[], &[]).with_http_send(http.clone()); + let provider = reqsign_aws_core::IMDSv2CredentialProvider::new(); + let credential = provider.provide_credential(&ctx).await?.unwrap(); + assert_eq!(credential.access_key_id, "key"); + assert_eq!(credential.session_token.as_deref(), Some("session")); + assert_eq!(http.requests.lock().unwrap().len(), 3); + assert!(!format!("{provider:?}").contains("secret-imds-token")); + Ok(()) +} + +#[tokio::test] +async fn metadata_tokens_are_reused_only_for_the_same_endpoint() -> Result<()> { + let http = Metadata::default(); + let ctx = context(&[], &[]).with_http_send(http.clone()); + let provider = IMDSv2RegionProvider::new(); + provider.resolve_region(&ctx).await?; + provider.resolve_region(&ctx).await?; + assert_eq!(http.requests.lock().unwrap().len(), 3); + let ctx = context( + &[], + &[("AWS_EC2_METADATA_SERVICE_ENDPOINT", "http://other.invalid")], + ) + .with_http_send(http.clone()); + provider.resolve_region(&ctx).await?; + let requests = http.requests.lock().unwrap(); + assert_eq!(requests.len(), 5); + assert_eq!(requests[3], "http://other.invalid/latest/api/token"); + Ok(()) +} diff --git a/services/aws-v4/src/lib.rs b/services/aws-v4/src/lib.rs index 1defd448..fb3cb5ce 100644 --- a/services/aws-v4/src/lib.rs +++ b/services/aws-v4/src/lib.rs @@ -173,6 +173,7 @@ pub use reqsign_aws_core::{ DefaultCredentialProviderBuilder, ECSCredentialProvider, EnvCredentialProvider, IMDSv2CredentialProvider, ProfileCredentialProvider, StaticCredentialProvider, }; +pub use reqsign_aws_core::{IMDSv2RegionProvider, Profile, SharedConfig}; #[cfg(not(target_arch = "wasm32"))] pub use reqsign_aws_core::{ProcessCredentialProvider, SSOCredentialProvider}; diff --git a/services/aws-v4a/src/lib.rs b/services/aws-v4a/src/lib.rs index 4475abf0..0523cf8f 100644 --- a/services/aws-v4a/src/lib.rs +++ b/services/aws-v4a/src/lib.rs @@ -29,6 +29,7 @@ pub use reqsign_aws_core::{ EnvCredentialProvider, IMDSv2CredentialProvider, ProfileCredentialProvider, StaticCredentialProvider, }; +pub use reqsign_aws_core::{IMDSv2RegionProvider, Profile, SharedConfig}; #[cfg(not(target_arch = "wasm32"))] pub use reqsign_aws_core::{ProcessCredentialProvider, SSOCredentialProvider}; pub use sign_request::RequestSigner; From 3457f3ed812e98e463cf9e2007f5df7984094b75 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 3 Oct 2026 01:32:45 +0800 Subject: [PATCH 2/5] feat(aws): refresh named SSO session access tokens --- core/src/api.rs | 102 +++- services/aws-core/Cargo.toml | 1 + .../src/provide_credential/default.rs | 7 +- .../aws-core/src/provide_credential/sso.rs | 440 +++++++++----- .../src/provide_credential/sso/tests.rs | 560 ++++++++++++++++++ 5 files changed, 968 insertions(+), 142 deletions(-) create mode 100644 services/aws-core/src/provide_credential/sso/tests.rs diff --git a/core/src/api.rs b/core/src/api.rs index eb4b5d03..61ef2f2b 100644 --- a/core/src/api.rs +++ b/core/src/api.rs @@ -373,7 +373,8 @@ where /// /// This is a generic implementation that can be used by any service to chain multiple /// credential providers together. The chain will try each provider in order until one -/// returns credentials or all providers have been exhausted. +/// returns credentials or all providers have been exhausted. Errors are skipped +/// unless a provider was added with [`Self::push_with_error_propagation`]. /// /// # Example /// @@ -405,7 +406,12 @@ where /// # } /// ``` pub struct ProvideCredentialChain { - providers: Vec>>, + providers: Vec>, +} + +struct CredentialProviderEntry { + provider: Box>, + propagate_error: bool, } impl ProvideCredentialChain @@ -421,7 +427,27 @@ where /// Add a credential provider to the chain. pub fn push(mut self, provider: impl ProvideCredential + 'static) -> Self { - self.providers.push(Box::new(provider)); + self.providers.push(CredentialProviderEntry { + provider: Box::new(provider), + propagate_error: false, + }); + self + } + + /// Add a provider whose errors stop credential resolution. + /// + /// `Ok(None)` still tries the next provider. Use this when a configured + /// identity must not silently fall back to another identity on failure. + /// Providers added with [`Self::push`], [`Self::push_front`], or + /// [`Self::from_vec`] retain their fallback-on-error behavior. + pub fn push_with_error_propagation( + mut self, + provider: impl ProvideCredential + 'static, + ) -> Self { + self.providers.push(CredentialProviderEntry { + provider: Box::new(provider), + propagate_error: true, + }); self } @@ -432,13 +458,27 @@ where mut self, provider: impl ProvideCredential + 'static, ) -> Self { - self.providers.insert(0, Box::new(provider)); + self.providers.insert( + 0, + CredentialProviderEntry { + provider: Box::new(provider), + propagate_error: false, + }, + ); self } /// Create a credential provider chain from a vector of providers. pub fn from_vec(providers: Vec>>) -> Self { - Self { providers } + Self { + providers: providers + .into_iter() + .map(|provider| CredentialProviderEntry { + provider, + propagate_error: false, + }) + .collect(), + } } /// Get the number of providers in the chain. @@ -479,7 +519,8 @@ where type Credential = C; async fn provide_credential(&self, ctx: &Context) -> Result> { - for provider in &self.providers { + for entry in &self.providers { + let provider = &entry.provider; log::debug!("Trying credential provider: {provider:?}"); match provider.provide_credential_dyn(ctx).await { @@ -492,6 +533,9 @@ where continue; } Err(e) => { + if entry.propagate_error { + return Err(e); + } log::warn!("Error loading credential from provider {provider:?}: {e:?}"); // Continue to next provider on error continue; @@ -507,6 +551,52 @@ where mod tests { use super::*; + #[derive(Debug)] + struct ChainResult(Result>); + + impl ProvideCredential for ChainResult { + type Credential = u8; + + async fn provide_credential(&self, _: &Context) -> Result> { + match &self.0 { + Ok(value) => Ok(*value), + Err(_) => Err(crate::Error::credential_invalid( + "selected identity expired", + )), + } + } + } + + #[test] + fn chain_error_propagation_is_opt_in_and_absence_still_falls_through() { + futures::executor::block_on(async { + let failure = || ChainResult(Err(crate::Error::credential_invalid("expired"))); + let fallback = || ChainResult(Ok(Some(42))); + let ctx = Context::new(); + let ordinary = ProvideCredentialChain::new() + .push(failure()) + .push(fallback()); + assert_eq!(ordinary.provide_credential(&ctx).await.unwrap(), Some(42)); + + let selected = ProvideCredentialChain::new() + .push_with_error_propagation(failure()) + .push(fallback()); + let error = selected.provide_credential(&ctx).await.unwrap_err(); + assert_eq!(error.kind(), crate::ErrorKind::CredentialInvalid); + assert_eq!(error.to_string(), "selected identity expired"); + + let absent = ProvideCredentialChain::new() + .push_with_error_propagation(ChainResult(Ok(None))) + .push(fallback()); + assert_eq!(absent.provide_credential(&ctx).await.unwrap(), Some(42)); + + let present = ProvideCredentialChain::new() + .push_with_error_propagation(fallback()) + .push_with_error_propagation(failure()); + assert_eq!(present.provide_credential(&ctx).await.unwrap(), Some(42)); + }); + } + #[derive(Clone, Debug)] struct ExactCredential { valid_at: Timestamp, diff --git a/services/aws-core/Cargo.toml b/services/aws-core/Cargo.toml index 4ce24227..e5218766 100644 --- a/services/aws-core/Cargo.toml +++ b/services/aws-core/Cargo.toml @@ -27,6 +27,7 @@ repository.workspace = true rust-version.workspace = true [dependencies] +asyncband = { workspace = true } bytes = { workspace = true } form_urlencoded = { workspace = true } hex = { workspace = true } diff --git a/services/aws-core/src/provide_credential/default.rs b/services/aws-core/src/provide_credential/default.rs index 7a905165..7e94e198 100644 --- a/services/aws-core/src/provide_credential/default.rs +++ b/services/aws-core/src/provide_credential/default.rs @@ -35,6 +35,9 @@ use reqsign_core::{Context, ProvideCredential, ProvideCredentialChain, Result}; /// 5. Process credentials /// 6. ECS (IAM Roles for Tasks) & Container credentials /// 7. EC2 IMDSv2 +/// +/// Once SSO configuration is selected, SSO errors stop resolution to avoid +/// silently switching identities. An absent SSO configuration continues the chain. #[derive(Debug)] pub struct DefaultCredentialProvider { chain: ProvideCredentialChain, @@ -253,7 +256,7 @@ impl DefaultCredentialProviderBuilder { #[cfg(not(target_arch = "wasm32"))] { if let Some(p) = self.sso { - chain = chain.push(p); + chain = chain.push_with_error_propagation(p); } } @@ -641,7 +644,7 @@ mod tests { .expect_err("selected SSO profile must be loaded before cache lookup"); assert_eq!(ErrorKind::ConfigInvalid, error.kind()); assert_eq!( - "No valid SSO token found. Please run 'aws sso login' first", + "SSO token cache not found. Please run 'aws sso login' for the selected profile", error.to_string() ); diff --git a/services/aws-core/src/provide_credential/sso.rs b/services/aws-core/src/provide_credential/sso.rs index 9fb6f287..6b2edee6 100644 --- a/services/aws-core/src/provide_credential/sso.rs +++ b/services/aws-core/src/provide_credential/sso.rs @@ -16,18 +16,19 @@ // under the License. use crate::Credential; +use asyncband::mutex::Mutex; use http::{Method, Request, StatusCode}; use ini::Ini; -use log::{debug, warn}; use reqsign_core::time::Timestamp; use reqsign_core::{Context, Error, ProvideCredential, Result}; use serde::Deserialize; +use std::collections::HashMap; +use std::sync::Arc; const AWS_SSO_ACCOUNT_ID: &str = "sso_account_id"; const AWS_SSO_REGION: &str = "sso_region"; const AWS_SSO_ROLE_NAME: &str = "sso_role_name"; const AWS_SSO_START_URL: &str = "sso_start_url"; -#[allow(dead_code)] const AWS_SSO_SESSION_NAME: &str = "sso_session"; /// SSO Credentials Provider @@ -39,11 +40,29 @@ const AWS_SSO_SESSION_NAME: &str = "sso_session"; /// SSO configuration is typically stored in ~/.aws/config under a profile: /// ```ini /// [profile my-sso-profile] -/// sso_start_url = https://my-sso-portal.awsapps.com/start -/// sso_region = us-east-1 +/// sso_session = my-session /// sso_account_id = 123456789012 /// sso_role_name = MyRole +/// +/// [sso-session my-session] +/// sso_start_url = https://my-sso-portal.awsapps.com/start +/// sso_region = us-east-1 /// ``` +/// +/// Named sessions refresh expired access tokens using cached OIDC registration +/// and refresh tokens. Refreshed material is kept in memory for this provider +/// and its clones; independently constructed providers do not coordinate refreshes. +/// The AWS CLI cache is read through `Context` and is never written. On restart, +/// the on-disk registration and refresh token must still be usable. When an +/// in-memory token expires, a newer disk token (for example after `aws sso login`) +/// replaces it. Recreate the provider to discard its in-memory state immediately. +/// +/// Legacy inline `sso_start_url` / `sso_region` profiles and configuration supplied +/// directly through the setters remain non-refreshable. This provider never +/// starts interactive login and does not resolve assume-role profile chains. +/// When assembling a custom credential chain, use +/// [`reqsign_core::ProvideCredentialChain::push_with_error_propagation`] to keep +/// SSO failures from falling back to an unrelated identity. #[derive(Debug, Clone)] pub struct SSOCredentialProvider { profile: Option, @@ -52,6 +71,7 @@ pub struct SSOCredentialProvider { sso_role_name: Option, sso_start_url: Option, sso_endpoint: Option, // Allow custom endpoint for testing + tokens: Arc>>, } impl Default for SSOCredentialProvider { @@ -70,6 +90,7 @@ impl SSOCredentialProvider { sso_role_name: None, sso_start_url: None, sso_endpoint: None, + tokens: Arc::new(Mutex::new(HashMap::new())), } } @@ -109,7 +130,7 @@ impl SSOCredentialProvider { self } - async fn load_sso_config(&self, ctx: &Context) -> Result { + async fn load_sso_config(&self, ctx: &Context) -> Result> { // If all fields are provided directly, use them if let (Some(account_id), Some(region), Some(role_name), Some(start_url)) = ( &self.sso_account_id, @@ -117,12 +138,14 @@ impl SSOCredentialProvider { &self.sso_role_name, &self.sso_start_url, ) { - return Ok(SSOConfig { + return Ok(Some(SSOConfig { sso_account_id: account_id.clone(), sso_region: region.clone(), sso_role_name: role_name.clone(), sso_start_url: start_url.clone(), - }); + session_name: None, + config_path: None, + })); } // Otherwise, load from config file @@ -135,7 +158,11 @@ impl SSOCredentialProvider { self.load_from_config_file(ctx, &profile_name).await } - async fn load_from_config_file(&self, ctx: &Context, profile: &str) -> Result { + async fn load_from_config_file( + &self, + ctx: &Context, + profile: &str, + ) -> Result> { // Load AWS config file let config_path = ctx .env_var("AWS_CONFIG_FILE") @@ -144,18 +171,19 @@ impl SSOCredentialProvider { let expanded_path = if config_path.starts_with("~/") { match ctx.expand_home_dir(&config_path) { Some(expanded) => expanded, - None => return Err(Error::config_invalid("failed to expand home directory")), + None => return Ok(None), } } else { config_path }; - let content = ctx.file_read(&expanded_path).await.map_err(|_| { - Error::config_invalid(format!("failed to read config file: {expanded_path}")) - })?; + let content = match ctx.file_read(&expanded_path).await { + Ok(content) => content, + Err(_) => return Ok(None), + }; let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)) - .map_err(|e| Error::config_invalid(format!("failed to parse config file: {e}")))?; + .map_err(|_| Error::config_invalid("failed to parse AWS SSO config file"))?; let profile_section = if profile == "default" { profile.to_string() @@ -163,73 +191,203 @@ impl SSOCredentialProvider { format!("profile {profile}") }; - let section = conf.section(Some(profile_section)).ok_or_else(|| { - Error::config_invalid(format!("profile '{profile}' not found in config")) - })?; - - // Check if this profile has SSO configuration - let sso_account_id = section.get(AWS_SSO_ACCOUNT_ID).ok_or_else(|| { - Error::config_invalid(format!("missing {AWS_SSO_ACCOUNT_ID} in profile")) - })?; - - let sso_region = section - .get(AWS_SSO_REGION) - .ok_or_else(|| Error::config_invalid(format!("missing {AWS_SSO_REGION} in profile")))?; - - let sso_role_name = section.get(AWS_SSO_ROLE_NAME).ok_or_else(|| { - Error::config_invalid(format!("missing {AWS_SSO_ROLE_NAME} in profile")) - })?; + let Some(section) = conf.section(Some(profile_section)) else { + return Ok(None); + }; + if ![ + AWS_SSO_ACCOUNT_ID, + AWS_SSO_REGION, + AWS_SSO_ROLE_NAME, + AWS_SSO_START_URL, + AWS_SSO_SESSION_NAME, + ] + .iter() + .any(|key| section.contains_key(key)) + { + return Ok(None); + } - let sso_start_url = section.get(AWS_SSO_START_URL).ok_or_else(|| { - Error::config_invalid(format!("missing {AWS_SSO_START_URL} in profile")) - })?; + let required = |section: &ini::Properties, key: &str| -> Result { + section + .get(key) + .filter(|v| !v.is_empty()) + .map(str::to_owned) + .ok_or_else(|| Error::config_invalid(format!("missing {key} in SSO configuration"))) + }; + let session_name = section.get(AWS_SSO_SESSION_NAME).map(str::to_owned); + let session = match &session_name { + Some(name) => conf + .section(Some(format!("sso-session {name}"))) + .ok_or_else(|| Error::config_invalid("referenced SSO session not found"))?, + None => section, + }; + let region = required(session, AWS_SSO_REGION)?; + let start_url = required(session, AWS_SSO_START_URL)?; + if session_name.is_some() { + for (key, value) in [(AWS_SSO_REGION, ®ion), (AWS_SSO_START_URL, &start_url)] { + if section.get(key).is_some_and(|v| v != value) { + return Err(Error::config_invalid(format!( + "{key} conflicts with named SSO session" + ))); + } + } + } + Ok(Some(SSOConfig { + sso_account_id: required(section, AWS_SSO_ACCOUNT_ID)?, + sso_region: region, + sso_role_name: required(section, AWS_SSO_ROLE_NAME)?, + sso_start_url: start_url, + session_name, + config_path: Some(expanded_path), + })) + } - Ok(SSOConfig { - sso_account_id: sso_account_id.to_string(), - sso_region: sso_region.to_string(), - sso_role_name: sso_role_name.to_string(), - sso_start_url: sso_start_url.to_string(), - }) + async fn read_cached_token(ctx: &Context, path: &str) -> Result> { + match ctx.file_read(path).await { + Ok(content) => serde_json::from_slice(&content) + .map(Some) + .map_err(|_| login_required("invalid SSO token cache")), + Err(_) => Ok(None), + } } - async fn find_cached_token( + async fn access_token( &self, ctx: &Context, - start_url: &str, - ) -> Result> { - // Get home directory and build cache path - let home_dir = ctx - .home_dir() - .ok_or_else(|| Error::config_invalid("HOME directory not found".to_string()))?; - - let cache_dir = home_dir.join(".aws").join("sso").join("cache"); - - // Generate cache file name (SHA1 hash of start URL) - let cache_key = hex_sha1(start_url.as_bytes()); - let cache_file = cache_dir.join(format!("{cache_key}.json")); - - debug!("looking for SSO token cache at: {cache_file:?}"); - - match ctx.file_read(&cache_file.to_string_lossy()).await { - Ok(content) => { - let token: CachedToken = serde_json::from_slice(&content).map_err(|e| { - Error::unexpected(format!("failed to parse SSO token cache: {e}")) - })?; - - // Check if token is expired - let expires_at = token.expires_at.parse::()?; - if expires_at <= Timestamp::now() { - warn!("SSO token is expired"); - return Ok(None); - } + config: &SSOConfig, + now: impl Fn() -> Timestamp + Send + Sync, + ) -> Result { + let key = config.token_key(ctx)?; + if config.session_name.is_none() { + let token = Self::read_cached_token(ctx, &key.path) + .await? + .ok_or_else(|| login_required("SSO token cache not found"))?; + return if token.is_valid_at(now())? { + Ok(token.access_token) + } else { + Err(login_required("legacy SSO token has expired")) + }; + } - Ok(Some(token)) + // Hold the provider lock across refresh so clones cannot exchange the + // same refresh token concurrently, including when the service rotates it. + let mut tokens = self.tokens.lock().await; + if let Some(token) = tokens.get(&key) { + if token.is_valid_at(now())? { + return Ok(token.access_token.clone()); } - Err(_) => { - debug!("SSO token cache not found"); - Ok(None) + } + if let Some(disk) = Self::read_cached_token(ctx, &key.path).await? { + let replace = match tokens.get(&key) { + Some(current) => disk.expiration()? > current.expiration()?, + None => true, + }; + if replace { + tokens.insert(key.clone(), disk); } } + let token = tokens + .get_mut(&key) + .ok_or_else(|| login_required("SSO token cache not found"))?; + if !token.is_valid_at(now())? { + self.refresh_token(ctx, config, token, &now).await?; + } + if !token.is_valid_at(now())? { + return Err(login_required("refreshed SSO access token has expired")); + } + Ok(token.access_token.clone()) + } + + async fn refresh_token( + &self, + ctx: &Context, + config: &SSOConfig, + token: &mut CachedToken, + now: impl Fn() -> Timestamp + Send + Sync, + ) -> Result<()> { + let required = |value: &Option| { + value + .as_ref() + .filter(|v| !v.is_empty()) + .cloned() + .ok_or_else(|| login_required("SSO refresh material is missing")) + }; + let registration_expiry = required(&token.registration_expires_at)? + .parse::() + .map_err(|_| login_required("invalid SSO registration expiration"))?; + let started = now(); + if registration_expiry <= started { + return Err(login_required("SSO client registration has expired")); + } + let body = serde_json::json!({ + "grantType": "refresh_token", + "clientId": required(&token.client_id)?, + "clientSecret": required(&token.client_secret)?, + "refreshToken": required(&token.refresh_token)?, + }); + let req = Request::builder() + .method(Method::POST) + .uri(format!( + "https://oidc.{}.amazonaws.com/token", + config.sso_region + )) + .header(http::header::CONTENT_TYPE, "application/json") + .body(bytes::Bytes::from(body.to_string())) + .map_err(|_| Error::config_invalid("failed to build SSO refresh request"))?; + // HTTP adapters and server error bodies can contain the submitted secrets. + let response = ctx.http_send(req).await.map_err(|_| { + Error::unexpected("SSO token refresh transport failed").set_retryable(true) + })?; + if response.status() != StatusCode::OK { + let status = response.status(); + if status.is_server_error() || status == StatusCode::TOO_MANY_REQUESTS { + return Err( + Error::unexpected(format!("SSO token refresh returned {status}")) + .set_retryable(true), + ); + } + return Err(login_required(&format!( + "SSO token refresh rejected ({status})" + ))); + } + let refreshed: RefreshResponse = serde_json::from_slice(response.body()) + .map_err(|_| login_required("invalid SSO token refresh response"))?; + if refreshed.access_token.is_empty() + || refreshed.expires_in <= 0 + || refreshed + .refresh_token + .as_ref() + .is_some_and(|v| v.is_empty()) + { + return Err(login_required("invalid SSO token refresh response")); + } + let expires_at = started + .as_second() + .checked_add(refreshed.expires_in) + .and_then(|seconds| Timestamp::from_second(seconds).ok()) + .ok_or_else(|| login_required("invalid SSO token expiration"))?; + // Commit the complete response before attempting GetRoleCredentials. Its + // failure must not lose a rotated refresh token or extend token lifetime. + token.access_token = refreshed.access_token; + token.expires_at = expires_at.to_string(); + if let Some(refresh_token) = refreshed.refresh_token { + token.refresh_token = Some(refresh_token); + } + Ok(()) + } + + async fn provide_with_clock( + &self, + ctx: &Context, + now: impl Fn() -> Timestamp + Send + Sync, + ) -> Result> { + let Some(config) = self.load_sso_config(ctx).await? else { + return Ok(None); + }; + let token = self.access_token(ctx, &config, now).await?; + self.get_role_credentials(ctx, &config, &token) + .await + .map(Some) } async fn get_role_credentials( @@ -258,17 +416,20 @@ impl SSOCredentialProvider { let url = format!("{endpoint}?{params}"); + let mut bearer = http::HeaderValue::from_str(access_token) + .map_err(|_| login_required("invalid SSO access token"))?; + bearer.set_sensitive(true); let req = Request::builder() .method(Method::GET) .uri(&url) - .header("x-amz-sso_bearer_token", access_token) + .header("x-amz-sso_bearer_token", bearer) .body(bytes::Bytes::new()) .map_err(|e| Error::unexpected(format!("failed to build request: {e}")))?; let resp = ctx .http_send(req) .await - .map_err(|e| Error::unexpected(format!("failed to fetch SSO credentials: {e}")))?; + .map_err(|_| Error::unexpected("failed to fetch SSO credentials"))?; if resp.status() != StatusCode::OK { return Err(Error::unexpected(format!( @@ -279,7 +440,7 @@ impl SSOCredentialProvider { let body = resp.into_body(); let creds: SSOCredentialResponse = serde_json::from_slice(&body) - .map_err(|e| Error::unexpected(format!("failed to parse SSO credentials: {e}")))?; + .map_err(|_| Error::unexpected("failed to parse SSO credentials"))?; let role_creds = creds.role_credentials; let expires_in = Timestamp::from_millisecond(role_creds.expiration) @@ -300,14 +461,81 @@ struct SSOConfig { sso_region: String, sso_role_name: String, sso_start_url: String, + session_name: Option, + config_path: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TokenKey { + path: String, + config_path: Option, + session_name: Option, + start_url: String, + region: String, +} + +impl SSOConfig { + fn token_key(&self, ctx: &Context) -> Result { + let home = ctx + .home_dir() + .ok_or_else(|| Error::config_invalid("HOME directory not found"))?; + let name = self.session_name.as_deref().unwrap_or(&self.sso_start_url); + let path = home + .join(".aws") + .join("sso") + .join("cache") + .join(format!("{}.json", hex_sha1(name.as_bytes()))); + Ok(TokenKey { + path: path.to_string_lossy().into_owned(), + config_path: self.config_path.clone(), + session_name: self.session_name.clone(), + start_url: self.sso_start_url.clone(), + region: self.sso_region.clone(), + }) + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] struct CachedToken { - #[serde(rename = "accessToken")] access_token: String, - #[serde(rename = "expiresAt")] expires_at: String, + refresh_token: Option, + client_id: Option, + client_secret: Option, + registration_expires_at: Option, +} + +impl std::fmt::Debug for CachedToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CachedToken").finish_non_exhaustive() + } +} + +impl CachedToken { + fn expiration(&self) -> Result { + self.expires_at + .parse() + .map_err(|_| login_required("invalid SSO token expiration")) + } + + fn is_valid_at(&self, now: Timestamp) -> Result { + Ok(!self.access_token.is_empty() && self.expiration()? > now) + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RefreshResponse { + access_token: String, + expires_in: i64, + refresh_token: Option, +} + +fn login_required(reason: &str) -> Error { + Error::config_invalid(format!( + "{reason}. Please run 'aws sso login' for the selected profile" + )) } #[derive(Debug, Deserialize)] @@ -328,35 +556,7 @@ impl ProvideCredential for SSOCredentialProvider { type Credential = Credential; async fn provide_credential(&self, ctx: &Context) -> Result> { - let config = match self.load_sso_config(ctx).await { - Ok(c) => c, - Err(_) => { - debug!("SSO configuration not found"); - return Ok(None); - } - }; - - debug!( - "SSO config loaded: account={}, role={}", - config.sso_account_id, config.sso_role_name - ); - - // Find cached SSO token - let token = self - .find_cached_token(ctx, &config.sso_start_url) - .await? - .ok_or_else(|| { - Error::config_invalid( - "No valid SSO token found. Please run 'aws sso login' first".to_string(), - ) - })?; - - // Exchange token for role credentials - let creds = self - .get_role_credentials(ctx, &config, &token.access_token) - .await?; - - Ok(Some(creds)) + self.provide_with_clock(ctx, Timestamp::now).await } } @@ -369,32 +569,4 @@ fn hex_sha1(data: &[u8]) -> String { } #[cfg(test)] -mod tests { - use super::*; - use reqsign_core::StaticEnv; - use reqsign_file_read_tokio::TokioFileRead; - use reqsign_http_send_reqwest::ReqwestHttpSend; - use std::collections::HashMap; - - #[tokio::test] - async fn test_sso_provider_no_config() { - let ctx = Context::new() - .with_file_read(TokioFileRead) - .with_http_send(ReqwestHttpSend::default()); - let ctx = ctx.with_env(StaticEnv { - home_dir: Some(std::path::PathBuf::from("/home/test")), - envs: HashMap::new(), - }); - - let provider = SSOCredentialProvider::new(); - let result = provider.provide_credential(&ctx).await.unwrap(); - assert!(result.is_none()); - } - - #[test] - fn test_sha1_hash() { - let url = "https://my-sso-portal.awsapps.com/start"; - let hash = hex_sha1(url.as_bytes()); - assert_eq!(hash.len(), 40); // SHA1 produces 40 hex characters - } -} +mod tests; diff --git a/services/aws-core/src/provide_credential/sso/tests.rs b/services/aws-core/src/provide_credential/sso/tests.rs new file mode 100644 index 00000000..d5ebcdfd --- /dev/null +++ b/services/aws-core/src/provide_credential/sso/tests.rs @@ -0,0 +1,560 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::*; +use bytes::Bytes; +use http::Response; +use reqsign_core::{FileRead, HttpSend, StaticEnv}; +use serde_json::{Value, json}; +use std::collections::VecDeque; +use std::future::{Future, poll_fn}; +use std::pin::pin; +use std::sync::Mutex as StdMutex; +use std::task::Poll; +use tokio::sync::Notify; + +const START_URL: &str = "https://example.awsapps.com/start"; + +#[derive(Debug, Default)] +struct Files(StdMutex>>); + +impl FileRead for Files { + async fn file_read(&self, path: &str) -> Result> { + self.0 + .lock() + .unwrap() + .get(path) + .cloned() + .ok_or_else(|| Error::unexpected("file not found")) + } +} + +impl Files { + fn put(&self, path: &str, content: impl Into>) { + self.0 + .lock() + .unwrap() + .insert(path.to_string(), content.into()); + } + + fn token(&self, home: &str, name: &str, value: Value) { + self.put( + &format!("{home}/.aws/sso/cache/{}.json", hex_sha1(name.as_bytes())), + value.to_string(), + ); + } +} + +#[derive(Debug, Default)] +struct Http { + requests: StdMutex>>, + responses: StdMutex>>>, + refresh_gate: Option, +} + +impl HttpSend for Http { + async fn http_send(&self, request: Request) -> Result> { + let refreshing = request.method() == Method::POST; + self.requests.lock().unwrap().push(request); + if refreshing { + if let Some(gate) = &self.refresh_gate { + gate.notified().await; + } + } + self.responses + .lock() + .unwrap() + .pop_front() + .expect("unexpected HTTP request") + } +} + +impl Http { + fn reply(&self, status: u16, value: Value) { + self.responses + .lock() + .unwrap() + .push_back(Ok(Response::builder() + .status(status) + .body(Bytes::from(value.to_string())) + .unwrap())); + } + + fn role(&self) { + self.reply( + 200, + json!({"roleCredentials": { + "accessKeyId": "role-key", "secretAccessKey": "role-secret", + "sessionToken": "role-token", "expiration": 2000000000000i64 + }}), + ); + } + + fn refresh(&self, token: &str, refresh: Option<&str>) { + self.reply( + 200, + json!({"accessToken": token, "expiresIn": 60, "refreshToken": refresh}), + ); + } + + fn count(&self) -> usize { + self.requests.lock().unwrap().len() + } +} + +fn config(session: &str) -> String { + format!( + "[profile test]\nsso_session = {session}\nsso_account_id = 123456789012\nsso_role_name = Reader\n\n[sso-session {session}]\nsso_region = us-east-1\nsso_start_url = {START_URL}\n" + ) +} + +fn cached(expires: &str) -> Value { + json!({"accessToken": "cached-access-secret", "expiresAt": expires, + "refreshToken": "original-refresh-secret", "clientId": "client-id", + "clientSecret": "client-secret", "registrationExpiresAt": "2099-01-01T00:00:00Z"}) +} + +fn timestamp() -> Timestamp { + "2026-01-01T00:00:00Z".parse().unwrap() +} + +fn context(files: &Arc, http: &Arc, home: &str, config_path: &str) -> Context { + Context::new() + .with_file_read(files.clone()) + .with_http_send(http.clone()) + .with_env(StaticEnv { + home_dir: Some(home.into()), + envs: HashMap::from([ + ("AWS_CONFIG_FILE".into(), config_path.into()), + ("AWS_PROFILE".into(), "test".into()), + ]), + }) +} + +fn fixture(token: Value) -> (Context, Arc, Arc) { + let files = Arc::new(Files::default()); + let http = Arc::new(Http::default()); + files.put("/config", config("one")); + files.token("/test", "one", token); + (context(&files, &http, "/test", "/config"), files, http) +} + +#[tokio::test] +async fn unexpired_token_is_used_without_refresh_material() { + let (ctx, _, http) = fixture(json!({ + "accessToken": "cached-access-secret", "expiresAt": "2099-01-01T00:00:00Z" + })); + http.role(); + let credential = SSOCredentialProvider::new() + .provide_credential(&ctx) + .await + .unwrap() + .unwrap(); + assert_eq!(credential.access_key_id, "role-key"); + assert_eq!( + credential.expires_in.unwrap(), + Timestamp::from_millisecond(2000000000000).unwrap() + ); + let requests = http.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method(), Method::GET); + assert_eq!( + requests[0].uri().to_string(), + "https://portal.sso.us-east-1.amazonaws.com/federation/credentials?role_name=Reader&account_id=123456789012" + ); + assert_eq!( + requests[0].headers()["x-amz-sso_bearer_token"], + "cached-access-secret" + ); + assert!(requests[0].headers()["x-amz-sso_bearer_token"].is_sensitive()); +} + +#[tokio::test] +async fn refresh_rotation_survives_role_failure_and_subsequent_refresh() { + let (ctx, files, http) = fixture(cached("2025-01-01T00:00:00Z")); + let disk_before = files.0.lock().unwrap().clone(); + let provider = SSOCredentialProvider::new(); + http.refresh("new-access-secret", Some("rotated-refresh-secret")); + http.reply(503, json!({"secret": "do-not-log"})); + assert!(provider.provide_with_clock(&ctx, timestamp).await.is_err()); + http.role(); + provider + .clone() + .provide_with_clock(&ctx, || timestamp() + std::time::Duration::from_secs(59)) + .await + .unwrap(); + http.refresh("next-access-secret", None); + http.role(); + provider + .provide_with_clock(&ctx, || timestamp() + std::time::Duration::from_secs(60)) + .await + .unwrap(); + http.refresh("third-access-secret", None); + http.role(); + provider + .provide_with_clock(&ctx, || timestamp() + std::time::Duration::from_secs(120)) + .await + .unwrap(); + let requests = http.requests.lock().unwrap(); + assert_eq!(requests.len(), 7); + assert_eq!( + requests[0].uri(), + "https://oidc.us-east-1.amazonaws.com/token" + ); + assert_eq!( + requests[0].headers()[http::header::CONTENT_TYPE], + "application/json" + ); + let body: Value = serde_json::from_slice(requests[0].body()).unwrap(); + assert_eq!( + body, + json!({"grantType": "refresh_token", "clientId": "client-id", "clientSecret": "client-secret", "refreshToken": "original-refresh-secret"}) + ); + for index in [3, 5] { + let body: Value = serde_json::from_slice(requests[index].body()).unwrap(); + assert_eq!(body["refreshToken"], "rotated-refresh-secret"); + } + for (index, token) in [ + (1, "new-access-secret"), + (2, "new-access-secret"), + (4, "next-access-secret"), + (6, "third-access-secret"), + ] { + assert_eq!(requests[index].headers()["x-amz-sso_bearer_token"], token); + } + assert_eq!(*files.0.lock().unwrap(), disk_before); + let debug = format!("{provider:?}"); + for secret in [ + "cached-access-secret", + "new-access-secret", + "third-access-secret", + "rotated-refresh-secret", + "client-secret", + ] { + assert!(!debug.contains(secret)); + } +} + +#[tokio::test] +async fn concurrent_clones_exchange_refresh_token_once() { + let files = Arc::new(Files::default()); + files.put("/config", config("one")); + files.token("/test", "one", cached("2025-01-01T00:00:00Z")); + let http = Arc::new(Http { + refresh_gate: Some(Notify::new()), + ..Http::default() + }); + http.refresh("new-token", Some("rotated")); + http.role(); + http.role(); + let ctx = context(&files, &http, "/test", "/config"); + let provider = SSOCredentialProvider::new(); + let clone = provider.clone(); + let mut first = pin!(provider.provide_with_clock(&ctx, timestamp)); + let mut second = pin!(clone.provide_with_clock(&ctx, timestamp)); + poll_fn(|cx| { + assert!(first.as_mut().poll(cx).is_pending()); + assert!(second.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + assert_eq!(http.count(), 1); + http.refresh_gate.as_ref().unwrap().notify_one(); + let (a, b) = tokio::join!(first, second); + assert!(a.unwrap().is_some()); + assert!(b.unwrap().is_some()); + assert_eq!(http.count(), 3); +} + +#[tokio::test] +async fn missing_or_expired_refresh_material_requires_login_without_http() { + for field in [ + "refreshToken", + "clientId", + "clientSecret", + "registrationExpiresAt", + ] { + for value in [Value::Null, json!("")] { + let mut token = cached("2025-01-01T00:00:00Z"); + token[field] = value; + let (ctx, _, http) = fixture(token); + let error = SSOCredentialProvider::new() + .provide_with_clock(&ctx, timestamp) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("SSO refresh material is missing") + ); + assert!(error.to_string().contains("aws sso login")); + assert_eq!(http.count(), 0); + } + } + for expiry in [ + "2026-01-01T00:00:00Z", + "2025-01-01T00:00:00Z", + "invalid-client-secret", + ] { + let mut token = cached("2025-01-01T00:00:00Z"); + token["registrationExpiresAt"] = json!(expiry); + let (ctx, _, http) = fixture(token); + let error = SSOCredentialProvider::new() + .provide_with_clock(&ctx, timestamp) + .await + .unwrap_err(); + assert!(error.to_string().contains("aws sso login")); + assert!(!format!("{error:?}").contains("invalid-client-secret")); + assert_eq!(http.count(), 0); + } +} + +#[tokio::test] +async fn rejected_or_malformed_refresh_never_uses_expired_token_or_leaks_secrets() { + for (status, body) in [ + ( + 400, + json!({"error": "invalid_grant", "error_description": "client-secret"}), + ), + (401, json!({"error": "invalid_client"})), + ( + 200, + json!({"accessToken": "new-secret", "expiresIn": "client-secret"}), + ), + (200, json!({"accessToken": "new-secret", "expiresIn": 0})), + (200, json!({"accessToken": "new-secret", "expiresIn": -1})), + ( + 200, + json!({"accessToken": "new-secret", "expiresIn": i64::MAX}), + ), + (200, json!({"accessToken": "", "expiresIn": 60})), + ] { + let (ctx, _, http) = fixture(cached("2025-01-01T00:00:00Z")); + http.reply(status, body); + let error = SSOCredentialProvider::new() + .provide_with_clock(&ctx, timestamp) + .await + .unwrap_err(); + assert!(error.to_string().contains("aws sso login")); + assert!(!format!("{error:?}").contains("client-secret")); + assert!(!format!("{error:?}").contains("new-secret")); + assert_eq!(http.count(), 1); + } +} + +#[tokio::test] +async fn transport_failure_can_retry_without_losing_refresh_material() { + let (ctx, _, http) = fixture(cached("2025-01-01T00:00:00Z")); + http.responses + .lock() + .unwrap() + .push_back(Err(Error::unexpected("client-secret"))); + let provider = SSOCredentialProvider::new(); + let error = provider + .provide_with_clock(&ctx, timestamp) + .await + .unwrap_err(); + assert!(!format!("{error:?}").contains("client-secret")); + http.refresh("new-token", None); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + assert_eq!(http.count(), 3); +} + +#[tokio::test] +async fn sessions_with_same_start_url_have_separate_cache_and_refresh_state() { + let (ctx, files, http) = fixture(cached("2025-01-01T00:00:00Z")); + let mut other = cached("2025-01-01T00:00:00Z"); + other["refreshToken"] = json!("session-two-refresh"); + files.token("/test", "two", other); + let provider = SSOCredentialProvider::new(); + http.refresh("session-one-access", None); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + files.put("/config", config("two")); + http.refresh("session-two-access", None); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + files.put("/config", config("one")); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + let requests = http.requests.lock().unwrap(); + let body: Value = serde_json::from_slice(requests[2].body()).unwrap(); + assert_eq!(body["refreshToken"], "session-two-refresh"); + assert_eq!( + requests[3].headers()["x-amz-sso_bearer_token"], + "session-two-access" + ); + assert_eq!( + requests[4].headers()["x-amz-sso_bearer_token"], + "session-one-access" + ); + assert_eq!(requests.len(), 5); +} + +#[tokio::test] +async fn cache_paths_and_config_sources_isolate_clones() { + let (ctx, files, http) = fixture(cached("2099-01-01T00:00:00Z")); + let provider = SSOCredentialProvider::new(); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + let mut other = cached("2099-01-01T00:00:00Z"); + other["accessToken"] = json!("other-home"); + files.token("/other", "one", other); + http.role(); + provider + .clone() + .provide_with_clock(&context(&files, &http, "/other", "/config"), timestamp) + .await + .unwrap(); + files.put("/other-config", config("one")); + let mut other = cached("2099-01-01T00:00:00Z"); + other["accessToken"] = json!("other-config"); + files.token("/test", "one", other); + http.role(); + provider + .provide_with_clock(&context(&files, &http, "/test", "/other-config"), timestamp) + .await + .unwrap(); + let requests = http.requests.lock().unwrap(); + assert_eq!( + requests[1].headers()["x-amz-sso_bearer_token"], + "other-home" + ); + assert_eq!( + requests[2].headers()["x-amz-sso_bearer_token"], + "other-config" + ); +} + +#[tokio::test] +async fn legacy_inline_config_uses_url_cache_and_never_refreshes() { + let (ctx, files, http) = fixture(cached("2099-01-01T00:00:00Z")); + files.put("/config", format!("[profile test]\nsso_start_url = {START_URL}\nsso_region = us-east-1\nsso_account_id = 123\nsso_role_name = Reader\n")); + files.token("/test", START_URL, cached("2026-01-01T00:00:01Z")); + let provider = SSOCredentialProvider::new(); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + let error = provider + .provide_with_clock(&ctx, || timestamp() + std::time::Duration::from_secs(1)) + .await + .unwrap_err(); + assert!(error.to_string().contains("legacy SSO token has expired")); + assert_eq!(http.count(), 1); +} + +#[tokio::test] +async fn named_session_never_falls_back_to_url_cache_or_another_provider() { + let (ctx, files, http) = fixture(cached("2025-01-01T00:00:00Z")); + files.token("/test", START_URL, cached("2099-01-01T00:00:00Z")); + http.reply(400, json!({"error": "invalid_grant"})); + let provider = crate::DefaultCredentialProvider::builder() + .no_env() + .no_profile() + .no_web_identity() + .no_process() + .no_imds() + .with_profile("test") + .build(); + // The default chain must preserve the SSO error instead of returning None. + let error = provider.provide_credential(&ctx).await.unwrap_err(); + assert!(error.to_string().contains("SSO token refresh rejected")); + assert_eq!(http.count(), 1); + http.refresh("default-chain-access", None); + http.role(); + assert_eq!( + provider + .provide_credential(&ctx) + .await + .unwrap() + .unwrap() + .access_key_id, + "role-key" + ); + assert_eq!(http.count(), 3); +} + +#[tokio::test] +async fn missing_named_cache_does_not_use_legacy_url_cache() { + let (ctx, files, http) = fixture(cached("2099-01-01T00:00:00Z")); + files.put("/config", config("missing")); + files.token("/test", START_URL, cached("2099-01-01T00:00:00Z")); + let error = SSOCredentialProvider::new() + .provide_credential(&ctx) + .await + .unwrap_err(); + assert!(error.to_string().contains("SSO token cache not found")); + assert!(error.to_string().contains("aws sso login")); + assert_eq!(http.count(), 0); +} + +#[tokio::test] +async fn named_session_config_errors_are_not_absence() { + for contents in [ + "[profile test]\nsso_session = missing\nsso_account_id = 123\nsso_role_name = Reader\n" + .to_string(), + config("one").replace("sso_account_id = 123456789012", ""), + config("one").replace("sso_region = us-east-1", ""), + config("one").replace( + "sso_session = one", + "sso_session = one\nsso_region = us-west-2", + ), + ] { + let (ctx, files, http) = fixture(cached("2099-01-01T00:00:00Z")); + files.put("/config", contents); + assert!( + SSOCredentialProvider::new() + .provide_credential(&ctx) + .await + .is_err() + ); + assert_eq!(http.count(), 0); + } + let (ctx, files, _) = fixture(cached("2099-01-01T00:00:00Z")); + files.put("/config", "[profile test]\nregion = us-east-1\n"); + assert!( + SSOCredentialProvider::new() + .provide_credential(&ctx) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn newer_login_cache_replaces_expired_in_memory_token() { + let (ctx, files, http) = fixture(cached("2025-01-01T00:00:00Z")); + let provider = SSOCredentialProvider::new(); + http.refresh("refreshed-access", None); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + files.token( + "/test", + "one", + json!({"accessToken": "new-login", "expiresAt": "2099-01-01T00:00:00Z"}), + ); + http.role(); + provider + .provide_with_clock(&ctx, || timestamp() + std::time::Duration::from_secs(60)) + .await + .unwrap(); + assert_eq!(http.count(), 3); + assert_eq!( + http.requests.lock().unwrap()[2].headers()["x-amz-sso_bearer_token"], + "new-login" + ); +} From cb961d9eb73aa2438de9b77f441ed5afdcb52f12 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 3 Oct 2026 01:35:47 +0800 Subject: [PATCH 3/5] fix(aws): preserve process config interpretation during loader reuse --- services/aws-core/src/config.rs | 43 +++++++++++++------ .../src/provide_credential/process.rs | 6 ++- services/aws-core/tests/shared_config.rs | 33 ++++++++++++++ 3 files changed, 68 insertions(+), 14 deletions(-) diff --git a/services/aws-core/src/config.rs b/services/aws-core/src/config.rs index 2edcd3f9..c65bda53 100644 --- a/services/aws-core/src/config.rs +++ b/services/aws-core/src/config.rs @@ -100,11 +100,21 @@ impl SharedConfig { } pub(crate) async fn load_config_file(&self, ctx: &Context) -> Result { + self.load_config_file_with_options(ctx, aws_parse_options()) + .await + } + + pub(crate) async fn load_config_file_with_options( + &self, + ctx: &Context, + options: ParseOption, + ) -> Result { read_ini( ctx, self.config_file.as_deref(), "AWS_CONFIG_FILE", "~/.aws/config", + options, ) .await } @@ -115,6 +125,7 @@ impl SharedConfig { self.credentials_file.as_deref(), "AWS_SHARED_CREDENTIALS_FILE", "~/.aws/credentials", + aws_parse_options(), ) .await } @@ -220,7 +231,23 @@ pub(crate) fn config_section(profile: &str) -> String { } } -async fn read_ini(ctx: &Context, explicit: Option<&str>, env: &str, default: &str) -> Result { +fn aws_parse_options() -> ParseOption { + // Preserve literal commands and keep nested settings distinct from profile properties. + ParseOption { + enabled_quote: false, + enabled_escape: false, + enabled_indented_mutiline_value: true, + ..Default::default() + } +} + +async fn read_ini( + ctx: &Context, + explicit: Option<&str>, + env: &str, + default: &str, + options: ParseOption, +) -> Result { let path = explicit .map(str::to_owned) .or_else(|| ctx.env_var(env)) @@ -233,16 +260,6 @@ async fn read_ini(ctx: &Context, explicit: Option<&str>, env: &str, default: &st }; let content = std::str::from_utf8(&content) .map_err(|_| Error::config_invalid("AWS shared configuration is not UTF-8"))?; - // AWS values, especially credential_process commands, contain literal quotes and backslashes. - // Keep nested service settings distinct from top-level profile properties. - Ini::load_from_str_opt( - content, - ParseOption { - enabled_quote: false, - enabled_escape: false, - enabled_indented_mutiline_value: true, - ..Default::default() - }, - ) - .map_err(|_| Error::config_invalid("failed to parse AWS shared configuration")) + Ini::load_from_str_opt(content, options) + .map_err(|_| Error::config_invalid("failed to parse AWS shared configuration")) } diff --git a/services/aws-core/src/provide_credential/process.rs b/services/aws-core/src/provide_credential/process.rs index 29506357..e70012ad 100644 --- a/services/aws-core/src/provide_credential/process.rs +++ b/services/aws-core/src/provide_credential/process.rs @@ -92,7 +92,11 @@ impl ProcessCredentialProvider { } async fn load_command_from_config(&self, ctx: &Context, profile: &str) -> Result { - let conf = crate::SharedConfig::new().load_config_file(ctx).await?; + // The current command tokenizer relies on INI quote handling. Preserve that + // interpretation when sharing file loading with other profile consumers. + let conf = crate::SharedConfig::new() + .load_config_file_with_options(ctx, ini::ParseOption::default()) + .await?; let profile_section = crate::config::config_section(profile); let section = conf.section(Some(profile_section)).ok_or_else(|| { diff --git a/services/aws-core/tests/shared_config.rs b/services/aws-core/tests/shared_config.rs index fefd090e..ecd5b179 100644 --- a/services/aws-core/tests/shared_config.rs +++ b/services/aws-core/tests/shared_config.rs @@ -480,3 +480,36 @@ async fn metadata_tokens_are_reused_only_for_the_same_endpoint() -> Result<()> { assert_eq!(requests[3], "http://other.invalid/latest/api/token"); Ok(()) } + +#[cfg(not(target_arch = "wasm32"))] +#[tokio::test] +async fn sharing_config_loading_preserves_existing_process_invocation() -> Result<()> { + use reqsign_core::{CommandExecute, CommandOutput, ProvideCredential}; + #[derive(Debug)] + struct Command; + impl CommandExecute for Command { + async fn command_execute(&self, program: &str, args: &[&str]) -> Result { + assert_eq!(program, "helper"); + assert_eq!(args, ["--profile", "selected"]); + Ok(CommandOutput { + status: 0, + stdout: br#"{"Version":1,"AccessKeyId":"key","SecretAccessKey":"secret"}"#.to_vec(), + stderr: vec![], + }) + } + } + let ctx = context( + &[( + "/config", + "[profile selected]\ncredential_process = \"helper\" --profile selected\n", + )], + &[("AWS_PROFILE", "selected"), ("AWS_CONFIG_FILE", "/config")], + ) + .with_command_execute(Command); + let credential = reqsign_aws_core::ProcessCredentialProvider::new() + .provide_credential(&ctx) + .await? + .unwrap(); + assert_eq!(credential.access_key_id, "key"); + Ok(()) +} From c607cbc16445e58c895e86bc78355063966c8184 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sat, 3 Oct 2026 20:59:04 +0800 Subject: [PATCH 4/5] fix(aws): preserve SSO config literals and retry throttling --- .../aws-core/src/provide_credential/sso.rs | 37 ++++++- .../src/provide_credential/sso/tests.rs | 103 ++++++++++++++++++ 2 files changed, 136 insertions(+), 4 deletions(-) diff --git a/services/aws-core/src/provide_credential/sso.rs b/services/aws-core/src/provide_credential/sso.rs index 6b2edee6..7b6da6a7 100644 --- a/services/aws-core/src/provide_credential/sso.rs +++ b/services/aws-core/src/provide_credential/sso.rs @@ -18,7 +18,7 @@ use crate::Credential; use asyncband::mutex::Mutex; use http::{Method, Request, StatusCode}; -use ini::Ini; +use ini::{Ini, ParseOption}; use reqsign_core::time::Timestamp; use reqsign_core::{Context, Error, ProvideCredential, Result}; use serde::Deserialize; @@ -182,8 +182,16 @@ impl SSOCredentialProvider { Err(_) => return Ok(None), }; - let conf = Ini::load_from_str(&String::from_utf8_lossy(&content)) - .map_err(|_| Error::config_invalid("failed to parse AWS SSO config file"))?; + // AWS config values are literal, including commands in unselected profiles. + let conf = Ini::load_from_str_opt( + &String::from_utf8_lossy(&content), + ParseOption { + enabled_quote: false, + enabled_escape: false, + ..Default::default() + }, + ) + .map_err(|_| Error::config_invalid("failed to parse AWS SSO config file"))?; let profile_section = if profile == "default" { profile.to_string() @@ -340,7 +348,15 @@ impl SSOCredentialProvider { })?; if response.status() != StatusCode::OK { let status = response.status(); - if status.is_server_error() || status == StatusCode::TOO_MANY_REQUESTS { + // CreateToken also reports throttling as HTTP 400. Only recognize + // the allowlisted code; never retain or expose the error description. + let slow_down = status == StatusCode::BAD_REQUEST + && serde_json::from_slice::(response.body()) + .is_ok_and(|error| matches!(error.error, RefreshErrorCode::SlowDown)); + if status == StatusCode::TOO_MANY_REQUESTS || slow_down { + return Err(Error::rate_limited("SSO token refresh was throttled")); + } + if status.is_server_error() { return Err( Error::unexpected(format!("SSO token refresh returned {status}")) .set_retryable(true), @@ -524,6 +540,19 @@ impl CachedToken { } } +#[derive(Deserialize)] +struct RefreshError { + error: RefreshErrorCode, +} + +#[derive(Deserialize)] +enum RefreshErrorCode { + #[serde(rename = "slow_down")] + SlowDown, + #[serde(other)] + Other, +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct RefreshResponse { diff --git a/services/aws-core/src/provide_credential/sso/tests.rs b/services/aws-core/src/provide_credential/sso/tests.rs index d5ebcdfd..10cd18c9 100644 --- a/services/aws-core/src/provide_credential/sso/tests.rs +++ b/services/aws-core/src/provide_credential/sso/tests.rs @@ -331,6 +331,10 @@ async fn rejected_or_malformed_refresh_never_uses_expired_token_or_leaks_secrets json!({"error": "invalid_grant", "error_description": "client-secret"}), ), (401, json!({"error": "invalid_client"})), + ( + 400, + json!({"error": "client-secret", "error_description": "new-secret"}), + ), ( 200, json!({"accessToken": "new-secret", "expiresIn": "client-secret"}), @@ -558,3 +562,102 @@ async fn newer_login_cache_replaces_expired_in_memory_token() { "new-login" ); } + +#[derive(Debug, Default)] +struct ProcessCommand(std::sync::atomic::AtomicUsize); + +impl reqsign_core::CommandExecute for ProcessCommand { + async fn command_execute( + &self, + program: &str, + args: &[&str], + ) -> Result { + assert_eq!(program, "helper"); + assert!(args.is_empty()); + self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(reqsign_core::CommandOutput { + status: 0, + stdout: + br#"{"Version":1,"AccessKeyId":"process-key","SecretAccessKey":"process-secret"}"# + .to_vec(), + stderr: vec![], + }) + } +} + +#[tokio::test] +async fn default_chain_preserves_literal_config_and_selected_sso_errors() { + for command in [ + r"helper --certificate C:\x509\certificate.pem", + r#"helper --argument "unterminated"#, + ] { + for selected_sso in [false, true] { + let (ctx, files, http) = fixture(cached("2099-01-01T00:00:00Z")); + let sso = if selected_sso { + "sso_session = missing\n" + } else { + "" + }; + files.put("/config", format!( + "[default]\nregion = us-east-1\n{sso}\n[profile unselected]\ncredential_process = {command}\n" + )); + let process = Arc::new(ProcessCommand::default()); + let ctx = ctx.with_command_execute(process.clone()); + let provider = crate::DefaultCredentialProvider::builder() + .no_env() + .no_profile() + .no_web_identity() + .no_ecs() + .no_imds() + .process(crate::ProcessCredentialProvider::new().with_command("helper")) + .with_profile("default") + .build(); + let result = provider.provide_credential(&ctx).await; + if selected_sso { + let error = result.unwrap_err(); + assert_eq!(error.kind(), reqsign_core::ErrorKind::ConfigInvalid); + assert_eq!(error.to_string(), "referenced SSO session not found"); + assert_eq!(process.0.load(std::sync::atomic::Ordering::SeqCst), 0); + } else { + assert_eq!(result.unwrap().unwrap().access_key_id, "process-key"); + assert_eq!(process.0.load(std::sync::atomic::Ordering::SeqCst), 1); + } + assert_eq!(http.count(), 0); + } + } +} + +#[tokio::test] +async fn refresh_throttling_preserves_material_for_retry() { + for status in [400, 429] { + let (ctx, _, http) = fixture(cached("2025-01-01T00:00:00Z")); + http.reply( + status, + json!({ + "error": "slow_down", "error_description": "client-secret original-refresh-secret" + }), + ); + let provider = SSOCredentialProvider::new(); + let error = provider + .provide_with_clock(&ctx, timestamp) + .await + .unwrap_err(); + assert_eq!(error.kind(), reqsign_core::ErrorKind::RateLimited); + assert!(error.is_retryable()); + assert!(!error.to_string().contains("aws sso login")); + for secret in ["client-secret", "original-refresh-secret"] { + assert!(!format!("{error:?}").contains(secret)); + } + assert_eq!(http.count(), 1); + http.refresh("new-access", Some("rotated-refresh")); + http.role(); + provider.provide_with_clock(&ctx, timestamp).await.unwrap(); + let requests = http.requests.lock().unwrap(); + assert_eq!(requests.len(), 3); + assert_eq!(requests[0].body(), requests[1].body()); + assert_eq!( + requests[2].headers()["x-amz-sso_bearer_token"], + "new-access" + ); + } +} From 2c468e7656554cec2349911399647850abd8c547 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 5 Oct 2026 12:56:49 +0800 Subject: [PATCH 5/5] refactor(aws): own token refresh state per SSO session --- core/src/api.rs | 102 +----- services/aws-core/src/config.rs | 28 +- .../src/provide_credential/default.rs | 44 ++- .../aws-core/src/provide_credential/sso.rs | 246 +------------- .../src/provide_credential/sso/session.rs | 305 ++++++++++++++++++ .../src/provide_credential/sso/tests.rs | 111 +++++++ 6 files changed, 476 insertions(+), 360 deletions(-) create mode 100644 services/aws-core/src/provide_credential/sso/session.rs diff --git a/core/src/api.rs b/core/src/api.rs index 61ef2f2b..eb4b5d03 100644 --- a/core/src/api.rs +++ b/core/src/api.rs @@ -373,8 +373,7 @@ where /// /// This is a generic implementation that can be used by any service to chain multiple /// credential providers together. The chain will try each provider in order until one -/// returns credentials or all providers have been exhausted. Errors are skipped -/// unless a provider was added with [`Self::push_with_error_propagation`]. +/// returns credentials or all providers have been exhausted. /// /// # Example /// @@ -406,12 +405,7 @@ where /// # } /// ``` pub struct ProvideCredentialChain { - providers: Vec>, -} - -struct CredentialProviderEntry { - provider: Box>, - propagate_error: bool, + providers: Vec>>, } impl ProvideCredentialChain @@ -427,27 +421,7 @@ where /// Add a credential provider to the chain. pub fn push(mut self, provider: impl ProvideCredential + 'static) -> Self { - self.providers.push(CredentialProviderEntry { - provider: Box::new(provider), - propagate_error: false, - }); - self - } - - /// Add a provider whose errors stop credential resolution. - /// - /// `Ok(None)` still tries the next provider. Use this when a configured - /// identity must not silently fall back to another identity on failure. - /// Providers added with [`Self::push`], [`Self::push_front`], or - /// [`Self::from_vec`] retain their fallback-on-error behavior. - pub fn push_with_error_propagation( - mut self, - provider: impl ProvideCredential + 'static, - ) -> Self { - self.providers.push(CredentialProviderEntry { - provider: Box::new(provider), - propagate_error: true, - }); + self.providers.push(Box::new(provider)); self } @@ -458,27 +432,13 @@ where mut self, provider: impl ProvideCredential + 'static, ) -> Self { - self.providers.insert( - 0, - CredentialProviderEntry { - provider: Box::new(provider), - propagate_error: false, - }, - ); + self.providers.insert(0, Box::new(provider)); self } /// Create a credential provider chain from a vector of providers. pub fn from_vec(providers: Vec>>) -> Self { - Self { - providers: providers - .into_iter() - .map(|provider| CredentialProviderEntry { - provider, - propagate_error: false, - }) - .collect(), - } + Self { providers } } /// Get the number of providers in the chain. @@ -519,8 +479,7 @@ where type Credential = C; async fn provide_credential(&self, ctx: &Context) -> Result> { - for entry in &self.providers { - let provider = &entry.provider; + for provider in &self.providers { log::debug!("Trying credential provider: {provider:?}"); match provider.provide_credential_dyn(ctx).await { @@ -533,9 +492,6 @@ where continue; } Err(e) => { - if entry.propagate_error { - return Err(e); - } log::warn!("Error loading credential from provider {provider:?}: {e:?}"); // Continue to next provider on error continue; @@ -551,52 +507,6 @@ where mod tests { use super::*; - #[derive(Debug)] - struct ChainResult(Result>); - - impl ProvideCredential for ChainResult { - type Credential = u8; - - async fn provide_credential(&self, _: &Context) -> Result> { - match &self.0 { - Ok(value) => Ok(*value), - Err(_) => Err(crate::Error::credential_invalid( - "selected identity expired", - )), - } - } - } - - #[test] - fn chain_error_propagation_is_opt_in_and_absence_still_falls_through() { - futures::executor::block_on(async { - let failure = || ChainResult(Err(crate::Error::credential_invalid("expired"))); - let fallback = || ChainResult(Ok(Some(42))); - let ctx = Context::new(); - let ordinary = ProvideCredentialChain::new() - .push(failure()) - .push(fallback()); - assert_eq!(ordinary.provide_credential(&ctx).await.unwrap(), Some(42)); - - let selected = ProvideCredentialChain::new() - .push_with_error_propagation(failure()) - .push(fallback()); - let error = selected.provide_credential(&ctx).await.unwrap_err(); - assert_eq!(error.kind(), crate::ErrorKind::CredentialInvalid); - assert_eq!(error.to_string(), "selected identity expired"); - - let absent = ProvideCredentialChain::new() - .push_with_error_propagation(ChainResult(Ok(None))) - .push(fallback()); - assert_eq!(absent.provide_credential(&ctx).await.unwrap(), Some(42)); - - let present = ProvideCredentialChain::new() - .push_with_error_propagation(fallback()) - .push_with_error_propagation(failure()); - assert_eq!(present.provide_credential(&ctx).await.unwrap(), Some(42)); - }); - } - #[derive(Clone, Debug)] struct ExactCredential { valid_at: Timestamp, diff --git a/services/aws-core/src/config.rs b/services/aws-core/src/config.rs index 3abe0e63..43148346 100644 --- a/services/aws-core/src/config.rs +++ b/services/aws-core/src/config.rs @@ -109,25 +109,17 @@ impl SharedConfig { } pub(crate) async fn load_config_file(&self, ctx: &Context) -> Result { - read_ini( - ctx, - self.config_file.as_deref(), - "AWS_CONFIG_FILE", - "~/.aws/config", - aws_parse_options(), - ) - .await + read_ini(ctx, self.config_file_path(ctx)).await } pub(crate) async fn load_credentials_file(&self, ctx: &Context) -> Result { - read_ini( + let path = file_path( ctx, self.credentials_file.as_deref(), "AWS_SHARED_CREDENTIALS_FILE", "~/.aws/credentials", - aws_parse_options(), - ) - .await + ); + read_ini(ctx, path).await } /// Read and merge the selected profile. This never contacts IMDS, even when enabled. @@ -241,14 +233,8 @@ fn aws_parse_options() -> ParseOption { } } -async fn read_ini( - ctx: &Context, - explicit: Option<&str>, - env: &str, - default: &str, - options: ParseOption, -) -> Result { - let Some(path) = file_path(ctx, explicit, env, default) else { +async fn read_ini(ctx: &Context, path: Option) -> Result { + let Some(path) = path else { return Ok(Ini::new()); }; let Ok(content) = ctx.file_read(&path).await else { @@ -256,7 +242,7 @@ async fn read_ini( }; let content = std::str::from_utf8(&content) .map_err(|_| Error::config_invalid("AWS shared configuration is not UTF-8"))?; - Ini::load_from_str_opt(content, options) + Ini::load_from_str_opt(content, aws_parse_options()) .map_err(|_| Error::config_invalid("failed to parse AWS shared configuration")) } diff --git a/services/aws-core/src/provide_credential/default.rs b/services/aws-core/src/provide_credential/default.rs index 7e94e198..0e5b5252 100644 --- a/services/aws-core/src/provide_credential/default.rs +++ b/services/aws-core/src/provide_credential/default.rs @@ -41,6 +41,9 @@ use reqsign_core::{Context, ProvideCredential, ProvideCredentialChain, Result}; #[derive(Debug)] pub struct DefaultCredentialProvider { chain: ProvideCredentialChain, + #[cfg(not(target_arch = "wasm32"))] + sso: Option, + fallback: ProvideCredentialChain, } impl Default for DefaultCredentialProvider { @@ -62,7 +65,12 @@ impl DefaultCredentialProvider { /// Create with a custom credential chain. pub fn with_chain(chain: ProvideCredentialChain) -> Self { - Self { chain } + Self { + chain, + #[cfg(not(target_arch = "wasm32"))] + sso: None, + fallback: ProvideCredentialChain::new(), + } } /// Add a credential provider to the front of the default chain. @@ -253,40 +261,50 @@ impl DefaultCredentialProviderBuilder { chain = chain.push(p); } - #[cfg(not(target_arch = "wasm32"))] - { - if let Some(p) = self.sso { - chain = chain.push_with_error_propagation(p); - } - } + let mut fallback = ProvideCredentialChain::new(); if let Some(p) = self.web_identity { - chain = chain.push(p); + fallback = fallback.push(p); } #[cfg(not(target_arch = "wasm32"))] { if let Some(p) = self.process { - chain = chain.push(p); + fallback = fallback.push(p); } } if let Some(p) = self.ecs { - chain = chain.push(p); + fallback = fallback.push(p); } if let Some(p) = self.imds { - chain = chain.push(p); + fallback = fallback.push(p); } - DefaultCredentialProvider::with_chain(chain) + DefaultCredentialProvider { + chain, + #[cfg(not(target_arch = "wasm32"))] + sso: self.sso, + fallback, + } } } impl ProvideCredential for DefaultCredentialProvider { type Credential = Credential; async fn provide_credential(&self, ctx: &Context) -> Result> { - self.chain.provide_credential(ctx).await + if let Some(credential) = self.chain.provide_credential(ctx).await? { + return Ok(Some(credential)); + } + // SSO selects an identity. Its failure must not switch to a later source. + #[cfg(not(target_arch = "wasm32"))] + if let Some(sso) = &self.sso { + if let Some(credential) = sso.provide_credential(ctx).await? { + return Ok(Some(credential)); + } + } + self.fallback.provide_credential(ctx).await } } diff --git a/services/aws-core/src/provide_credential/sso.rs b/services/aws-core/src/provide_credential/sso.rs index e32bf8f6..627c048b 100644 --- a/services/aws-core/src/provide_credential/sso.rs +++ b/services/aws-core/src/provide_credential/sso.rs @@ -21,6 +21,9 @@ use http::{Method, Request, StatusCode}; use reqsign_core::time::Timestamp; use reqsign_core::{Context, Error, ProvideCredential, Result}; use serde::Deserialize; + +mod session; +use session::{SessionKey, SsoSession, login_required}; use std::collections::HashMap; use std::sync::Arc; @@ -59,9 +62,8 @@ const AWS_SSO_SESSION_NAME: &str = "sso_session"; /// Legacy inline `sso_start_url` / `sso_region` profiles and configuration supplied /// directly through the setters remain non-refreshable. This provider never /// starts interactive login and does not resolve assume-role profile chains. -/// When assembling a custom credential chain, use -/// [`reqsign_core::ProvideCredentialChain::push_with_error_propagation`] to keep -/// SSO failures from falling back to an unrelated identity. +/// The AWS default provider propagates SSO errors. A custom credential chain +/// retains its own error policy. #[derive(Debug, Clone)] pub struct SSOCredentialProvider { profile: Option, @@ -70,7 +72,7 @@ pub struct SSOCredentialProvider { sso_role_name: Option, sso_start_url: Option, sso_endpoint: Option, // Allow custom endpoint for testing - tokens: Arc>>, + sessions: Arc>>>, } impl Default for SSOCredentialProvider { @@ -89,7 +91,7 @@ impl SSOCredentialProvider { sso_role_name: None, sso_start_url: None, sso_endpoint: None, - tokens: Arc::new(Mutex::new(HashMap::new())), + sessions: Arc::new(Mutex::new(HashMap::new())), } } @@ -212,146 +214,24 @@ impl SSOCredentialProvider { })) } - async fn read_cached_token(ctx: &Context, path: &str) -> Result> { - match ctx.file_read(path).await { - Ok(content) => serde_json::from_slice(&content) - .map(Some) - .map_err(|_| login_required("invalid SSO token cache")), - Err(_) => Ok(None), - } - } - async fn access_token( &self, ctx: &Context, config: &SSOConfig, now: impl Fn() -> Timestamp + Send + Sync, ) -> Result { - let key = config.token_key(ctx)?; + let key = SessionKey::from_config(ctx, config)?; if config.session_name.is_none() { - let token = Self::read_cached_token(ctx, &key.path) - .await? - .ok_or_else(|| login_required("SSO token cache not found"))?; - return if token.is_valid_at(now())? { - Ok(token.access_token) - } else { - Err(login_required("legacy SSO token has expired")) - }; + return session::legacy_access_token(ctx, &key.path, now()).await; } - - // Hold the provider lock across refresh so clones cannot exchange the - // same refresh token concurrently, including when the service rotates it. - let mut tokens = self.tokens.lock().await; - if let Some(token) = tokens.get(&key) { - if token.is_valid_at(now())? { - return Ok(token.access_token.clone()); - } - } - if let Some(disk) = Self::read_cached_token(ctx, &key.path).await? { - let replace = match tokens.get(&key) { - Some(current) => disk.expiration()? > current.expiration()?, - None => true, - }; - if replace { - tokens.insert(key.clone(), disk); - } - } - let token = tokens - .get_mut(&key) - .ok_or_else(|| login_required("SSO token cache not found"))?; - if !token.is_valid_at(now())? { - self.refresh_token(ctx, config, token, &now).await?; - } - if !token.is_valid_at(now())? { - return Err(login_required("refreshed SSO access token has expired")); - } - Ok(token.access_token.clone()) - } - - async fn refresh_token( - &self, - ctx: &Context, - config: &SSOConfig, - token: &mut CachedToken, - now: impl Fn() -> Timestamp + Send + Sync, - ) -> Result<()> { - let required = |value: &Option| { - value - .as_ref() - .filter(|v| !v.is_empty()) - .cloned() - .ok_or_else(|| login_required("SSO refresh material is missing")) + let session = { + let mut sessions = self.sessions.lock().await; + sessions + .entry(key.clone()) + .or_insert_with(|| Arc::new(SsoSession::new(key))) + .clone() }; - let registration_expiry = required(&token.registration_expires_at)? - .parse::() - .map_err(|_| login_required("invalid SSO registration expiration"))?; - let started = now(); - if registration_expiry <= started { - return Err(login_required("SSO client registration has expired")); - } - let body = serde_json::json!({ - "grantType": "refresh_token", - "clientId": required(&token.client_id)?, - "clientSecret": required(&token.client_secret)?, - "refreshToken": required(&token.refresh_token)?, - }); - let req = Request::builder() - .method(Method::POST) - .uri(format!( - "https://oidc.{}.amazonaws.com/token", - config.sso_region - )) - .header(http::header::CONTENT_TYPE, "application/json") - .body(bytes::Bytes::from(body.to_string())) - .map_err(|_| Error::config_invalid("failed to build SSO refresh request"))?; - // HTTP adapters and server error bodies can contain the submitted secrets. - let response = ctx.http_send(req).await.map_err(|_| { - Error::unexpected("SSO token refresh transport failed").set_retryable(true) - })?; - if response.status() != StatusCode::OK { - let status = response.status(); - // CreateToken also reports throttling as HTTP 400. Only recognize - // the allowlisted code; never retain or expose the error description. - let slow_down = status == StatusCode::BAD_REQUEST - && serde_json::from_slice::(response.body()) - .is_ok_and(|error| matches!(error.error, RefreshErrorCode::SlowDown)); - if status == StatusCode::TOO_MANY_REQUESTS || slow_down { - return Err(Error::rate_limited("SSO token refresh was throttled")); - } - if status.is_server_error() { - return Err( - Error::unexpected(format!("SSO token refresh returned {status}")) - .set_retryable(true), - ); - } - return Err(login_required(&format!( - "SSO token refresh rejected ({status})" - ))); - } - let refreshed: RefreshResponse = serde_json::from_slice(response.body()) - .map_err(|_| login_required("invalid SSO token refresh response"))?; - if refreshed.access_token.is_empty() - || refreshed.expires_in <= 0 - || refreshed - .refresh_token - .as_ref() - .is_some_and(|v| v.is_empty()) - { - return Err(login_required("invalid SSO token refresh response")); - } - let expires_at = started - .as_second() - .checked_add(refreshed.expires_in) - .and_then(|seconds| Timestamp::from_second(seconds).ok()) - .ok_or_else(|| login_required("invalid SSO token expiration"))?; - // Commit the complete response before attempting GetRoleCredentials. Its - // failure must not lose a rotated refresh token or extend token lifetime. - token.access_token = refreshed.access_token; - token.expires_at = expires_at.to_string(); - if let Some(refresh_token) = refreshed.refresh_token { - token.refresh_token = Some(refresh_token); - } - Ok(()) + session.access_token(ctx, now).await } async fn provide_with_clock( @@ -443,92 +323,6 @@ struct SSOConfig { config_path: Option, } -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -struct TokenKey { - path: String, - config_path: Option, - session_name: Option, - start_url: String, - region: String, -} - -impl SSOConfig { - fn token_key(&self, ctx: &Context) -> Result { - let home = ctx - .home_dir() - .ok_or_else(|| Error::config_invalid("HOME directory not found"))?; - let name = self.session_name.as_deref().unwrap_or(&self.sso_start_url); - let path = home - .join(".aws") - .join("sso") - .join("cache") - .join(format!("{}.json", hex_sha1(name.as_bytes()))); - Ok(TokenKey { - path: path.to_string_lossy().into_owned(), - config_path: self.config_path.clone(), - session_name: self.session_name.clone(), - start_url: self.sso_start_url.clone(), - region: self.sso_region.clone(), - }) - } -} - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct CachedToken { - access_token: String, - expires_at: String, - refresh_token: Option, - client_id: Option, - client_secret: Option, - registration_expires_at: Option, -} - -impl std::fmt::Debug for CachedToken { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("CachedToken").finish_non_exhaustive() - } -} - -impl CachedToken { - fn expiration(&self) -> Result { - self.expires_at - .parse() - .map_err(|_| login_required("invalid SSO token expiration")) - } - - fn is_valid_at(&self, now: Timestamp) -> Result { - Ok(!self.access_token.is_empty() && self.expiration()? > now) - } -} - -#[derive(Deserialize)] -struct RefreshError { - error: RefreshErrorCode, -} - -#[derive(Deserialize)] -enum RefreshErrorCode { - #[serde(rename = "slow_down")] - SlowDown, - #[serde(other)] - Other, -} - -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct RefreshResponse { - access_token: String, - expires_in: i64, - refresh_token: Option, -} - -fn login_required(reason: &str) -> Error { - Error::config_invalid(format!( - "{reason}. Please run 'aws sso login' for the selected profile" - )) -} - #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] struct SSOCredentialResponse { @@ -551,13 +345,5 @@ impl ProvideCredential for SSOCredentialProvider { } } -// Simple SHA1 implementation for cache key generation -fn hex_sha1(data: &[u8]) -> String { - use sha1::{Digest, Sha1}; - let mut hasher = Sha1::new(); - hasher.update(data); - hex::encode(hasher.finalize()) -} - #[cfg(test)] mod tests; diff --git a/services/aws-core/src/provide_credential/sso/session.rs b/services/aws-core/src/provide_credential/sso/session.rs new file mode 100644 index 00000000..965e51fd --- /dev/null +++ b/services/aws-core/src/provide_credential/sso/session.rs @@ -0,0 +1,305 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Session-owned access tokens and OIDC refresh state. Disk cache records are +//! inputs only; refreshed material remains in memory for the session lifetime. + +use asyncband::mutex::Mutex; +use http::{Method, Request, StatusCode}; +use reqsign_core::time::Timestamp; +use reqsign_core::{Context, Error, Result}; +use serde::Deserialize; + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(super) struct SessionKey { + pub(super) path: String, + config_path: Option, + session_name: Option, + start_url: String, + region: String, +} + +impl SessionKey { + pub fn from_config(ctx: &Context, config: &super::SSOConfig) -> Result { + let home = ctx + .home_dir() + .ok_or_else(|| Error::config_invalid("HOME directory not found"))?; + let name = config + .session_name + .as_deref() + .unwrap_or(&config.sso_start_url); + let path = home + .join(".aws") + .join("sso") + .join("cache") + .join(format!("{}.json", hex_sha1(name.as_bytes()))); + Ok(SessionKey { + path: path.to_string_lossy().into_owned(), + config_path: config.config_path.clone(), + session_name: config.session_name.clone(), + start_url: config.sso_start_url.clone(), + region: config.sso_region.clone(), + }) + } +} + +#[derive(Debug)] +pub(super) struct SsoSession { + key: SessionKey, + token: Mutex>, +} + +impl SsoSession { + pub fn new(key: SessionKey) -> Self { + Self { + key, + token: Mutex::new(None), + } + } + + pub async fn access_token( + &self, + ctx: &Context, + now: impl Fn() -> Timestamp + Send + Sync, + ) -> Result { + // Only this session is locked across I/O. Clones share refresh state, + // while other sessions can load or refresh their tokens independently. + let mut state = self.token.lock().await; + if let Some(token) = state.as_ref() { + if token.is_valid_at(now()) { + return Ok(token.access_token.clone()); + } + } + if let Some(disk) = SessionToken::read(ctx, &self.key.path).await? { + if state + .as_ref() + .is_none_or(|current| disk.expires_at > current.expires_at) + { + *state = Some(disk); + } + } + let token = state + .as_mut() + .ok_or_else(|| login_required("SSO token cache not found"))?; + if !token.is_valid_at(now()) { + // Replace only on success. Retain rotated material even if a later + // role exchange fails, and preserve current state on refresh errors. + *token = token.refresh_token(ctx, &self.key.region, &now).await?; + } + if !token.is_valid_at(now()) { + return Err(login_required("refreshed SSO access token has expired")); + } + Ok(token.access_token.clone()) + } +} + +pub(super) async fn legacy_access_token( + ctx: &Context, + path: &str, + now: Timestamp, +) -> Result { + let token = SessionToken::read(ctx, path) + .await? + .ok_or_else(|| login_required("SSO token cache not found"))?; + if !token.is_valid_at(now) { + return Err(login_required("legacy SSO token has expired")); + } + Ok(token.access_token) +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct CachedTokenFile { + access_token: String, + expires_at: String, + refresh_token: Option, + client_id: Option, + client_secret: Option, + registration_expires_at: Option, +} + +impl CachedTokenFile { + fn into_token(self) -> Result { + let expires_at = self + .expires_at + .parse() + .map_err(|_| login_required("invalid SSO token expiration"))?; + // Invalid refresh material must not prevent use of a valid access token. + let refresh = (|| { + Some(RefreshMaterial { + token: self.refresh_token.filter(|s| !s.is_empty())?, + client_id: self.client_id.filter(|s| !s.is_empty())?, + client_secret: self.client_secret.filter(|s| !s.is_empty())?, + expires_at: self.registration_expires_at?.parse().ok()?, + }) + })(); + Ok(SessionToken { + access_token: self.access_token, + expires_at, + refresh, + }) + } +} + +struct SessionToken { + access_token: String, + expires_at: Timestamp, + refresh: Option, +} + +impl std::fmt::Debug for SessionToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionToken").finish_non_exhaustive() + } +} + +#[derive(Clone)] +struct RefreshMaterial { + token: String, + client_id: String, + client_secret: String, + expires_at: Timestamp, +} + +impl SessionToken { + fn is_valid_at(&self, now: Timestamp) -> bool { + !self.access_token.is_empty() && self.expires_at > now + } + + async fn read(ctx: &Context, path: &str) -> Result> { + let content = match ctx.file_read(path).await { + Ok(content) => content, + Err(_) => return Ok(None), + }; + serde_json::from_slice::(&content) + .map_err(|_| login_required("invalid SSO token cache"))? + .into_token() + .map(Some) + } + + async fn refresh_token( + &self, + ctx: &Context, + region: &str, + now: impl Fn() -> Timestamp + Send + Sync, + ) -> Result { + let material = self + .refresh + .as_ref() + .ok_or_else(|| login_required("SSO refresh material is missing or invalid"))?; + let started = now(); + if material.expires_at <= started { + return Err(login_required("SSO client registration has expired")); + } + let body = serde_json::json!({ + "grantType": "refresh_token", + "clientId": material.client_id, + "clientSecret": material.client_secret, + "refreshToken": material.token, + }); + let req = Request::builder() + .method(Method::POST) + .uri(format!("https://oidc.{}.amazonaws.com/token", region)) + .header(http::header::CONTENT_TYPE, "application/json") + .body(bytes::Bytes::from(body.to_string())) + .map_err(|_| Error::config_invalid("failed to build SSO refresh request"))?; + // HTTP adapters and server error bodies can contain the submitted secrets. + let response = ctx.http_send(req).await.map_err(|_| { + Error::unexpected("SSO token refresh transport failed").set_retryable(true) + })?; + if response.status() != StatusCode::OK { + let status = response.status(); + // CreateToken also reports throttling as HTTP 400. Only recognize + // the allowlisted code; never retain or expose the error description. + let slow_down = status == StatusCode::BAD_REQUEST + && serde_json::from_slice::(response.body()) + .is_ok_and(|error| matches!(error.error, RefreshErrorCode::SlowDown)); + if status == StatusCode::TOO_MANY_REQUESTS || slow_down { + return Err(Error::rate_limited("SSO token refresh was throttled")); + } + if status.is_server_error() { + return Err( + Error::unexpected(format!("SSO token refresh returned {status}")) + .set_retryable(true), + ); + } + return Err(login_required(&format!( + "SSO token refresh rejected ({status})" + ))); + } + let refreshed: RefreshResponse = serde_json::from_slice(response.body()) + .map_err(|_| login_required("invalid SSO token refresh response"))?; + if refreshed.access_token.is_empty() + || refreshed.expires_in <= 0 + || refreshed + .refresh_token + .as_ref() + .is_some_and(|v| v.is_empty()) + { + return Err(login_required("invalid SSO token refresh response")); + } + let expires_at = started + .as_second() + .checked_add(refreshed.expires_in) + .and_then(|seconds| Timestamp::from_second(seconds).ok()) + .ok_or_else(|| login_required("invalid SSO token expiration"))?; + let mut material = material.clone(); + if let Some(token) = refreshed.refresh_token { + material.token = token; + } + Ok(Self { + access_token: refreshed.access_token, + expires_at, + refresh: Some(material), + }) + } +} + +#[derive(Deserialize)] +struct RefreshError { + error: RefreshErrorCode, +} + +#[derive(Deserialize)] +enum RefreshErrorCode { + #[serde(rename = "slow_down")] + SlowDown, + #[serde(other)] + Other, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct RefreshResponse { + access_token: String, + expires_in: i64, + refresh_token: Option, +} + +pub(super) fn login_required(reason: &str) -> Error { + Error::config_invalid(format!( + "{reason}. Please run 'aws sso login' for the selected profile" + )) +} + +// Simple SHA1 implementation for cache key generation +pub(super) fn hex_sha1(data: &[u8]) -> String { + use sha1::{Digest, Sha1}; + let mut hasher = Sha1::new(); + hasher.update(data); + hex::encode(hasher.finalize()) +} diff --git a/services/aws-core/src/provide_credential/sso/tests.rs b/services/aws-core/src/provide_credential/sso/tests.rs index 10cd18c9..3a05e66a 100644 --- a/services/aws-core/src/provide_credential/sso/tests.rs +++ b/services/aws-core/src/provide_credential/sso/tests.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use super::session::hex_sha1; use super::*; use bytes::Bytes; use http::Response; @@ -661,3 +662,113 @@ async fn refresh_throttling_preserves_material_for_retry() { ); } } + +#[tokio::test] +async fn another_session_can_finish_while_refresh_is_blocked() { + let files = Arc::new(Files::default()); + files.put( + "/config", + format!( + "{}\n{}", + config("one"), + config("two").replace("[profile test]", "[profile second]") + ), + ); + files.token("/test", "one", cached("2025-01-01T00:00:00Z")); + files.token("/test", "two", cached("2025-01-01T00:00:00Z")); + let blocked_http = Arc::new(Http { + refresh_gate: Some(Notify::new()), + ..Http::default() + }); + blocked_http.refresh("one-access", None); + blocked_http.role(); + let independent_http = Arc::new(Http::default()); + independent_http.refresh("two-access", None); + independent_http.role(); + let ctx = context(&files, &blocked_http, "/test", "/config"); + let second_ctx = context(&files, &independent_http, "/test", "/config"); + let provider = SSOCredentialProvider::new(); + let second_provider = provider.clone().with_profile("second"); + let mut blocked = pin!(provider.provide_with_clock(&ctx, timestamp)); + let mut independent = pin!(second_provider.provide_with_clock(&second_ctx, timestamp)); + poll_fn(|cx| { + assert!(blocked.as_mut().poll(cx).is_pending()); + let Poll::Ready(result) = independent.as_mut().poll(cx) else { + panic!("another session must not wait for the blocked refresh"); + }; + assert_eq!(result.unwrap().unwrap().access_key_id, "role-key"); + Poll::Ready(()) + }) + .await; + assert_eq!(blocked_http.count(), 1); + assert_eq!(independent_http.count(), 2); + blocked_http.refresh_gate.as_ref().unwrap().notify_one(); + assert!(blocked.await.unwrap().is_some()); +} + +#[derive(Debug)] +struct FailedProvider; + +impl ProvideCredential for FailedProvider { + type Credential = Credential; + + async fn provide_credential(&self, _: &Context) -> Result> { + Err(Error::config_invalid("custom provider failed")) + } +} + +#[tokio::test] +async fn default_provider_keeps_slot_removal_precedence_and_custom_chain_policy() { + let (ctx, files, _) = fixture(cached("2099-01-01T00:00:00Z")); + files.put("/config", "[profile test]\nsso_session = missing\n"); + let process = Arc::new(ProcessCommand::default()); + let ctx = ctx.with_command_execute(process.clone()); + let builder = || { + crate::DefaultCredentialProvider::builder() + .no_env() + .no_profile() + .no_web_identity() + .no_ecs() + .no_imds() + .process(crate::ProcessCredentialProvider::new().with_command("helper")) + }; + assert!(builder().build().provide_credential(&ctx).await.is_err()); + assert_eq!(process.0.load(std::sync::atomic::Ordering::SeqCst), 0); + assert_eq!( + builder() + .no_sso() + .build() + .provide_credential(&ctx) + .await + .unwrap() + .unwrap() + .access_key_id, + "process-key" + ); + let first = builder() + .build() + .push_front(crate::StaticCredentialProvider::new("first-key", "secret")); + assert_eq!( + first + .provide_credential(&ctx) + .await + .unwrap() + .unwrap() + .access_key_id, + "first-key" + ); + let chain = reqsign_core::ProvideCredentialChain::new() + .push(FailedProvider) + .push(crate::StaticCredentialProvider::new("custom-key", "secret")); + let custom = crate::DefaultCredentialProvider::with_chain(chain); + assert_eq!( + custom + .provide_credential(&ctx) + .await + .unwrap() + .unwrap() + .access_key_id, + "custom-key" + ); + assert_eq!(process.0.load(std::sync::atomic::Ordering::SeqCst), 1); +}