diff --git a/crates/ghostlight-dungeon/Cargo.toml b/crates/ghostlight-dungeon/Cargo.toml index ff99b00..0692839 100644 --- a/crates/ghostlight-dungeon/Cargo.toml +++ b/crates/ghostlight-dungeon/Cargo.toml @@ -43,4 +43,5 @@ zeroize.workspace = true [dev-dependencies] tempfile = "3" +tokio = { workspace = true, features = ["test-util"] } tower = { version = "0.5", features = ["util"] } diff --git a/crates/ghostlight-dungeon/src/heimdall.rs b/crates/ghostlight-dungeon/src/heimdall.rs index 5177cc9..9b9e27b 100644 --- a/crates/ghostlight-dungeon/src/heimdall.rs +++ b/crates/ghostlight-dungeon/src/heimdall.rs @@ -45,10 +45,35 @@ pub(crate) struct HeimdallDenied { /// Heimdall's signing keys could not be fetched, so a receipt could not be /// judged either way. Distinct from a receipt that positively fails /// verification: the refresh loop keeps the receipt for this and revokes for -/// that. +/// that. It carries an error class and never the fetch error itself, whose +/// text names the key-set URL: an error never echoes an input value. #[derive(Debug, thiserror::Error)] -#[error("Heimdall signing keys unavailable: {0}")] -pub(crate) struct VerificationUnavailable(pub(crate) String); +#[error("Heimdall signing keys unavailable: the key-set fetch {class}")] +pub(crate) struct VerificationUnavailable { + class: &'static str, +} + +impl VerificationUnavailable { + fn of_fetch(error: &reqwest::Error) -> Self { + let class = if error.is_timeout() { + "timed out" + } else if error.is_connect() { + "could not connect" + } else if error.is_status() { + "returned an error status" + } else if error.is_decode() { + "returned an unreadable key set" + } else { + "failed" + }; + Self { class } + } + + #[cfg(test)] + pub(crate) fn new(class: &'static str) -> Self { + Self { class } + } +} /// How long a fetched signing-key set is trusted before it is fetched again. const JWKS_TTL: Duration = Duration::from_secs(300); @@ -491,6 +516,13 @@ impl HeimdallClient { } } + /// The same client against another issuer, for tests that stand up their own. + #[cfg(test)] + pub(crate) fn with_issuer(mut self, issuer: impl Into) -> Self { + self.issuer = issuer.into(); + self + } + pub async fn begin(&self, idempotency_key: &str) -> anyhow::Result { let payload = BeginCommand { provider: "discord", @@ -606,15 +638,14 @@ impl HeimdallClient { )?)) } - async fn fetch_jwks(&self) -> anyhow::Result { - Ok(self - .http + async fn fetch_jwks(&self) -> Result { + self.http .get(format!("{}/.well-known/jwks.json", self.issuer)) .send() .await? .error_for_status()? .json() - .await?) + .await } async fn verify_access(&self, token: &str) -> anyhow::Result { @@ -630,7 +661,7 @@ impl HeimdallClient { let set = self .fetch_jwks() .await - .map_err(|error| VerificationUnavailable(format!("{error:#}")))?; + .map_err(|error| VerificationUnavailable::of_fetch(&error))?; let jwk = set.find(&kid).cloned(); self.jwks .lock() @@ -1143,10 +1174,190 @@ fn required_env(name: &str) -> anyhow::Result { Ok(value.to_owned()) } +/// A local stand-in for Heimdall's issuer: it serves the key set (or a 5xx) +/// on a loopback port and counts the requests it receives. The key is the +/// RFC 8037 appendix A.1 Ed25519 test key. +#[cfg(test)] +pub(crate) mod test_issuer { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + }; + + pub(crate) const KID: &str = "k1"; + const PRIVATE_D: &str = "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A"; + const PUBLIC_X: &str = "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"; + + #[derive(Clone, Copy)] + pub(crate) enum Answer { + KeySet, + ServerError, + } + + pub(crate) struct Issuer { + pub(crate) url: String, + fetches: Arc, + } + + impl Issuer { + pub(crate) async fn start(answer: Answer) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let fetches = Arc::new(AtomicUsize::new(0)); + let counted = fetches.clone(); + let key_set = serde_json::json!({ + "keys": [{"kty": "OKP", "crv": "Ed25519", "kid": KID, "x": PUBLIC_X}] + }) + .to_string(); + tokio::spawn(async move { + while let Ok((mut stream, _)) = listener.accept().await { + let mut request = Vec::new(); + let mut chunk = [0_u8; 1024]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(read) => request.extend_from_slice(&chunk[..read]), + } + } + counted.fetch_add(1, Ordering::SeqCst); + let (status, body) = match answer { + Answer::KeySet => ("200 OK", key_set.as_str()), + Answer::ServerError => ("503 Service Unavailable", ""), + }; + let response = format!( + "HTTP/1.1 {status}\r\ncontent-type: application/json\r\n\ + content-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } + }); + Self { url, fetches } + } + + pub(crate) fn fetches(&self) -> usize { + self.fetches.load(Ordering::SeqCst) + } + + /// The URL of a loopback port nothing listens on. + pub(crate) fn closed_url() -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + format!("http://{}", listener.local_addr().unwrap()) + } + } + + /// An access token Heimdall's key signs for `issuer`, with the claims + /// `verify_access` demands. + pub(crate) fn signed_token(issuer: &str) -> String { + let mut der = vec![ + 0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, + 0x04, 0x20, + ]; + der.extend(URL_SAFE_NO_PAD.decode(PRIVATE_D).unwrap()); + let pem = format!( + "-----BEGIN PRIVATE KEY-----\n{}\n-----END PRIVATE KEY-----\n", + STANDARD.encode(der) + ); + let mut header = jsonwebtoken::Header::new(Algorithm::EdDSA); + header.kid = Some(KID.into()); + let claims = serde_json::json!({ + "iss": issuer, "aud": APP_SLUG, "sub": "account-1", "sid": "session-1", + "exp": Utc::now().timestamp() + 300, "typ": "heimdall_access", + "account_id": "account-1", "access_revision": 1, + "capabilities": ["app_access"], "app": {"slug": APP_SLUG}, + }); + jsonwebtoken::encode( + &header, + &claims, + &jsonwebtoken::EncodingKey::from_ed_pem(pem.as_bytes()).unwrap(), + ) + .unwrap() + } +} + #[cfg(test)] mod tests { use super::*; + use super::test_issuer::{Answer, Issuer, signed_token}; + + #[tokio::test] + async fn a_key_fetch_failure_is_typed_unavailable_and_names_no_url() { + let closed = Issuer::closed_url(); + let failing = Issuer::start(Answer::ServerError).await; + for (base, class) in [ + (closed, "could not connect"), + (failing.url.clone(), "returned an error status"), + ] { + let canary = format!("{base}/canary-q7x1"); + let client = HeimdallClient::fixture().with_issuer(canary.clone()); + let error = client + .verify_access(&signed_token(&canary)) + .await + .unwrap_err(); + assert!(error.downcast_ref::().is_some(), "{error:#}"); + for text in [format!("{error}"), format!("{error:#}"), format!("{error:?}")] { + assert!(text.contains("signing keys unavailable"), "{text}"); + assert!(text.contains(class), "{text}"); + for leaked in [canary.as_str(), base.as_str(), "canary-q7x1", "127.0.0.1"] { + assert!(!text.contains(leaked), "{leaked} leaked into {text}"); + } + } + } + } + + #[tokio::test] + async fn the_key_set_is_fetched_once_within_its_ttl_and_again_after_it() { + let issuer = Issuer::start(Answer::KeySet).await; + let client = HeimdallClient::fixture().with_issuer(issuer.url.clone()); + let token = signed_token(&issuer.url); + client.verify_access(&token).await.unwrap(); + assert_eq!(issuer.fetches(), 1); + client.verify_access(&token).await.unwrap(); + assert_eq!(issuer.fetches(), 1, "a second verify within the ttl reads the cache"); + let age = |seconds: u64| { + client.jwks.lock().unwrap().fetched.as_mut().unwrap().0 = + Instant::now().checked_sub(Duration::from_secs(seconds)).unwrap(); + }; + age(299); + client.verify_access(&token).await.unwrap(); + assert_eq!(issuer.fetches(), 1, "299 s old is still fresh"); + age(301); + client.verify_access(&token).await.unwrap(); + assert_eq!(issuer.fetches(), 2, "301 s old is fetched again"); + client.verify_access(&token).await.unwrap(); + assert_eq!(issuer.fetches(), 2, "and the new fetch is cached"); + } + + #[tokio::test] + async fn a_token_signed_under_any_other_algorithm_is_refused_before_any_key_fetch() { + let issuer = Issuer::start(Answer::KeySet).await; + let client = HeimdallClient::fixture().with_issuer(issuer.url.clone()); + let part = |text: &str| URL_SAFE_NO_PAD.encode(text); + for alg in [ + "none", "HS256", "HS384", "HS512", "RS256", "RS384", "PS256", "ES256", "ES384", + ] { + let token = format!( + "{}.{}.{}", + part(&format!(r#"{{"alg":"{alg}","typ":"JWT","kid":"k1"}}"#)), + part(r#"{"sub":"account-1"}"#), + URL_SAFE_NO_PAD.encode([0xAB_u8; 32]), + ); + let error = client.verify_access(&token).await.unwrap_err(); + assert!(error.downcast_ref::().is_none(), "{alg}"); + if alg != "none" { + assert!( + error.to_string().contains("unsupported signing algorithm"), + "{alg}: {error:#}" + ); + } + assert_eq!(issuer.fetches(), 0, "{alg} reached the key fetch"); + } + } + fn boundary(endpoint: &str) -> HeimdallCommandBoundaryRecord { HeimdallCommandBoundaryRecord { schema: "heimdall.command_boundary.v1".into(), @@ -1187,22 +1398,6 @@ mod tests { assert!(cache.key("k1", start + JWKS_TTL).is_none(), "stale at the ttl"); } - #[tokio::test] - async fn a_token_signed_with_another_algorithm_is_refused_before_any_key_fetch() { - let token = jsonwebtoken::encode( - &jsonwebtoken::Header::new(Algorithm::HS256), - &serde_json::json!({"sub": "x"}), - &jsonwebtoken::EncodingKey::from_secret(b"secret"), - ) - .unwrap(); - let error = HeimdallClient::fixture() - .verify_access(&token) - .await - .unwrap_err(); - assert!(error.to_string().contains("unsupported signing algorithm"), "{error:#}"); - assert!(error.downcast_ref::().is_none()); - } - #[test] fn a_denied_status_is_the_typed_denial_and_any_other_status_passes() { let diagnostics = vec!["Idempotency key was reused".to_owned()]; diff --git a/crates/ghostlight-dungeon/src/runtime.rs b/crates/ghostlight-dungeon/src/runtime.rs index e1574f4..fbd1db6 100644 --- a/crates/ghostlight-dungeon/src/runtime.rs +++ b/crates/ghostlight-dungeon/src/runtime.rs @@ -1863,23 +1863,21 @@ fn session_cookie(raw: &str, expires_at: DateTime, now: DateTime) -> H } async fn maintain_session_refresh(state: AppState) -> anyhow::Result<()> { - run_refresh_loop( - &state.sessions, - state.heimdall.as_ref(), - Duration::from_secs(60), - ) - .await + run_refresh_loop(&state.sessions, state.heimdall.as_ref()).await } -/// Runs `refresh_due_sessions` once per `period` for as long as the daemon -/// serves. The receipts it holds are Heimdall refreshes already spent -/// upstream whose verification could not yet be judged. +/// How often the refresh loop scans for sessions due for refresh. Decided +/// here and nowhere else. +const SESSION_REFRESH_PERIOD: Duration = Duration::from_secs(60); + +/// Runs `refresh_due_sessions` once per `SESSION_REFRESH_PERIOD` for as long +/// as the daemon serves. The receipts it holds are Heimdall refreshes already +/// spent upstream whose verification could not yet be judged. async fn run_refresh_loop( sessions: &Mutex, heimdall: &impl SessionRefresher, - period: Duration, ) -> anyhow::Result<()> { - let mut interval = tokio::time::interval(period); + let mut interval = tokio::time::interval(SESSION_REFRESH_PERIOD); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut unverified = HashMap::new(); loop { @@ -6481,6 +6479,9 @@ mod refresh_loop_tests { moving: Option>>, pending: StdMutex>, calls: StdMutex>, + /// Verify receipts with this real client, against the access token + /// each receipt carries, instead of from the verdict queue. + real: Option<(HeimdallClient, String)>, } impl ScriptedHeimdall { @@ -6496,6 +6497,11 @@ mod refresh_loop_tests { self } + fn verifying_for_real(mut self, client: HeimdallClient, token: String) -> Self { + self.real = Some((client, token)); + self + } + fn verifications(&self) -> usize { *self.verifications.lock().unwrap() } @@ -6566,7 +6572,7 @@ mod refresh_loop_tests { error: None, account: None, session: None, - access_token: None, + access_token: self.real.as_ref().map(|(_, token)| token.clone()), refresh_token: None, refresh: None, shared_capabilities: Vec::new(), @@ -6577,12 +6583,15 @@ mod refresh_loop_tests { async fn verify_refresh( &self, - _completion: AuthCompletionReceipt, + completion: AuthCompletionReceipt, ) -> anyhow::Result { *self.verifications.lock().unwrap() += 1; + if let Some((client, _)) = &self.real { + return client.verify_refresh(completion).await; + } match self.verdicts.lock().unwrap().pop_front() { Some(Verdict::Unavailable) => { - Err(VerificationUnavailable("jwks fetch timed out".into()).into()) + Err(VerificationUnavailable::new("timed out").into()) } Some(Verdict::Invalid) => bail!("Heimdall access token signature is invalid"), None => Ok(self.pending.lock().unwrap().take().expect("scripted receipt")), @@ -6814,6 +6823,12 @@ mod refresh_loop_tests { heimdall.moving = Some(sessions.clone()); pass(&sessions, &heimdall).await; assert!(is_live(&sessions, &cookie).await); + // The newer state wins: the next refresh spends its claim, not the + // stale receipt's. + heimdall.moving = None; + heimdall.replies.lock().unwrap().push_back(Reply::Timeout); + pass(&sessions, &heimdall).await; + assert_eq!(heimdall.calls()[1].0, "claim-moved"); } #[tokio::test] @@ -6832,16 +6847,82 @@ mod refresh_loop_tests { assert!(!is_live(&sessions, &cookie).await); } - #[tokio::test] - async fn the_refresh_loop_runs_a_pass_on_every_tick() { + /// Lets the loop under test run everything that is ready, without moving + /// the paused clock. + async fn settle() { + for _ in 0..16 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(start_paused = true)] + async fn the_refresh_loop_runs_a_pass_every_sixty_seconds_and_keeps_going() { let (_directory, sessions, _cookie) = due_session(1); - let heimdall = ScriptedHeimdall::answering((0..200).map(|_| Reply::Timeout).collect()); - let ran = tokio::time::timeout( - Duration::from_millis(400), - run_refresh_loop(&sessions, &heimdall, Duration::from_millis(10)), - ) - .await; - assert!(ran.is_err(), "the loop never returns"); - assert!(heimdall.calls().len() >= 2, "{:?}", heimdall.calls()); + let heimdall = ScriptedHeimdall::answering((0..64).map(|_| Reply::Timeout).collect()); + let observed = async { + settle().await; + assert_eq!(heimdall.calls().len(), 1, "the first pass runs at once"); + tokio::time::advance(Duration::from_millis(59_999)).await; + settle().await; + assert_eq!(heimdall.calls().len(), 1, "no pass before sixty seconds"); + tokio::time::advance(Duration::from_millis(1)).await; + settle().await; + assert_eq!(heimdall.calls().len(), 2, "the second pass at sixty seconds"); + for passes in 3..=12 { + tokio::time::advance(Duration::from_secs(60)).await; + settle().await; + assert_eq!(heimdall.calls().len(), passes); + } + }; + tokio::select! { + ran = run_refresh_loop(&sessions, &heimdall) => panic!("the loop returned: {ran:?}"), + () = observed => {} + } + } + + /// A refresh receipt verified by the real client against `issuer_url`, + /// whose key-set fetch fails. Returns whether the session stayed live, + /// how many receipts were kept, and the log the pass wrote. + async fn pass_through_real_verification(issuer_url: &str) -> (bool, usize, String) { + let (_directory, sessions, cookie) = due_session(1); + let client = HeimdallClient::fixture().with_issuer(issuer_url); + let token = heimdall::test_issuer::signed_token(issuer_url); + let heimdall = ScriptedHeimdall::answering(vec![Reply::Receipt { + revision: 1, + claim: "claim-1", + }]) + .verifying_for_real(client, token); + let captured = Captured::default(); + let subscriber = tracing_subscriber::fmt() + .with_writer(captured.clone()) + .with_ansi(false) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + let mut unverified = HashMap::new(); + refresh_due_sessions(&sessions, &heimdall, &mut unverified).await.unwrap(); + let logged = String::from_utf8(captured.0.lock().unwrap().clone()).unwrap(); + (is_live(&sessions, &cookie).await, unverified.len(), logged) + } + + #[tokio::test] + async fn a_key_fetch_failure_in_the_real_verification_keeps_the_session_and_the_receipt() { + let failing = heimdall::test_issuer::Issuer::start(heimdall::test_issuer::Answer::ServerError) + .await; + for issuer_url in [heimdall::test_issuer::Issuer::closed_url(), failing.url.clone()] { + let (live, kept, logged) = pass_through_real_verification(&issuer_url).await; + assert!(live, "{issuer_url}: a key-fetch failure signed the session out"); + assert_eq!(kept, 1, "{issuer_url}: the spent receipt was not kept"); + assert!(logged.contains("could not be verified yet"), "{logged}"); + } + } + + #[tokio::test] + async fn the_unverified_warning_names_the_error_class_and_no_url() { + let canary = format!("{}/canary-q7x1", heimdall::test_issuer::Issuer::closed_url()); + let (_, _, logged) = pass_through_real_verification(&canary).await; + assert!(logged.contains("could not connect"), "{logged}"); + for leaked in [canary.as_str(), "canary-q7x1", "127.0.0.1", "jwks.json"] { + assert!(!logged.contains(leaked), "{leaked} leaked into the log: {logged}"); + } } }