#![cfg(not(target_arch = "wasm32"))]
use anyhow::Result;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, RwLock};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::oneshot;
use tokio::time::Duration;
const DEFAULT_AUTHORIZE_BASE: &str = "https://pluto.openrtc.app/sso/authorize";
const DEFAULT_FIREBASE_API_KEY: &str = "AIzaSyA62Krj-7ZYFT5xjrTUq7mXana41Ahj_mM";
#[derive(Debug, Clone)]
pub struct SsoConfig {
pub authorize_base_url: String,
pub callback_host: String,
pub callback_port: u16,
}
impl Default for SsoConfig {
fn default() -> Self {
Self {
authorize_base_url: DEFAULT_AUTHORIZE_BASE.to_string(),
callback_host: "127.0.0.1".to_string(),
callback_port: 0,
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SsoCallbackPayload {
pub state: String,
pub code: Option<String>,
pub token: Option<String>,
pub error: Option<String>,
pub raw_query: String,
}
pub struct SsoSession {
pub authorize_url: String,
pub callback_url: String,
pub state: String,
result_rx: oneshot::Receiver<Result<SsoCallbackPayload>>,
shutdown_tx: Option<oneshot::Sender<()>>,
}
#[derive(Debug, Clone)]
pub struct PlutoSsoConfig {
pub session: SsoConfig,
pub firebase_api_key: String,
pub callback_timeout: Duration,
pub open_browser: bool,
}
impl Default for PlutoSsoConfig {
fn default() -> Self {
Self {
session: SsoConfig::default(),
firebase_api_key: DEFAULT_FIREBASE_API_KEY.to_string(),
callback_timeout: Duration::from_secs(300),
open_browser: true,
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PlutoAuthSession {
pub custom_token: String,
pub id_token: String,
pub refresh_token: Option<String>,
pub expires_in_seconds: Option<i64>,
pub user_id: Option<String>,
}
#[derive(Clone, Default)]
pub struct ManagedAuthState {
session: Arc<RwLock<Option<PlutoAuthSession>>>,
}
impl ManagedAuthState {
pub fn new() -> Self {
Self::default()
}
pub fn token_provider(&self) -> Box<dyn Fn() -> Option<String> + Send + Sync> {
let session = self.session.clone();
Box::new(move || {
session
.read()
.ok()
.and_then(|guard| guard.as_ref().map(|value| value.id_token.clone()))
})
}
pub fn current_session(&self) -> Option<PlutoAuthSession> {
self.session.read().ok().and_then(|guard| guard.clone())
}
pub fn set_session(&self, session: PlutoAuthSession) {
if let Ok(mut guard) = self.session.write() {
*guard = Some(session);
}
}
pub fn clear(&self) {
if let Ok(mut guard) = self.session.write() {
*guard = None;
}
}
pub async fn sign_in_with_pluto(&self, config: PlutoSsoConfig) -> Result<PlutoAuthSession> {
let session = sign_in_with_pluto(config).await?;
self.set_session(session.clone());
Ok(session)
}
}
impl SsoSession {
pub async fn wait_for_callback(self, timeout: Duration) -> Result<SsoCallbackPayload> {
let received = tokio::time::timeout(timeout, self.result_rx)
.await
.map_err(|_| anyhow::anyhow!("sso callback timed out"))?;
received.map_err(|_| anyhow::anyhow!("sso callback channel closed"))?
}
pub fn cancel(mut self) {
if let Some(tx) = self.shutdown_tx.take() {
let _ = tx.send(());
}
}
}
pub async fn start_sso_session(config: SsoConfig) -> Result<SsoSession> {
let bind_addr: SocketAddr = format!("{}:{}", config.callback_host, config.callback_port)
.parse()
.map_err(|e| anyhow::anyhow!("invalid callback bind address: {}", e))?;
let listener = TcpListener::bind(bind_addr)
.await
.map_err(|e| anyhow::anyhow!("failed to bind callback listener: {}", e))?;
let local_addr = listener
.local_addr()
.map_err(|e| anyhow::anyhow!("failed to read local callback address: {}", e))?;
let callback_url = format!("http://{}/callback", local_addr);
let state = uuid::Uuid::new_v4().to_string();
let authorize_url = build_authorize_url(&config.authorize_base_url, &callback_url, &state);
let expected_state = state.clone();
let (result_tx, result_rx) = oneshot::channel::<Result<SsoCallbackPayload>>();
let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>();
tokio::spawn(async move {
tokio::select! {
_ = &mut shutdown_rx => {}
accept_res = listener.accept() => {
match accept_res {
Ok((mut socket, _)) => {
let result = read_and_validate_callback(&mut socket, &expected_state).await;
let (status_code, body) = if result.is_ok() {
(
"200 OK",
"<html><body><h3>Sign-in complete</h3><p>You can close this tab.</p></body></html>",
)
} else {
(
"400 Bad Request",
"<html><body><h3>Sign-in failed</h3><p>You can close this tab.</p></body></html>",
)
};
let response = format!(
"HTTP/1.1 {}\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
status_code,
body.len(),
body
);
let _ = socket.write_all(response.as_bytes()).await;
let _ = socket.shutdown().await;
let _ = result_tx.send(result);
}
Err(error) => {
let _ = result_tx.send(Err(anyhow::anyhow!(
"failed to accept callback connection: {}",
error
)));
}
}
}
}
});
Ok(SsoSession {
authorize_url,
callback_url,
state,
result_rx,
shutdown_tx: Some(shutdown_tx),
})
}
pub async fn sign_in_with_pluto(config: PlutoSsoConfig) -> Result<PlutoAuthSession> {
let sso_session = start_sso_session(config.session.clone()).await?;
let authorize_url = sso_session.authorize_url.clone();
if config.open_browser {
open::that(&authorize_url).map_err(|error| {
anyhow::anyhow!("failed opening Pluto SSO authorize URL: {}", error)
})?;
}
let callback = sso_session
.wait_for_callback(config.callback_timeout)
.await?;
complete_pluto_sign_in(callback, &config.firebase_api_key).await
}
pub async fn complete_pluto_sign_in(
callback: SsoCallbackPayload,
firebase_api_key: &str,
) -> Result<PlutoAuthSession> {
if let Some(error) = callback.error {
return Err(anyhow::anyhow!("pluto sso failed: {}", error));
}
let custom_token = callback
.token
.map(|token| token.trim().to_string())
.filter(|token| !token.is_empty())
.ok_or_else(|| anyhow::anyhow!("pluto sso callback did not include a custom token"))?;
exchange_custom_token(&custom_token, firebase_api_key).await
}
pub async fn exchange_custom_token(
custom_token: &str,
firebase_api_key: &str,
) -> Result<PlutoAuthSession> {
#[derive(serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct RequestPayload<'a> {
token: &'a str,
return_secure_token: bool,
}
#[derive(serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct ResponsePayload {
id_token: String,
refresh_token: Option<String>,
expires_in: Option<String>,
user_id: Option<String>,
}
let url = format!(
"https://identitytoolkit.googleapis.com/v1/accounts:signInWithCustomToken?key={}",
firebase_api_key
);
let response = reqwest::Client::new()
.post(url)
.json(&RequestPayload {
token: custom_token,
return_secure_token: true,
})
.send()
.await
.map_err(|error| anyhow::anyhow!("custom token exchange request failed: {}", error))?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
return Err(anyhow::anyhow!(
"custom token exchange failed status={} body={}",
status,
body
));
}
let payload: ResponsePayload = response.json().await.map_err(|error| {
anyhow::anyhow!("failed parsing custom token exchange response: {}", error)
})?;
Ok(PlutoAuthSession {
custom_token: custom_token.to_string(),
id_token: payload.id_token,
refresh_token: payload.refresh_token,
expires_in_seconds: payload
.expires_in
.as_deref()
.and_then(|value| value.parse::<i64>().ok()),
user_id: payload.user_id,
})
}
fn build_authorize_url(base: &str, redirect_uri: &str, state: &str) -> String {
let delimiter = if base.contains('?') { "&" } else { "?" };
format!(
"{}{}redirect_uri={}&state={}",
base,
delimiter,
percent_encode(redirect_uri),
percent_encode(state)
)
}
async fn read_and_validate_callback(
socket: &mut TcpStream,
expected_state: &str,
) -> Result<SsoCallbackPayload> {
let mut buf = [0u8; 8192];
let read = socket
.read(&mut buf)
.await
.map_err(|e| anyhow::anyhow!("failed reading callback request: {}", e))?;
if read == 0 {
return Err(anyhow::anyhow!("empty callback request"));
}
let request = String::from_utf8_lossy(&buf[..read]);
let first_line = request
.lines()
.next()
.ok_or_else(|| anyhow::anyhow!("invalid callback request line"))?;
let mut parts = first_line.split_whitespace();
let _method = parts.next().unwrap_or_default();
let target = parts.next().unwrap_or_default();
let query = target
.split_once('?')
.map(|(_, q)| q)
.unwrap_or_default()
.to_string();
let params = parse_query(&query);
let state = params.get("state").cloned().unwrap_or_default();
if state != expected_state {
return Err(anyhow::anyhow!("invalid callback state"));
}
Ok(SsoCallbackPayload {
state,
code: params.get("code").cloned(),
token: params.get("token").cloned(),
error: params.get("error").cloned(),
raw_query: query,
})
}
fn parse_query(query: &str) -> HashMap<String, String> {
let mut out = HashMap::new();
for pair in query.split('&') {
if pair.is_empty() {
continue;
}
let (key, value) = pair.split_once('=').unwrap_or((pair, ""));
out.insert(percent_decode(key), percent_decode(value));
}
out
}
fn percent_encode(input: &str) -> String {
let mut out = String::new();
for b in input.as_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!("%{:02X}", b)),
}
}
out
}
fn percent_decode(input: &str) -> String {
let bytes = input.as_bytes();
let mut out = Vec::with_capacity(bytes.len());
let mut index = 0usize;
while index < bytes.len() {
match bytes[index] {
b'+' => {
out.push(b' ');
index += 1;
}
b'%' if index + 2 < bytes.len() => {
let hi = bytes[index + 1] as char;
let lo = bytes[index + 2] as char;
if let (Some(hi), Some(lo)) = (hi.to_digit(16), lo.to_digit(16)) {
out.push(((hi << 4) | lo) as u8);
index += 3;
} else {
out.push(bytes[index]);
index += 1;
}
}
ch => {
out.push(ch);
index += 1;
}
}
}
String::from_utf8_lossy(&out).to_string()
}
#[cfg(test)]
mod tests {
use super::{parse_query, percent_decode, percent_encode, ManagedAuthState, PlutoAuthSession};
#[test]
fn percent_roundtrip() {
let input = "http://127.0.0.1:4242/callback?x=1 2";
let encoded = percent_encode(input);
let decoded = percent_decode(&encoded);
assert_eq!(decoded, input);
}
#[test]
fn parses_query_map() {
let parsed = parse_query("code=abc&state=s1&error=");
assert_eq!(parsed.get("code").map(String::as_str), Some("abc"));
assert_eq!(parsed.get("state").map(String::as_str), Some("s1"));
assert_eq!(parsed.get("error").map(String::as_str), Some(""));
}
#[test]
fn managed_auth_state_exposes_id_token() {
let state = ManagedAuthState::new();
let provider = state.token_provider();
assert_eq!(provider(), None);
state.set_session(PlutoAuthSession {
custom_token: "custom-token".to_string(),
id_token: "id-token".to_string(),
refresh_token: Some("refresh-token".to_string()),
expires_in_seconds: Some(3600),
user_id: Some("user-123".to_string()),
});
assert_eq!(provider().as_deref(), Some("id-token"));
assert_eq!(
state.current_session().and_then(|value| value.user_id),
Some("user-123".to_string())
);
}
}