postrust_auth/jwks.rs
1//! Keys fetched from a JWKS URL (`PGRST_JWT_JWKS_URL`), and kept current.
2//!
3//! An identity provider -- Supabase Auth (GoTrue), Auth0, Clerk, Cognito,
4//! Keycloak -- publishes the public keys its tokens are signed with as a JSON
5//! Web Key Set at a well-known URL, and rotates them: a new key is published,
6//! tokens start being signed with it, and the old one is withdrawn once no
7//! token signed by it can still be valid. A server that read the set once, at
8//! start-up, would refuse every token signed after the first rotation. So the
9//! set is fetched again, on a schedule, and whenever a token names a `kid`
10//! this server has never seen.
11//!
12//! # Where the waiting happens
13//!
14//! Verifying a token is synchronous ([`crate::validate_token`]), and stays
15//! so: the keys a request verifies against are a snapshot taken from
16//! [`JwtKeys`], swapped whole when a fetch brings a new set. The only
17//! asynchronous step is [`Jwks::ensure_for_token`] (reached through
18//! [`crate::ensure_keys`]), which a host awaits *before* the synchronous
19//! check. It reads the token's header, and if the snapshot already holds a key
20//! the header selects it returns at once without awaiting anything -- the
21//! common case, every request but the few after a rotation. Only an unknown
22//! `kid`, or keys never loaded, or (on a host with no background task) keys
23//! past their refresh time, make it fetch.
24//!
25//! # Fetching
26//!
27//! The HTTP request itself is the host's: [`Fetch`] is implemented with
28//! reqwest by the server and with the Workers Fetch API by the Worker, so
29//! this crate stays free of an HTTP client and builds for wasm32. What is
30//! fetched, and whether the answer is trusted, is decided here, once for every
31//! host:
32//!
33//! - **HTTPS only**, unless the host is loopback (`localhost`, `127.0.0.0/8`,
34//! `::1`) -- a development IdP, or a sidecar. A key set fetched over plain
35//! HTTP is a key set anyone on the path can replace with their own, and then
36//! sign any token they like. Checked when the URL is configured, and again
37//! for every redirect.
38//! - **No redirect to another host.** The URL the operator configured names
39//! the party trusted to say which keys are valid; a redirect elsewhere is
40//! that party delegating trust in a way nobody reviewed. A redirect on the
41//! same host (`/jwks` to `/jwks/`, http to https on loopback) is followed,
42//! at most [`MAX_REDIRECTS`] times. The [`Fetch`] implementation must not
43//! follow redirects itself, so that this rule is the only one.
44//! - **[`FETCH_TIMEOUT`] and [`MAX_JWKS_BYTES`]**, enforced by the [`Fetch`]
45//! implementation, which is the only party that can stop reading: a JWKS
46//! endpoint that hangs or streams forever must not hold a request, or the
47//! refresh task, hostage. Real key sets are a few kilobytes.
48//! - The body is parsed with [`KeySet::from_jwk_json`], exactly as a JWKS in
49//! `PGRST_JWT_SECRET` is. A response that is not a 200, or not a set this
50//! server can use, is a failed refresh.
51//!
52//! # Failing
53//!
54//! A failed refresh keeps the last good set and logs a warning: an identity
55//! provider having a bad minute must not log every user out. If no set has
56//! ever loaded, every token the JWKS would have verified is refused -- fail
57//! closed -- and [`Jwks::status`] says so, which the server's readiness
58//! endpoints report as not ready. Start-up does not fail: an IdP outage that
59//! crash-looped every replica would turn a partial outage into a total one.
60//!
61//! # Refreshing
62//!
63//! A successful fetch is trusted for its `Cache-Control: max-age`, clamped to
64//! [`MIN_MAX_AGE`]..=[`MAX_MAX_AGE`], or for the configured refresh interval
65//! (`PGRST_JWT_JWKS_REFRESH`, default [`DEFAULT_REFRESH`]) when the response
66//! says nothing. The provider knows how often it rotates; the clamp keeps a
67//! `max-age=0` from turning into a fetch per request and a `max-age` of a year
68//! from making a withdrawn key valid for a year. A failed fetch is retried
69//! after 1 s, 2 s, 4 s... up to [`MAX_RETRY`] (and never later than the
70//! refresh interval).
71//!
72//! A token whose `kid` is unknown triggers one fetch, at most once per
73//! [`UNKNOWN_KID_INTERVAL`] -- anyone can mint a token with a random `kid`,
74//! and without the limit every such request would be a request to the
75//! identity provider. Requests that arrive while that fetch is in flight wait
76//! for it rather than each starting their own (natively; a Worker's requests
77//! never wait on one another, see `Jwks::exclusive`); if the `kid` is still
78//! unknown afterwards the token is refused with PGRST301, as any token is
79//! whose key this server does not hold.
80//!
81//! # With `PGRST_JWT_SECRET`
82//!
83//! Both may be set, and a token is accepted if either verifies it: the keys
84//! in force are the union of the secret's and the fetched set's. That is the
85//! deployment where a backend mints HS256 service tokens with a shared secret
86//! while users arrive with tokens from the identity provider -- Supabase's own
87//! legacy `service_role` key is exactly that. Neither source can widen the
88//! other: a key still admits only its own algorithms, and a token naming a
89//! `kid` still selects only keys carrying that `kid` (the plain-string secret,
90//! which has no `kid`, has always accepted any, and still does).
91
92use std::sync::atomic::{AtomicBool, Ordering};
93use std::sync::{Arc, Mutex};
94
95use jsonwebtoken::Algorithm;
96use web_time::{Duration, Instant, SystemTime, UNIX_EPOCH};
97
98use crate::{JwtKeys, KeySet};
99use address::Address;
100
101mod address;
102
103/// How long one fetch may take, connecting and reading included.
104pub const FETCH_TIMEOUT: Duration = Duration::from_secs(5);
105
106/// The largest JWKS response read, in bytes: 1 MiB.
107pub const MAX_JWKS_BYTES: usize = 1 << 20;
108
109/// How many same-host redirects are followed before the fetch is a failure.
110pub const MAX_REDIRECTS: usize = 3;
111
112/// How long a fetched set is used without being fetched again, when neither
113/// the response nor `PGRST_JWT_JWKS_REFRESH` says otherwise: five minutes.
114pub const DEFAULT_REFRESH: Duration = Duration::from_secs(300);
115
116/// The shortest `Cache-Control: max-age` honoured. Shorter, `no-cache` and
117/// `no-store` all mean this.
118pub const MIN_MAX_AGE: Duration = Duration::from_secs(60);
119
120/// The longest `Cache-Control: max-age` honoured: a day.
121pub const MAX_MAX_AGE: Duration = Duration::from_secs(86_400);
122
123/// How often a token with an unknown `kid` may cause a fetch.
124pub const UNKNOWN_KID_INTERVAL: Duration = Duration::from_secs(30);
125
126/// The longest wait before retrying a failed fetch.
127pub const MAX_RETRY: Duration = Duration::from_secs(30);
128
129/// `Send + Sync` natively, nothing at all on wasm32, where a Worker's
130/// JavaScript objects are neither and nothing runs on another thread anyway.
131#[cfg(not(target_arch = "wasm32"))]
132pub trait MaybeSendSync: Send + Sync {}
133#[cfg(not(target_arch = "wasm32"))]
134impl<T: Send + Sync + ?Sized> MaybeSendSync for T {}
135
136/// `Send + Sync` natively, nothing at all on wasm32. See the native definition.
137#[cfg(target_arch = "wasm32")]
138pub trait MaybeSendSync {}
139#[cfg(target_arch = "wasm32")]
140impl<T: ?Sized> MaybeSendSync for T {}
141
142/// What one `GET` answered.
143#[derive(Clone, Debug)]
144#[non_exhaustive]
145pub struct FetchResponse {
146 /// The HTTP status.
147 pub status: u16,
148 /// The `Location` header, for a redirect.
149 pub location: Option<String>,
150 /// The `Cache-Control` header.
151 pub cache_control: Option<String>,
152 /// The body, no longer than [`MAX_JWKS_BYTES`].
153 pub body: Vec<u8>,
154}
155
156impl FetchResponse {
157 /// A response with this status and body, and no headers.
158 pub fn new(status: u16, body: Vec<u8>) -> Self {
159 Self {
160 status,
161 location: None,
162 cache_control: None,
163 body,
164 }
165 }
166
167 /// With a `Location` header.
168 pub fn with_location(mut self, location: Option<String>) -> Self {
169 self.location = location;
170 self
171 }
172
173 /// With a `Cache-Control` header.
174 pub fn with_cache_control(mut self, cache_control: Option<String>) -> Self {
175 self.cache_control = cache_control;
176 self
177 }
178}
179
180/// A fetch or a refresh that failed, and why, for the log.
181#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
182#[error("{0}")]
183pub struct FetchError(String);
184
185impl FetchError {
186 /// A failure described by `reason`.
187 pub fn new(reason: impl Into<String>) -> Self {
188 Self(reason.into())
189 }
190}
191
192/// An HTTP client, as much of one as fetching a JWKS needs.
193///
194/// Implementations must, so that the rules in the [module documentation](self)
195/// hold on every host:
196///
197/// - **not follow redirects**, but answer a 3xx with its `Location`;
198/// - give up after [`FETCH_TIMEOUT`];
199/// - stop reading, and fail, once a body exceeds [`MAX_JWKS_BYTES`].
200#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
201#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
202pub trait Fetch: MaybeSendSync {
203 /// `GET` an absolute `http` or `https` URL this module has already
204 /// checked.
205 async fn get(&self, url: &str) -> Result<FetchResponse, FetchError>;
206}
207
208/// A copy of the fetched set somewhere other processes can read it.
209///
210/// For a host whose processes come and go faster than the set changes: a
211/// Cloudflare Worker keeps it in KV, so that an isolate starting cold reads
212/// what another fetched a minute ago instead of asking the identity provider
213/// again. A stored copy is used only while it is fresh, and never to answer
214/// an unknown `kid` -- that question is only answered by the provider.
215#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
216#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
217pub trait JwksStore: MaybeSendSync {
218 /// The stored copy for `url`, if there is one.
219 async fn load(&self, url: &str) -> Option<StoredJwks>;
220 /// Store `jwks` for `url`. It may be dropped once `ttl` has passed.
221 /// Failing is not an error worth more than a log line.
222 async fn save(&self, url: &str, jwks: &StoredJwks, ttl: Duration);
223}
224
225/// A fetched set, as a [`JwksStore`] keeps it.
226#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
227pub struct StoredJwks {
228 /// The JWKS document, as fetched.
229 pub body: String,
230 /// Until when it may be used without fetching again, in seconds since the
231 /// Unix epoch.
232 pub fresh_until: u64,
233}
234
235/// What a [`Jwks`] has done so far: the numbers a metric or a readiness check
236/// reads. A snapshot, taken by [`Jwks::status`].
237#[derive(Clone, Debug)]
238#[non_exhaustive]
239pub struct JwksStatus {
240 /// Whether a set has ever been fetched. Until it has, every token the
241 /// JWKS would have verified is refused.
242 pub loaded: bool,
243 /// How many keys the fetched set holds.
244 pub keys: usize,
245 /// Successful fetches, a fresh copy read from a [`JwksStore`] included.
246 pub refreshes: u64,
247 /// Failed fetches.
248 pub failures: u64,
249 /// Failed fetches since the last successful one.
250 pub consecutive_failures: u32,
251 /// How many times a fetch brought a set different from the one held.
252 pub changes: u64,
253 /// When the last successful fetch was.
254 pub last_success: Option<SystemTime>,
255 /// How long ago the set held was last confirmed by a successful fetch.
256 pub age: Option<Duration>,
257 /// How long a fetched set is used for: the last response's `max-age`,
258 /// clamped, or the configured interval.
259 pub refresh_interval: Duration,
260 /// How long until the next fetch is due.
261 pub next_refresh_in: Duration,
262 /// Why the last fetch failed, while it is the most recent outcome.
263 pub last_error: Option<String>,
264}
265
266/// A JWKS URL that cannot be fetched from.
267#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
268#[error("{0}")]
269pub struct JwksUrlError(String);
270
271/// Check a JWKS URL against the rules in the [module documentation](self):
272/// absolute, `https`, or `http` to a loopback host.
273pub fn check_url(url: &str) -> Result<(), JwksUrlError> {
274 parse_url(url).map(drop)
275}
276
277fn parse_url(url: &str) -> Result<Address, JwksUrlError> {
278 let parsed = Address::parse(url).map_err(|e| JwksUrlError(format!("{url:?}: {e}")))?;
279 check_parsed(&parsed)?;
280 Ok(parsed)
281}
282
283fn check_parsed(url: &Address) -> Result<(), JwksUrlError> {
284 if url.is_https() || url.is_loopback() {
285 return Ok(());
286 }
287 Err(JwksUrlError(format!(
288 "{url}: a key set fetched over plain HTTP can be replaced by anyone on the path; \
289 use https (http is accepted only for localhost, 127.0.0.0/8 and [::1])"
290 )))
291}
292
293/// How long a response says it may be used for: its `max-age`, clamped, or
294/// [`MIN_MAX_AGE`] for `no-cache` and `no-store`; `None` when it says nothing.
295fn max_age(cache_control: Option<&str>) -> Option<Duration> {
296 let mut found = None;
297 for directive in cache_control?.split(',') {
298 let directive = directive.trim();
299 let (name, value) = directive.split_once('=').unwrap_or((directive, ""));
300 match name.trim().to_ascii_lowercase().as_str() {
301 "no-cache" | "no-store" => return Some(MIN_MAX_AGE),
302 "max-age" => {
303 if let Ok(secs) = value.trim().trim_matches('"').parse::<u64>() {
304 found = Some(Duration::from_secs(secs));
305 }
306 }
307 _ => {}
308 }
309 }
310 found.map(|age| age.clamp(MIN_MAX_AGE, MAX_MAX_AGE))
311}
312
313/// How long to wait after the `failures`th failure in a row.
314fn retry_after(failures: u32, interval: Duration) -> Duration {
315 let backoff = Duration::from_secs(1u64 << failures.saturating_sub(1).min(5));
316 backoff.min(MAX_RETRY).min(interval)
317}
318
319fn unix_now() -> u64 {
320 SystemTime::now()
321 .duration_since(UNIX_EPOCH)
322 .map(|d| d.as_secs())
323 .unwrap_or(0)
324}
325
326/// Builds a [`Jwks`]; see [`Jwks::builder`].
327pub struct JwksBuilder {
328 url: String,
329 fetch: Box<dyn Fetch>,
330 fixed: Option<Arc<KeySet>>,
331 refresh: Duration,
332 unknown_kid_interval: Duration,
333 store: Option<Box<dyn JwksStore>>,
334}
335
336impl JwksBuilder {
337 /// Accept the keys of `PGRST_JWT_SECRET` as well as the fetched ones.
338 /// See "With `PGRST_JWT_SECRET`" in the [module documentation](self).
339 pub fn fixed(mut self, keys: Option<&JwtKeys>) -> Self {
340 self.fixed = keys.map(JwtKeys::current);
341 self
342 }
343
344 /// How long a set is used when the response carries no `max-age`
345 /// (`PGRST_JWT_JWKS_REFRESH`). Default [`DEFAULT_REFRESH`]; zero is
346 /// read as one second.
347 pub fn refresh(mut self, interval: Duration) -> Self {
348 self.refresh = interval.max(Duration::from_secs(1));
349 self
350 }
351
352 /// How often an unknown `kid` may cause a fetch. Default
353 /// [`UNKNOWN_KID_INTERVAL`]; for tests.
354 pub fn unknown_kid_interval(mut self, interval: Duration) -> Self {
355 self.unknown_kid_interval = interval;
356 self
357 }
358
359 /// Keep a copy of every fetched set in `store`, and read a fresh one from
360 /// there before fetching.
361 pub fn store(mut self, store: impl JwksStore + 'static) -> Self {
362 self.store = Some(Box::new(store));
363 self
364 }
365
366 /// The [`Jwks`], or the reason its URL cannot be fetched from.
367 pub fn build(self) -> Result<Jwks, JwksUrlError> {
368 let url = parse_url(&self.url)?;
369 let keys = JwtKeys::new(match &self.fixed {
370 Some(fixed) => KeySet::union(fixed, &KeySet::empty()),
371 None => KeySet::empty(),
372 });
373 // On wasm32 the fetcher is a JavaScript object and `Inner` is not
374 // `Send`; nothing there runs on another thread, and one type for both
375 // targets is worth more than the `Rc` the lint suggests.
376 #[cfg_attr(target_arch = "wasm32", allow(clippy::arc_with_non_send_sync))]
377 Ok(Jwks {
378 inner: Arc::new(Inner {
379 url_text: url.to_string(),
380 url,
381 fetch: self.fetch,
382 store: self.store,
383 keys,
384 fixed: self.fixed,
385 unknown_kid_interval: self.unknown_kid_interval,
386 background: AtomicBool::new(false),
387 state: Mutex::new(State {
388 body: None,
389 keys: 0,
390 last_success: None,
391 interval: self.refresh,
392 configured: self.refresh,
393 next_due: Instant::now(),
394 consecutive_failures: 0,
395 refreshes: 0,
396 failures: 0,
397 changes: 0,
398 last_error: None,
399 last_unknown_kid: None,
400 }),
401 refreshing: futures::lock::Mutex::new(()),
402 }),
403 })
404 }
405}
406
407/// The keys at a JWKS URL, fetched and kept current. See the
408/// [module documentation](self).
409///
410/// Cloning shares it. [`Jwks::keys`] is the handle a [`crate::JwtConfig`]
411/// verifies with; [`crate::JwtConfig::with_jwks`] sets both.
412#[derive(Clone)]
413pub struct Jwks {
414 inner: Arc<Inner>,
415}
416
417struct Inner {
418 url: Address,
419 /// `url`, as text, for [`Jwks::url`].
420 url_text: String,
421 fetch: Box<dyn Fetch>,
422 store: Option<Box<dyn JwksStore>>,
423 /// The keys in force: the fixed ones and the last good fetched set.
424 keys: JwtKeys,
425 fixed: Option<Arc<KeySet>>,
426 unknown_kid_interval: Duration,
427 /// Whether a background task keeps the set fresh. When none does, a
428 /// request finding the set due for a refresh does it.
429 background: AtomicBool,
430 state: Mutex<State>,
431 /// Held for the length of a fetch, so that only one runs at a time and a
432 /// request needing its outcome can wait for it (natively; see
433 /// [`Jwks::exclusive`]). An async lock with no runtime behind it: the
434 /// futures crate's.
435 refreshing: futures::lock::Mutex<()>,
436}
437
438struct State {
439 /// The document the set in force was parsed from. A fetch bringing the
440 /// same bytes changes nothing -- in particular, it does not empty the
441 /// token cache.
442 body: Option<String>,
443 keys: usize,
444 last_success: Option<(Instant, SystemTime)>,
445 interval: Duration,
446 configured: Duration,
447 next_due: Instant,
448 consecutive_failures: u32,
449 refreshes: u64,
450 failures: u64,
451 changes: u64,
452 last_error: Option<String>,
453 last_unknown_kid: Option<Instant>,
454}
455
456impl std::fmt::Debug for Jwks {
457 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
458 f.debug_struct("Jwks")
459 .field("url", &self.inner.url_text)
460 .field("keys", &self.inner.keys)
461 .finish_non_exhaustive()
462 }
463}
464
465/// Why a fetch is being made, which decides whether a stored copy will do.
466#[derive(Clone, Copy, PartialEq, Eq)]
467enum Reason {
468 /// Due, or never loaded: a fresh stored copy is as good as a fetch.
469 AnySource,
470 /// A `kid` nobody here has seen, or a schedule writing the store for
471 /// everyone else. Only the provider can answer.
472 ProviderOnly,
473}
474
475impl Jwks {
476 /// Start building one for `url`, fetched through `fetch`.
477 pub fn builder(url: impl Into<String>, fetch: impl Fetch + 'static) -> JwksBuilder {
478 JwksBuilder {
479 url: url.into(),
480 fetch: Box::new(fetch),
481 fixed: None,
482 refresh: DEFAULT_REFRESH,
483 unknown_kid_interval: UNKNOWN_KID_INTERVAL,
484 store: None,
485 }
486 }
487
488 /// The keys in force, for [`crate::JwtConfig::keys`].
489 pub fn keys(&self) -> &JwtKeys {
490 &self.inner.keys
491 }
492
493 /// The URL the keys are fetched from.
494 pub fn url(&self) -> &str {
495 &self.inner.url_text
496 }
497
498 /// Whether a set has ever been fetched.
499 pub fn is_loaded(&self) -> bool {
500 self.state().last_success.is_some()
501 }
502
503 /// What has happened so far.
504 pub fn status(&self) -> JwksStatus {
505 let state = self.state();
506 let now = Instant::now();
507 JwksStatus {
508 loaded: state.last_success.is_some(),
509 keys: state.keys,
510 refreshes: state.refreshes,
511 failures: state.failures,
512 consecutive_failures: state.consecutive_failures,
513 changes: state.changes,
514 last_success: state.last_success.map(|(_, at)| at),
515 age: state
516 .last_success
517 .map(|(at, _)| now.saturating_duration_since(at)),
518 refresh_interval: state.interval,
519 next_refresh_in: state.next_due.saturating_duration_since(now),
520 last_error: state.last_error.clone(),
521 }
522 }
523
524 /// How long until the next fetch is due.
525 pub fn next_refresh_in(&self) -> Duration {
526 self.state()
527 .next_due
528 .saturating_duration_since(Instant::now())
529 }
530
531 /// Fetch the set now, whenever it was last fetched, waiting for any fetch
532 /// already in flight first. The outcome is also in [`Jwks::status`].
533 ///
534 /// Always from the provider, never from a [`JwksStore`]: this is what a
535 /// schedule calls, and what writes the store for everyone else.
536 pub async fn refresh(&self) -> Result<(), FetchError> {
537 let _fetching = self.exclusive().await;
538 self.fetch_and_install(Reason::ProviderOnly).await
539 }
540
541 /// Keep the set fresh until the end of time: fetch whenever the next
542 /// fetch is due, sleeping with `sleep` in between.
543 ///
544 /// For a host that runs background tasks -- the server spawns this on its
545 /// runtime with `tokio::time::sleep`. Once it runs, requests no longer
546 /// refresh a set that is merely due; they still fetch for an unknown
547 /// `kid`.
548 pub async fn refresh_forever<S, F>(&self, mut sleep: S)
549 where
550 S: FnMut(Duration) -> F,
551 F: std::future::Future<Output = ()>,
552 {
553 self.inner.background.store(true, Ordering::Relaxed);
554 loop {
555 sleep(self.next_refresh_in()).await;
556 // The outcome is logged and kept in the status; the schedule
557 // follows from it either way.
558 let _ = self.refresh().await;
559 }
560 }
561
562 /// Make sure the keys a token needs are here, fetching if they may not
563 /// be. Never fails: a token still without a key is refused by the
564 /// synchronous check that follows, as PGRST301.
565 ///
566 /// Returns without awaiting anything when the set in force holds a key
567 /// for the token's header -- unless no background task keeps the set
568 /// fresh and it is due, when the request that notices fetches it (and
569 /// one that finds a fetch already running goes on with the set it has).
570 pub async fn ensure_for_token(&self, token: &str) {
571 // A header that does not parse is refused by the check that follows,
572 // and no fetch could change that.
573 let Ok(header) = jsonwebtoken::decode_header(token) else {
574 return;
575 };
576 let kid = header.kid.as_deref();
577 if self.inner.keys.current().selects(kid, header.alg) {
578 if !self.inner.background.load(Ordering::Relaxed) && self.due() {
579 self.refresh_if_due().await;
580 }
581 return;
582 }
583 self.refresh_for(kid, header.alg).await;
584 }
585
586 /// The right to fetch, waiting for a fetch already in flight to finish.
587 ///
588 /// Natively, a request that needs a fetch's outcome waits for the one in
589 /// flight rather than starting its own. On wasm32 it never waits: in a
590 /// Worker every request is its own I/O context, and a request woken by
591 /// another's fetch completing would go on running -- and doing its own
592 /// I/O -- inside the other's context, which the Workers runtime refuses
593 /// and can kill the isolate for. So there a request that finds a fetch in
594 /// flight makes its own (the unknown-`kid` limit still applies, and a
595 /// cold isolate's fetches are usually reads of the [`JwksStore`]).
596 async fn exclusive(&self) -> Option<futures::lock::MutexGuard<'_, ()>> {
597 #[cfg(not(target_arch = "wasm32"))]
598 return Some(self.inner.refreshing.lock().await);
599 #[cfg(target_arch = "wasm32")]
600 return self.inner.refreshing.try_lock();
601 }
602
603 fn state(&self) -> std::sync::MutexGuard<'_, State> {
604 self.inner.state.lock().unwrap_or_else(|e| e.into_inner())
605 }
606
607 fn due(&self) -> bool {
608 Instant::now() >= self.state().next_due
609 }
610
611 /// A lazy host's scheduled refresh, done by the request that noticed it
612 /// was due. One request does it; the rest go on with the set they have.
613 async fn refresh_if_due(&self) {
614 let Some(_fetching) = self.inner.refreshing.try_lock() else {
615 return;
616 };
617 if self.due() {
618 let _ = self.fetch_and_install(Reason::AnySource).await;
619 }
620 }
621
622 /// A token selects no key: fetch, unless one was fetched for the same
623 /// reason too recently, and wait for the outcome.
624 async fn refresh_for(&self, kid: Option<&str>, alg: Algorithm) {
625 let _fetching = self.exclusive().await;
626 // Whoever held the lock may have brought the key.
627 if self.inner.keys.current().selects(kid, alg) {
628 return;
629 }
630 let reason = {
631 let mut state = self.state();
632 if state.last_success.is_none() && Instant::now() >= state.next_due {
633 // Never loaded, and due to be tried again: the schedule
634 // already rate-limits this, so it does not spend the
635 // unknown-kid allowance.
636 Reason::AnySource
637 } else {
638 let now = Instant::now();
639 if let Some(last) = state.last_unknown_kid {
640 if now.saturating_duration_since(last) < self.inner.unknown_kid_interval {
641 return;
642 }
643 }
644 state.last_unknown_kid = Some(now);
645 Reason::ProviderOnly
646 }
647 };
648 let _ = self.fetch_and_install(reason).await;
649 }
650
651 /// Fetch -- or read a fresh stored copy, where `reason` allows -- and
652 /// install what comes back, keeping the books either way. The caller
653 /// holds `refreshing`.
654 async fn fetch_and_install(&self, reason: Reason) -> Result<(), FetchError> {
655 if reason == Reason::AnySource {
656 if let Some(store) = &self.inner.store {
657 if let Some(stored) = store.load(self.url()).await {
658 let now = unix_now();
659 if stored.fresh_until > now && self.install(&stored.body).is_ok() {
660 self.succeeded(Duration::from_secs(stored.fresh_until - now));
661 return Ok(());
662 }
663 }
664 }
665 }
666
667 let fetched = self.download().await.and_then(|(body, max_age)| {
668 self.install(&body)?;
669 Ok((body, max_age))
670 });
671 match fetched {
672 Ok((body, max_age)) => {
673 let interval = max_age.unwrap_or(self.state().configured);
674 self.succeeded(interval);
675 if let Some(store) = &self.inner.store {
676 let stored = StoredJwks {
677 body,
678 fresh_until: unix_now() + interval.as_secs(),
679 };
680 store.save(self.url(), &stored, interval).await;
681 }
682 Ok(())
683 }
684 Err(e) => {
685 let loaded = {
686 let mut state = self.state();
687 state.failures += 1;
688 state.consecutive_failures += 1;
689 state.last_error = Some(e.to_string());
690 state.next_due =
691 Instant::now() + retry_after(state.consecutive_failures, state.interval);
692 state.last_success.is_some()
693 };
694 if loaded {
695 tracing::warn!(
696 "PGRST_JWT_JWKS_URL: refreshing the keys from {} failed, keeping the \
697 ones fetched before: {e}",
698 self.url()
699 );
700 } else {
701 tracing::warn!(
702 "PGRST_JWT_JWKS_URL: fetching the keys from {} failed, and no keys \
703 from it have loaded yet, so tokens signed by them are refused: {e}",
704 self.url()
705 );
706 }
707 Err(e)
708 }
709 }
710 }
711
712 fn succeeded(&self, interval: Duration) {
713 let mut state = self.state();
714 let now = Instant::now();
715 state.last_success = Some((now, SystemTime::now()));
716 state.refreshes += 1;
717 state.consecutive_failures = 0;
718 state.last_error = None;
719 state.interval = interval;
720 state.next_due = now + interval;
721 }
722
723 /// Parse `body` and put it in force, unless it is what is in force
724 /// already. Parsing only when it changed also keeps the warnings about
725 /// keys set aside to one per change rather than one per refresh.
726 fn install(&self, body: &str) -> Result<(), FetchError> {
727 if self.state().body.as_deref() == Some(body) {
728 return Ok(());
729 }
730 let fetched = KeySet::from_jwk_json(body)
731 .map_err(|e| FetchError::new(format!("the response is not a usable JWKS: {e}")))?;
732 let count = fetched.len();
733 let set = match &self.inner.fixed {
734 Some(fixed) => KeySet::union(fixed, &fetched),
735 None => fetched,
736 };
737 let mut state = self.state();
738 self.inner.keys.replace(set);
739 state.body = Some(body.to_string());
740 state.keys = count;
741 state.changes += 1;
742 tracing::info!(
743 "PGRST_JWT_JWKS_URL: {count} key(s) in force from {}",
744 self.inner.url
745 );
746 Ok(())
747 }
748
749 /// `GET` the URL, following same-host redirects, and return the body and
750 /// how long it says it may be used for.
751 async fn download(&self) -> Result<(String, Option<Duration>), FetchError> {
752 let mut url = self.inner.url.clone();
753 for _ in 0..=MAX_REDIRECTS {
754 let response = self.inner.fetch.get(&url.to_string()).await?;
755 match response.status {
756 200 => {
757 if response.body.len() > MAX_JWKS_BYTES {
758 return Err(FetchError::new(format!(
759 "the response is larger than {MAX_JWKS_BYTES} bytes"
760 )));
761 }
762 let body = String::from_utf8(response.body)
763 .map_err(|_| FetchError::new("the response is not UTF-8"))?;
764 return Ok((body, max_age(response.cache_control.as_deref())));
765 }
766 301 | 302 | 303 | 307 | 308 => {
767 let location = response.location.ok_or_else(|| {
768 FetchError::new(format!("HTTP {} with no Location", response.status))
769 })?;
770 let next = url.join(&location).map_err(|e| {
771 FetchError::new(format!("a redirect to {location:?} is not followed: {e}"))
772 })?;
773 if next.host() != url.host() {
774 return Err(FetchError::new(format!(
775 "a redirect to another host ({next}) is not followed"
776 )));
777 }
778 check_parsed(&next).map_err(|e| FetchError::new(e.to_string()))?;
779 url = next;
780 }
781 status => return Err(FetchError::new(format!("HTTP {status} from {url}"))),
782 }
783 }
784 Err(FetchError::new(format!(
785 "more than {MAX_REDIRECTS} redirects"
786 )))
787 }
788}
789
790#[cfg(test)]
791mod tests;