From 629ac4ba9227a4b6da855e1ffd26aa3b2625218f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 3 Sep 2026 09:28:08 +0000 Subject: [PATCH] test: pin JWKS single-flight with a concurrent cold burst A stub issuer that counts inbound connections, ten concurrent cached_jwks calls, and an assertion that exactly one fetch is made. Pins the property the caching layer exists for, which no test covered before. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01VF3v6LA9399gNphmZaHYG3 --- backend/windmill-common/src/guest_jwt.rs | 45 ++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/backend/windmill-common/src/guest_jwt.rs b/backend/windmill-common/src/guest_jwt.rs index 32a8340185..797e19b55a 100644 --- a/backend/windmill-common/src/guest_jwt.rs +++ b/backend/windmill-common/src/guest_jwt.rs @@ -469,4 +469,49 @@ mod tests { fn a_non_key_pem_is_rejected() { assert!(decoding_key_from_pem("-----BEGIN CERTIFICATE-----\nnope\n-----END CERTIFICATE-----").is_err()); } + + #[tokio::test] + async fn a_concurrent_cold_burst_makes_one_jwks_fetch() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::AsyncWriteExt; + // The stub listens on loopback, which SSRF validation refuses without this. + unsafe { std::env::set_var("ALLOW_PRIVATE_GUEST_JWKS_URLS", "true") }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let body = format!( + r#"{{"keys":[{{"kty":"EC","crv":"P-256","kid":"k1","x":"{PUB1_X}","y":"{PUB1_Y}"}}]}}"# + ); + let hits = std::sync::Arc::new(AtomicUsize::new(0)); + let hits_srv = hits.clone(); + tokio::spawn(async move { + loop { + let (mut sock, _) = listener.accept().await.unwrap(); + hits_srv.fetch_add(1, Ordering::SeqCst); + let resp = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + // Delay so the other callers pile onto the single-flight lock before the + // leader's fetch returns. + tokio::time::sleep(Duration::from_millis(150)).await; + let _ = sock.write_all(resp.as_bytes()).await; + let _ = sock.shutdown().await; + } + }); + let url = format!("http://127.0.0.1:{}/jwks.json", addr.port()); + let mut handles = Vec::new(); + for _ in 0..10 { + let u = url.clone(); + handles.push(tokio::spawn(async move { cached_jwks(&u).await.map(|e| e.keys.len()) })); + } + for h in handles { + assert_eq!(h.await.unwrap().unwrap(), 1, "each caller resolves the one key"); + } + assert_eq!( + hits.load(Ordering::SeqCst), + 1, + "single-flight: a concurrent cold burst makes one fetch" + ); + } }