use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use serde::Deserialize;
use sha2::{Digest, Sha256};
use crate::http::client::HttpClient;
use crate::http::send_prepared;
use crate::request::auth::{OAuthAuth, OAuthGrantType};
use crate::request::{Method, Request};
use crate::SendraError;
const EXPIRY_MARGIN: Duration = Duration::from_secs(30);
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct CacheKey {
token_url: String,
client_id: String,
grant_type: OAuthGrantType,
scope: Option<String>,
}
impl CacheKey {
fn from(auth: &OAuthAuth) -> Self {
Self {
token_url: auth.token_url.clone(),
client_id: auth.client_id.clone(),
grant_type: auth.grant_type,
scope: auth.scope.clone(),
}
}
}
enum CacheEntry {
Token {
access_token: String,
expires_at: Option<Instant>,
},
Failed(String),
}
pub struct OAuthTokenCache {
entries: Mutex<HashMap<CacheKey, CacheEntry>>,
}
impl OAuthTokenCache {
pub fn new() -> Self {
Self {
entries: Mutex::new(HashMap::new()),
}
}
}
impl Default for OAuthTokenCache {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for OAuthTokenCache {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let count = self
.entries
.lock()
.map(|entries| entries.len())
.unwrap_or(0);
f.debug_struct("OAuthTokenCache")
.field("entries", &count)
.finish()
}
}
impl OAuthTokenCache {
pub fn insert_token(&self, auth: &OAuthAuth, access_token: String, expires_in: Option<u64>) {
let key = CacheKey::from(auth);
let expires_at = expires_in.map(|secs| Instant::now() + Duration::from_secs(secs));
let mut entries = self
.entries
.lock()
.expect("the cache mutex is never held across a panic");
entries.insert(
key,
CacheEntry::Token {
access_token,
expires_at,
},
);
}
}
#[derive(Deserialize)]
struct TokenResponse {
access_token: String,
#[serde(default)]
expires_in: Option<u64>,
}
pub async fn acquire_token(
auth: &OAuthAuth,
client: &HttpClient,
cache: &OAuthTokenCache,
) -> Result<String, SendraError> {
let key = CacheKey::from(auth);
{
let entries = cache
.entries
.lock()
.expect("the cache mutex is never held across a panic");
match entries.get(&key) {
Some(CacheEntry::Token {
access_token,
expires_at,
}) => {
let still_valid = match expires_at {
Some(expires_at) => Instant::now() + EXPIRY_MARGIN < *expires_at,
None => true,
};
if still_valid {
return Ok(access_token.clone());
}
}
Some(CacheEntry::Failed(reason)) => {
return Err(SendraError::OAuthAcquisition {
token_url: auth.token_url.clone(),
reason: reason.clone(),
});
}
None => {}
}
}
if auth.grant_type == OAuthGrantType::AuthorizationCode {
let reason = "grant_type: authorization_code requires an interactive browser login — \
trigger it from the auth editor, then re-run this request"
.to_string();
let mut entries = cache
.entries
.lock()
.expect("the cache mutex is never held across a panic");
entries.insert(key, CacheEntry::Failed(reason.clone()));
return Err(SendraError::OAuthAcquisition {
token_url: auth.token_url.clone(),
reason,
});
}
match acquire_fresh(auth, client).await {
Ok((access_token, expires_in)) => {
let expires_at = expires_in.map(|secs| Instant::now() + Duration::from_secs(secs));
let mut entries = cache
.entries
.lock()
.expect("the cache mutex is never held across a panic");
entries.insert(
key,
CacheEntry::Token {
access_token: access_token.clone(),
expires_at,
},
);
Ok(access_token)
}
Err(reason) => {
let mut entries = cache
.entries
.lock()
.expect("the cache mutex is never held across a panic");
entries.insert(key, CacheEntry::Failed(reason.clone()));
Err(SendraError::OAuthAcquisition {
token_url: auth.token_url.clone(),
reason,
})
}
}
}
async fn acquire_fresh(
auth: &OAuthAuth,
client: &HttpClient,
) -> Result<(String, Option<u64>), String> {
let mut form: Vec<(String, String)> = vec![
(
"grant_type".to_string(),
auth.grant_type.as_str().to_string(),
),
("client_id".to_string(), auth.client_id.clone()),
];
if let Some(client_secret) = &auth.client_secret {
form.push(("client_secret".to_string(), client_secret.clone()));
}
if let Some(scope) = &auth.scope {
form.push(("scope".to_string(), scope.clone()));
}
if auth.grant_type == OAuthGrantType::Password {
form.push((
"username".to_string(),
auth.username.clone().unwrap_or_default(),
));
form.push((
"password".to_string(),
auth.password.clone().unwrap_or_default(),
));
}
post_token_form(auth, client, form).await
}
pub async fn exchange_authorization_code(
auth: &OAuthAuth,
client: &HttpClient,
code: &str,
code_verifier: &str,
) -> Result<(String, Option<u64>), SendraError> {
let redirect_uri = auth.redirect_uri.clone().unwrap_or_default();
let mut form: Vec<(String, String)> = vec![
(
"grant_type".to_string(),
OAuthGrantType::AuthorizationCode.as_str().to_string(),
),
("code".to_string(), code.to_string()),
("redirect_uri".to_string(), redirect_uri),
("client_id".to_string(), auth.client_id.clone()),
("code_verifier".to_string(), code_verifier.to_string()),
];
if let Some(client_secret) = &auth.client_secret {
form.push(("client_secret".to_string(), client_secret.clone()));
}
post_token_form(auth, client, form)
.await
.map_err(|reason| SendraError::OAuthAcquisition {
token_url: auth.token_url.clone(),
reason,
})
}
async fn post_token_form(
auth: &OAuthAuth,
client: &HttpClient,
form: Vec<(String, String)>,
) -> Result<(String, Option<u64>), String> {
let body = serde_urlencoded::to_string(&form)
.expect("a Vec<(String, String)> always encodes as x-www-form-urlencoded pairs");
let request = Request {
name: None,
method: Method::Post,
url: auth.token_url.clone(),
headers: vec![(
"Content-Type".to_string(),
"application/x-www-form-urlencoded".to_string(),
)],
query: Vec::new(),
body: Some(body),
json: None,
body_file: None,
form: Vec::new(),
multipart: Vec::new(),
auth: None,
assertions: None,
pre_request: None,
post_request: None,
capture: None,
retry: None,
};
let response = send_prepared(&request, client)
.await
.map_err(|err| err.to_string())?;
if !(200..300).contains(&response.status) {
return Err(format!(
"token endpoint responded {} {}: {}",
response.status,
response.status_text,
truncate(&response.body)
));
}
let parsed: TokenResponse = serde_json::from_str(&response.body).map_err(|err| {
format!(
"could not parse the token response as JSON: {err} (body: {})",
truncate(&response.body)
)
})?;
Ok((parsed.access_token, parsed.expires_in))
}
pub struct PkcePair {
pub verifier: String,
pub challenge: String,
}
pub fn generate_pkce() -> PkcePair {
let mut bytes = [0u8; 32];
bytes[..16].copy_from_slice(uuid::Uuid::new_v4().as_bytes());
bytes[16..].copy_from_slice(uuid::Uuid::new_v4().as_bytes());
let verifier = URL_SAFE_NO_PAD.encode(bytes);
let challenge = URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes()));
PkcePair {
verifier,
challenge,
}
}
pub fn generate_state() -> String {
uuid::Uuid::new_v4().to_string()
}
pub fn build_authorization_url(
auth: &OAuthAuth,
state: &str,
code_challenge: &str,
) -> Result<String, SendraError> {
let authorization_url = auth.authorization_url.clone().unwrap_or_default();
let mut url = reqwest::Url::parse(&authorization_url).map_err(|source| {
SendraError::OAuthAuthorizationUrl {
authorization_url: authorization_url.clone(),
reason: source.to_string(),
}
})?;
{
let mut pairs = url.query_pairs_mut();
pairs.append_pair("response_type", "code");
pairs.append_pair("client_id", &auth.client_id);
pairs.append_pair(
"redirect_uri",
auth.redirect_uri.as_deref().unwrap_or_default(),
);
pairs.append_pair("code_challenge", code_challenge);
pairs.append_pair("code_challenge_method", "S256");
pairs.append_pair("state", state);
if let Some(scope) = &auth.scope {
pairs.append_pair("scope", scope);
}
}
Ok(url.into())
}
fn truncate(body: &str) -> String {
const MAX_CHARS: usize = 200;
if body.chars().count() <= MAX_CHARS {
body.to_string()
} else {
format!("{}...", body.chars().take(MAX_CHARS).collect::<String>())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::Config;
use crate::http::client::build_client;
use std::io::{BufRead, BufReader, Write};
use std::net::{SocketAddr, TcpListener};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
fn oauth_auth(token_url: &str) -> OAuthAuth {
OAuthAuth {
grant_type: OAuthGrantType::ClientCredentials,
token_url: token_url.to_string(),
client_id: "client-id".to_string(),
client_secret: Some("client-secret".to_string()),
scope: None,
username: None,
password: None,
authorization_url: None,
redirect_uri: None,
}
}
fn client() -> HttpClient {
build_client(&Config::default()).expect("a client builds")
}
struct TokenServer {
addr: SocketAddr,
hits: Arc<AtomicUsize>,
}
impl TokenServer {
fn start(response: Vec<u8>) -> Self {
let listener = TcpListener::bind("127.0.0.1:0").expect("an ephemeral port is free");
let addr = listener.local_addr().expect("the listener has an address");
let hits = Arc::new(AtomicUsize::new(0));
let counted = hits.clone();
std::thread::spawn(move || {
for stream in listener.incoming() {
let Ok(stream) = stream else { continue };
let mut writer = stream.try_clone().expect("the socket clones");
let mut reader = BufReader::new(stream);
let mut request_line = String::new();
if reader.read_line(&mut request_line).unwrap_or(0) == 0 {
continue;
}
let mut content_length = 0usize;
loop {
let mut header = String::new();
match reader.read_line(&mut header) {
Ok(0) | Err(_) => break,
Ok(_) if header == "\r\n" => break,
Ok(_) => {
if let Some((name, value)) = header.split_once(':') {
if name.trim().eq_ignore_ascii_case("content-length") {
content_length = value.trim().parse().unwrap_or(0);
}
}
}
}
}
let mut body = vec![0u8; content_length];
if content_length > 0 {
use std::io::Read;
let _ = reader.read_exact(&mut body);
}
counted.fetch_add(1, Ordering::SeqCst);
if writer.write_all(&response).is_err() {
continue;
}
let _ = writer.flush();
}
});
Self { addr, hits }
}
fn token_url(&self) -> String {
format!("http://{}/token", self.addr)
}
fn hits(&self) -> usize {
self.hits.load(Ordering::SeqCst)
}
}
fn token_response(body: &'static str) -> Vec<u8> {
format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
body.len()
)
.into_bytes()
}
#[tokio::test]
async fn client_credentials_acquires_a_token() {
let server = TokenServer::start(token_response(
r#"{"access_token": "abc123", "token_type": "Bearer"}"#,
));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
let token = acquire_token(&auth, &client, &cache)
.await
.expect("the mock token endpoint answers");
assert_eq!(token, "abc123");
}
#[tokio::test]
async fn password_grant_acquires_a_token() {
let server = TokenServer::start(token_response(r#"{"access_token": "pwd-token"}"#));
let auth = OAuthAuth {
grant_type: OAuthGrantType::Password,
username: Some("ada".to_string()),
password: Some("s3cr3t".to_string()),
..oauth_auth(&server.token_url())
};
let client = client();
let cache = OAuthTokenCache::new();
let token = acquire_token(&auth, &client, &cache)
.await
.expect("the mock token endpoint answers");
assert_eq!(token, "pwd-token");
}
#[tokio::test]
async fn a_token_is_reused_across_requests_sharing_the_same_config() {
let server = TokenServer::start(token_response(r#"{"access_token": "shared"}"#));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
for _ in 0..3 {
let token = acquire_token(&auth, &client, &cache)
.await
.expect("acquires or reuses successfully");
assert_eq!(token, "shared");
}
assert_eq!(
server.hits(),
1,
"three requests through one config must acquire exactly one token"
);
}
#[tokio::test]
async fn a_token_with_no_expires_in_is_reused_indefinitely() {
let server = TokenServer::start(token_response(r#"{"access_token": "no-expiry"}"#));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
for _ in 0..5 {
acquire_token(&auth, &client, &cache)
.await
.expect("acquires or reuses successfully");
}
assert_eq!(
server.hits(),
1,
"omitting `expires_in` must be treated as not expiring for this run, not \
reacquired on every call"
);
}
#[tokio::test]
async fn an_expired_cached_token_triggers_reacquisition() {
let server = TokenServer::start(token_response(
r#"{"access_token": "still-first", "expires_in": 0}"#,
));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
acquire_token(&auth, &client, &cache)
.await
.expect("the first acquisition succeeds");
acquire_token(&auth, &client, &cache)
.await
.expect("reacquisition against the same, still-up server succeeds");
assert_eq!(
server.hits(),
2,
"an expired cached token must trigger a fresh acquisition"
);
}
#[tokio::test]
async fn a_non_2xx_token_response_is_a_typed_acquisition_error() {
let server = TokenServer::start(
b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 20\r\n\r\n{\"error\":\"denied\"}\r\n"
.to_vec(),
);
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
let err = acquire_token(&auth, &client, &cache)
.await
.expect_err("a 401 must not be treated as success");
match err {
SendraError::OAuthAcquisition { reason, .. } => {
assert!(reason.contains("401"), "got {reason}");
}
other => panic!("expected OAuthAcquisition, got {other:?}"),
}
}
#[tokio::test]
async fn a_malformed_token_response_is_a_typed_acquisition_error() {
let server = TokenServer::start(token_response("not json"));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
let err = acquire_token(&auth, &client, &cache)
.await
.expect_err("a non-JSON body must not be treated as success");
assert!(matches!(err, SendraError::OAuthAcquisition { .. }));
}
#[tokio::test]
async fn a_failed_acquisition_is_remembered_and_not_retried() {
let server = TokenServer::start(token_response("not json"));
let auth = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
assert!(acquire_token(&auth, &client, &cache).await.is_err());
assert!(acquire_token(&auth, &client, &cache).await.is_err());
assert_eq!(
server.hits(),
1,
"a config that already failed once in this run must not be retried against \
the endpoint for a second request"
);
}
#[tokio::test]
async fn two_different_scopes_are_cached_separately() {
let server = TokenServer::start(token_response(r#"{"access_token": "tok"}"#));
let base = oauth_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
let scoped_a = OAuthAuth {
scope: Some("read".to_string()),
..base.clone()
};
let scoped_b = OAuthAuth {
scope: Some("write".to_string()),
..base
};
acquire_token(&scoped_a, &client, &cache)
.await
.expect("acquires");
acquire_token(&scoped_b, &client, &cache)
.await
.expect("acquires");
assert_eq!(
server.hits(),
2,
"two different scopes must not share one cache entry"
);
}
fn authorization_code_auth(token_url: &str) -> OAuthAuth {
OAuthAuth {
grant_type: OAuthGrantType::AuthorizationCode,
token_url: token_url.to_string(),
client_id: "client-id".to_string(),
client_secret: None,
scope: None,
username: None,
password: None,
authorization_url: Some("https://auth.example.com/authorize".to_string()),
redirect_uri: Some("http://127.0.0.1:8899/callback".to_string()),
}
}
#[tokio::test]
async fn authorization_code_never_reaches_the_token_endpoint_via_acquire_token() {
let server = TokenServer::start(token_response(r#"{"access_token": "unused"}"#));
let auth = authorization_code_auth(&server.token_url());
let client = client();
let cache = OAuthTokenCache::new();
let err = acquire_token(&auth, &client, &cache)
.await
.expect_err("no code is available for an automatic acquisition");
match err {
SendraError::OAuthAcquisition { reason, .. } => {
assert!(reason.contains("interactive"), "got {reason}");
}
other => panic!("expected OAuthAcquisition, got {other:?}"),
}
assert_eq!(
server.hits(),
0,
"acquire_token must never hit the token endpoint for authorization_code — there is \
no code to send"
);
}
#[tokio::test]
async fn a_token_inserted_via_insert_token_is_served_by_acquire_token_afterward() {
let server = TokenServer::start(token_response(r#"{"access_token": "unused"}"#));
let auth = authorization_code_auth(&server.token_url());
let cache = OAuthTokenCache::new();
cache.insert_token(&auth, "interactively-acquired".to_string(), Some(3600));
let client = client();
let token = acquire_token(&auth, &client, &cache)
.await
.expect("a token inserted via insert_token must be served like any other");
assert_eq!(token, "interactively-acquired");
assert_eq!(
server.hits(),
0,
"serving an inserted token must never itself hit the token endpoint"
);
}
#[test]
fn generate_pkce_produces_a_verifier_and_a_matching_s256_challenge() {
let pkce = generate_pkce();
assert_eq!(
pkce.verifier.len(),
43,
"32 bytes base64url-no-pad encodes to exactly 43 characters"
);
assert!(
pkce.verifier
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
"the verifier must use only RFC 7636's unreserved base64url characters: {}",
pkce.verifier
);
let expected_challenge = URL_SAFE_NO_PAD.encode(Sha256::digest(pkce.verifier.as_bytes()));
assert_eq!(pkce.challenge, expected_challenge);
}
#[test]
fn generate_pkce_never_repeats_across_calls() {
let a = generate_pkce();
let b = generate_pkce();
assert_ne!(a.verifier, b.verifier);
assert_ne!(a.challenge, b.challenge);
}
#[test]
fn generate_state_never_repeats_across_calls() {
assert_ne!(generate_state(), generate_state());
}
#[test]
fn build_authorization_url_includes_every_required_parameter() {
let mut auth = authorization_code_auth("https://auth.example.com/token");
auth.scope = Some("read write".to_string());
let url = build_authorization_url(&auth, "csrf-state", "the-challenge")
.expect("a valid authorization_url must build");
let parsed = reqwest::Url::parse(&url).expect("the result is itself a valid URL");
let pairs: std::collections::HashMap<_, _> = parsed.query_pairs().into_owned().collect();
assert_eq!(pairs.get("response_type").map(String::as_str), Some("code"));
assert_eq!(
pairs.get("client_id").map(String::as_str),
Some("client-id")
);
assert_eq!(
pairs.get("redirect_uri").map(String::as_str),
Some("http://127.0.0.1:8899/callback")
);
assert_eq!(
pairs.get("code_challenge").map(String::as_str),
Some("the-challenge")
);
assert_eq!(
pairs.get("code_challenge_method").map(String::as_str),
Some("S256")
);
assert_eq!(pairs.get("state").map(String::as_str), Some("csrf-state"));
assert_eq!(pairs.get("scope").map(String::as_str), Some("read write"));
}
#[test]
fn build_authorization_url_rejects_an_unparseable_authorization_url() {
let mut auth = authorization_code_auth("https://auth.example.com/token");
auth.authorization_url = Some("not a url".to_string());
let err = build_authorization_url(&auth, "state", "challenge").expect_err(
"an unparseable authorization_url must be rejected before any network call",
);
assert!(matches!(err, SendraError::OAuthAuthorizationUrl { .. }));
}
#[tokio::test]
async fn exchange_authorization_code_acquires_a_token() {
let server = TokenServer::start(token_response(r#"{"access_token": "exchanged"}"#));
let auth = authorization_code_auth(&server.token_url());
let client = client();
let (token, expires_in) =
exchange_authorization_code(&auth, &client, "the-code", "the-verifier")
.await
.expect("the mock token endpoint answers");
assert_eq!(token, "exchanged");
assert_eq!(expires_in, None);
}
#[tokio::test]
async fn exchange_authorization_code_surfaces_a_non_2xx_response_as_a_typed_error() {
let server = TokenServer::start(
b"HTTP/1.1 400 Bad Request\r\nContent-Length: 20\r\n\r\n{\"error\":\"invalid\"}\r\n"
.to_vec(),
);
let auth = authorization_code_auth(&server.token_url());
let client = client();
let err = exchange_authorization_code(&auth, &client, "the-code", "the-verifier")
.await
.expect_err("a 400 must not be treated as success");
match err {
SendraError::OAuthAcquisition { reason, .. } => {
assert!(reason.contains("400"), "got {reason}");
}
other => panic!("expected OAuthAcquisition, got {other:?}"),
}
}
}