use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use std::time::{Duration, Instant};
use openidconnect::core::{
CoreAuthenticationFlow, CoreClient, CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier,
CoreJsonWebKey, CoreJsonWebKeySet, CoreJwsSigningAlgorithm, CoreProviderMetadata,
CoreUserInfoClaims,
};
use openidconnect::reqwest as oidc_reqwest;
use openidconnect::{
AuthorizationCode, ClaimsVerificationError, ClientId, ClientSecret, CsrfToken,
EndpointMaybeSet, EndpointNotSet, EndpointSet, IssuerUrl, JsonWebKeySetUrl, Nonce,
OAuth2TokenResponse, PkceCodeChallenge, PkceCodeVerifier, RedirectUrl, Scope,
SignatureVerificationError, SubjectIdentifier, TokenResponse,
};
use parking_lot::RwLock;
use url::Url;
use crate::error::{Error, Result};
pub const DEFAULT_UPSTREAM_SCOPES: &[&str] = &["openid", "email", "profile"];
const JWKS_MIN_TTL: Duration = Duration::from_secs(300);
const JWKS_MAX_TTL: Duration = Duration::from_secs(6 * 60 * 60);
const JWKS_DEFAULT_TTL: Duration = Duration::from_secs(60 * 60);
const JWKS_MIN_REFETCH_INTERVAL: Duration = Duration::from_secs(60);
struct JwksCache {
keys: Vec<CoreJsonWebKey>,
expires_at: Instant,
last_fetch: Instant,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct UpstreamProvider {
pub slug: String,
pub issuer: String,
pub client_id: String,
pub client_secret: String,
pub scopes: Vec<String>,
pub auth_params: BTreeMap<String, String>,
}
#[derive(Clone, Debug)]
pub struct UpstreamUserInfo {
pub provider: String,
pub subject: String,
pub email: Option<String>,
pub email_verified: bool,
pub name: Option<String>,
pub picture: Option<String>,
pub raw_claims: serde_json::Value,
}
pub struct OidcClient {
inner: CoreClient<
EndpointSet,
EndpointNotSet,
EndpointNotSet,
EndpointNotSet,
EndpointMaybeSet,
EndpointMaybeSet,
>,
provider: UpstreamProvider,
redirect_uri: RedirectUrl,
jwks_uri: JsonWebKeySetUrl,
signing_algs: Vec<CoreJwsSigningAlgorithm>,
jwks: Arc<RwLock<JwksCache>>,
jwks_http: reqwest::Client,
}
impl OidcClient {
pub fn provider(&self) -> &UpstreamProvider {
&self.provider
}
pub fn redirect_uri(&self) -> &RedirectUrl {
&self.redirect_uri
}
pub fn start_login(&self, state: CsrfToken) -> StartedLogin {
let (pkce_challenge, pkce_verifier) = PkceCodeChallenge::new_random_sha256();
let mut request = self.inner.authorize_url(
CoreAuthenticationFlow::AuthorizationCode,
move || state,
Nonce::new_random,
);
for scope in &self.provider.scopes {
if scope == "openid" {
continue;
}
request = request.add_scope(Scope::new(scope.clone()));
}
for (k, v) in &self.provider.auth_params {
request = request.add_extra_param(k.clone(), v.clone());
}
let (url, csrf_token, nonce) = request.set_pkce_challenge(pkce_challenge).url();
StartedLogin {
url,
csrf_token,
nonce,
pkce_verifier,
}
}
pub async fn complete_login(
&self,
code: String,
pkce_verifier: PkceCodeVerifier,
nonce: Nonce,
) -> Result<UpstreamUserInfo> {
let http = build_oidc_http_client(HttpClientOptions::default())?;
let token_response = self
.inner
.exchange_code(AuthorizationCode::new(code))
.map_err(|e| Error::Oidc(format!("exchange_code config: {e}")))?
.set_pkce_verifier(pkce_verifier)
.request_async(&http)
.await
.map_err(|e| Error::Oidc(format!("token exchange: {e}")))?;
let id_token = token_response
.id_token()
.ok_or_else(|| Error::Oidc("upstream returned no id_token".to_string()))?;
let claims = self.verify_id_token_claims(id_token, &nonce).await?;
let subject = claims.subject().to_string();
let mut email = claims.email().map(|e| e.to_string());
let mut email_verified = claims.email_verified().unwrap_or(false);
let mut name = claims
.name()
.and_then(|map| map.get(None))
.map(|n| n.to_string());
let mut picture = claims
.picture()
.and_then(|map| map.get(None))
.map(|u| u.to_string());
let mut raw_claims =
serde_json::to_value(claims).unwrap_or_else(|_| serde_json::json!({"sub": subject}));
if let Ok(req) = self.inner.user_info(
token_response.access_token().clone(),
Some(SubjectIdentifier::new(subject.clone())),
) && let Ok(userinfo) = req.request_async(&http).await
{
let user_claims: CoreUserInfoClaims = userinfo;
if email.is_none() {
email = user_claims.email().map(|e| e.to_string());
}
if !email_verified {
email_verified = user_claims.email_verified().unwrap_or(email_verified);
}
if name.is_none() {
name = user_claims
.name()
.and_then(|map| map.get(None))
.map(|n| n.to_string());
}
if picture.is_none() {
picture = user_claims
.picture()
.and_then(|map| map.get(None))
.map(|u| u.to_string());
}
if let Ok(userinfo_value) = serde_json::to_value(&user_claims) {
merge_json(&mut raw_claims, userinfo_value);
}
}
Ok(UpstreamUserInfo {
provider: self.provider.slug.clone(),
subject,
email,
email_verified,
name,
picture,
raw_claims,
})
}
async fn verify_id_token_claims<'t>(
&self,
id_token: &'t CoreIdToken,
nonce: &Nonce,
) -> Result<&'t CoreIdTokenClaims> {
let mut refreshed = false;
let keys = if self.jwks.read().expires_at <= Instant::now() {
match self.refresh_jwks().await {
Ok(fresh) => {
refreshed = true;
fresh
}
Err(e) => {
tracing::warn!(
slug = %self.provider.slug,
error = %e,
"proactive jwks refresh failed; verifying against cached keys"
);
self.jwks.read().keys.clone()
}
}
} else {
self.jwks.read().keys.clone()
};
let verifier = self.id_token_verifier(keys)?;
match id_token.claims(&verifier, nonce) {
Ok(claims) => Ok(claims),
Err(e)
if is_unknown_signing_key(&e) && !refreshed && self.reactive_refetch_allowed() =>
{
tracing::info!(
slug = %self.provider.slug,
"id_token kid not in cached jwks; refetching upstream keys (likely rotation)"
);
let fresh = self.refresh_jwks().await?;
let verifier = self.id_token_verifier(fresh)?;
id_token
.claims(&verifier, nonce)
.map_err(|e| Error::Oidc(format!("id_token verify: {e}")))
}
Err(e) => Err(Error::Oidc(format!("id_token verify: {e}"))),
}
}
fn id_token_verifier(&self, keys: Vec<CoreJsonWebKey>) -> Result<CoreIdTokenVerifier<'static>> {
let issuer = IssuerUrl::new(self.provider.issuer.clone())
.map_err(|e| Error::Oidc(format!("issuer url {}: {e}", self.provider.issuer)))?;
let client_id = ClientId::new(self.provider.client_id.clone());
let jwks = CoreJsonWebKeySet::new(keys);
let verifier = if self.provider.client_secret.is_empty() {
CoreIdTokenVerifier::new_public_client(client_id, issuer, jwks)
} else {
CoreIdTokenVerifier::new_confidential_client(
client_id,
ClientSecret::new(self.provider.client_secret.clone()),
issuer,
jwks,
)
};
Ok(verifier.set_allowed_algs(self.signing_algs.clone()))
}
fn reactive_refetch_allowed(&self) -> bool {
self.jwks.read().last_fetch.elapsed() >= JWKS_MIN_REFETCH_INTERVAL
}
async fn refresh_jwks(&self) -> Result<Vec<CoreJsonWebKey>> {
let uri = self.jwks_uri.url().to_string();
let resp = self
.jwks_http
.get(&uri)
.send()
.await
.map_err(|e| Error::Oidc(format!("fetch jwks {uri}: {e}")))?
.error_for_status()
.map_err(|e| Error::Oidc(format!("fetch jwks {uri}: {e}")))?;
let ttl = jwks_cache_ttl(resp.headers());
let fetched: CoreJsonWebKeySet = resp
.json()
.await
.map_err(|e| Error::Oidc(format!("parse jwks {uri}: {e}")))?;
let keys = fetched.keys().clone();
let now = Instant::now();
{
let mut cache = self.jwks.write();
cache.keys = keys.clone();
cache.expires_at = now + ttl;
cache.last_fetch = now;
}
tracing::debug!(
slug = %self.provider.slug,
ttl_secs = ttl.as_secs(),
keys = keys.len(),
"refreshed upstream jwks"
);
Ok(keys)
}
}
pub struct StartedLogin {
pub url: Url,
pub csrf_token: CsrfToken,
pub nonce: Nonce,
pub pkce_verifier: PkceCodeVerifier,
}
#[derive(Clone, Default)]
pub struct OidcRegistry {
inner: Arc<RwLock<HashMap<String, Arc<OidcClient>>>>,
}
impl OidcRegistry {
pub fn new() -> Self {
Self::default()
}
pub async fn add(&self, provider: UpstreamProvider, redirect_uri: Url) -> Result<()> {
let issuer = IssuerUrl::new(provider.issuer.clone())
.map_err(|e| Error::Oidc(format!("issuer url {}: {e}", provider.issuer)))?;
let http = build_oidc_http_client(HttpClientOptions::default())?;
let metadata = CoreProviderMetadata::discover_async(issuer, &http)
.await
.map_err(|e| Error::Oidc(format!("discover {}: {e}", provider.slug)))?;
let jwks_uri = metadata.jwks_uri().clone();
let signing_algs = metadata.id_token_signing_alg_values_supported().clone();
let initial_keys = metadata.jwks().keys().clone();
let redirect = RedirectUrl::new(redirect_uri.to_string())
.map_err(|e| Error::Oidc(format!("redirect_uri {redirect_uri}: {e}")))?;
let client_secret = if provider.client_secret.is_empty() {
None
} else {
Some(ClientSecret::new(provider.client_secret.clone()))
};
let inner = CoreClient::from_provider_metadata(
metadata,
ClientId::new(provider.client_id.clone()),
client_secret,
)
.set_redirect_uri(redirect.clone());
let jwks_http = reqwest::Client::builder()
.timeout(Duration::from_secs(10))
.build()
.map_err(|e| Error::Oidc(format!("build jwks http client: {e}")))?;
let now = Instant::now();
let client = OidcClient {
inner,
provider: provider.clone(),
redirect_uri: redirect,
jwks_uri,
signing_algs,
jwks: Arc::new(RwLock::new(JwksCache {
keys: initial_keys,
expires_at: now + JWKS_DEFAULT_TTL,
last_fetch: now,
})),
jwks_http,
};
self.inner
.write()
.insert(provider.slug.clone(), Arc::new(client));
Ok(())
}
pub fn client(&self, slug: &str) -> Option<Arc<OidcClient>> {
self.inner.read().get(slug).cloned()
}
pub fn slugs(&self) -> Vec<String> {
self.inner.read().keys().cloned().collect()
}
pub fn remove(&self, slug: &str) -> bool {
self.inner.write().remove(slug).is_some()
}
pub fn len(&self) -> usize {
self.inner.read().len()
}
pub fn is_empty(&self) -> bool {
self.inner.read().is_empty()
}
}
#[derive(Clone, Debug)]
pub struct HttpClientOptions {
pub connect_timeout_secs: u64,
pub request_timeout_secs: u64,
}
impl Default for HttpClientOptions {
fn default() -> Self {
Self {
connect_timeout_secs: 5,
request_timeout_secs: 10,
}
}
}
pub fn build_oidc_http_client(opts: HttpClientOptions) -> Result<oidc_reqwest::Client> {
oidc_reqwest::ClientBuilder::new()
.redirect(oidc_reqwest::redirect::Policy::none())
.connect_timeout(Duration::from_secs(opts.connect_timeout_secs))
.timeout(Duration::from_secs(opts.request_timeout_secs))
.build()
.map_err(|e| Error::Oidc(format!("build oidc http client: {e}")))
}
fn jwks_cache_ttl(headers: &reqwest::header::HeaderMap) -> Duration {
headers
.get(reqwest::header::CACHE_CONTROL)
.and_then(|v| v.to_str().ok())
.and_then(parse_max_age_secs)
.map(|secs| Duration::from_secs(secs).clamp(JWKS_MIN_TTL, JWKS_MAX_TTL))
.unwrap_or(JWKS_DEFAULT_TTL)
}
fn parse_max_age_secs(cache_control: &str) -> Option<u64> {
cache_control.split(',').find_map(|directive| {
directive
.trim()
.strip_prefix("max-age")?
.trim_start()
.strip_prefix('=')?
.trim()
.parse::<u64>()
.ok()
})
}
fn is_unknown_signing_key(e: &ClaimsVerificationError) -> bool {
matches!(
e,
ClaimsVerificationError::SignatureVerification(SignatureVerificationError::NoMatchingKey)
)
}
fn merge_json(target: &mut serde_json::Value, src: serde_json::Value) {
match (target, src) {
(serde_json::Value::Object(a), serde_json::Value::Object(b)) => {
for (k, v) in b {
merge_json(a.entry(k).or_insert(serde_json::Value::Null), v);
}
}
(slot, src) => {
if !src.is_null() {
*slot = src;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn registry_starts_empty() {
let reg = OidcRegistry::new();
assert!(reg.is_empty());
assert_eq!(reg.len(), 0);
assert!(reg.client("google").is_none());
assert!(reg.slugs().is_empty());
}
#[test]
fn merge_json_merges_objects_and_keeps_existing_on_null() {
let mut a = serde_json::json!({"email": "a@x", "groups": ["a"]});
let b = serde_json::json!({"email": serde_json::Value::Null, "name": "Alice"});
merge_json(&mut a, b);
assert_eq!(a["email"], "a@x");
assert_eq!(a["name"], "Alice");
assert_eq!(a["groups"], serde_json::json!(["a"]));
}
#[test]
fn upstream_provider_record_is_clonable() {
let p = UpstreamProvider {
slug: "google".to_string(),
issuer: "https://accounts.google.com".to_string(),
client_id: "client".to_string(),
client_secret: "secret".to_string(),
scopes: vec!["openid".to_string(), "email".to_string()],
auth_params: BTreeMap::new(),
};
let dup = p.clone();
assert_eq!(p, dup);
}
#[test]
fn parses_max_age_from_cache_control() {
assert_eq!(parse_max_age_secs("max-age=3600"), Some(3600));
assert_eq!(
parse_max_age_secs("public, max-age=600, must-revalidate"),
Some(600)
);
assert_eq!(parse_max_age_secs("max-age = 42"), Some(42));
assert_eq!(parse_max_age_secs("private, no-cache"), None);
assert_eq!(parse_max_age_secs("s-maxage=120"), None);
}
#[test]
fn jwks_ttl_clamps_and_defaults() {
use reqwest::header::{CACHE_CONTROL, HeaderMap, HeaderValue};
assert_eq!(jwks_cache_ttl(&HeaderMap::new()), JWKS_DEFAULT_TTL);
let ttl = |v: &'static str| {
let mut h = HeaderMap::new();
h.insert(CACHE_CONTROL, HeaderValue::from_static(v));
jwks_cache_ttl(&h)
};
assert_eq!(ttl("max-age=10"), JWKS_MIN_TTL);
assert_eq!(ttl("public, max-age=999999"), JWKS_MAX_TTL);
assert_eq!(ttl("max-age=1800"), Duration::from_secs(1800));
}
#[test]
fn only_missing_key_triggers_refetch() {
assert!(is_unknown_signing_key(
&ClaimsVerificationError::SignatureVerification(
SignatureVerificationError::NoMatchingKey
)
));
assert!(!is_unknown_signing_key(
&ClaimsVerificationError::SignatureVerification(
SignatureVerificationError::CryptoError("bad sig".into())
)
));
assert!(!is_unknown_signing_key(&ClaimsVerificationError::Expired(
"stale".into()
)));
}
#[tokio::test]
async fn discover_unreachable_issuer_returns_oidc_error() {
let reg = OidcRegistry::new();
let provider = UpstreamProvider {
slug: "ghost".to_string(),
issuer: "http://127.0.0.1:1/oidc".to_string(),
client_id: "client".to_string(),
client_secret: "secret".to_string(),
scopes: vec!["openid".to_string()],
auth_params: BTreeMap::new(),
};
let redirect = Url::parse("https://example.com/login/ghost/callback").unwrap();
let result = reg.add(provider, redirect).await;
assert!(matches!(result, Err(Error::Oidc(_))));
}
}