From e8584e595b7a18b36df5bd054d69ad5d89433539 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:03:23 +0200 Subject: [PATCH] swarm-queue-client: audience-scoped tokens + a blocking mint for a non-reactor caller --- swarm-controller/src/auth.rs | 2 +- swarm-queue-client/Cargo.toml | 11 +- swarm-queue-client/src/lib.rs | 187 ++++++++++++++++++++++++++++++---- 3 files changed, 179 insertions(+), 21 deletions(-) diff --git a/swarm-controller/src/auth.rs b/swarm-controller/src/auth.rs index f08cbb7a..e53634a0 100644 --- a/swarm-controller/src/auth.rs +++ b/swarm-controller/src/auth.rs @@ -98,7 +98,7 @@ impl AuthBridge { // configured CA, if any) — deliberately not `self.http`, which is // the bridge's own client and has nothing to do with authelia's // token endpoint's trust anchors. - let token = swarm_queue_client::mint_token_for(&self.queue_cfg) + let token = swarm_queue_client::mint_token_for(&self.queue_cfg, None) .await .context("minting a bearer token for swarm-authelia-bridge")?; diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index b0b63588..3798b01d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -35,7 +35,16 @@ notices = ["async-nats/jetstream"] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature # above - the connect itself needs none of them. async-nats.workspace = true -reqwest.workspace = true +# `blocking` on top of the workspace default (`form`/`json`/`rustls`) — +# `mint_token_for_blocking` needs `reqwest::blocking::Client` for a caller +# with no tokio reactor to `.await` an async request on (an OTLP exporter's +# `HttpClient` impl, see that function's doc). Declared here rather than +# left to arrive transitively from a consumer that happens to pull in +# `reqwest`'s blocking feature some other way — this crate already has a +# recorded case of exactly that kind of accidental compile (see the `kv` +# feature's comment above), and `cargo check -p swarm-queue-client` alone +# must not depend on what else is in the build. +reqwest = { workspace = true, features = ["blocking"] } serde.workspace = true serde_json.workspace = true # A library, so its errors are a matchable enum rather than an opaque diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 5e5d956e..1165142e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -316,7 +316,11 @@ impl QueueConfig { /// authenticates as itself. Authelia refuses the `openid` scope for this grant /// (a machine client receives an access token and never an id-token), so no /// scope is requested. -async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result { +async fn mint_token( + http: &reqwest::Client, + cfg: &QueueConfig, + audience: Option<&str>, +) -> Result { // Read per call rather than caching: the file is small, and a cached // secret would survive a rotation that the operator believes took effect. let secret = tokio::fs::read_to_string(&cfg.client_secret_file) @@ -326,19 +330,35 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result Result { + // Why the body and not just the status: see `Error::TokenRefused`. if !status.is_success() { - return Err(Error::TokenRefused { status, body }); + return Err(Error::TokenRefused { + status, + body: body.to_owned(), + }); } - let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?; + let parsed: TokenResponse = serde_json::from_str(body).map_err(Error::TokenResponse)?; Ok(CachedToken { token: parsed.access_token, expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in), @@ -364,14 +384,45 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result, ) -> reqwest::RequestBuilder { + let mut form = vec![("grant_type", "client_credentials")]; + if let Some(audience) = audience { + form.push(("audience", audience)); + } http.post(&cfg.token_endpoint) .basic_auth(&cfg.client_id, Some(secret)) - .form(&[("grant_type", "client_credentials")]) + .form(&form) +} + +/// Read + parse `path` as a PEM root certificate. Shared by the async and +/// blocking HTTP client builders below — `reqwest::Certificate::from_pem` +/// takes the same bytes regardless of which client type ends up trusting +/// it, so the CA loading has no reason to exist twice. +fn load_ca_cert(path: &std::path::Path) -> Result { + let pem = std::fs::read(path).map_err(|source| Error::CaFile { + path: path.display().to_string(), + source, + })?; + reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { + path: path.display().to_string(), + source, + }) } /// Build the HTTP client used to reach `cfg.token_endpoint`, trusting @@ -382,15 +433,7 @@ fn token_request( fn build_http_client(cfg: &QueueConfig) -> Result { let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10)); if let Some(path) = &cfg.ca_file { - let pem = std::fs::read(path).map_err(|source| Error::CaFile { - path: path.display().to_string(), - source, - })?; - let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { - path: path.display().to_string(), - source, - })?; - builder = builder.add_root_certificate(cert); + builder = builder.add_root_certificate(load_ca_cert(path)?); } builder.build().map_err(Error::HttpClient) } @@ -408,9 +451,70 @@ fn build_http_client(cfg: &QueueConfig) -> Result { /// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints /// per call, same as this crate did before the reconnect-storm fix added the /// cache. -pub async fn mint_token_for(cfg: &QueueConfig) -> Result { +/// +/// `audience` is passed straight to [`token_request`] — see its doc for why +/// it is optional and when a caller needs it. `None` reproduces this +/// function's behaviour before the parameter existed, so every caller from +/// before that added it (the queue connect path, `auth.rs`'s bridge client) +/// is unaffected. +pub async fn mint_token_for(cfg: &QueueConfig, audience: Option<&str>) -> Result { let http = build_http_client(cfg)?; - Ok(mint_token(&http, cfg).await?.token) + Ok(mint_token(&http, cfg, audience).await?.token) +} + +/// Blocking sibling of [`mint_token_for`], for a caller with no tokio +/// reactor to `.await` on. +/// +/// That caller is real, not hypothetical: an OTLP exporter's `HttpClient` +/// implementation runs on the `PeriodicReader`'s own thread, which this +/// workspace deliberately keeps reactor-free (`reqwest-blocking-client` was +/// chosen over the async client for exactly that reason — see +/// `swarm-controller::vcs_metrics`'s module doc). Calling the async +/// [`mint_token_for`] from there would need a runtime that thread does not +/// have; this function uses `reqwest::blocking::Client` end to end instead, +/// so it never needs one. +/// +/// Same request shape and same no-caching behaviour as [`mint_token_for`] — +/// see that function's doc for why both of those are the right call here. +pub fn mint_token_for_blocking(cfg: &QueueConfig, audience: Option<&str>) -> Result { + let http = build_blocking_http_client(cfg)?; + + // Read per call rather than caching — see `mint_token`'s identical + // comment on the async path; the reasoning does not change with the + // client type. + let secret = + std::fs::read_to_string(&cfg.client_secret_file).map_err(|source| Error::ClientSecret { + path: cfg.client_secret_file.display().to_string(), + source, + })?; + + let mut form = vec![("grant_type", "client_credentials")]; + if let Some(audience) = audience { + form.push(("audience", audience)); + } + let response = http + .post(&cfg.token_endpoint) + .basic_auth(&cfg.client_id, Some(secret.trim())) + .form(&form) + .send() + .map_err(Error::TokenRequest)?; + + let status = response.status(); + let body = response.text().unwrap_or_default(); + Ok(parse_token_response(status, &body)?.token) +} + +/// Blocking counterpart to [`build_http_client`] — same CA-trust logic +/// (via the shared [`load_ca_cert`]), a `reqwest::blocking::Client` instead +/// of the async one. See [`mint_token_for_blocking`] for why a caller needs +/// this rather than the async builder. +fn build_blocking_http_client(cfg: &QueueConfig) -> Result { + let mut builder = + reqwest::blocking::Client::builder().timeout(std::time::Duration::from_secs(10)); + if let Some(path) = &cfg.ca_file { + builder = builder.add_root_certificate(load_ca_cert(path)?); + } + builder.build().map_err(Error::HttpClient) } /// Fail fast unless the client is actually connected. @@ -487,7 +591,7 @@ pub async fn connect(cfg: QueueConfig) -> Result { let token = if let Some(token) = reuse { token } else { - let minted = mint_token(&http, &cfg) + let minted = mint_token(&http, &cfg, None) .await // The callback's error type carries a string, so the // source chain would be lost; flatten it rather than @@ -631,7 +735,7 @@ mod tests { #[test] fn the_token_request_authenticates_with_http_basic() { - let req = token_request(&offline_client(), &token_cfg(), "s3cret") + let req = token_request(&offline_client(), &token_cfg(), "s3cret", None) .build() .expect("the token request must build"); @@ -667,4 +771,49 @@ mod tests { "the client id belongs in the Basic credentials, got: {body}" ); } + + /// The queue/bridge shape (`audience: None`) must stay unchanged by the + /// parameter's addition — no `audience` field appears in the body at + /// all, not even empty. + #[test] + fn no_audience_means_no_audience_field() { + let req = token_request(&offline_client(), &token_cfg(), "s3cret", None) + .build() + .expect("the token request must build"); + let body = std::str::from_utf8( + req.body() + .and_then(reqwest::Body::as_bytes) + .expect("the request has an in-memory body"), + ) + .expect("the body is utf-8"); + assert!( + !body.contains("audience"), + "omitting the audience must not even send an empty field, got: {body}" + ); + } + + /// A caller that DOES ask for an audience gets it in the form body, + /// verbatim — this is the half `swarm-otel`'s `oidc/` + /// authenticators actually check. + #[test] + fn an_audience_is_sent_verbatim() { + let req = token_request( + &offline_client(), + &token_cfg(), + "s3cret", + Some("https://otel.example/swarm"), + ) + .build() + .expect("the token request must build"); + let body = std::str::from_utf8( + req.body() + .and_then(reqwest::Body::as_bytes) + .expect("the request has an in-memory body"), + ) + .expect("the body is utf-8"); + assert!( + body.contains("audience=https%3A%2F%2Fotel.example%2Fswarm"), + "the requested audience must reach the form body, got: {body}" + ); + } }