use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use jsonwebtoken::Algorithm;
use web_time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use crate::{JwtKeys, KeySet};
use address::Address;
mod address;
pub const FETCH_TIMEOUT: Duration = Duration::from_secs(5);
pub const MAX_JWKS_BYTES: usize = 1 << 20;
pub const MAX_REDIRECTS: usize = 3;
pub const DEFAULT_REFRESH: Duration = Duration::from_secs(300);
pub const MIN_MAX_AGE: Duration = Duration::from_secs(60);
pub const MAX_MAX_AGE: Duration = Duration::from_secs(86_400);
pub const UNKNOWN_KID_INTERVAL: Duration = Duration::from_secs(30);
pub const MAX_RETRY: Duration = Duration::from_secs(30);
#[cfg(not(target_arch = "wasm32"))]
pub trait MaybeSendSync: Send + Sync {}
#[cfg(not(target_arch = "wasm32"))]
impl<T: Send + Sync + ?Sized> MaybeSendSync for T {}
#[cfg(target_arch = "wasm32")]
pub trait MaybeSendSync {}
#[cfg(target_arch = "wasm32")]
impl<T: ?Sized> MaybeSendSync for T {}
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct FetchResponse {
pub status: u16,
pub location: Option<String>,
pub cache_control: Option<String>,
pub body: Vec<u8>,
}
impl FetchResponse {
pub fn new(status: u16, body: Vec<u8>) -> Self {
Self {
status,
location: None,
cache_control: None,
body,
}
}
pub fn with_location(mut self, location: Option<String>) -> Self {
self.location = location;
self
}
pub fn with_cache_control(mut self, cache_control: Option<String>) -> Self {
self.cache_control = cache_control;
self
}
}
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
#[error("{0}")]
pub struct FetchError(String);
impl FetchError {
pub fn new(reason: impl Into<String>) -> Self {
Self(reason.into())
}
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
pub trait Fetch: MaybeSendSync {
async fn get(&self, url: &str) -> Result<FetchResponse, FetchError>;
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
pub trait JwksStore: MaybeSendSync {
async fn load(&self, url: &str) -> Option<StoredJwks>;
async fn save(&self, url: &str, jwks: &StoredJwks, ttl: Duration);
}
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StoredJwks {
pub body: String,
pub fresh_until: u64,
}
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct JwksStatus {
pub loaded: bool,
pub keys: usize,
pub refreshes: u64,
pub failures: u64,
pub consecutive_failures: u32,
pub changes: u64,
pub last_success: Option<SystemTime>,
pub age: Option<Duration>,
pub refresh_interval: Duration,
pub next_refresh_in: Duration,
pub last_error: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
#[error("{0}")]
pub struct JwksUrlError(String);
pub fn check_url(url: &str) -> Result<(), JwksUrlError> {
parse_url(url).map(drop)
}
fn parse_url(url: &str) -> Result<Address, JwksUrlError> {
let parsed = Address::parse(url).map_err(|e| JwksUrlError(format!("{url:?}: {e}")))?;
check_parsed(&parsed)?;
Ok(parsed)
}
fn check_parsed(url: &Address) -> Result<(), JwksUrlError> {
if url.is_https() || url.is_loopback() {
return Ok(());
}
Err(JwksUrlError(format!(
"{url}: a key set fetched over plain HTTP can be replaced by anyone on the path; \
use https (http is accepted only for localhost, 127.0.0.0/8 and [::1])"
)))
}
fn max_age(cache_control: Option<&str>) -> Option<Duration> {
let mut found = None;
for directive in cache_control?.split(',') {
let directive = directive.trim();
let (name, value) = directive.split_once('=').unwrap_or((directive, ""));
match name.trim().to_ascii_lowercase().as_str() {
"no-cache" | "no-store" => return Some(MIN_MAX_AGE),
"max-age" => {
if let Ok(secs) = value.trim().trim_matches('"').parse::<u64>() {
found = Some(Duration::from_secs(secs));
}
}
_ => {}
}
}
found.map(|age| age.clamp(MIN_MAX_AGE, MAX_MAX_AGE))
}
fn retry_after(failures: u32, interval: Duration) -> Duration {
let backoff = Duration::from_secs(1u64 << failures.saturating_sub(1).min(5));
backoff.min(MAX_RETRY).min(interval)
}
fn unix_now() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
pub struct JwksBuilder {
url: String,
fetch: Box<dyn Fetch>,
fixed: Option<Arc<KeySet>>,
refresh: Duration,
unknown_kid_interval: Duration,
store: Option<Box<dyn JwksStore>>,
}
impl JwksBuilder {
pub fn fixed(mut self, keys: Option<&JwtKeys>) -> Self {
self.fixed = keys.map(JwtKeys::current);
self
}
pub fn refresh(mut self, interval: Duration) -> Self {
self.refresh = interval.max(Duration::from_secs(1));
self
}
pub fn unknown_kid_interval(mut self, interval: Duration) -> Self {
self.unknown_kid_interval = interval;
self
}
pub fn store(mut self, store: impl JwksStore + 'static) -> Self {
self.store = Some(Box::new(store));
self
}
pub fn build(self) -> Result<Jwks, JwksUrlError> {
let url = parse_url(&self.url)?;
let keys = JwtKeys::new(match &self.fixed {
Some(fixed) => KeySet::union(fixed, &KeySet::empty()),
None => KeySet::empty(),
});
#[cfg_attr(target_arch = "wasm32", allow(clippy::arc_with_non_send_sync))]
Ok(Jwks {
inner: Arc::new(Inner {
url_text: url.to_string(),
url,
fetch: self.fetch,
store: self.store,
keys,
fixed: self.fixed,
unknown_kid_interval: self.unknown_kid_interval,
background: AtomicBool::new(false),
state: Mutex::new(State {
body: None,
keys: 0,
last_success: None,
interval: self.refresh,
configured: self.refresh,
next_due: Instant::now(),
consecutive_failures: 0,
refreshes: 0,
failures: 0,
changes: 0,
last_error: None,
last_unknown_kid: None,
}),
refreshing: futures::lock::Mutex::new(()),
}),
})
}
}
#[derive(Clone)]
pub struct Jwks {
inner: Arc<Inner>,
}
struct Inner {
url: Address,
url_text: String,
fetch: Box<dyn Fetch>,
store: Option<Box<dyn JwksStore>>,
keys: JwtKeys,
fixed: Option<Arc<KeySet>>,
unknown_kid_interval: Duration,
background: AtomicBool,
state: Mutex<State>,
refreshing: futures::lock::Mutex<()>,
}
struct State {
body: Option<String>,
keys: usize,
last_success: Option<(Instant, SystemTime)>,
interval: Duration,
configured: Duration,
next_due: Instant,
consecutive_failures: u32,
refreshes: u64,
failures: u64,
changes: u64,
last_error: Option<String>,
last_unknown_kid: Option<Instant>,
}
impl std::fmt::Debug for Jwks {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Jwks")
.field("url", &self.inner.url_text)
.field("keys", &self.inner.keys)
.finish_non_exhaustive()
}
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum Reason {
AnySource,
ProviderOnly,
}
impl Jwks {
pub fn builder(url: impl Into<String>, fetch: impl Fetch + 'static) -> JwksBuilder {
JwksBuilder {
url: url.into(),
fetch: Box::new(fetch),
fixed: None,
refresh: DEFAULT_REFRESH,
unknown_kid_interval: UNKNOWN_KID_INTERVAL,
store: None,
}
}
pub fn keys(&self) -> &JwtKeys {
&self.inner.keys
}
pub fn url(&self) -> &str {
&self.inner.url_text
}
pub fn is_loaded(&self) -> bool {
self.state().last_success.is_some()
}
pub fn status(&self) -> JwksStatus {
let state = self.state();
let now = Instant::now();
JwksStatus {
loaded: state.last_success.is_some(),
keys: state.keys,
refreshes: state.refreshes,
failures: state.failures,
consecutive_failures: state.consecutive_failures,
changes: state.changes,
last_success: state.last_success.map(|(_, at)| at),
age: state
.last_success
.map(|(at, _)| now.saturating_duration_since(at)),
refresh_interval: state.interval,
next_refresh_in: state.next_due.saturating_duration_since(now),
last_error: state.last_error.clone(),
}
}
pub fn next_refresh_in(&self) -> Duration {
self.state()
.next_due
.saturating_duration_since(Instant::now())
}
pub async fn refresh(&self) -> Result<(), FetchError> {
let _fetching = self.exclusive().await;
self.fetch_and_install(Reason::ProviderOnly).await
}
pub async fn refresh_forever<S, F>(&self, mut sleep: S)
where
S: FnMut(Duration) -> F,
F: std::future::Future<Output = ()>,
{
self.inner.background.store(true, Ordering::Relaxed);
loop {
sleep(self.next_refresh_in()).await;
let _ = self.refresh().await;
}
}
pub async fn ensure_for_token(&self, token: &str) {
let Ok(header) = jsonwebtoken::decode_header(token) else {
return;
};
let kid = header.kid.as_deref();
if self.inner.keys.current().selects(kid, header.alg) {
if !self.inner.background.load(Ordering::Relaxed) && self.due() {
self.refresh_if_due().await;
}
return;
}
self.refresh_for(kid, header.alg).await;
}
async fn exclusive(&self) -> Option<futures::lock::MutexGuard<'_, ()>> {
#[cfg(not(target_arch = "wasm32"))]
return Some(self.inner.refreshing.lock().await);
#[cfg(target_arch = "wasm32")]
return self.inner.refreshing.try_lock();
}
fn state(&self) -> std::sync::MutexGuard<'_, State> {
self.inner.state.lock().unwrap_or_else(|e| e.into_inner())
}
fn due(&self) -> bool {
Instant::now() >= self.state().next_due
}
async fn refresh_if_due(&self) {
let Some(_fetching) = self.inner.refreshing.try_lock() else {
return;
};
if self.due() {
let _ = self.fetch_and_install(Reason::AnySource).await;
}
}
async fn refresh_for(&self, kid: Option<&str>, alg: Algorithm) {
let _fetching = self.exclusive().await;
if self.inner.keys.current().selects(kid, alg) {
return;
}
let reason = {
let mut state = self.state();
if state.last_success.is_none() && Instant::now() >= state.next_due {
Reason::AnySource
} else {
let now = Instant::now();
if let Some(last) = state.last_unknown_kid {
if now.saturating_duration_since(last) < self.inner.unknown_kid_interval {
return;
}
}
state.last_unknown_kid = Some(now);
Reason::ProviderOnly
}
};
let _ = self.fetch_and_install(reason).await;
}
async fn fetch_and_install(&self, reason: Reason) -> Result<(), FetchError> {
if reason == Reason::AnySource {
if let Some(store) = &self.inner.store {
if let Some(stored) = store.load(self.url()).await {
let now = unix_now();
if stored.fresh_until > now && self.install(&stored.body).is_ok() {
self.succeeded(Duration::from_secs(stored.fresh_until - now));
return Ok(());
}
}
}
}
let fetched = self.download().await.and_then(|(body, max_age)| {
self.install(&body)?;
Ok((body, max_age))
});
match fetched {
Ok((body, max_age)) => {
let interval = max_age.unwrap_or(self.state().configured);
self.succeeded(interval);
if let Some(store) = &self.inner.store {
let stored = StoredJwks {
body,
fresh_until: unix_now() + interval.as_secs(),
};
store.save(self.url(), &stored, interval).await;
}
Ok(())
}
Err(e) => {
let loaded = {
let mut state = self.state();
state.failures += 1;
state.consecutive_failures += 1;
state.last_error = Some(e.to_string());
state.next_due =
Instant::now() + retry_after(state.consecutive_failures, state.interval);
state.last_success.is_some()
};
if loaded {
tracing::warn!(
"PGRST_JWT_JWKS_URL: refreshing the keys from {} failed, keeping the \
ones fetched before: {e}",
self.url()
);
} else {
tracing::warn!(
"PGRST_JWT_JWKS_URL: fetching the keys from {} failed, and no keys \
from it have loaded yet, so tokens signed by them are refused: {e}",
self.url()
);
}
Err(e)
}
}
}
fn succeeded(&self, interval: Duration) {
let mut state = self.state();
let now = Instant::now();
state.last_success = Some((now, SystemTime::now()));
state.refreshes += 1;
state.consecutive_failures = 0;
state.last_error = None;
state.interval = interval;
state.next_due = now + interval;
}
fn install(&self, body: &str) -> Result<(), FetchError> {
if self.state().body.as_deref() == Some(body) {
return Ok(());
}
let fetched = KeySet::from_jwk_json(body)
.map_err(|e| FetchError::new(format!("the response is not a usable JWKS: {e}")))?;
let count = fetched.len();
let set = match &self.inner.fixed {
Some(fixed) => KeySet::union(fixed, &fetched),
None => fetched,
};
let mut state = self.state();
self.inner.keys.replace(set);
state.body = Some(body.to_string());
state.keys = count;
state.changes += 1;
tracing::info!(
"PGRST_JWT_JWKS_URL: {count} key(s) in force from {}",
self.inner.url
);
Ok(())
}
async fn download(&self) -> Result<(String, Option<Duration>), FetchError> {
let mut url = self.inner.url.clone();
for _ in 0..=MAX_REDIRECTS {
let response = self.inner.fetch.get(&url.to_string()).await?;
match response.status {
200 => {
if response.body.len() > MAX_JWKS_BYTES {
return Err(FetchError::new(format!(
"the response is larger than {MAX_JWKS_BYTES} bytes"
)));
}
let body = String::from_utf8(response.body)
.map_err(|_| FetchError::new("the response is not UTF-8"))?;
return Ok((body, max_age(response.cache_control.as_deref())));
}
301 | 302 | 303 | 307 | 308 => {
let location = response.location.ok_or_else(|| {
FetchError::new(format!("HTTP {} with no Location", response.status))
})?;
let next = url.join(&location).map_err(|e| {
FetchError::new(format!("a redirect to {location:?} is not followed: {e}"))
})?;
if next.host() != url.host() {
return Err(FetchError::new(format!(
"a redirect to another host ({next}) is not followed"
)));
}
check_parsed(&next).map_err(|e| FetchError::new(e.to_string()))?;
url = next;
}
status => return Err(FetchError::new(format!("HTTP {status} from {url}"))),
}
}
Err(FetchError::new(format!(
"more than {MAX_REDIRECTS} redirects"
)))
}
}
#[cfg(test)]
mod tests;