diff --git a/.github/workflows/windows-smoke.yml b/.github/workflows/windows-smoke.yml index 05f7827..f6d8fb9 100644 --- a/.github/workflows/windows-smoke.yml +++ b/.github/workflows/windows-smoke.yml @@ -9,12 +9,14 @@ name: windows-smoke # sign-in starts in the background and hands back the approval link. It then # answers the way the approval page does when the developer clicks Deny, which # shows the sign-in outlived the command that started it and that the -# browser's callback reaches it. +# browser's callback reaches it, even with another program on one loopback +# address of the first callback port. on: workflow_dispatch: pull_request: paths: - cli/elevenlabs/workflow/onboard/** + - cli/elevenlabs/workflow/loopback.rs - .github/workflows/windows-smoke.yml jobs: @@ -82,6 +84,19 @@ jobs: fi } + # Another program on one loopback address of the first callback + # port. Browsers dial [::1] first for `localhost`, so a sign-in that + # left [::1]:8484 to it would never get the callback. + node -e "require('http').createServer((q, s) => s.end('another program')).listen(8484, '::1')" > /dev/null 2>&1 & + for _ in $(seq 1 50); do + [[ "$(curl -sg http://[::1]:8484/ || true)" == "another program" ]] && break + sleep 0.2 + done + if [[ "$(curl -sg http://[::1]:8484/ || true)" != "another program" ]]; then + echo "::error::The other program didn't start on [::1]:8484" + exit 1 + fi + step "$CLI onboard init" run # Starts the sign-in, waits for an approval that never comes, then says to keep waiting. step "$(jq -r .run <<< "$reply")" wait @@ -94,7 +109,12 @@ jobs: # What the approval page does when the developer clicks Deny. callback=$(node -p 'new URL(process.argv[1]).searchParams.get("redirect_uri")' "$url") - curl -sS -o /dev/null -w "Callback answered %{http_code}\n" "$callback?error=access_denied" + if [[ "$callback" != "http://localhost:8483/callback" ]]; then + echo "::error::The sign-in should have moved off 8484, it listens at $callback" + exit 1 + fi + state=$(node -p 'new URL(process.argv[1]).searchParams.get("state")' "$url") + curl -sS -o /dev/null -w "Callback answered %{http_code}\n" "$callback?error=access_denied&state=$state" step "$(jq -r .run <<< "$reply")" ask_user if ! jq -e '.say | contains("declined")' <<< "$reply" > /dev/null; then diff --git a/Cargo.lock b/Cargo.lock index 6462f53..6b54ec2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -525,7 +525,7 @@ checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2" [[package]] name = "elevenlabs-cli" -version = "1.7.0" +version = "1.7.1" dependencies = [ "anyhow", "arboard", @@ -2047,9 +2047,9 @@ dependencies = [ [[package]] name = "rustls-webpki" -version = "0.103.15" +version = "0.103.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" dependencies = [ "ring", "rustls-pki-types", diff --git a/Cargo.toml b/Cargo.toml index ae8f8d8..ab9f5b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "elevenlabs-cli" -version = "1.7.0" +version = "1.7.1" edition = "2021" description = "CLI for the ElevenLabs API" license = "MIT" diff --git a/cli/elevenlabs/workflow/loopback.rs b/cli/elevenlabs/workflow/loopback.rs new file mode 100644 index 0000000..8074971 --- /dev/null +++ b/cli/elevenlabs/workflow/loopback.rs @@ -0,0 +1,629 @@ +//! The browser's way back to a sign-in: the loopback listener the approval +//! page redirects to, with the PKCE exchange around it. `onboard` uses it in +//! place of the framework's listener, which is generated code; `auth login` +//! can move to it once the framework lets a CLI replace its declared flow. +//! +//! - It listens on both loopback addresses of one port, 127.0.0.1 and [::1]: +//! a browser may dial either for `localhost`. Listening on only one leaves +//! the other free for any program, and the browser then reaches that +//! program instead, with nothing telling the sign-in. A port where either +//! address is taken is skipped for the next registered one. +//! - Each connection is read on its own and with a time limit, so one that +//! never sends a request holds up nothing. +//! - A callback for another sign-in (a tab left open by an earlier one) is +//! answered with the error page, and the wait goes on: an old tab can't end +//! the sign-in the newest one is for. + +use std::io::ErrorKind; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::time::Duration; + +use base64::Engine; +use fern_cli_sdk::auth::{active_store, TokenBundle}; +use fern_cli_sdk::error::CliError; +use serde::Deserialize; +use sha2::Digest; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::mpsc; +use tokio::task::JoinSet; + +/// The path of every registered redirect URI. +const CALLBACK_PATH: &str = "/callback"; +/// How long one connection may take to send its request. +const READ_TIMEOUT: Duration = Duration::from_secs(10); +/// What the error page says when the login failed after the callback; the +/// reason is in the worker's log. +const AFTER_CALLBACK_FAILED: &str = + "The CLI could not complete the login. Check your terminal for details."; +/// What the error page says to a tab whose sign-in isn't the one waiting. +const NOT_THIS_SIGN_IN: &str = + "This approval is from an earlier sign-in, so the CLI didn't use it. Approve in the newest tab."; + +/// One browser sign-in: what `auth login` declares in `main.rs`, and what to +/// add to it. +pub(super) struct Login<'a> { + pub cli_name: &'a str, + pub scheme: &'a str, + pub client_id: &'a str, + pub authorize_url: &'a str, + pub token_url: &'a str, + pub scopes: &'a [String], + /// Host and ports of the registered redirect URIs. + pub redirect_host: &'a str, + pub redirect_ports: &'a [u16], + pub success_url: &'a str, + pub error_url: &'a str, + /// Added to the authorize URL. + pub params: &'a [(&'a str, &'a str)], + pub no_browser: bool, + /// How long the approval may take. + pub timeout: Duration, +} + +/// Sign in through the browser and store the token where `auth login` does. +pub(super) async fn sign_in(login: Login<'_>) -> Result<(), CliError> { + let verifier = random_token(); + let state = random_token(); + let (listeners, port) = bind(login.redirect_ports).await?; + let redirect_uri = format!("http://{}:{port}{CALLBACK_PATH}", login.redirect_host); + let url = authorize_link(&login, &redirect_uri, &state, &challenge(&verifier)); + // `Session::url` reads the link from the `URL:` line. + eprintln!("Opening browser to authenticate…\n URL: {url}\n Listening on {redirect_uri}"); + if !login.no_browser { + let _ = webbrowser::open(&url); + } + let (code, mut browser) = + wait_for_callback(listeners, &state, login.error_url, login.timeout).await?; + // The browser waits on the result page until the token is stored. + let stored = async { + let token = exchange( + login.token_url, + login.client_id, + &code, + &verifier, + &redirect_uri, + ) + .await?; + active_store().set(login.cli_name, login.scheme, &token.to_keyring_value()?) + }; + match stored.await { + Ok(()) => { + let _ = redirect(&mut browser, login.success_url).await; + Ok(()) + } + Err(e) => { + let page = error_page(login.error_url, "server_error", Some(AFTER_CALLBACK_FAILED)); + let _ = redirect(&mut browser, &page).await; + Err(e) + } + } +} + +/// Listen on both loopback addresses of the first port in `ports` where +/// neither is taken. +pub(super) async fn bind(ports: &[u16]) -> Result<(Vec, u16), CliError> { + let mut why = Vec::new(); + for &port in ports { + match bind_both(port).await { + Ok(listeners) => return Ok((listeners, port)), + Err(e) => why.push(format!("{port}: {e}")), + } + } + Err(CliError::Auth(format!( + "Could not listen for the browser on any sign-in port ({}). Is another sign-in running?", + why.join("; ") + ))) +} + +async fn bind_both(port: u16) -> std::io::Result> { + let mut listeners = Vec::new(); + let mut missing = None; + for ip in [ + IpAddr::V4(Ipv4Addr::LOCALHOST), + IpAddr::V6(Ipv6Addr::LOCALHOST), + ] { + match TcpListener::bind(SocketAddr::new(ip, port)).await { + Ok(listener) => listeners.push(listener), + // No such loopback address or family here (IPv6 turned off), so + // no browser dials it. + Err(e) if family_missing(&e) => missing = Some(e), + // Anything else and this port won't do: in use, reserved (Windows + // reports a port in an excluded range as access denied), or out of + // file handles, which says nothing about who holds the port. + Err(e) => return Err(e), + } + } + match missing { + Some(e) if listeners.is_empty() => Err(e), + _ => Ok(listeners), + } +} + +/// The error for an address family the OS doesn't have. +#[cfg(unix)] +const NO_FAMILY: i32 = libc::EAFNOSUPPORT; +#[cfg(windows)] +const NO_FAMILY: i32 = 10047; // WSAEAFNOSUPPORT + +/// This machine has no such loopback address or family at all, as opposed to +/// the port being taken or the bind failing for some other reason. +fn family_missing(e: &std::io::Error) -> bool { + e.kind() == ErrorKind::AddrNotAvailable || e.raw_os_error() == Some(NO_FAMILY) +} + +/// What a request to the listener turned out to be. +#[derive(Debug, PartialEq)] +pub(super) enum Callback { + /// Not the callback path: a favicon, a probe. + Other, + /// A callback this sign-in can't use: another sign-in's `state`, or none. + NotOurs, + Code(String), + /// The page's error, e.g. `access_denied` when the developer clicked Deny. + Error { + error: String, + description: Option, + }, +} + +/// `target` is the request target, `/callback?code=…&state=…`. +pub(super) fn classify(target: &str, state: &str) -> Callback { + let (path, query) = target.split_once('?').unwrap_or((target, "")); + if path != CALLBACK_PATH { + return Callback::Other; + } + let (mut code, mut got_state, mut error, mut description) = (None, None, None, None); + for (key, value) in form_urlencoded::parse(query.as_bytes()) { + let value = Some(value.into_owned()); + match key.as_ref() { + "code" => code = value, + "state" => got_state = value, + "error" => error = value, + "error_description" => description = value, + _ => {} + } + } + match (got_state, code, error) { + (Some(got), _, _) if got != state => Callback::NotOurs, + (Some(_), _, Some(error)) => Callback::Error { error, description }, + (Some(_), Some(code), None) => Callback::Code(code), + // Without a `state`, nothing from ElevenLabs: the page's Deny and the + // server's error redirects always carry it, so any page could have + // sent this, to end the sign-in or to put words in the agent's reply. + _ => Callback::NotOurs, + } +} + +/// Wait for this sign-in's callback. A code comes back with the browser's +/// connection still open, so the browser is answered once the login is done. +pub(super) async fn wait_for_callback( + listeners: Vec, + state: &str, + error_url: &str, + timeout: Duration, +) -> Result<(String, TcpStream), CliError> { + let (tx, mut rx) = mpsc::channel::<(String, TcpStream)>(16); + // Dropped on return, which stops every listener. + let mut accepting = JoinSet::new(); + for listener in listeners { + let tx = tx.clone(); + accepting.spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + // Out of file handles, say: try again shortly. + tokio::time::sleep(Duration::from_millis(50)).await; + continue; + }; + let tx = tx.clone(); + tokio::spawn(async move { + if let Some(target) = request_target(&mut socket).await { + let _ = tx.send((target, socket)).await; + } + }); + } + }); + } + drop(tx); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let Ok(Some((target, mut socket))) = tokio::time::timeout_at(deadline, rx.recv()).await + else { + return Err(CliError::Auth(format!( + "Timed out waiting for the OAuth callback after {}s.", + timeout.as_secs() + ))); + }; + match classify(&target, state) { + Callback::Other => { + let _ = respond(&mut socket, "404 Not Found", &[], "not found").await; + } + Callback::NotOurs => { + let page = error_page(error_url, "invalid_request", Some(NOT_THIS_SIGN_IN)); + let _ = redirect(&mut socket, &page).await; + } + Callback::Code(code) => return Ok((code, socket)), + Callback::Error { error, description } => { + let page = error_page(error_url, &error, description.as_deref()); + let _ = redirect(&mut socket, &page).await; + return Err(CliError::Auth(match description { + Some(d) => format!("Authorization server returned error: {error} ({d})"), + None => format!("Authorization server returned error: {error}"), + })); + } + } + } +} + +/// The request target of the connection's first line, or `None` when no +/// request arrives in time. The headers are read to their end, however long: +/// closing a connection with some still unread resets it, and the browser can +/// lose the response. +async fn request_target(socket: &mut TcpStream) -> Option { + let read = async { + let mut first = Vec::new(); + // The last bytes read, to find the blank line across reads. + let mut end = Vec::new(); + let mut chunk = [0u8; 4096]; + loop { + let n = socket.read(&mut chunk).await.ok()?; + if n == 0 { + break; + } + let room = 8 * 1024 - first.len().min(8 * 1024); + first.extend_from_slice(&chunk[..n.min(room)]); + end.extend_from_slice(&chunk[..n]); + if end.windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + end.drain(..end.len().saturating_sub(3)); + } + Some(first) + }; + let head = tokio::time::timeout(READ_TIMEOUT, read).await.ok()??; + let line = String::from_utf8_lossy(&head); + let mut parts = line.lines().next()?.split_whitespace(); + let _method = parts.next()?; + parts.next().map(str::to_string) +} + +async fn respond( + socket: &mut TcpStream, + status: &str, + headers: &[(&str, &str)], + body: &str, +) -> std::io::Result<()> { + let mut response = format!("HTTP/1.1 {status}\r\n"); + for (name, value) in headers { + response.push_str(&format!("{name}: {value}\r\n")); + } + response.push_str(&format!( + "Content-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + )); + socket.write_all(response.as_bytes()).await?; + socket.shutdown().await +} + +async fn redirect(socket: &mut TcpStream, location: &str) -> std::io::Result<()> { + respond(socket, "302 Found", &[("Location", location)], "").await +} + +fn error_page(url: &str, error: &str, description: Option<&str>) -> String { + let mut query = form_urlencoded::Serializer::new(String::new()); + query.append_pair("error", error); + if let Some(description) = description { + query.append_pair("error_description", description); + } + let separator = if url.contains('?') { '&' } else { '?' }; + format!("{url}{separator}{}", query.finish()) +} + +fn authorize_link(login: &Login<'_>, redirect_uri: &str, state: &str, challenge: &str) -> String { + let mut query = form_urlencoded::Serializer::new(String::new()); + query + .append_pair("response_type", "code") + .append_pair("client_id", login.client_id) + .append_pair("redirect_uri", redirect_uri) + .append_pair("state", state) + .append_pair("code_challenge", challenge) + .append_pair("code_challenge_method", "S256") + .append_pair("scope", &login.scopes.join(" ")); + for (key, value) in login.params { + query.append_pair(key, value); + } + let separator = if login.authorize_url.contains('?') { + '&' + } else { + '?' + }; + format!("{}{separator}{}", login.authorize_url, query.finish()) +} + +/// 64 random bytes, base64url: a PKCE verifier (RFC 7636 §4.1), or a `state`. +fn random_token() -> String { + use rand::RngCore; + let mut bytes = [0u8; 64]; + rand::thread_rng().fill_bytes(&mut bytes); + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes) +} + +fn challenge(verifier: &str) -> String { + base64::engine::general_purpose::URL_SAFE_NO_PAD + .encode(sha2::Sha256::digest(verifier.as_bytes())) +} + +/// Trade the code for a token, as `auth login` does. +async fn exchange( + token_url: &str, + client_id: &str, + code: &str, + verifier: &str, + redirect_uri: &str, +) -> Result { + #[derive(Deserialize)] + struct Token { + access_token: String, + refresh_token: Option, + expires_in: Option, + } + let failed = |e: String| CliError::Auth(format!("PKCE token exchange failed: {e}")); + let response = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(30)) + .build() + .map_err(|e| failed(e.to_string()))? + .post(token_url) + .form(&[ + ("grant_type", "authorization_code"), + ("code", code), + ("code_verifier", verifier), + ("client_id", client_id), + ("redirect_uri", redirect_uri), + ]) + .send() + .await + .map_err(|e| failed(e.to_string()))?; + let status = response.status(); + let body = response.text().await.map_err(|e| failed(e.to_string()))?; + if !status.is_success() { + let detail = serde_json::from_str::(&body) + .ok() + .and_then(|v| { + v.get("error_description") + .or_else(|| v.get("error")) + .and_then(|d| d.as_str().map(str::to_string)) + }) + .unwrap_or_else(|| body.chars().take(200).collect()); + return Err(failed(format!("HTTP {status}: {detail}"))); + } + let token: Token = + serde_json::from_str(&body).map_err(|e| failed(format!("token response: {e}")))?; + Ok(TokenBundle::from_token_response( + &token.access_token, + token.refresh_token.as_deref(), + token.expires_in, + )) +} + +#[cfg(test)] +mod tests { + use std::io::{Read, Write}; + use std::net::{Ipv6Addr, TcpListener, TcpStream}; + use std::time::Duration; + + use super::{bind, classify, wait_for_callback, Callback}; + + const ERROR_PAGE: &str = "https://example.test/error"; + + /// Without an IPv6 loopback (some containers) these tests can't run. + fn no_ipv6() -> bool { + let missing = TcpListener::bind((Ipv6Addr::LOCALHOST, 0)).is_err(); + if missing { + eprintln!("skipped: no IPv6 loopback here"); + } + missing + } + + /// A port free on both loopback addresses, a moment ago. + fn free_port() -> u16 { + loop { + let port = TcpListener::bind(("127.0.0.1", 0)) + .unwrap() + .local_addr() + .unwrap() + .port(); + if TcpListener::bind((Ipv6Addr::LOCALHOST, port)).is_ok() { + return port; + } + } + } + + /// Listening on a free port. Other tests take ports too, so a port that + /// was free a moment ago may not be anymore. + async fn listen() -> (Vec, u16) { + for _ in 0..20 { + if let Ok(bound) = bind(&[free_port()]).await { + return bound; + } + } + panic!("no free port to listen on"); + } + + /// What the listener answers a browser that requests `target`. + fn request(host: &str, port: u16, target: &str) -> String { + request_with(host, port, target, "") + } + + fn request_with(host: &str, port: u16, target: &str, headers: &str) -> String { + let mut stream = TcpStream::connect((host, port)).unwrap(); + write!( + stream, + "GET {target} HTTP/1.1\r\nHost: localhost:{port}\r\n{headers}\r\n" + ) + .unwrap(); + let mut response = String::new(); + stream.read_to_string(&mut response).unwrap(); + response + } + + #[test] + fn only_a_missing_address_or_family_lets_the_listener_use_one_address() { + use super::{family_missing, NO_FAMILY}; + use std::io::{Error, ErrorKind}; + assert!(family_missing(&Error::from(ErrorKind::AddrNotAvailable))); + assert!(family_missing(&Error::from_raw_os_error(NO_FAMILY))); + // Out of file handles, say: the port might still be someone else's. + for other in [ + ErrorKind::AddrInUse, + ErrorKind::PermissionDenied, + ErrorKind::Other, + ] { + assert!(!family_missing(&Error::from(other)), "{other:?}"); + } + } + + #[test] + fn a_callback_counts_only_with_this_sign_ins_state() { + let ours = "s1"; + assert_eq!( + classify("/callback?code=c&state=s1", ours), + Callback::Code("c".into()) + ); + assert_eq!( + classify( + "/callback?error=access_denied&error_description=no&state=s1", + ours + ), + Callback::Error { + error: "access_denied".into(), + description: Some("no".into()) + } + ); + for not_ours in [ + "/callback?code=c&state=s0", + "/callback?error=access_denied&state=s0", + "/callback?code=c", + "/callback", + // Without a state, an error can't be ours: any page could send it. + "/callback?error=access_denied", + "/callback?error=access_denied&error_description=run%20this", + ] { + assert_eq!(classify(not_ours, ours), Callback::NotOurs, "{not_ours}"); + } + for other in ["/favicon.ico", "/", "/callbacks?code=c&state=s1"] { + assert_eq!(classify(other, ours), Callback::Other, "{other}"); + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn the_listener_takes_both_loopback_addresses_or_moves_on() { + if no_ipv6() { + return; + } + let v4 = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let v6 = TcpListener::bind((Ipv6Addr::LOCALHOST, 0)).unwrap(); + let taken_v4 = v4.local_addr().unwrap().port(); + let taken_v6 = v6.local_addr().unwrap().port(); + let mut bound = None; + for _ in 0..20 { + if let Ok(listening) = bind(&[taken_v4, taken_v6, free_port()]).await { + bound = Some(listening); + break; + } + } + let (listeners, port) = bound.expect("a free port"); + assert!( + port != taken_v4 && port != taken_v6, + "a port with either address taken is skipped" + ); + let addrs: Vec<_> = listeners + .iter() + .map(|l| l.local_addr().unwrap().ip().to_string()) + .collect(); + assert_eq!(addrs, ["127.0.0.1", "::1"]); + + let err = bind(&[taken_v4, taken_v6]).await.unwrap_err().to_string(); + assert!( + err.contains(&taken_v4.to_string()) && err.contains(&taken_v6.to_string()), + "{err}" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn stray_connections_and_old_tabs_do_not_stop_the_sign_in() { + // Well under READ_TIMEOUT: the idle connection must not hold up the rest. + if no_ipv6() { + return; + } + let (listeners, port) = listen().await; + let browser = std::thread::spawn(move || { + // A connection that never sends a request, left open throughout. + let _idle = TcpStream::connect(("127.0.0.1", port)).unwrap(); + for _ in 0..10 { + drop(TcpStream::connect(("::1", port)).unwrap()); + } + let favicon = request("::1", port, "/favicon.ico"); + // Cookies for localhost come along whatever the port: long headers. + let old_tab = request_with( + "127.0.0.1", + port, + "/callback?code=old&state=earlier", + &format!("Cookie: c={}\r\n", "x".repeat(40_000)), + ); + let this_tab = request("::1", port, "/callback?code=new&state=current"); + (favicon, old_tab, this_tab) + }); + let (code, mut socket) = + wait_for_callback(listeners, "current", ERROR_PAGE, Duration::from_secs(3)) + .await + .unwrap(); + assert_eq!(code, "new"); + tokio::io::AsyncWriteExt::shutdown(&mut socket) + .await + .unwrap(); + let (favicon, old_tab, this_tab) = browser.join().unwrap(); + assert!(favicon.starts_with("HTTP/1.1 404"), "{favicon}"); + assert!(old_tab.starts_with("HTTP/1.1 302"), "{old_tab}"); + assert!( + old_tab.contains( + "Location: https://example.test/error?error=invalid_request&error_description=" + ), + "{old_tab}" + ); + // The browser is answered once the login is done; here, nothing was written. + assert_eq!(this_tab, ""); + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_decline_ends_the_sign_in_and_silence_times_out() { + if no_ipv6() { + return; + } + let (listeners, port) = listen().await; + let browser = std::thread::spawn(move || { + request( + "localhost", + port, + "/callback?error=access_denied&error_description=declined&state=current", + ) + }); + let err = wait_for_callback(listeners, "current", ERROR_PAGE, Duration::from_secs(20)) + .await + .unwrap_err() + .to_string(); + assert!( + err.contains("returned error: access_denied (declined)"), + "{err}" + ); + let page = browser.join().unwrap(); + assert!(page.contains("Location: https://example.test/error?error=access_denied&error_description=declined"), "{page}"); + + let (listeners, _) = listen().await; + let err = wait_for_callback(listeners, "current", ERROR_PAGE, Duration::from_millis(200)) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("Timed out waiting"), "{err}"); + } +} diff --git a/cli/elevenlabs/workflow/mod.rs b/cli/elevenlabs/workflow/mod.rs index 8eea3fd..64b340d 100644 --- a/cli/elevenlabs/workflow/mod.rs +++ b/cli/elevenlabs/workflow/mod.rs @@ -19,6 +19,7 @@ mod components; mod feedback; mod flows_wait; mod intent; +mod loopback; mod onboard; mod project; mod residency; diff --git a/cli/elevenlabs/workflow/onboard/auth.rs b/cli/elevenlabs/workflow/onboard/auth.rs index 632bf19..aefa960 100644 --- a/cli/elevenlabs/workflow/onboard/auth.rs +++ b/cli/elevenlabs/workflow/onboard/auth.rs @@ -3,14 +3,17 @@ //! that a key on the clipboard has to pass. use std::sync::OnceLock; +use std::time::Duration; -use fern_cli_sdk::auth::{active_store, LoginContext, LoginFlow, PkceLoginFlow, TokenBundle}; +use fern_cli_sdk::auth::{active_store, TokenBundle}; use fern_cli_sdk::error::CliError; use fern_cli_sdk::openapi::AppContext; use serde_json::Value; +use super::super::loopback; use super::super::say::run_async as block_on; use super::super::settings; +use super::session::SIGN_IN_MINUTES; pub(super) const CLI_NAME: &str = "elevenlabs"; @@ -96,21 +99,21 @@ pub(super) fn sign_in( .and_then(loopback) .unwrap_or_else(|| config.token_url.trim_end_matches("/v1/oauth/token").into()); let token_url = format!("{oauth_base}/v1/oauth/token"); - PkceLoginFlow::new(&config.scheme) - .client_id(&config.client_id) - .authorization_url(authorize_url) - .token_url(&token_url) - .scopes(&config.scopes) - .redirect_host(&config.redirect_host) - .redirect_ports(config.redirect_ports.iter().copied()) - .success_redirect_url(&config.success_url) - .error_redirect_url(&config.error_url) - .authorization_params(vec![("key_handoff", "clipboard".to_string())]) - .run(&LoginContext { - cli_name: CLI_NAME.to_string(), - no_browser, - })?; - Ok(()) + block_on(loopback::sign_in(loopback::Login { + cli_name: CLI_NAME, + scheme: &config.scheme, + client_id: &config.client_id, + authorize_url, + token_url: &token_url, + scopes: &config.scopes, + redirect_host: &config.redirect_host, + redirect_ports: &config.redirect_ports, + success_url: &config.success_url, + error_url: &config.error_url, + params: &[("key_handoff", "clipboard")], + no_browser, + timeout: Duration::from_secs(SIGN_IN_MINUTES * 60), + })) } pub(super) fn stored_access_token() -> Option { diff --git a/cli/elevenlabs/workflow/onboard/session.rs b/cli/elevenlabs/workflow/onboard/session.rs index 30dc58a..7513576 100644 --- a/cli/elevenlabs/workflow/onboard/session.rs +++ b/cli/elevenlabs/workflow/onboard/session.rs @@ -15,8 +15,8 @@ //! command's process tree; it also runs in its own process group. //! //! One session per project folder, under `~/.elevenlabs/onboard/`, and one -//! sign-in waiting for the browser on the machine at a time: two would share -//! the callback port. +//! sign-in waiting for the browser on the machine at a time, so the tab the +//! developer approves is the one for the project they're in. use std::ffi::OsString; use std::fs::{File, OpenOptions}; @@ -32,7 +32,7 @@ use super::clipboard::Clipboard; use super::env_file; use super::reply::{Next, Reply}; -/// How long the sign-in waits for the approval page: the framework's callback timeout. +/// How long the sign-in waits for the approval page, as `auth login` does. pub(super) const SIGN_IN_MINUTES: u64 = 5; /// How long a new worker may take to take its lock before starting counts as failed. const STARTUP: Duration = Duration::from_secs(10); @@ -152,9 +152,9 @@ impl Session { } /// Stop any other project's sign-in still waiting for the browser. Only - /// one may listen for the browser's callback: two can hold the same port - /// (one on IPv4, one on IPv6), and the callback then reaches the wrong one. - /// One past the browser no longer listens, and is left to store its key. + /// one waits at a time, so the tab the developer approves is the one for + /// the project they're in. One past the browser no longer listens, and is + /// left to store its key. /// The stopped one's agent is told why, so it asks before signing in again. pub(super) fn stop_others(&self) { let Some(root) = self.dir.parent() else { @@ -197,10 +197,7 @@ impl Session { .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(log); - detach(&mut command); - command - .spawn() - .map(drop) + spawn_detached(&mut command) .map_err(|e| CliError::Other(anyhow::anyhow!("start the sign-in worker: {e}"))) } @@ -324,17 +321,31 @@ fn keep_stdio_from_children() { fn keep_stdio_from_children() {} #[cfg(unix)] -fn detach(command: &mut Command) { +fn spawn_detached(command: &mut Command) -> std::io::Result<()> { use std::os::unix::process::CommandExt; - command.process_group(0); + command.process_group(0).spawn().map(drop) } +/// Some agents run each command in a Job Object and end the job, and every +/// process in it, when the command is stopped or times out. The worker +/// leaves the job when the job allows it; otherwise it starts in the job. #[cfg(windows)] -fn detach(command: &mut Command) { +fn spawn_detached(command: &mut Command) -> std::io::Result<()> { use std::os::windows::process::CommandExt; const DETACHED_PROCESS: u32 = 0x0000_0008; const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; - command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP); + const CREATE_BREAKAWAY_FROM_JOB: u32 = 0x0100_0000; + const ERROR_ACCESS_DENIED: i32 = 5; + let detached = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP; + match command + .creation_flags(detached | CREATE_BREAKAWAY_FROM_JOB) + .spawn() + { + Err(e) if e.raw_os_error() == Some(ERROR_ACCESS_DENIED) => { + command.creation_flags(detached).spawn().map(drop) + } + spawned => spawned.map(drop), + } } /// End a worker and wait briefly for it to let go of its port. Only ever one