use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Redirect};
use axum::Json;
use base64::Engine;
use serde::Deserialize;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use uuid::Uuid;
use super::auth::{
generate_api_key, generate_token, mark_email_verified, rotate_api_key, sha256_hex, upsert_user,
AppState,
};
use super::config::Config;
use super::helpers::internal_error;
const DISCOVERY_TTL: Duration = Duration::from_hours(1);
static ISSUER_CACHE: Mutex<Option<HashMap<String, (Value, Value, Instant)>>> = Mutex::new(None);
fn cached_issuer_meta(issuer: &str) -> Option<(Value, Value)> {
let guard = match ISSUER_CACHE.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard
.as_ref()
.and_then(|m| m.get(issuer))
.filter(|(_, _, at)| at.elapsed() < DISCOVERY_TTL)
.map(|(d, j, _)| (d.clone(), j.clone()))
}
fn store_issuer_meta(issuer: &str, discovery: Value, jwks: Value) {
let mut guard = match ISSUER_CACHE.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard
.get_or_insert_with(HashMap::new)
.insert(issuer.to_string(), (discovery, jwks, Instant::now()));
}
fn issuer_meta(issuer: &str) -> Result<(Value, Value), String> {
if let Some(hit) = cached_issuer_meta(issuer) {
return Ok(hit);
}
let discovery_url = format!("{issuer}/.well-known/openid-configuration");
let discovery = http_get_json(&discovery_url)?;
let doc_issuer = discovery
.get("issuer")
.and_then(Value::as_str)
.unwrap_or_default()
.trim_end_matches('/');
if doc_issuer != issuer {
return Err(format!(
"discovery issuer mismatch: configured {issuer}, document says {doc_issuer}"
));
}
let jwks_uri = discovery
.get("jwks_uri")
.and_then(Value::as_str)
.ok_or_else(|| "discovery document has no jwks_uri".to_string())?;
let jwks = http_get_json(jwks_uri)?;
store_issuer_meta(issuer, discovery.clone(), jwks.clone());
Ok((discovery, jwks))
}
fn http_get_json(url: &str) -> Result<Value, String> {
if !url.starts_with("https://") {
return Err(format!("refusing non-https url: {url}"));
}
let body = ureq::get(url)
.header("Accept", "application/json")
.call()
.map_err(|e| format!("GET {url}: {e}"))?
.into_body()
.read_to_string()
.map_err(|e| format!("read {url}: {e}"))?;
serde_json::from_str(&body).map_err(|e| format!("parse {url}: {e}"))
}
pub(super) struct SsoOrg {
pub issuer: String,
pub client_id: String,
pub sso_required: bool,
pub owner_email: Option<String>,
}
pub(super) async fn lookup_sso_org(cfg: &Config, domain: &str) -> Result<Option<SsoOrg>, String> {
let (Some(base), Some(key)) = (
cfg.billing_base_url.clone(),
cfg.billing_internal_key.clone(),
) else {
return Ok(None);
};
let url = format!("{base}/api/billing/sso/lookup/{domain}");
let resp = tokio::task::spawn_blocking(move || {
let agent: ureq::Agent = ureq::Agent::config_builder()
.http_status_as_error(false)
.build()
.into();
let resp = agent
.get(&url)
.header("X-Internal-Key", &key)
.call()
.map_err(|e| e.to_string())?;
let status = resp.status().as_u16();
let body = resp
.into_body()
.read_to_string()
.map_err(|e| e.to_string())?;
Ok::<_, String>((status, body))
})
.await
.map_err(|e| format!("join: {e}"))?;
let (status, body) = match resp {
Ok(v) => v,
Err(e) => {
tracing::warn!(error = %e, "sso lookup unreachable — treating as no-SSO");
return Ok(None);
}
};
if status == 404 {
return Ok(None);
}
if status != 200 {
return Err(format!("sso lookup returned {status}"));
}
let v: Value = serde_json::from_str(&body).map_err(|e| format!("sso lookup parse: {e}"))?;
Ok(Some(SsoOrg {
issuer: v
.get("issuer")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
client_id: v
.get("client_id")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
sso_required: v
.get("sso_required")
.and_then(Value::as_bool)
.unwrap_or(false),
owner_email: v
.get("owner_email")
.and_then(Value::as_str)
.map(str::to_string),
}))
}
pub(super) async fn password_login_blocked(cfg: &Config, email: &str) -> bool {
let Some(domain) = email.rsplit('@').next().filter(|d| d.contains('.')) else {
return false;
};
match lookup_sso_org(cfg, &domain.to_ascii_lowercase()).await {
Ok(Some(org)) => org.sso_required && org.owner_email.as_deref() != Some(email),
_ => false,
}
}
fn fetch_exchange_secret(
base: &str,
key: &str,
domain: &str,
) -> Result<(String, String, String), String> {
let url = format!("{base}/api/billing/sso/exchange/{domain}");
let body = ureq::get(&url)
.header("X-Internal-Key", key)
.call()
.map_err(|e| format!("exchange secret: {e}"))?
.into_body()
.read_to_string()
.map_err(|e| format!("exchange secret read: {e}"))?;
let v: Value = serde_json::from_str(&body).map_err(|e| format!("exchange parse: {e}"))?;
let get = |k: &str| -> Result<String, String> {
v.get(k)
.and_then(Value::as_str)
.map(str::to_string)
.ok_or_else(|| format!("exchange secret: missing {k}"))
};
Ok((get("issuer")?, get("client_id")?, get("client_secret")?))
}
#[derive(Deserialize)]
pub(super) struct StartBody {
pub email: String,
}
pub(super) async fn sso_start(
State(state): State<AppState>,
Json(body): Json<StartBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
let email = body.email.trim().to_lowercase();
let Some(domain) = email
.rsplit('@')
.next()
.filter(|d| d.contains('.') && !d.is_empty())
.map(str::to_string)
else {
return Err((StatusCode::BAD_REQUEST, "Invalid email".into()));
};
let org = lookup_sso_org(&state.cfg, &domain).await.map_err(|e| {
tracing::error!(error = %e, "sso start lookup failed");
(
StatusCode::BAD_GATEWAY,
"SSO is temporarily unavailable".to_string(),
)
})?;
let Some(org) = org else {
return Ok(Json(json!({ "sso": false })));
};
let issuer = org.issuer.clone();
let meta = tokio::task::spawn_blocking(move || issuer_meta(&issuer))
.await
.map_err(internal_error)?
.map_err(|e| {
tracing::error!(error = %e, issuer = %org.issuer, "sso discovery failed");
(
StatusCode::BAD_GATEWAY,
"The identity provider is not reachable right now".to_string(),
)
})?;
let authorize_endpoint = meta
.0
.get("authorization_endpoint")
.and_then(Value::as_str)
.ok_or((
StatusCode::BAD_GATEWAY,
"The identity provider publishes no authorization endpoint".to_string(),
))?
.to_string();
let state_token = generate_token();
let nonce = generate_token();
let pkce_verifier = generate_token();
let pkce_challenge = base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(Sha256::digest(pkce_verifier.as_bytes()));
let client = state.pool.get().await.map_err(internal_error)?;
client
.execute(
"DELETE FROM sso_login_states WHERE created_at < NOW() - INTERVAL '10 minutes'",
&[],
)
.await
.map_err(internal_error)?;
client
.execute(
"INSERT INTO sso_login_states (state_sha256, email_domain, nonce, pkce_verifier)
VALUES ($1, $2, $3, $4)",
&[&sha256_hex(&state_token), &domain, &nonce, &pkce_verifier],
)
.await
.map_err(internal_error)?;
let redirect_uri = format!(
"{}/api/auth/sso/callback",
state.cfg.api_base_url.trim_end_matches('/')
);
let authorize_url = format!(
"{authorize_endpoint}{}response_type=code&client_id={}&redirect_uri={}&scope={}&state={}&nonce={}&code_challenge={}&code_challenge_method=S256&login_hint={}",
if authorize_endpoint.contains('?') { "&" } else { "?" },
urlencode(&org.client_id),
urlencode(&redirect_uri),
urlencode("openid email profile"),
urlencode(&state_token),
urlencode(&nonce),
urlencode(&pkce_challenge),
urlencode(&email),
);
Ok(Json(json!({ "sso": true, "redirect_url": authorize_url })))
}
#[derive(Deserialize)]
pub(super) struct CallbackQuery {
pub code: Option<String>,
pub state: Option<String>,
pub error: Option<String>,
}
fn error_redirect(cfg: &Config, reason: &str) -> Redirect {
Redirect::to(&format!(
"{}/login/?sso_error={reason}",
cfg.public_base_url.trim_end_matches('/')
))
}
pub(super) async fn sso_callback(
State(state): State<AppState>,
Query(q): Query<CallbackQuery>,
) -> impl IntoResponse {
if let Some(e) = q.error.as_deref() {
tracing::info!(idp_error = %e, "sso callback: idp returned error");
return error_redirect(&state.cfg, "idp_denied");
}
let (Some(code), Some(state_token)) = (q.code.as_deref(), q.state.as_deref()) else {
return error_redirect(&state.cfg, "missing_params");
};
let row = match state.pool.get().await {
Ok(client) => client
.query_opt(
"DELETE FROM sso_login_states
WHERE state_sha256 = $1 AND created_at > NOW() - INTERVAL '10 minutes'
RETURNING email_domain, nonce, pkce_verifier",
&[&sha256_hex(state_token)],
)
.await
.ok()
.flatten(),
Err(_) => None,
};
let Some(row) = row else {
return error_redirect(&state.cfg, "expired");
};
let domain: String = row.get(0);
let expected_nonce: String = row.get(1);
let pkce_verifier: String = row.get(2);
let (Some(base), Some(key)) = (
state.cfg.billing_base_url.clone(),
state.cfg.billing_internal_key.clone(),
) else {
return error_redirect(&state.cfg, "unavailable");
};
let code = code.to_string();
let domain_for_exchange = domain.clone();
let redirect_uri = format!(
"{}/api/auth/sso/callback",
state.cfg.api_base_url.trim_end_matches('/')
);
let exchanged = tokio::task::spawn_blocking(move || {
exchange_and_verify(
&base,
&key,
&domain_for_exchange,
&code,
&redirect_uri,
&pkce_verifier,
&expected_nonce,
)
})
.await;
let (email, org_id) = match exchanged {
Ok(Ok(v)) => v,
Ok(Err(e)) => {
tracing::warn!(error = %e, %domain, "sso callback verification failed");
return error_redirect(&state.cfg, "verify_failed");
}
Err(e) => {
tracing::error!(error = %e, "sso callback join error");
return error_redirect(&state.cfg, "unavailable");
}
};
if email.rsplit('@').next().map(str::to_ascii_lowercase) != Some(domain.clone()) {
tracing::warn!(%domain, "sso callback: email outside org domain");
return error_redirect(&state.cfg, "verify_failed");
}
let (user_id, _created) = match upsert_user(&state.pool, &email, None).await {
Ok(v) => v,
Err(e) => {
tracing::error!(error = %e, "sso callback: upsert user failed");
return error_redirect(&state.cfg, "unavailable");
}
};
if let Err(e) = mark_email_verified(&state.pool, user_id).await {
tracing::warn!(error = %e, "sso callback: mark verified failed");
}
let api_key = generate_api_key();
if let Err(e) = rotate_api_key(&state.pool, user_id, &sha256_hex(&api_key)).await {
tracing::error!(error = %e, "sso callback: rotate key failed");
return error_redirect(&state.cfg, "unavailable");
}
{
let cfg = state.cfg.clone();
let email = email.clone();
tokio::spawn(async move {
jit_membership(&cfg, org_id, user_id, &email).await;
});
}
let handoff = generate_token();
let stored = match state.pool.get().await {
Ok(client) => client
.execute(
"INSERT INTO sso_handoff_codes (code_sha256, user_id, api_key, email)
VALUES ($1, $2, $3, $4)",
&[&sha256_hex(&handoff), &user_id, &api_key, &email],
)
.await
.is_ok(),
Err(_) => false,
};
if !stored {
return error_redirect(&state.cfg, "unavailable");
}
Redirect::to(&format!(
"{}/login/?sso_handoff={handoff}",
state.cfg.public_base_url.trim_end_matches('/')
))
}
fn exchange_and_verify(
billing_base: &str,
internal_key: &str,
domain: &str,
code: &str,
redirect_uri: &str,
pkce_verifier: &str,
expected_nonce: &str,
) -> Result<(String, Uuid), String> {
let (issuer, client_id, client_secret) =
fetch_exchange_secret(billing_base, internal_key, domain)?;
let org_id = {
let url = format!("{billing_base}/api/billing/sso/lookup/{domain}");
let body = ureq::get(&url)
.header("X-Internal-Key", internal_key)
.call()
.map_err(|e| format!("org lookup: {e}"))?
.into_body()
.read_to_string()
.map_err(|e| format!("org lookup read: {e}"))?;
serde_json::from_str::<Value>(&body)
.ok()
.and_then(|v| {
v.get("org_id")
.and_then(Value::as_str)
.and_then(|s| Uuid::parse_str(s).ok())
})
.ok_or("org lookup: missing org_id")?
};
let (discovery, jwks) = issuer_meta(&issuer)?;
let token_endpoint = discovery
.get("token_endpoint")
.and_then(Value::as_str)
.ok_or("discovery: no token_endpoint")?;
let form = format!(
"grant_type=authorization_code&code={}&redirect_uri={}&client_id={}&client_secret={}&code_verifier={}",
urlencode(code),
urlencode(redirect_uri),
urlencode(&client_id),
urlencode(&client_secret),
urlencode(pkce_verifier),
);
let body = ureq::post(token_endpoint)
.header("Content-Type", "application/x-www-form-urlencoded")
.header("Accept", "application/json")
.send(form.as_bytes())
.map_err(|e| format!("token exchange: {e}"))?
.into_body()
.read_to_string()
.map_err(|e| format!("token exchange read: {e}"))?;
let tokens: Value =
serde_json::from_str(&body).map_err(|e| format!("token exchange parse: {e}"))?;
let id_token = tokens
.get("id_token")
.and_then(Value::as_str)
.ok_or("token response has no id_token")?;
let claims = verify_id_token(id_token, &jwks, &issuer, &client_id)?;
let nonce = claims.get("nonce").and_then(Value::as_str).unwrap_or("");
if nonce != expected_nonce {
return Err("nonce mismatch".into());
}
let email = claims
.get("email")
.and_then(Value::as_str)
.map(str::to_lowercase)
.ok_or("id token has no email claim")?;
if claims.get("email_verified").and_then(Value::as_bool) == Some(false) {
return Err("idp reports email_verified=false".into());
}
Ok((email, org_id))
}
fn verify_id_token(
id_token: &str,
jwks: &Value,
issuer: &str,
client_id: &str,
) -> Result<Value, String> {
use jsonwebtoken::{decode, decode_header, Algorithm, DecodingKey, Validation};
let header = decode_header(id_token).map_err(|e| format!("jwt header: {e}"))?;
let kid = header.kid.as_deref();
let keys = jwks
.get("keys")
.and_then(Value::as_array)
.ok_or("jwks has no keys")?;
let jwk = keys
.iter()
.find(|k| kid.is_some() && k.get("kid").and_then(Value::as_str) == kid)
.or_else(|| (keys.len() == 1).then(|| &keys[0]))
.ok_or("no matching jwk for token kid")?;
let kty = jwk.get("kty").and_then(Value::as_str).unwrap_or("");
let (decoding_key, algorithm) = match kty {
"RSA" => {
let n = jwk
.get("n")
.and_then(Value::as_str)
.ok_or("rsa jwk missing n")?;
let e = jwk
.get("e")
.and_then(Value::as_str)
.ok_or("rsa jwk missing e")?;
(
DecodingKey::from_rsa_components(n, e).map_err(|e| format!("rsa key: {e}"))?,
header.alg,
)
}
"EC" => {
let x = jwk
.get("x")
.and_then(Value::as_str)
.ok_or("ec jwk missing x")?;
let y = jwk
.get("y")
.and_then(Value::as_str)
.ok_or("ec jwk missing y")?;
(
DecodingKey::from_ec_components(x, y).map_err(|e| format!("ec key: {e}"))?,
header.alg,
)
}
other => return Err(format!("unsupported jwk kty {other}")),
};
if !matches!(
algorithm,
Algorithm::RS256
| Algorithm::RS384
| Algorithm::RS512
| Algorithm::PS256
| Algorithm::PS384
| Algorithm::PS512
| Algorithm::ES256
| Algorithm::ES384
) {
return Err(format!("refusing token alg {algorithm:?}"));
}
let mut validation = Validation::new(algorithm);
validation.set_issuer(&[issuer]);
validation.set_audience(&[client_id]);
validation.leeway = 60;
let data = decode::<Value>(id_token, &decoding_key, &validation)
.map_err(|e| format!("jwt verify: {e}"))?;
Ok(data.claims)
}
async fn jit_membership(cfg: &Config, org_id: Uuid, user_id: Uuid, email: &str) {
let (Some(base), Some(key)) = (
cfg.billing_base_url.clone(),
cfg.billing_internal_key.clone(),
) else {
return;
};
let payload = json!({ "org_id": org_id, "user_id": user_id, "email": email });
let Ok(body) = serde_json::to_vec(&payload) else {
return;
};
let _ = tokio::task::spawn_blocking(move || {
let result = ureq::post(format!("{base}/api/billing/sso/jit").as_str())
.header("X-Internal-Key", &key)
.header("Content-Type", "application/json")
.send(&body);
if let Err(e) = result {
tracing::warn!(error = %e, "sso jit membership failed");
}
})
.await;
}
#[derive(Deserialize)]
pub(super) struct HandoffBody {
pub code: String,
}
pub(super) async fn sso_handoff(
State(state): State<AppState>,
Json(body): Json<HandoffBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
let code = body.code.trim();
if code.len() != 64 || !code.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err((StatusCode::BAD_REQUEST, "Invalid handoff code".into()));
}
let client = state.pool.get().await.map_err(internal_error)?;
client
.execute(
"DELETE FROM sso_handoff_codes WHERE created_at < NOW() - INTERVAL '60 seconds'",
&[],
)
.await
.map_err(internal_error)?;
let row = client
.query_opt(
"DELETE FROM sso_handoff_codes
WHERE code_sha256 = $1 AND created_at > NOW() - INTERVAL '60 seconds'
RETURNING user_id, api_key, email",
&[&sha256_hex(code)],
)
.await
.map_err(internal_error)?;
let Some(row) = row else {
return Err((
StatusCode::UNAUTHORIZED,
"This sign-in link has expired — please try again".into(),
));
};
let user_id: Uuid = row.get(0);
let api_key: String = row.get(1);
let email: String = row.get(2);
Ok(Json(json!({
"api_key": api_key,
"user_id": user_id,
"email": email,
"email_verified": true,
})))
}
fn urlencode(s: &str) -> String {
let mut out = String::with_capacity(s.len() * 3);
for b in s.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(b as char);
}
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn urlencode_covers_reserved_chars() {
assert_eq!(urlencode("abc-_.~XYZ09"), "abc-_.~XYZ09");
assert_eq!(urlencode("a b&c=d"), "a%20b%26c%3Dd");
assert_eq!(urlencode("https://x/y?z"), "https%3A%2F%2Fx%2Fy%3Fz");
assert_eq!(urlencode("ü"), "%C3%BC");
}
#[test]
fn pkce_challenge_is_base64url_of_sha256() {
let verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk";
let challenge = base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(Sha256::digest(verifier.as_bytes()));
assert_eq!(challenge, "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM");
}
#[test]
fn id_token_verification_rejects_garbage_and_alg_confusion() {
let jwks = json!({ "keys": [{ "kty": "RSA", "kid": "k1", "n": "AQAB", "e": "AQAB" }] });
assert!(verify_id_token("not.a.jwt", &jwks, "https://iss", "cid").is_err());
let hs_header = base64::engine::general_purpose::URL_SAFE_NO_PAD
.encode(r#"{"alg":"HS256","typ":"JWT","kid":"k1"}"#);
let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r"{}");
let forged = format!("{hs_header}.{payload}.c2ln");
let err = verify_id_token(&forged, &jwks, "https://iss", "cid").unwrap_err();
assert!(err.contains("refusing token alg"), "got: {err}");
}
#[test]
fn email_domain_extraction_is_strict() {
let email = "user@sub.acme.com";
assert_eq!(
email.rsplit('@').next().map(str::to_ascii_lowercase),
Some("sub.acme.com".to_string())
);
}
}