swarm-queue-client: audience-scoped tokens + a blocking mint for a non-reactor caller

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")?;
 

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

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<CachedToken, Error> {
+async fn mint_token(
+    http: &reqwest::Client,
+    cfg: &QueueConfig,
+    audience: Option<&str>,
+) -> Result<CachedToken, Error> {
     // 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<CachedT
             source,
         })?;
 
-    let response = token_request(http, cfg, secret.trim())
+    let response = token_request(http, cfg, secret.trim(), audience)
         .send()
         .await
         .map_err(Error::TokenRequest)?;
 
-    // Why the body and not just the status: see `Error::TokenRefused`.
     let status = response.status();
     let body = response.text().await.unwrap_or_default();
+    parse_token_response(status, &body)
+}
+
+/// Turn a raw token-endpoint response into a [`CachedToken`] — the half of
+/// [`mint_token`] that is pure and therefore shared with the blocking sibling
+/// below: neither the success/failure check nor the JSON shape cares which
+/// HTTP client fetched the bytes.
+///
+/// Takes the body already read out, not the response itself: `reqwest`'s
+/// blocking and async response types have no common trait for that, and
+/// forcing one here would mean this function is not actually callable from
+/// both callers, which defeats the reason it exists.
+fn parse_token_response(status: reqwest::StatusCode, body: &str) -> Result<CachedToken, Error> {
+    // 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<CachedT
 ///
 /// Split out of [`mint_token`] so the request's *shape* is testable without a
 /// running identity provider — see the tests at the bottom of this file.
+///
+/// `audience` is RFC 8707 resource-indicator territory, and it is opt-in:
+/// omitted, this identity gets whatever audience authelia defaults a
+/// scopeless `client_credentials` grant to (the queue connection's own
+/// case — it has always worked without asking). A caller proving this
+/// identity to a SPECIFIC audience-checked receiver — the swarm-otel
+/// `oidc/swarm` authenticator being the first one — has to ask for it by
+/// name, the same way `swarm-otel.nix`'s own prometheus scrape config
+/// already does per target (`endpoint_params.audience`): a token minted
+/// without asking carries `aud: []`, and an audience-checked receiver
+/// refuses that just as readily as the wrong one.
 fn token_request(
     http: &reqwest::Client,
     cfg: &QueueConfig,
     secret: &str,
+    audience: Option<&str>,
 ) -> 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<reqwest::Certificate, Error> {
+    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<reqwest::Client, Error> {
     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<reqwest::Client, Error> {
 /// 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<String, Error> {
+///
+/// `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<String, Error> {
     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<String, Error> {
+    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<reqwest::blocking::Client, Error> {
+    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<async_nats::Client, Error> {
             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/<owner>`
+    /// 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}"
+        );
+    }
 }