diff --git a/Cargo.lock b/Cargo.lock index 5bec56d2..a8fe08f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -173,6 +173,7 @@ dependencies = [ "hex", "hyper-rustls", "mimalloc", + "quick-xml 0.32.0", "protoc-bin-vendored", "rayon", "rust-s3", diff --git a/Cargo.toml b/Cargo.toml index b39d97bb..8e46e28d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,6 +31,8 @@ gcp_auth = { version = "=0.12.3", default-features = false, features = ["webpki- hex = "0.4" hyper-rustls = { version = "0.27", default-features = false, features = ["ring"] } mimalloc = { version = "0.1", optional = true } +# Already resolved for aws-creds; parses STS answers in web_identity.rs. +quick-xml = { version = "0.32", features = ["serialize"] } rayon = "1.10" rustls-pki-types = "1.15" rust-s3 = { version = "=0.36.0", default-features = false, features = ["sync-rustls-tls"] } diff --git a/README.md b/README.md index 67a155b3..99f02d71 100644 --- a/README.md +++ b/README.md @@ -408,6 +408,13 @@ identity on EKS, IMDS), or pin the region with `--s3Region`. `--s3Endpoint` plus `--s3ForcePathStyle` point the client at an S3-compatible store (MinIO, LocalStack) for local testing. +S3-compatible stores that serve STS themselves (MinIO, SeaweedFS) can take a web identity token +too, such as a Kubernetes ServiceAccount token. Set `--s3StsEndpoint` (or `AWS_ENDPOINT_URL_STS`) +along with `AWS_ROLE_ARN` and `AWS_WEB_IDENTITY_TOKEN_FILE`, and `serve` sends +`AssumeRoleWithWebIdentity` to that endpoint instead of `sts.amazonaws.com`. It renews the session +five minutes before it expires, re-reading the token file each time, since kubelet rotates +projected tokens. `AWS_ROLE_SESSION_NAME` names the session (default `bazel-diff`). + S3 errors never fail a request: a failed read is treated as a cache miss (the revision is regenerated — slower, but correct) and a failed write leaves the entry local-only, so an S3 outage degrades throughput rather than availability. Concurrent replicas racing to hash the same @@ -877,6 +884,8 @@ Options: --s3ForcePathStyle + --s3StsEndpoint + STS endpoint where AWS_WEB_IDENTITY_TOKEN_FILE is exchanged for AWS_ROLE_ARN's credentials (MinIO, SeaweedFS) [env: AWS_ENDPOINT_URL_STS=] --gcsBucket Google Cloud Storage bucket used as the shared cache tier (Application Default Credentials) --gcsPrefix diff --git a/src/lib.rs b/src/lib.rs index 81dd1219..2d69ce03 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,3 +19,4 @@ pub mod proto { include!(concat!(env!("OUT_DIR"), "/proto.rs")); } pub mod server; +pub mod web_identity; diff --git a/src/main.rs b/src/main.rs index f0991bdf..34fde5a4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -471,6 +471,10 @@ struct ServeArgs { #[arg(long = "s3ForcePathStyle")] s3_force_path_style: bool, + /// STS endpoint where AWS_WEB_IDENTITY_TOKEN_FILE is exchanged for AWS_ROLE_ARN's credentials (MinIO, SeaweedFS) + #[arg(long = "s3StsEndpoint", env = "AWS_ENDPOINT_URL_STS")] + s3_sts_endpoint: Option, + /// Google Cloud Storage bucket used as the shared cache tier (Application Default Credentials) #[arg(long = "gcsBucket", conflicts_with_all = ["s3_bucket", "nexus_url"])] gcs_bucket: Option, @@ -1142,6 +1146,7 @@ impl ServeArgs { s3_region: self.s3_region.clone(), s3_endpoint: self.s3_endpoint.clone(), s3_force_path_style: self.s3_force_path_style, + s3_sts_endpoint: self.s3_sts_endpoint.clone(), gcs_bucket: self.gcs_bucket.clone(), gcs_prefix: self.gcs_prefix.clone(), nexus_url: self.nexus_url.clone(), diff --git a/src/server.rs b/src/server.rs index f0c9cc10..200a6e28 100644 --- a/src/server.rs +++ b/src/server.rs @@ -8,6 +8,7 @@ use crate::model::{ }; use crate::module_graph::{bzlmod_enabled_for, impacted_with_module_changes}; use crate::partial::{self, Classification, Glob, TreeIndex, TreeKind}; +use crate::web_identity::WebIdentitySessions; use anyhow::{anyhow, bail, Context, Result}; use flate2::read::GzDecoder; use flate2::write::GzEncoder; @@ -19,6 +20,7 @@ use s3::request::ResponseData; use s3::{Bucket, Region}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; +use std::borrow::Cow; use std::collections::{BTreeSet, HashSet, VecDeque}; use std::fmt; use std::fs; @@ -65,6 +67,9 @@ pub struct ServerConfig { pub s3_prefix: String, pub s3_region: Option, pub s3_endpoint: Option, + /// STS endpoint for `AssumeRoleWithWebIdentity`. When set, the S3 tier signs with sessions + /// exchanged there (see [`WebIdentitySessions`]) instead of the default provider chain. + pub s3_sts_endpoint: Option, pub s3_force_path_style: bool, pub gcs_bucket: Option, pub gcs_prefix: String, @@ -302,7 +307,7 @@ impl Drop for ActiveQuery { /// a miss (the revision is regenerated) and a failed write leaves the entry local-only, so an /// outage of the shared store costs throughput, never availability. enum RemoteCache { - S3(S3Cache), + S3(Box), Gcs(Box), Nexus(Box), } @@ -310,7 +315,7 @@ enum RemoteCache { impl RemoteCache { fn new(config: &ServerConfig) -> Result> { if let Some(bucket_name) = &config.s3_bucket { - return Ok(Some(Self::S3(S3Cache::new(config, bucket_name)?))); + return Ok(Some(Self::S3(Box::new(S3Cache::new(config, bucket_name)?)))); } if let Some(bucket_name) = &config.gcs_bucket { return Ok(Some(Self::Gcs(Box::new(GcsCache::new( @@ -414,6 +419,9 @@ fn gunzip(data: &[u8]) -> std::io::Result> { struct S3Cache { bucket: Box, prefix: String, + /// Set when credentials come from a configured STS endpoint; every request then signs with + /// the current session. + sessions: Option, } impl S3Cache { @@ -431,7 +439,19 @@ impl S3Cache { }, None => region_name.parse().context("invalid S3 region")?, }; - let credentials = Credentials::default().context("failed to resolve AWS credentials")?; + let sessions = config + .s3_sts_endpoint + .as_deref() + .map(WebIdentitySessions::from_env) + .transpose()?; + // A failed first exchange fails startup, as an unresolvable default chain does; later + // renewals degrade like any other S3 failure. + let credentials = match &sessions { + Some(sessions) => sessions + .credentials() + .context("failed to assume the S3 role at the STS endpoint")?, + None => Credentials::default().context("failed to resolve AWS credentials")?, + }; let mut bucket = Bucket::new(bucket_name, region, credentials).context("failed to create S3 client")?; if config.s3_force_path_style { @@ -440,9 +460,21 @@ impl S3Cache { Ok(Self { bucket, prefix: normalize_s3_prefix(&config.s3_prefix), + sessions, }) } + /// The bucket to send the next request with, signed by the current web-identity session when + /// one is configured. + fn signing_bucket(&self) -> Result> { + let Some(sessions) = &self.sessions else { + return Ok(Cow::Borrowed(&self.bucket)); + }; + let mut bucket = (*self.bucket).clone(); + bucket.set_credentials(sessions.credentials()?); + Ok(Cow::Owned(bucket)) + } + fn object_key(&self, name: &str) -> String { format!("{}{name}", self.prefix) } @@ -455,13 +487,16 @@ impl S3Cache { /// stack fails reading that absent body -- `attohttpc: Io Error: unexpected end of file`. A GET /// miss carries a real `NoSuchKey` body, so it reads back cleanly. fn get(&self, name: &str) -> Fetched { - let failure = match self.bucket.get_object(self.object_key(name)) { + let response = self + .signing_bucket() + .and_then(|bucket| bucket.get_object(self.object_key(name)).map_err(Into::into)); + let failure = match response { Ok(response) if response.status_code() == 200 => { return Fetched::Found(response.to_vec()) } Ok(response) if response.status_code() == 404 => return Fetched::Missing, Ok(response) => http_failure(&response), - Err(error) => error.to_string(), + Err(error) => format!("{error:#}"), }; eprintln!( "[Warn] S3 cache read of {} failed (treating as a miss): {failure}", @@ -471,10 +506,15 @@ impl S3Cache { } fn put(&self, name: &str, data: &[u8]) { - let failure = match self.bucket.put_object(self.object_key(name), data) { + let response = self.signing_bucket().and_then(|bucket| { + bucket + .put_object(self.object_key(name), data) + .map_err(Into::into) + }); + let failure = match response { Ok(response) if (200..300).contains(&response.status_code()) => return, Ok(response) => http_failure(&response), - Err(error) => error.to_string(), + Err(error) => format!("{error:#}"), }; eprintln!( "[Warn] S3 cache write of {} failed (entry not shared): {failure}", @@ -2163,6 +2203,7 @@ mod tests { s3_prefix: String::new(), s3_region: None, s3_endpoint: None, + s3_sts_endpoint: None, s3_force_path_style: false, gcs_bucket: None, gcs_prefix: String::new(), @@ -2356,15 +2397,78 @@ mod tests { .unwrap() .with_path_style(); ( - RemoteCache::S3(S3Cache { + RemoteCache::S3(Box::new(S3Cache { bucket, prefix: normalize_s3_prefix(prefix), - }), + sessions: None, + })), receiver, handle, ) } + #[test] + fn s3_cache_signs_with_the_web_identity_session() { + let sts = Server::http("127.0.0.1:0").unwrap(); + let sts_endpoint = format!("http://{}/", sts.server_addr().to_ip().unwrap()); + let sts_thread = thread::spawn(move || { + let answer = "\ + session-key\ + session-secret\ + session-token\ + 2999-01-01T00:00:00Z\ + "; + let request = sts.recv().unwrap(); + request.respond(Response::from_string(answer)).unwrap(); + }); + let s3 = Server::http("127.0.0.1:0").unwrap(); + let s3_endpoint = format!("http://{}", s3.server_addr().to_ip().unwrap()); + let s3_thread = thread::spawn(move || { + let request = s3.recv().unwrap(); + let token = request + .headers() + .iter() + .find(|header| header.field.equiv("x-amz-security-token")) + .map(|header| header.value.to_string()); + request + .respond(Response::from_data(Vec::new()).with_status_code(StatusCode(404))) + .unwrap(); + token + }); + let token_dir = tempfile::tempdir().unwrap(); + let token_file = token_dir.path().join("token"); + fs::write(&token_file, "service-account-token").unwrap(); + let sessions = WebIdentitySessions::new( + &sts_endpoint, + "arn:aws:iam::role/cache".to_owned(), + token_file, + "bazel-diff".to_owned(), + ) + .unwrap(); + let placeholder = Credentials::new(Some("key"), Some("secret"), None, None, None).unwrap(); + let bucket = Bucket::new( + "bucket", + Region::Custom { + region: "us-east-1".to_owned(), + endpoint: s3_endpoint, + }, + placeholder, + ) + .unwrap() + .with_path_style(); + let cache = S3Cache { + bucket, + prefix: String::new(), + sessions: Some(sessions), + }; + + let fetched = cache.get("entry.json.gz"); + + sts_thread.join().unwrap(); + assert!(matches!(fetched, Fetched::Missing)); + assert_eq!(s3_thread.join().unwrap().as_deref(), Some("session-token")); + } + #[test] fn parses_compound_duration() { assert_eq!(parse_duration("1d12h30m").unwrap().as_secs(), 131_400); diff --git a/src/web_identity.rs b/src/web_identity.rs new file mode 100644 index 00000000..6b0de178 --- /dev/null +++ b/src/web_identity.rs @@ -0,0 +1,405 @@ +//! Temporary S3 credentials from `AssumeRoleWithWebIdentity` at a configurable STS endpoint. +//! +//! `aws-creds` sends web-identity exchanges only to `https://sts.amazonaws.com/`. S3-compatible +//! stores (MinIO, SeaweedFS) serve STS themselves, so with an STS endpoint configured the S3 tier +//! makes the exchange here and renews the session before it expires. + +use anyhow::{bail, Context, Result}; +use s3::creds::{Credentials, Rfc3339OffsetDateTime}; +use serde::Deserialize; +use std::fs; +use std::path::PathBuf; +use std::sync::{Mutex, PoisonError}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use url::{form_urlencoded, Url}; + +/// How long before a session expires it is replaced, so no request is signed with credentials +/// that lapse in flight. +pub const RENEWAL_MARGIN: Duration = Duration::from_secs(5 * 60); + +const DEFAULT_SESSION_NAME: &str = "bazel-diff"; +const STS_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +const STS_READ_TIMEOUT: Duration = Duration::from_secs(30); + +/// Exchanges a web identity token (a Kubernetes ServiceAccount token, say) for S3 credentials and +/// reuses them until [`RENEWAL_MARGIN`] before they expire. +pub struct WebIdentitySessions { + endpoint: Url, + role_arn: String, + token_file: PathBuf, + session_name: String, + http: attohttpc::Session, + current: Mutex>, +} + +struct Session { + credentials: Credentials, + expires_at: SystemTime, +} + +impl WebIdentitySessions { + /// Reads the role, token file and session name from `AWS_ROLE_ARN`, + /// `AWS_WEB_IDENTITY_TOKEN_FILE` and `AWS_ROLE_SESSION_NAME`, as the AWS SDKs do. + pub fn from_env(endpoint: &str) -> Result { + let role_arn = + std::env::var("AWS_ROLE_ARN").context("an STS endpoint requires AWS_ROLE_ARN")?; + let token_file = std::env::var_os("AWS_WEB_IDENTITY_TOKEN_FILE") + .context("an STS endpoint requires AWS_WEB_IDENTITY_TOKEN_FILE")?; + let session_name = std::env::var("AWS_ROLE_SESSION_NAME") + .unwrap_or_else(|_| DEFAULT_SESSION_NAME.to_owned()); + Self::new(endpoint, role_arn, token_file.into(), session_name) + } + + /// Exchanges the token in `token_file` for `role_arn`'s credentials at `endpoint`. + pub fn new( + endpoint: &str, + role_arn: String, + token_file: PathBuf, + session_name: String, + ) -> Result { + let endpoint = + Url::parse(endpoint).with_context(|| format!("invalid STS endpoint {endpoint:?}"))?; + if !matches!(endpoint.scheme(), "http" | "https") { + bail!("STS endpoint must be http or https: {endpoint}"); + } + let mut http = attohttpc::Session::new(); + http.connect_timeout(STS_CONNECT_TIMEOUT); + http.read_timeout(STS_READ_TIMEOUT); + // The token travels in the request body; following a redirect would replay it to + // whatever host the `Location` names. + http.follow_redirects(false); + Ok(Self { + endpoint, + role_arn, + token_file, + session_name, + http, + current: Mutex::new(None), + }) + } + + /// Credentials to sign the next request with: the current session's, or a new session's when + /// the current one expires within [`RENEWAL_MARGIN`]. Concurrent callers wait for a single + /// exchange rather than each making their own. + pub fn credentials(&self) -> Result { + let mut current = self.current.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(session) = current.as_ref() { + if !needs_renewal(session.expires_at, SystemTime::now()) { + return Ok(session.credentials.clone()); + } + } + let session = self.assume_role()?; + let credentials = session.credentials.clone(); + *current = Some(session); + Ok(credentials) + } + + /// One `AssumeRoleWithWebIdentity` call. The token file is read every time because kubelet + /// rotates a projected ServiceAccount token before it expires. + fn assume_role(&self) -> Result { + let token = fs::read_to_string(&self.token_file).with_context(|| { + format!( + "failed to read the web identity token {}", + self.token_file.display() + ) + })?; + // A form post rather than query parameters keeps the token out of access logs. + let form = form_urlencoded::Serializer::new(String::new()) + .append_pair("Action", "AssumeRoleWithWebIdentity") + .append_pair("Version", "2011-06-15") + .append_pair("RoleArn", &self.role_arn) + .append_pair("RoleSessionName", &self.session_name) + .append_pair("WebIdentityToken", token.trim()) + .finish(); + let response = self + .http + .post(self.endpoint.as_str()) + .header( + attohttpc::header::CONTENT_TYPE, + "application/x-www-form-urlencoded", + ) + .text(form) + .send() + .with_context(|| format!("failed to reach STS at {}", self.endpoint))?; + let status = response.status(); + let body = response.text().unwrap_or_default(); + if !status.is_success() { + bail!( + "STS at {} refused {}: HTTP {} {}", + self.endpoint, + self.role_arn, + status.as_u16(), + body.trim() + ); + } + parse_session(&body) + } +} + +/// Whether a session expiring at `expires_at` is too close to expiry to sign with at `now`. +fn needs_renewal(expires_at: SystemTime, now: SystemTime) -> bool { + now + RENEWAL_MARGIN >= expires_at +} + +/// The part of an `AssumeRoleWithWebIdentity` answer this reads. Other elements, which +/// S3-compatible stores fill in to varying degrees, are ignored. +#[derive(Deserialize)] +#[serde(rename_all = "PascalCase")] +struct AssumeRoleWithWebIdentityResponse { + assume_role_with_web_identity_result: AssumeRoleWithWebIdentityResult, +} + +#[derive(Deserialize)] +#[serde(rename_all = "PascalCase")] +struct AssumeRoleWithWebIdentityResult { + credentials: StsCredentials, +} + +#[derive(Deserialize)] +#[serde(rename_all = "PascalCase")] +struct StsCredentials { + access_key_id: String, + secret_access_key: String, + #[serde(default)] + session_token: Option, + expiration: Rfc3339OffsetDateTime, +} + +fn parse_session(body: &str) -> Result { + let response: AssumeRoleWithWebIdentityResponse = + quick_xml::de::from_str(body).context("STS answered without credentials")?; + let sts = response.assume_role_with_web_identity_result.credentials; + let expires_after = u64::try_from(sts.expiration.unix_timestamp()).unwrap_or_default(); + Ok(Session { + // No `expiration`: `rust-s3` checks it before every request and, once it passes, replaces + // the credentials from the default provider chain, which would send the token to AWS. + // Renewal is this module's job, ahead of expiry. + credentials: Credentials { + access_key: Some(sts.access_key_id), + secret_key: Some(sts.secret_access_key), + security_token: None, + session_token: sts.session_token, + expiration: None, + }, + expires_at: UNIX_EPOCH + Duration::from_secs(expires_after), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc; + use std::thread; + use tiny_http::{Response, Server, StatusCode}; + + struct CapturedExchange { + method: String, + url: String, + content_type: String, + form: Vec<(String, String)>, + } + + fn sts_answer(access_key: &str, expiration: &str) -> String { + format!( + r#" + + + {access_key} + secret-{access_key} + token-{access_key} + {expiration} + + +"# + ) + } + + fn capture(request: &mut tiny_http::Request) -> CapturedExchange { + let mut body = String::new(); + request.as_reader().read_to_string(&mut body).unwrap(); + let content_type = request + .headers() + .iter() + .find(|header| header.field.equiv("Content-Type")) + .map(|header| header.value.to_string()) + .unwrap_or_default(); + CapturedExchange { + method: request.method().to_string(), + url: request.url().to_owned(), + content_type, + form: form_urlencoded::parse(body.as_bytes()) + .into_owned() + .collect(), + } + } + + /// A local STS that answers each exchange with the next of `responses` and reports what it + /// was sent. + fn sts_with_responses( + responses: Vec<(u16, String)>, + ) -> ( + String, + mpsc::Receiver, + thread::JoinHandle<()>, + ) { + let server = Server::http("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/", server.server_addr().to_ip().unwrap()); + let (sender, receiver) = mpsc::channel(); + let handle = thread::spawn(move || { + for (status, body) in responses { + let mut request = server.recv().unwrap(); + sender.send(capture(&mut request)).unwrap(); + request + .respond(Response::from_string(body).with_status_code(StatusCode(status))) + .unwrap(); + } + }); + (endpoint, receiver, handle) + } + + fn token_file(token: &str) -> (tempfile::TempDir, PathBuf) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("token"); + fs::write(&path, token).unwrap(); + (dir, path) + } + + fn sessions(endpoint: &str, token_file: PathBuf) -> WebIdentitySessions { + WebIdentitySessions::new( + endpoint, + "arn:aws:iam::role/cache".to_owned(), + token_file, + "bazel-diff".to_owned(), + ) + .unwrap() + } + + fn form_value<'a>(exchange: &'a CapturedExchange, name: &str) -> &'a str { + exchange + .form + .iter() + .find(|(key, _)| key == name) + .map(|(_, value)| value.as_str()) + .unwrap_or_else(|| panic!("form has no {name}")) + } + + #[test] + fn exchanges_the_token_with_a_form_post_to_the_configured_endpoint() { + let (endpoint, exchanges, handle) = + sts_with_responses(vec![(200, sts_answer("first", "2999-01-01T00:00:00Z"))]); + let (_dir, path) = token_file("service-account-token\n"); + + let credentials = sessions(&endpoint, path).credentials().unwrap(); + + handle.join().unwrap(); + let exchange = exchanges.recv().unwrap(); + assert_eq!(exchange.method, "POST"); + assert_eq!(exchange.url, "/"); + assert_eq!(exchange.content_type, "application/x-www-form-urlencoded"); + assert_eq!(form_value(&exchange, "Action"), "AssumeRoleWithWebIdentity"); + assert_eq!(form_value(&exchange, "Version"), "2011-06-15"); + assert_eq!(form_value(&exchange, "RoleArn"), "arn:aws:iam::role/cache"); + assert_eq!(form_value(&exchange, "RoleSessionName"), "bazel-diff"); + assert_eq!( + form_value(&exchange, "WebIdentityToken"), + "service-account-token" + ); + assert_eq!(credentials.access_key.as_deref(), Some("first")); + assert_eq!(credentials.secret_key.as_deref(), Some("secret-first")); + assert_eq!(credentials.session_token.as_deref(), Some("token-first")); + assert_eq!(credentials.expiration, None); + } + + #[test] + fn reuses_a_session_that_is_not_close_to_expiry() { + let (endpoint, exchanges, handle) = + sts_with_responses(vec![(200, sts_answer("first", "2999-01-01T00:00:00Z"))]); + let (_dir, path) = token_file("token"); + let sessions = sessions(&endpoint, path); + + let first = sessions.credentials().unwrap(); + let second = sessions.credentials().unwrap(); + + handle.join().unwrap(); + assert_eq!(exchanges.try_iter().count(), 1); + assert_eq!(first, second); + } + + #[test] + fn renews_an_expiring_session_with_the_current_token() { + let (endpoint, exchanges, handle) = sts_with_responses(vec![ + (200, sts_answer("first", "2000-01-01T00:00:00Z")), + (200, sts_answer("second", "2999-01-01T00:00:00Z")), + ]); + let (_dir, path) = token_file("old-token"); + let sessions = sessions(&endpoint, path.clone()); + + sessions.credentials().unwrap(); + fs::write(&path, "rotated-token").unwrap(); + let renewed = sessions.credentials().unwrap(); + + handle.join().unwrap(); + let tokens: Vec = exchanges + .try_iter() + .map(|exchange| form_value(&exchange, "WebIdentityToken").to_owned()) + .collect(); + assert_eq!(tokens, ["old-token", "rotated-token"]); + assert_eq!(renewed.access_key.as_deref(), Some("second")); + } + + #[test] + fn reports_a_refused_exchange_with_the_status_and_body() { + let (endpoint, _exchanges, handle) = + sts_with_responses(vec![(403, "AccessDenied: token expired".to_owned())]); + let (_dir, path) = token_file("token"); + + let error = sessions(&endpoint, path).credentials().unwrap_err(); + + handle.join().unwrap(); + let message = format!("{error:#}"); + assert!(message.contains("HTTP 403"), "{message}"); + assert!(message.contains("AccessDenied: token expired"), "{message}"); + } + + #[test] + fn reports_an_answer_without_credentials() { + let error = parse_session("").err(); + + assert!(error.is_some()); + } + + #[test] + fn reports_a_missing_token_file() { + let dir = tempfile::tempdir().unwrap(); + let sessions = sessions("http://127.0.0.1:9/", dir.path().join("absent")); + + let error = format!("{:#}", sessions.credentials().unwrap_err()); + + assert!(error.contains("web identity token"), "{error}"); + } + + #[test] + fn rejects_an_endpoint_that_is_not_http() { + let error = WebIdentitySessions::new( + "ftp://sts.example", + "role".to_owned(), + PathBuf::from("token"), + "bazel-diff".to_owned(), + ) + .err(); + + assert!(error.is_some()); + } + + #[test] + fn renews_within_the_margin_and_not_before() { + let now = UNIX_EPOCH + Duration::from_secs(1_000_000); + + assert!(needs_renewal(now, now)); + assert!(needs_renewal(now + RENEWAL_MARGIN, now)); + assert!(!needs_renewal( + now + RENEWAL_MARGIN + Duration::from_secs(1), + now + )); + } +} diff --git a/tools/readme_template.md b/tools/readme_template.md index 2b754ab4..33c1960b 100644 --- a/tools/readme_template.md +++ b/tools/readme_template.md @@ -408,6 +408,13 @@ identity on EKS, IMDS), or pin the region with `--s3Region`. `--s3Endpoint` plus `--s3ForcePathStyle` point the client at an S3-compatible store (MinIO, LocalStack) for local testing. +S3-compatible stores that serve STS themselves (MinIO, SeaweedFS) can take a web identity token +too, such as a Kubernetes ServiceAccount token. Set `--s3StsEndpoint` (or `AWS_ENDPOINT_URL_STS`) +along with `AWS_ROLE_ARN` and `AWS_WEB_IDENTITY_TOKEN_FILE`, and `serve` sends +`AssumeRoleWithWebIdentity` to that endpoint instead of `sts.amazonaws.com`. It renews the session +five minutes before it expires, re-reading the token file each time, since kubelet rotates +projected tokens. `AWS_ROLE_SESSION_NAME` names the session (default `bazel-diff`). + S3 errors never fail a request: a failed read is treated as a cache miss (the revision is regenerated — slower, but correct) and a failed write leaves the entry local-only, so an S3 outage degrades throughput rather than availability. Concurrent replicas racing to hash the same