use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::process::{Command, Stdio};
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine;
use secrecy::SecretString;
use serde::Deserialize;
use sha2::{Digest, Sha256};
use url::form_urlencoded;
use super::store::{self, StoredToken};
use super::{build_client, classify_status, validate, AuthError, Session};
const CLIENT_ID: &str = "fluidattacks-cli";
const AUTHORIZE_ENDPOINT: &str = "https://app.fluidattacks.com/auth/oauth/authorize";
const TOKEN_ENDPOINT: &str = "https://app.fluidattacks.com/auth/oauth/token";
const LOGOUT_ENDPOINT: &str = "https://app.fluidattacks.com/auth/oauth/logout";
const LOOPBACK_HOST: &str = "127.0.0.1";
const CALLBACK_PATH: &str = "/callback";
const LOGIN_TIMEOUT: Duration = Duration::from_mins(5);
const POLL_INTERVAL: Duration = Duration::from_millis(200);
const READ_TIMEOUT: Duration = Duration::from_secs(5);
const REFRESH_SKEW_SECONDS: u64 = 60;
const SUCCESS_BODY: &str = "<html><body>Login complete. You can close this window.</body></html>";
const FAILURE_BODY: &str =
"<html><body>Login failed. Close this window and retry from the CLI.</body></html>";
pub fn authenticate_oauth() -> Result<Session, AuthError> {
let listener = bind_loopback()?;
let redirect_uri = redirect_uri(&listener)?;
let pkce = Pkce::new();
let state = new_opaque();
let url = build_authorize_url(&redirect_uri, &pkce.challenge, &state)?;
announce_login(&url);
open_browser(&url);
let code = await_callback(&listener, &state)?;
let tokens = exchange_code(&code, &pkce.verifier, &redirect_uri)?;
persist_session(tokens)
}
pub fn resolve_stored_session() -> Result<Session, AuthError> {
let Some(stored) = load_stored()? else {
return Err(AuthError::NotAuthenticated);
};
let access_token = if stored.is_access_expired(now(), REFRESH_SKEW_SECONDS) {
refresh_stored(&stored)?
} else {
stored.access_token
};
let email = validate(&access_token)?;
Ok(Session {
email,
token: SecretString::new(access_token.into_boxed_str()),
})
}
pub fn logout() -> Result<(), AuthError> {
if let Some(stored) = load_stored()? {
let _ = post_logout(&stored.refresh_token);
}
store::delete().map_err(local)
}
pub fn whoami() -> Result<Option<String>, AuthError> {
Ok(load_stored()?.map(|stored| stored.email))
}
struct Pkce {
verifier: String,
challenge: String,
}
impl Pkce {
fn new() -> Self {
let verifier = new_opaque();
let challenge = challenge_for(&verifier);
Self {
verifier,
challenge,
}
}
}
fn challenge_for(verifier: &str) -> String {
URL_SAFE_NO_PAD.encode(Sha256::digest(verifier.as_bytes()))
}
fn new_opaque() -> String {
let mut bytes = Vec::with_capacity(32);
bytes.extend_from_slice(uuid::Uuid::new_v4().as_bytes());
bytes.extend_from_slice(uuid::Uuid::new_v4().as_bytes());
URL_SAFE_NO_PAD.encode(&bytes)
}
fn bind_loopback() -> Result<TcpListener, AuthError> {
TcpListener::bind((LOOPBACK_HOST, 0)).map_err(local)
}
fn redirect_uri(listener: &TcpListener) -> Result<String, AuthError> {
let port = listener.local_addr().map_err(local)?.port();
Ok(format!("http://{LOOPBACK_HOST}:{port}{CALLBACK_PATH}"))
}
fn build_authorize_url(
redirect_uri: &str,
challenge: &str,
state: &str,
) -> Result<String, AuthError> {
let mut url = reqwest::Url::parse(AUTHORIZE_ENDPOINT).map_err(|err| local(err.to_string()))?;
url.query_pairs_mut()
.append_pair("response_type", "code")
.append_pair("client_id", CLIENT_ID)
.append_pair("redirect_uri", redirect_uri)
.append_pair("code_challenge", challenge)
.append_pair("code_challenge_method", "S256")
.append_pair("state", state);
Ok(String::from(url))
}
#[allow(clippy::print_stderr)]
fn announce_login(url: &str) {
eprintln!("Opening your browser to log in. If it does not open, visit:\n{url}");
}
fn open_browser(url: &str) {
let opener = if cfg!(target_os = "macos") {
"open"
} else {
"xdg-open"
};
let spawned = Command::new(opener)
.arg(url)
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn();
if let Err(err) = spawned {
tracing::debug!(error = %err, "could not launch a browser; the login URL was printed");
}
}
fn await_callback(listener: &TcpListener, state: &str) -> Result<String, AuthError> {
let deadline = Instant::now()
.checked_add(LOGIN_TIMEOUT)
.ok_or_else(|| local("clock overflow"))?;
listener.set_nonblocking(true).map_err(local)?;
while Instant::now() < deadline {
match listener.accept() {
Ok((stream, _)) => {
if let Some(code) = handle_stream(stream, state)? {
return Ok(code);
}
}
Err(ref err) if err.kind() == std::io::ErrorKind::WouldBlock => {
thread::sleep(POLL_INTERVAL);
}
Err(err) => return Err(local(err.to_string())),
}
}
Err(local("timed out waiting for the browser redirect"))
}
fn handle_stream(mut stream: TcpStream, expected_state: &str) -> Result<Option<String>, AuthError> {
let _ = stream.set_read_timeout(Some(READ_TIMEOUT));
let Some(request_line) = read_request_line(&stream) else {
return Ok(None);
};
let target = request_line.split_whitespace().nth(1).unwrap_or_default();
let params = parse_callback_query(target);
let outcome = classify_callback(¶ms, expected_state);
write_response(&mut stream, matches!(outcome, Ok(Some(_))));
outcome
}
fn read_request_line(stream: &TcpStream) -> Option<String> {
let mut line = String::new();
match BufReader::new(stream).read_line(&mut line) {
Ok(0) | Err(_) => None,
Ok(_) => Some(line),
}
}
#[derive(Default)]
struct CallbackParams {
code: Option<String>,
state: Option<String>,
error: Option<String>,
}
fn parse_callback_query(target: &str) -> CallbackParams {
let mut params = CallbackParams::default();
let Ok(url) = reqwest::Url::parse("http://127.0.0.1/").and_then(|base| base.join(target))
else {
return params;
};
for (key, value) in url.query_pairs() {
match key.as_ref() {
"code" => params.code = Some(value.into_owned()),
"state" => params.state = Some(value.into_owned()),
"error" => params.error = Some(value.into_owned()),
_ => {}
}
}
params
}
fn classify_callback(
params: &CallbackParams,
expected_state: &str,
) -> Result<Option<String>, AuthError> {
if params.state.as_deref() != Some(expected_state) {
return Ok(None);
}
if params.error.is_some() {
return Err(AuthError::Invalid);
}
Ok(params.code.clone())
}
fn write_response(stream: &mut TcpStream, ok: bool) {
let body = if ok { SUCCESS_BODY } else { FAILURE_BODY };
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/html; charset=utf-8\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(response.as_bytes());
let _ = stream.flush();
}
struct Tokens {
access_token: String,
refresh_token: String,
expires_in: u64,
}
fn exchange_code(code: &str, verifier: &str, redirect_uri: &str) -> Result<Tokens, AuthError> {
post_token(&[
("grant_type", "authorization_code"),
("code", code),
("code_verifier", verifier),
("client_id", CLIENT_ID),
("redirect_uri", redirect_uri),
])
}
fn refresh(refresh_token: &str) -> Result<Tokens, AuthError> {
post_token(&[
("grant_type", "refresh_token"),
("refresh_token", refresh_token),
])
}
fn post_token(form: &[(&str, &str)]) -> Result<Tokens, AuthError> {
let response = post_form(TOKEN_ENDPOINT, form)?;
let status = response.status();
if !status.is_success() {
return Err(classify_status(status.as_u16()));
}
let body = response
.text()
.map_err(|err| AuthError::Transport(err.without_url().to_string()))?;
parse_token_response(&body)
}
fn post_form(
endpoint: &str,
form: &[(&str, &str)],
) -> Result<reqwest::blocking::Response, AuthError> {
let body = form_urlencoded::Serializer::new(String::new())
.extend_pairs(form)
.finish();
build_client()?
.post(endpoint)
.header("Content-Type", "application/x-www-form-urlencoded")
.body(body)
.send()
.map_err(|err| AuthError::Transport(err.without_url().to_string()))
}
#[derive(Deserialize)]
struct TokenResponse {
access_token: Option<String>,
refresh_token: Option<String>,
expires_in: Option<u64>,
}
fn parse_token_response(body: &str) -> Result<Tokens, AuthError> {
let parsed: TokenResponse = serde_json::from_str(body)
.map_err(|_| AuthError::Transport("unexpected response from the platform".to_owned()))?;
match (parsed.access_token, parsed.refresh_token, parsed.expires_in) {
(Some(access_token), Some(refresh_token), Some(expires_in))
if !access_token.trim().is_empty() && !refresh_token.trim().is_empty() =>
{
Ok(Tokens {
access_token,
refresh_token,
expires_in,
})
}
_ => Err(AuthError::Invalid),
}
}
fn post_logout(refresh_token: &str) -> Result<(), AuthError> {
let _ = post_form(LOGOUT_ENDPOINT, &[("refresh_token", refresh_token)])?;
Ok(())
}
fn persist_session(tokens: Tokens) -> Result<Session, AuthError> {
let email = validate(&tokens.access_token)?;
save_tokens(&tokens, &email)?;
Ok(Session {
email,
token: SecretString::new(tokens.access_token.into_boxed_str()),
})
}
fn refresh_stored(stored: &StoredToken) -> Result<String, AuthError> {
let tokens = match refresh(&stored.refresh_token) {
Ok(tokens) => tokens,
Err(AuthError::Invalid) => {
let _ = store::delete();
return Err(AuthError::NotAuthenticated);
}
Err(err) => return Err(err),
};
save_tokens(&tokens, &stored.email)?;
Ok(tokens.access_token)
}
fn save_tokens(tokens: &Tokens, email: &str) -> Result<(), AuthError> {
store::save(&StoredToken {
access_token: tokens.access_token.clone(),
refresh_token: tokens.refresh_token.clone(),
expires_at: now().saturating_add(tokens.expires_in),
email: email.to_owned(),
})
.map_err(local)
}
fn load_stored() -> Result<Option<StoredToken>, AuthError> {
store::load().map_err(local)
}
#[allow(clippy::needless_pass_by_value)]
fn local(detail: impl ToString) -> AuthError {
AuthError::Local(detail.to_string())
}
fn now() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_secs())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn challenge_matches_rfc7636_vector() {
let verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk";
assert_eq!(
challenge_for(verifier),
"E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"
);
}
#[test]
fn opaque_values_are_url_safe_and_long() {
let value = new_opaque();
assert!(value.len() >= 43);
assert!(value
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'));
}
#[test]
fn opaque_values_differ_between_calls() {
assert_ne!(new_opaque(), new_opaque());
}
#[test]
fn authorize_url_carries_pkce_and_state() {
let url = build_authorize_url("http://127.0.0.1:1234/callback", "chal", "st").unwrap();
assert!(url.starts_with(AUTHORIZE_ENDPOINT));
assert!(url.contains("code_challenge_method=S256"));
assert!(url.contains("code_challenge=chal"));
assert!(url.contains("state=st"));
assert!(url.contains("client_id=fluidattacks-cli"));
assert!(url.contains("redirect_uri=http%3A%2F%2F127.0.0.1%3A1234%2Fcallback"));
}
#[test]
fn parse_query_extracts_code_and_state() {
let params = parse_callback_query("/callback?code=abc&state=xyz");
assert_eq!(params.code.as_deref(), Some("abc"));
assert_eq!(params.state.as_deref(), Some("xyz"));
assert!(params.error.is_none());
}
#[test]
fn parse_query_without_query_is_empty() {
let params = parse_callback_query("/favicon.ico");
assert!(params.code.is_none() && params.state.is_none());
}
#[test]
fn classify_requires_matching_state() {
let params = CallbackParams {
code: Some("c".to_owned()),
state: Some("good".to_owned()),
error: None,
};
assert_eq!(
classify_callback(¶ms, "good").unwrap(),
Some("c".to_owned())
);
assert_eq!(classify_callback(¶ms, "bad").unwrap(), None);
}
#[test]
fn classify_ignores_stray_requests() {
let params = CallbackParams::default();
assert_eq!(classify_callback(¶ms, "s").unwrap(), None);
}
#[test]
fn classify_rejects_provider_error_with_matching_state() {
let params = CallbackParams {
code: None,
state: Some("s".to_owned()),
error: Some("access_denied".to_owned()),
};
assert!(matches!(
classify_callback(¶ms, "s"),
Err(AuthError::Invalid)
));
}
#[test]
fn parse_token_response_ok() {
let body =
r#"{"access_token":"a","refresh_token":"r","expires_in":3600,"token_type":"Bearer"}"#;
let tokens = parse_token_response(body).unwrap();
assert_eq!(tokens.access_token, "a");
assert_eq!(tokens.refresh_token, "r");
assert_eq!(tokens.expires_in, 3600);
}
#[test]
fn parse_token_response_missing_or_blank_is_invalid() {
assert!(matches!(
parse_token_response(r#"{"access_token":"a"}"#),
Err(AuthError::Invalid)
));
assert!(matches!(
parse_token_response(r#"{"access_token":"","refresh_token":"r","expires_in":1}"#),
Err(AuthError::Invalid)
));
}
#[test]
fn parse_token_response_non_json_is_transport() {
assert!(matches!(
parse_token_response("<html>502</html>"),
Err(AuthError::Transport(_))
));
}
#[test]
fn pkce_challenge_matches_its_verifier() {
let pkce = Pkce::new();
assert_eq!(pkce.challenge, challenge_for(&pkce.verifier));
}
#[test]
fn redirect_uri_is_loopback_callback() {
let listener = bind_loopback().unwrap();
let uri = redirect_uri(&listener).unwrap();
assert!(uri.starts_with("http://127.0.0.1:"));
assert!(uri.ends_with("/callback"));
}
#[test]
fn local_error_wraps_detail() {
assert!(matches!(local("boom"), AuthError::Local(detail) if detail == "boom"));
}
#[test]
fn now_is_after_epoch() {
assert!(now() > 1_600_000_000);
}
#[test]
fn handle_stream_returns_code_on_matching_state() {
use std::io::Read;
let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
let addr = listener.local_addr().unwrap();
let client = thread::spawn(move || {
let mut stream = TcpStream::connect(addr).unwrap();
stream
.write_all(b"GET /callback?code=thecode&state=st HTTP/1.1\r\nHost: x\r\n\r\n")
.unwrap();
let mut buf = Vec::new();
let _ = stream.read_to_end(&mut buf);
buf
});
let (stream, _) = listener.accept().unwrap();
let code = handle_stream(stream, "st").unwrap();
let response = client.join().unwrap();
assert_eq!(code.as_deref(), Some("thecode"));
assert!(String::from_utf8_lossy(&response).contains("200 OK"));
}
#[test]
fn handle_stream_ignores_state_mismatch() {
use std::io::Read;
let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
let addr = listener.local_addr().unwrap();
let client = thread::spawn(move || {
let mut stream = TcpStream::connect(addr).unwrap();
stream
.write_all(b"GET /callback?code=c&state=wrong HTTP/1.1\r\n\r\n")
.unwrap();
let mut buf = Vec::new();
let _ = stream.read_to_end(&mut buf);
});
let (stream, _) = listener.accept().unwrap();
let outcome = handle_stream(stream, "expected");
client.join().unwrap();
assert_eq!(outcome.unwrap(), None);
}
#[test]
fn handle_stream_ignores_empty_connection() {
let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap();
let addr = listener.local_addr().unwrap();
let client = thread::spawn(move || {
let _stream = TcpStream::connect(addr).unwrap();
});
let (stream, _) = listener.accept().unwrap();
let outcome = handle_stream(stream, "expected");
client.join().unwrap();
assert_eq!(outcome.unwrap(), None);
}
}