GameCult/Ghostlight
Watch
9
0
Fork
You've already forked Ghostlight
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refresh tests pin the production edges: key-fetch typing, cache wiring, algorithm gate, cadence

Browse source 
VerificationUnavailable now carries an error class and never the fetch error,
whose text named the key-set URL. fetch_jwks returns the reqwest error so the
class is read from it, not from a string. run_refresh_loop loses its period
parameter: SESSION_REFRESH_PERIOD is decided in one place and a paused-clock
test observes it exactly and over twelve ticks.

Tests drive HeimdallClient::verify_access against a loopback issuer that
counts fetches (cache stored, read, ttl 299 s fresh and 301 s stale), a closed
port and a 503 (kept session and receipt, no URL in the error or the log line),
and nine wrong-algorithm tokens. The moved-session test also pins that the
newer claim is the one spent next.

Builds and passes under ygg-verify (28 of 28 in heimdall:: and refresh_loop_tests).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-09 20:13:25 +02:00
commit 26f533e8f4
3 changed files with 325 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

1
crates/ghostlight-dungeon/Cargo.toml
View file

@ -43,4 +43,5 @@ zeroize.workspace = true
[dev-dependencies]
tempfile = "3"
tokio = { workspace = true, features = ["test-util"] }
tower = { version = "0.5", features = ["util"] }

243
crates/ghostlight-dungeon/src/heimdall.rs
View file

@ -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<String>) -> Self {
self.issuer = issuer.into();
self
}
pub async fn begin(&self, idempotency_key: &str) -> anyhow::Result<AuthBeginReceipt> {
let payload = BeginCommand {
provider: "discord",
@ -606,15 +638,14 @@ impl HeimdallClient {
)?))
}
async fn fetch_jwks(&self) -> anyhow::Result<JwkSet> {
Ok(self
.http
async fn fetch_jwks(&self) -> Result<JwkSet, reqwest::Error> {
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<AccessClaims> {
@ -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<String> {
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<AtomicUsize>,
}
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::<VerificationUnavailable>().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::<VerificationUnavailable>().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::<VerificationUnavailable>().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()];

129
crates/ghostlight-dungeon/src/runtime.rs
View file

@ -1863,23 +1863,21 @@ fn session_cookie(raw: &str, expires_at: DateTime<Utc>, now: DateTime<Utc>) -> 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<AppSessionOwner>,
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<std::sync::Arc<Mutex<AppSessionOwner>>>,
pending: StdMutex<Option<VerifiedSessionRefresh>>,
calls: StdMutex<Vec<(String, String)>>,
/// 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<VerifiedSessionRefresh> {
*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}");
}
}
}