diff --git a/Cargo.lock b/Cargo.lock index fea30799..885fd1c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4895,6 +4895,7 @@ version = "0.1.0" dependencies = [ "anyhow", "async-nats", + "axum", "clap", "data-encoding", "futures-util", diff --git a/swarm-nats-auth/Cargo.toml b/swarm-nats-auth/Cargo.toml index e77b546a..99f03a91 100644 --- a/swarm-nats-auth/Cargo.toml +++ b/swarm-nats-auth/Cargo.toml @@ -39,6 +39,9 @@ tracing.workspace = true tracing-subscriber.workspace = true [dev-dependencies] +# A loopback stand-in `IdP` for `oidc_secret`'s test of a non-2xx +# introspection starting the re-read. +axum.workspace = true # A TEST ORACLE, not part of the production path. Neither JWT this crate emits # is expressible through it - `Claims` has no `aud`, which the response wrapper # needs (the server id) and the user token needs (the account name), and diff --git a/swarm-nats-auth/src/oidc_secret.rs b/swarm-nats-auth/src/oidc_secret.rs index 9adc29a1..f62b4ed3 100644 --- a/swarm-nats-auth/src/oidc_secret.rs +++ b/swarm-nats-auth/src/oidc_secret.rs @@ -14,6 +14,7 @@ use std::future::Future; use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; use anyhow::{Context, Result}; use futures_util::FutureExt; @@ -92,6 +93,11 @@ impl Source for Store { } } +/// The longest one re-read runs. The store client sets no timeout of its +/// own, and while a re-read runs every refusal waits on it instead of +/// starting another. +const REREAD_TIMEOUT: Duration = Duration::from_secs(10); + /// A re-read in progress, shared by every request waiting on it. type Reread = Shared>; @@ -167,12 +173,16 @@ impl OidcSecret { let source = Arc::clone(&self.source); let current = Arc::clone(&self.current); let task = tokio::spawn(async move { - match source.read().await { - Ok(secret) => *current.write().expect("unpoisoned") = secret, - Err(e) => tracing::warn!( + match tokio::time::timeout(REREAD_TIMEOUT, source.read()).await { + Ok(Ok(secret)) => *current.write().expect("unpoisoned") = secret, + Ok(Err(e)) => tracing::warn!( error = format!("{e:#}"), "could not re-read the OIDC client secret; keeping the one in use" ), + Err(_) => tracing::warn!( + timeout = ?REREAD_TIMEOUT, + "re-reading the OIDC client secret timed out; keeping the one in use" + ), } }); let handle = task.abort_handle(); @@ -394,4 +404,97 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 3); assert_eq!(reads.load(Ordering::SeqCst), 1); } + + /// A re-read that stalls is given up at [`REREAD_TIMEOUT`]: the secret in + /// use stays, and the next refusal starts a fresh re-read rather than + /// waiting on the stalled one. + #[tokio::test(start_paused = true)] + async fn a_stalled_re_read_is_dropped_at_its_timeout() { + let store = Fake { + secret: Some("new"), + delay: REREAD_TIMEOUT * 6, + reads: Arc::default(), + }; + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); + let calls = AtomicUsize::new(0); + + let first = secret + .introspect(budget(), idp("new", &calls)) + .await + .expect("answered"); + assert_eq!(first, None); + + tokio::time::sleep(REREAD_TIMEOUT).await; + assert_eq!(secret.current(), "old"); + let second = secret + .introspect(budget(), idp("new", &calls)) + .await + .expect("answered"); + assert_eq!(second, None); + assert_eq!(reads.load(Ordering::SeqCst), 2); + } + + /// An `IdP` on a loopback port that admits as `hive-alpha` a request + /// whose basic auth is `swarm-nats:{registered}`, and answers any other + /// with a 401. + async fn stub_idp(registered: &str) -> String { + use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION}; + + let expected = format!( + "Basic {}", + data_encoding::BASE64.encode(format!("swarm-nats:{registered}").as_bytes()) + ); + let app = axum::Router::new().fallback(move |headers: HeaderMap| async move { + if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) { + ( + StatusCode::OK, + r#"{"active":true,"client_id":"hive-alpha"}"#, + ) + } else { + (StatusCode::UNAUTHORIZED, "") + } + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let url = format!("http://{}/introspect", listener.local_addr().expect("addr")); + tokio::spawn(async move { axum::serve(listener, app).await }); + url + } + + /// The `IdP`'s 401 to the secret held since start is what starts the + /// re-read: driven through the real HTTP introspection, not an [`Answer`] + /// handed in. + #[tokio::test] + async fn an_idp_401_re_reads_the_secret_and_admits_with_the_new_one() { + let url = stub_idp("new").await; + let http = reqwest::Client::builder() + .no_proxy() + .build() + .expect("client"); + let store = Fake::holding("new"); + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); + let deadline = budget(); + let caller = secret + .introspect(deadline, |own| { + let (http, url) = (&http, &url); + async move { + crate::introspect::identify_caller( + http, + url, + "swarm-nats", + &own, + "a-token", + deadline, + ) + .await + } + }) + .await + .expect("answered"); + assert_eq!(caller.as_deref(), Some("hive-alpha")); + assert_eq!(reads.load(Ordering::SeqCst), 1); + } }