From 52aeb301d82a6192c60a64d7e25e93f3fea707b2 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Fri, 18 Sep 2026 18:40:19 +0800 Subject: [PATCH 1/7] fix: honor environment HTTP proxies for CDP WebSockets --- Cargo.lock | 2 + Cargo.toml | 2 + README.md | 17 +++ src/cdp.rs | 4 +- src/cdp/proxy.rs | 315 +++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 339 insertions(+), 1 deletion(-) create mode 100644 src/cdp/proxy.rs diff --git a/Cargo.lock b/Cargo.lock index eb854ff..e0b2c25 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1213,7 +1213,9 @@ dependencies = [ "base64 0.22.1", "clap", "dirs", + "httparse", "httpmock", + "hyper-util", "open", "rand 0.9.5", "reqwest", diff --git a/Cargo.toml b/Cargo.toml index 7e35b51..71ad759 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,8 @@ path = "src/main.rs" base64 = "0.22" clap = { version = "4.5", features = ["derive"] } dirs = "6" +httparse = "1" +hyper-util = { version = "0.1.20", features = ["client-proxy"] } open = "5" rand = "0.9" reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls-native-roots"] } diff --git a/README.md b/README.md index a53fd80..f68cb93 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,23 @@ are never printed. All commands emit one JSON document. Run `browser-cli --help` for the complete surface. +## Cloud runtime proxies + +Source builds route CDP WebSocket connections through the environment's HTTP +CONNECT proxy. `wss://` uses `HTTPS_PROXY` and `ws://` uses `HTTP_PROXY`, with +`ALL_PROXY` as the fallback; lowercase variables and `NO_PROXY` are handled by +the same proxy matcher used by the HTTP client. Target DNS is resolved by the +proxy. Proxy Basic authentication stays on CONNECT and is not forwarded to CDP. +TLS certificate and hostname checks remain enabled. A rejected proxy request +never falls back to a direct connection. + +This transport currently accepts `http://` proxies only; HTTPS-to-proxy and +SOCKS proxies return an explicit unsupported configuration error. The CONNECT +stage has a 15-second deadline and bounded headers; TLS/WebSocket handshake I/O +has a 15-second timeout. Existing direct connections are unchanged when no proxy +matches. These changes require a new CLI release; published 1.1.15 and 1.2.0 +binaries do not acquire them by updating Skill instructions. + ## Select a page in a multi-tab session Explicit page selection is introduced in version 1.2.0. Check that the installed diff --git a/src/cdp.rs b/src/cdp.rs index 6ff97e0..fa2873e 100644 --- a/src/cdp.rs +++ b/src/cdp.rs @@ -12,6 +12,8 @@ use tungstenite::{Message, WebSocket, stream::MaybeTlsStream}; use crate::{Error, Result}; +mod proxy; + pub struct Cdp { socket: WebSocket>, next_id: u64, @@ -51,7 +53,7 @@ impl Cdp { } fn connect_with_target(url: &str, requested_target: Option<&str>) -> Result { - let (socket, _) = tungstenite::connect(url)?; + let socket = proxy::connect(url)?; let mut client = Self { socket, next_id: 1, diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs new file mode 100644 index 0000000..ff6dc3a --- /dev/null +++ b/src/cdp/proxy.rs @@ -0,0 +1,315 @@ +//! Use the same environment proxy matcher as reqwest for CDP connections. +use std::{ + io::{Read, Write}, + net::{TcpStream, ToSocketAddrs}, + time::{Duration, Instant}, +}; + +use hyper_util::client::proxy::matcher::{Intercept, Matcher}; +use tungstenite::{ + WebSocket, + handshake::HandshakeError, + http::{Uri, uri::Scheme}, + stream::MaybeTlsStream, +}; + +use crate::{Error, Result}; + +const TIMEOUT: Duration = Duration::from_secs(15); + +pub(super) fn connect(url: &str) -> Result>> { + connect_with_matcher(url, &Matcher::from_env()) +} + +fn destination(url: &str) -> Result { + let uri: Uri = url + .parse() + .map_err(|_| Error::Config("invalid CDP URL".into()))?; + let scheme = match uri.scheme_str() { + Some("ws") => Scheme::HTTP, + Some("wss") => Scheme::HTTPS, + _ => return Err(Error::Config("CDP URL must use ws or wss".into())), + }; + if uri.host().is_none() { + return Err(Error::Config("CDP URL has no host".into())); + } + let mut parts = uri.into_parts(); + parts.scheme = Some(scheme); + Uri::from_parts(parts).map_err(|_| Error::Config("invalid CDP URL".into())) +} + +fn connect_with_matcher( + url: &str, + matcher: &Matcher, +) -> Result>> { + let destination = destination(url)?; + let Some(proxy) = matcher.intercept(&destination) else { + return Ok(tungstenite::connect(url)?.0); + }; + // ponytail: implement the HTTP CONNECT proxy used by cloud runtimes. Other + // proxy schemes fail explicitly; never silently retry a proxy failure direct. + if proxy.uri().scheme_str() != Some("http") { + return Err(Error::Config( + "CDP supports only http:// CONNECT proxies".into(), + )); + } + let host = proxy + .uri() + .host() + .ok_or_else(|| Error::Config("proxy has no host".into()))?; + let port = proxy.uri().port_u16().unwrap_or(80); + let deadline = Instant::now() + TIMEOUT; + let mut last_error = std::io::Error::other("proxy has no addresses"); + let mut connected = None; + for address in (host.trim_matches(['[', ']']), port).to_socket_addrs()? { + match TcpStream::connect_timeout(&address, remaining(deadline)?) { + Ok(stream) => { + connected = Some(stream); + break; + } + Err(error) => last_error = error, + } + } + let mut stream = connected.ok_or(last_error)?; + stream.set_write_timeout(Some(remaining(deadline)?))?; + establish_tunnel(&mut stream, &destination, &proxy, deadline)?; + stream.set_read_timeout(Some(TIMEOUT))?; + stream.set_write_timeout(Some(TIMEOUT))?; + // Keep a handle to restore normal CDP I/O after the TLS/WebSocket handshake. + let timeout_handle = stream.try_clone()?; + let (socket, _) = tungstenite::client_tls(url, stream).map_err(|error| match error { + HandshakeError::Failure(error) => Error::from(error), + HandshakeError::Interrupted(_) => { + Error::Cdp("proxy WebSocket handshake interrupted".into()) + } + })?; + timeout_handle.set_read_timeout(None)?; + timeout_handle.set_write_timeout(None)?; + Ok(socket) +} + +fn remaining(deadline: Instant) -> Result { + deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .ok_or_else(|| Error::Timeout("CDP proxy connection".into())) +} + +fn establish_tunnel( + stream: &mut TcpStream, + destination: &Uri, + proxy: &Intercept, + deadline: Instant, +) -> Result<()> { + let host = destination + .host() + .ok_or_else(|| Error::Config("CDP URL has no host".into()))?; + let port = destination + .port_u16() + .unwrap_or(if destination.scheme_str() == Some("https") { + 443 + } else { + 80 + }); + let authority = format!("{host}:{port}"); + write!( + stream, + "CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\n" + )?; + if let Some(auth) = proxy.basic_auth() { + stream.write_all(b"Proxy-Authorization: ")?; + stream.write_all(auth.as_bytes())?; + stream.write_all(b"\r\n")?; + } + stream.write_all(b"\r\n")?; + stream.flush()?; + // Read exactly the CONNECT headers so no TLS/WebSocket bytes are lost. + let mut bytes = Vec::new(); + while !bytes.ends_with(b"\r\n\r\n") { + if bytes.len() >= 16 * 1024 { + return Err(Error::Cdp( + "proxy CONNECT response headers too large".into(), + )); + } + stream.set_read_timeout(Some(remaining(deadline)?))?; + let mut byte = [0]; + stream.read_exact(&mut byte)?; + bytes.push(byte[0]); + } + let mut headers = [httparse::EMPTY_HEADER; 128]; + let mut response = httparse::Response::new(&mut headers); + if !matches!(response.parse(&bytes), Ok(httparse::Status::Complete(_))) { + return Err(Error::Cdp("invalid proxy CONNECT response".into())); + } + match response.code { + Some(200..=299) => Ok(()), + Some(code) => Err(Error::Cdp(format!( + "proxy CONNECT rejected with HTTP {code}" + ))), + None => Err(Error::Cdp("proxy CONNECT response has no status".into())), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{net::TcpListener, thread}; + use tungstenite::{ + Message, + handshake::server::{Request, Response}, + }; + + fn listener() -> TcpListener { + TcpListener::bind("127.0.0.1:0").unwrap() + } + + fn read_headers(stream: &mut TcpStream) -> String { + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut bytes = Vec::new(); + while !bytes.ends_with(b"\r\n\r\n") { + let mut byte = [0]; + stream.read_exact(&mut byte).unwrap(); + bytes.push(byte[0]); + } + String::from_utf8(bytes).unwrap() + } + + #[test] + #[allow( + clippy::result_large_err, + reason = "tungstenite fixes the callback's response error type" + )] + fn connect_uses_proxy_dns_and_keeps_proxy_credentials_out_of_websocket() { + let proxy = listener(); + let matcher = Matcher::builder() + .http(format!( + "http://test-user:test-password@{}", + proxy.local_addr().unwrap() + )) + .build(); + let server = thread::spawn(move || { + let (mut stream, _) = proxy.accept().unwrap(); + let headers = read_headers(&mut stream); + assert!(headers.starts_with("CONNECT browser.invalid:80 HTTP/1.1\r\n")); + assert!( + headers.contains("Proxy-Authorization: Basic dGVzdC11c2VyOnRlc3QtcGFzc3dvcmQ=\r\n") + ); + assert!(!headers.contains("private-token")); + stream + .write_all(b"HTTP/1.1 200 Connection established\r\n\r\n") + .unwrap(); + let mut ws = + tungstenite::accept_hdr(stream, |request: &Request, response: Response| { + assert_eq!(request.uri(), "/cdp?token=private-token"); + assert!(!request.headers().contains_key("Proxy-Authorization")); + Ok(response) + }) + .unwrap(); + assert_eq!(ws.read().unwrap().into_text().unwrap(), "probe"); + ws.send(Message::text("ok")).unwrap(); + }); + let mut ws = + connect_with_matcher("ws://browser.invalid/cdp?token=private-token", &matcher).unwrap(); + ws.send(Message::text("probe")).unwrap(); + assert_eq!(ws.read().unwrap().into_text().unwrap(), "ok"); + server.join().unwrap(); + } + + #[test] + fn secure_websockets_use_https_proxy_and_no_proxy_is_respected() { + let matcher = Matcher::builder() + .http("http://plain-proxy:8080") + .https("http://secure-proxy:8081") + .no("localhost,.internal.example,127.0.0.0/8") + .build(); + let proxy = matcher + .intercept(&destination("wss://browser.example/cdp").unwrap()) + .unwrap(); + assert_eq!(proxy.uri().host(), Some("secure-proxy")); + for url in [ + "ws://localhost/cdp", + "wss://api.internal.example/cdp", + "ws://127.0.0.1/cdp", + ] { + assert!(matcher.intercept(&destination(url).unwrap()).is_none()); + } + assert!( + matcher + .intercept(&destination("wss://notinternal.example/cdp").unwrap()) + .is_some() + ); + } + + #[test] + fn proxy_rejection_does_not_retry_direct_or_disclose_response() { + let proxy = listener(); + let origin = listener(); + origin.set_nonblocking(true).unwrap(); + let matcher = Matcher::builder() + .all(format!("http://{}", proxy.local_addr().unwrap())) + .build(); + let server = thread::spawn(move || { + let (mut stream, _) = proxy.accept().unwrap(); + read_headers(&mut stream); + stream + .write_all(b"HTTP/1.1 407 private-secret\r\nX-Secret: hidden\r\n\r\n") + .unwrap(); + }); + let error = connect_with_matcher( + &format!("ws://{}/?token=private-token", origin.local_addr().unwrap()), + &matcher, + ) + .unwrap_err(); + assert_eq!( + error.to_string(), + "CDP command failed: proxy CONNECT rejected with HTTP 407" + ); + assert_eq!( + origin.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + server.join().unwrap(); + } + + #[test] + fn unsupported_proxy_is_an_error_instead_of_direct_fallback() { + for scheme in ["socks5", "https"] { + let matcher = Matcher::builder() + .all(format!("{scheme}://user:private-secret@proxy.invalid:8080")) + .build(); + let error = + connect_with_matcher("wss://browser.invalid/?token=private-token", &matcher) + .unwrap_err(); + assert_eq!( + error.to_string(), + "configuration error: CDP supports only http:// CONNECT proxies" + ); + } + } + + #[test] + fn tunnel_headers_are_bounded_and_malformed_responses_are_not_echoed() { + for response in [ + b"invalid private-secret\r\n\r\n".to_vec(), + vec![b'x'; 16 * 1024], + ] { + let proxy = listener(); + let matcher = Matcher::builder() + .all(format!("http://{}", proxy.local_addr().unwrap())) + .build(); + let server = thread::spawn(move || { + let (mut stream, _) = proxy.accept().unwrap(); + read_headers(&mut stream); + stream.write_all(&response).unwrap(); + }); + let error = connect_with_matcher("ws://browser.invalid/", &matcher) + .unwrap_err() + .to_string(); + assert!(error.contains("proxy CONNECT response")); + assert!(!error.contains("private-secret")); + server.join().unwrap(); + } + } +} From af092b0974b0695b3aa7965050b795650c0b1d3d Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Fri, 18 Sep 2026 18:46:07 +0800 Subject: [PATCH 2/7] chore: prepare proxy fix for CLI 1.2.1 release --- Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 6 +++--- skills/lexmount-browser/scripts/bootstrap.ps1 | 2 +- skills/lexmount-browser/scripts/bootstrap.sh | 2 +- 5 files changed, 7 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e0b2c25..c6e2e79 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1208,7 +1208,7 @@ checksum = "db13adb97ab515a3691f56e4dbab09283d0b86cb45abd991d8634a9d6f501760" [[package]] name = "lexmount-browser" -version = "1.2.0" +version = "1.2.1" dependencies = [ "base64 0.22.1", "clap", diff --git a/Cargo.toml b/Cargo.toml index 71ad759..2538fa8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "lexmount-browser" -version = "1.2.0" +version = "1.2.1" edition = "2024" license = "MIT" description = "Native Rust SDK and CLI for Lexmount cloud browsers" diff --git a/README.md b/README.md index f68cb93..b47fcda 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ surface. ## Cloud runtime proxies -Source builds route CDP WebSocket connections through the environment's HTTP +Version 1.2.1 routes CDP WebSocket connections through the environment's HTTP CONNECT proxy. `wss://` uses `HTTPS_PROXY` and `ws://` uses `HTTP_PROXY`, with `ALL_PROXY` as the fallback; lowercase variables and `NO_PROXY` are handled by the same proxy matcher used by the HTTP client. Target DNS is resolved by the @@ -44,8 +44,8 @@ binaries do not acquire them by updating Skill instructions. Explicit page selection is introduced in version 1.2.0. Check that the installed binary's `browser-cli action --help` lists `--target-id`; the published 1.1.15 binary does not have it. The package version and both bootstrap scripts target -1.2.0 together. Merging or building this source does not publish release assets: -bootstrap can install 1.2.0 only after its binaries and checksums are published +1.2.1 together. Merging or building this source does not publish release assets: +bootstrap can install 1.2.1 only after its binaries and checksums are published to COS. Until then, use a source build for local verification. Every `action` command accepts an optional `--target-id`. Obtain the page's CDP diff --git a/skills/lexmount-browser/scripts/bootstrap.ps1 b/skills/lexmount-browser/scripts/bootstrap.ps1 index 2badbd2..315e27b 100644 --- a/skills/lexmount-browser/scripts/bootstrap.ps1 +++ b/skills/lexmount-browser/scripts/bootstrap.ps1 @@ -12,7 +12,7 @@ function Invoke-Tls12Download { } } -$version = if ($env:LEXMOUNT_BROWSER_CLI_VERSION) { $env:LEXMOUNT_BROWSER_CLI_VERSION } else { "1.2.0" } +$version = if ($env:LEXMOUNT_BROWSER_CLI_VERSION) { $env:LEXMOUNT_BROWSER_CLI_VERSION } else { "1.2.1" } $downloadBaseUrl = if ($env:LEXMOUNT_BROWSER_CLI_DOWNLOAD_BASE_URL) { $env:LEXMOUNT_BROWSER_CLI_DOWNLOAD_BASE_URL.TrimEnd('/') } else { "https://cli-bin-1377899528.cos.ap-nanjing.myqcloud.com/releases/browser-cli" } $architecture = if ($env:PROCESSOR_ARCHITEW6432) { $env:PROCESSOR_ARCHITEW6432 } else { $env:PROCESSOR_ARCHITECTURE } if ($architecture -ne "AMD64") { throw "Only Windows x64 is supported" } diff --git a/skills/lexmount-browser/scripts/bootstrap.sh b/skills/lexmount-browser/scripts/bootstrap.sh index f016089..fa15f4b 100755 --- a/skills/lexmount-browser/scripts/bootstrap.sh +++ b/skills/lexmount-browser/scripts/bootstrap.sh @@ -1,7 +1,7 @@ #!/bin/sh set -eu -version="${LEXMOUNT_BROWSER_CLI_VERSION:-1.2.0}" +version="${LEXMOUNT_BROWSER_CLI_VERSION:-1.2.1}" download_base_url="${LEXMOUNT_BROWSER_CLI_DOWNLOAD_BASE_URL:-https://cli-bin-1377899528.cos.ap-nanjing.myqcloud.com/releases/browser-cli}" repo="${download_base_url%/}/v${version}" case "$(uname -s)-$(uname -m)" in From 89334babc7a81cbbe531f4e9d490b1fc2f023548 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Fri, 18 Sep 2026 19:08:50 +0800 Subject: [PATCH 3/7] fix: preserve redirects across CDP proxy routes --- src/cdp/proxy.rs | 167 ++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 166 insertions(+), 1 deletion(-) diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs index ff6dc3a..4e8cdba 100644 --- a/src/cdp/proxy.rs +++ b/src/cdp/proxy.rs @@ -42,9 +42,35 @@ fn connect_with_matcher( url: &str, matcher: &Matcher, ) -> Result>> { + let mut url = url.to_owned(); + // Match tungstenite::connect's three-hop limit, but choose the route anew + // for every destination, including redirects to/from NO_PROXY hosts. + for attempt in 0..=3 { + match connect_once(&url, matcher) { + Err(Error::WebSocket(error)) if attempt < 3 => { + if let tungstenite::Error::Http(response) = error.as_ref() + && response.status().is_redirection() + && let Some(location) = response.headers().get("Location") + { + url = location + .to_str() + .map_err(|_| Error::Config("invalid CDP redirect Location".into()))? + .to_owned(); + continue; + } + return Err(Error::WebSocket(error)); + } + result => return result, + } + } + unreachable!("last connection attempt always returns") +} + +fn connect_once(url: &str, matcher: &Matcher) -> Result>> { let destination = destination(url)?; let Some(proxy) = matcher.intercept(&destination) else { - return Ok(tungstenite::connect(url)?.0); + // The outer loop owns redirects so a direct hop cannot skip proxy rules. + return Ok(tungstenite::client::connect_with_config(url, None, 0)?.0); }; // ponytail: implement the HTTP CONNECT proxy used by cloud runtimes. Other // proxy schemes fail explicitly; never silently retry a proxy failure direct. @@ -176,6 +202,145 @@ mod tests { String::from_utf8(bytes).unwrap() } + fn accept_with_timeout(listener: &TcpListener) -> TcpStream { + listener.set_nonblocking(true).unwrap(); + let deadline = Instant::now() + Duration::from_secs(3); + loop { + match listener.accept() { + Ok((stream, _)) => return stream, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "expected another connection"); + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("accept failed: {error}"), + } + } + } + + #[test] + fn redirects_rematch_proxy_and_no_proxy_for_every_hop() { + let proxy = listener(); + let direct = listener(); + let direct_url = format!("ws://{}/middle", direct.local_addr().unwrap()); + let matcher = Matcher::builder() + .http(format!("http://{}", proxy.local_addr().unwrap())) + .no("127.0.0.1") + .build(); + let proxy_server = thread::spawn(move || { + for host in ["entry.invalid", "next.invalid", "final.invalid"] { + let mut stream = accept_with_timeout(&proxy); + assert!( + read_headers(&mut stream) + .starts_with(&format!("CONNECT {host}:80 HTTP/1.1\r\n")) + ); + stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").unwrap(); + if host != "final.invalid" { + read_headers(&mut stream); + let location = if host == "entry.invalid" { + direct_url.as_str() + } else { + "ws://final.invalid/cdp" + }; + write!( + stream, + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\n\r\n" + ) + .unwrap(); + } else { + let mut socket = tungstenite::accept(stream).unwrap(); + socket.send(Message::text("redirected")).unwrap(); + } + } + }); + let direct_server = thread::spawn(move || { + let mut stream = accept_with_timeout(&direct); + let request = read_headers(&mut stream); + assert!(request.starts_with("GET /middle HTTP/1.1\r\n")); + assert!(!request.to_ascii_lowercase().contains("proxy-authorization")); + stream.write_all(b"HTTP/1.1 307 Temporary Redirect\r\nLocation: ws://next.invalid/cdp\r\nContent-Length: 0\r\n\r\n").unwrap(); + }); + let mut socket = connect_with_matcher("ws://entry.invalid/cdp", &matcher).unwrap(); + assert_eq!(socket.read().unwrap().into_text().unwrap(), "redirected"); + proxy_server.join().unwrap(); + direct_server.join().unwrap(); + } + + #[test] + fn redirects_stop_after_three_hops_for_proxy_and_direct_connections() { + for proxied in [true, false] { + let endpoint = listener(); + let address = endpoint.local_addr().unwrap(); + let matcher = if proxied { + Matcher::builder().all(format!("http://{address}")).build() + } else { + Matcher::builder().build() + }; + let url = if proxied { + "ws://loop.invalid/cdp".to_owned() + } else { + format!("ws://{address}/cdp") + }; + let location = url.clone(); + let server = thread::spawn(move || { + for _ in 0..4 { + let mut stream = accept_with_timeout(&endpoint); + if proxied { + assert!(read_headers(&mut stream).starts_with("CONNECT ")); + stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").unwrap(); + } + assert!(read_headers(&mut stream).starts_with("GET ")); + write!( + stream, + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\n\r\n" + ) + .unwrap(); + } + endpoint + }); + let error = connect_with_matcher(&url, &matcher).unwrap_err(); + assert!(matches!(error, Error::WebSocket(ref error) + if matches!(error.as_ref(), tungstenite::Error::Http(response) + if response.status().as_u16() == 302))); + let endpoint = server.join().unwrap(); + assert_eq!( + endpoint.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + } + } + + #[test] + fn missing_or_invalid_redirect_locations_fail_without_another_connection() { + for location in [ + "", + "Location: /relative\r\n", + "Location: https://browser.invalid/\r\n", + ] { + let proxy = listener(); + let matcher = Matcher::builder() + .all(format!("http://{}", proxy.local_addr().unwrap())) + .build(); + let server = thread::spawn(move || { + let mut stream = accept_with_timeout(&proxy); + read_headers(&mut stream); + stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").unwrap(); + read_headers(&mut stream); + write!( + stream, + "HTTP/1.1 302 Found\r\n{location}Content-Length: 0\r\n\r\n" + ) + .unwrap(); + proxy + }); + assert!(connect_with_matcher("ws://browser.invalid/", &matcher).is_err()); + let proxy = server.join().unwrap(); + assert_eq!( + proxy.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + } + } + #[test] #[allow( clippy::result_large_err, From 92e7c4bccbfc1289fe8e2c639800bf5d67d85123 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Fri, 18 Sep 2026 19:12:52 +0800 Subject: [PATCH 4/7] test: use blocking accepted sockets on Windows --- src/cdp/proxy.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs index 4e8cdba..fd0e403 100644 --- a/src/cdp/proxy.rs +++ b/src/cdp/proxy.rs @@ -207,7 +207,11 @@ mod tests { let deadline = Instant::now() + Duration::from_secs(3); loop { match listener.accept() { - Ok((stream, _)) => return stream, + Ok((stream, _)) => { + // Windows accepted sockets inherit the listener's mode. + stream.set_nonblocking(false).unwrap(); + return stream; + } Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { assert!(Instant::now() < deadline, "expected another connection"); thread::sleep(Duration::from_millis(10)); From 74daec62a5138fed5ef257f2ae5862351edf8a40 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Sun, 20 Sep 2026 13:49:31 +0800 Subject: [PATCH 5/7] fix: bound CDP connection setup and isolate proxy tests --- .github/workflows/ci.yml | 6 + README.md | 28 ++- src/cdp.rs | 5 +- src/cdp/proxy.rs | 374 ++++++++++++++++++++++++++++++---- src/cdp/proxy/resolver.rs | 209 +++++++++++++++++++ src/client.rs | 6 +- tests/page_targets.rs | 64 ++++++ tests/page_targets_browser.rs | 6 + tests/support/mod.rs | 54 ++++- 9 files changed, 691 insertions(+), 61 deletions(-) create mode 100644 src/cdp/proxy/resolver.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5a05f4c..e2bcc2c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,6 +15,12 @@ jobs: components: rustfmt, clippy - run: cargo fmt --all -- --check - run: cargo test --all-targets --locked + - name: Verify tests ignore inherited proxy configuration + run: >- + env -u NO_PROXY -u no_proxy + HTTP_PROXY=http://127.0.0.1:1 HTTPS_PROXY=http://127.0.0.1:1 ALL_PROXY=http://127.0.0.1:1 + http_proxy=http://127.0.0.1:1 https_proxy=http://127.0.0.1:1 all_proxy=http://127.0.0.1:1 + cargo test --all-targets --locked --offline - run: cargo clippy --all-targets --locked -- -D warnings - name: Verify release target contract run: | diff --git a/README.md b/README.md index b47fcda..95720f6 100644 --- a/README.md +++ b/README.md @@ -33,11 +33,29 @@ TLS certificate and hostname checks remain enabled. A rejected proxy request never falls back to a direct connection. This transport currently accepts `http://` proxies only; HTTPS-to-proxy and -SOCKS proxies return an explicit unsupported configuration error. The CONNECT -stage has a 15-second deadline and bounded headers; TLS/WebSocket handshake I/O -has a 15-second timeout. Existing direct connections are unchanged when no proxy -matches. These changes require a new CLI release; published 1.1.15 and 1.2.0 -binaries do not acquire them by updating Skill instructions. +SOCKS proxies return an explicit unsupported configuration error. Direct and +proxied connections share one 15-second network connection budget, covering +DNS, TCP, CONNECT and TLS/WebSocket handshake across all redirects. CONNECT +headers are limited to 16 KiB. Each blocking network operation uses the remaining +budget; a slow peer cannot restart it by sending another byte. + +OS DNS resolution preserves hosts/VPN configuration. Two process-wide workers +and four queue slots bound background work. The caller stops waiting at its +deadline; an in-flight OS lookup cannot be cancelled, and its late result cannot +open a connection. Expired queued lookups are skipped. If the pool is saturated, +new hostname lookups fail with `DNS resolver busy; retry later` until workers +recover. Numeric addresses bypass DNS. + +Connection timeouts exit the CLI with status 1 and a JSON error on stderr, e.g. +`{"ok":false,"error":"timeout","message":"request timed out: CDP connection (stage: proxy_dns, budget: 15s)"}`. +Stage names distinguish `proxy_dns`/`target_dns`, `proxy_tcp`/`target_tcp`, +`proxy_connect`, and `websocket_handshake`/`tls_websocket_handshake` without exposing +URLs or credentials. A connection timeout occurs before any CDP command is sent. +The budget ends at the WebSocket upgrade: REST session requests, CDP target +attachment, and later browser actions retain their existing timeout behavior. +It is not a deadline for an entire CLI command or Agent turn, nor does it trigger +automatic action retries. These changes require a new CLI release; published +1.1.15 and 1.2.0 binaries do not acquire them by updating Skill instructions. ## Select a page in a multi-tab session diff --git a/src/cdp.rs b/src/cdp.rs index fa2873e..deb4bad 100644 --- a/src/cdp.rs +++ b/src/cdp.rs @@ -1,21 +1,20 @@ use std::{ collections::VecDeque, fs, - net::TcpStream, path::Path, time::{Duration, Instant}, }; use base64::{Engine, engine::general_purpose::STANDARD}; use serde_json::{Value, json}; -use tungstenite::{Message, WebSocket, stream::MaybeTlsStream}; +use tungstenite::Message; use crate::{Error, Result}; mod proxy; pub struct Cdp { - socket: WebSocket>, + socket: proxy::Socket, next_id: u64, target_session_id: String, events: VecDeque, diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs index fd0e403..7e645fa 100644 --- a/src/cdp/proxy.rs +++ b/src/cdp/proxy.rs @@ -1,7 +1,7 @@ //! Use the same environment proxy matcher as reqwest for CDP connections. use std::{ - io::{Read, Write}, - net::{TcpStream, ToSocketAddrs}, + io::{self, Read, Write}, + net::TcpStream, time::{Duration, Instant}, }; @@ -15,9 +15,13 @@ use tungstenite::{ use crate::{Error, Result}; +mod resolver; + +pub(super) type Socket = WebSocket>; + const TIMEOUT: Duration = Duration::from_secs(15); -pub(super) fn connect(url: &str) -> Result>> { +pub(super) fn connect(url: &str) -> Result { connect_with_matcher(url, &Matcher::from_env()) } @@ -38,15 +42,16 @@ fn destination(url: &str) -> Result { Uri::from_parts(parts).map_err(|_| Error::Config("invalid CDP URL".into())) } -fn connect_with_matcher( - url: &str, - matcher: &Matcher, -) -> Result>> { +fn connect_with_matcher(url: &str, matcher: &Matcher) -> Result { + connect_with_deadline(url, matcher, Deadline::new(TIMEOUT)) +} + +fn connect_with_deadline(url: &str, matcher: &Matcher, deadline: Deadline) -> Result { let mut url = url.to_owned(); // Match tungstenite::connect's three-hop limit, but choose the route anew // for every destination, including redirects to/from NO_PROXY hosts. for attempt in 0..=3 { - match connect_once(&url, matcher) { + match connect_once(&url, matcher, deadline) { Err(Error::WebSocket(error)) if attempt < 3 => { if let tungstenite::Error::Http(response) = error.as_ref() && response.status().is_redirection() @@ -66,29 +71,44 @@ fn connect_with_matcher( unreachable!("last connection attempt always returns") } -fn connect_once(url: &str, matcher: &Matcher) -> Result>> { +fn connect_once(url: &str, matcher: &Matcher, deadline: Deadline) -> Result { let destination = destination(url)?; - let Some(proxy) = matcher.intercept(&destination) else { - // The outer loop owns redirects so a direct hop cannot skip proxy rules. - return Ok(tungstenite::client::connect_with_config(url, None, 0)?.0); - }; - // ponytail: implement the HTTP CONNECT proxy used by cloud runtimes. Other - // proxy schemes fail explicitly; never silently retry a proxy failure direct. - if proxy.uri().scheme_str() != Some("http") { + let proxy = matcher.intercept(&destination); + // ponytail: HTTP CONNECT only. Never silently retry a proxy failure direct. + if proxy + .as_ref() + .is_some_and(|p| p.uri().scheme_str() != Some("http")) + { return Err(Error::Config( "CDP supports only http:// CONNECT proxies".into(), )); } - let host = proxy - .uri() + let endpoint = proxy.as_ref().map_or(&destination, |p| p.uri()); + let host = endpoint .host() - .ok_or_else(|| Error::Config("proxy has no host".into()))?; - let port = proxy.uri().port_u16().unwrap_or(80); - let deadline = Instant::now() + TIMEOUT; - let mut last_error = std::io::Error::other("proxy has no addresses"); + .ok_or_else(|| Error::Config("missing host".into()))?; + let port = endpoint + .port_u16() + .unwrap_or(if endpoint.scheme_str() == Some("https") { + 443 + } else { + 80 + }); + let dns_stage = if proxy.is_some() { + "proxy_dns" + } else { + "target_dns" + }; + let addresses = resolver::resolve(host.trim_matches(['[', ']']), port, deadline, dns_stage)?; + let stage = if proxy.is_some() { + "proxy_tcp" + } else { + "target_tcp" + }; + let mut last_error = io::Error::other("endpoint has no addresses"); let mut connected = None; - for address in (host.trim_matches(['[', ']']), port).to_socket_addrs()? { - match TcpStream::connect_timeout(&address, remaining(deadline)?) { + for address in addresses { + match TcpStream::connect_timeout(&address, deadline.remaining(stage)?) { Ok(stream) => { connected = Some(stream); break; @@ -96,36 +116,161 @@ fn connect_once(url: &str, matcher: &Matcher) -> Result last_error = error, } } - let mut stream = connected.ok_or(last_error)?; - stream.set_write_timeout(Some(remaining(deadline)?))?; - establish_tunnel(&mut stream, &destination, &proxy, deadline)?; - stream.set_read_timeout(Some(TIMEOUT))?; - stream.set_write_timeout(Some(TIMEOUT))?; - // Keep a handle to restore normal CDP I/O after the TLS/WebSocket handshake. - let timeout_handle = stream.try_clone()?; - let (socket, _) = tungstenite::client_tls(url, stream).map_err(|error| match error { - HandshakeError::Failure(error) => Error::from(error), - HandshakeError::Interrupted(_) => { - Error::Cdp("proxy WebSocket handshake interrupted".into()) - } + deadline.remaining(stage)?; + let stream = connected.ok_or_else(|| deadline.normalize(Error::Io(last_error), stage))?; + stream.set_nodelay(true)?; + let mut stream = ConnectionStream { + stream, + deadline: Some(deadline.end), + }; + if let Some(proxy) = proxy { + establish_tunnel(&mut stream, &destination, &proxy) + .map_err(|error| deadline.normalize(error, "proxy_connect"))?; + } + let stage = if destination.scheme_str() == Some("https") { + "tls_websocket_handshake" + } else { + "websocket_handshake" + }; + deadline.remaining(stage)?; + let (mut socket, _) = tungstenite::client_tls(url, stream).map_err(|error| { + deadline.normalize( + match error { + HandshakeError::Failure(error) => Error::from(error), + HandshakeError::Interrupted(_) => { + Error::Cdp("WebSocket handshake interrupted".into()) + } + }, + stage, + ) })?; - timeout_handle.set_read_timeout(None)?; - timeout_handle.set_write_timeout(None)?; + deadline.remaining(stage)?; + let stream = match socket.get_mut() { + MaybeTlsStream::Plain(stream) => stream, + MaybeTlsStream::Rustls(stream) => &mut stream.sock, + _ => return Err(Error::Config("unsupported CDP TLS backend".into())), + }; + stream.stream.set_read_timeout(None)?; + stream.stream.set_write_timeout(None)?; + stream.deadline = None; Ok(socket) } -fn remaining(deadline: Instant) -> Result { +#[derive(Clone, Copy)] +struct Deadline { + end: Instant, + budget: Duration, +} + +impl Deadline { + fn new(budget: Duration) -> Self { + Self { + end: Instant::now() + budget, + budget, + } + } + + fn timeout(self, stage: &str) -> Error { + Error::Timeout(format!( + "CDP connection (stage: {stage}, budget: {:?})", + self.budget + )) + } + + fn remaining(self, stage: &str) -> Result { + remaining(self.end).map_err(|_| self.timeout(stage)) + } + + fn normalize(self, error: Error, stage: &str) -> Error { + let io = match &error { + Error::Io(error) => Some(error), + Error::WebSocket(error) => match error.as_ref() { + tungstenite::Error::Io(error) => Some(error), + _ => None, + }, + _ => None, + }; + if self.remaining(stage).is_err() + || io.is_some_and(|error| { + matches!( + error.kind(), + io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock + ) + }) + { + self.timeout(stage) + } else { + error + } + } +} + +fn remaining(deadline: Instant) -> io::Result { deadline .checked_duration_since(Instant::now()) .filter(|duration| !duration.is_zero()) - .ok_or_else(|| Error::Timeout("CDP proxy connection".into())) + .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "CDP connection deadline exceeded")) +} + +// Each underlying I/O uses the remaining budget, including rustls' handshake +// loops and peers that keep trickling bytes. Disabled after the WS upgrade. +#[derive(Debug)] +pub(super) struct ConnectionStream { + stream: TcpStream, + deadline: Option, +} + +impl Read for ConnectionStream { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + if let Some(deadline) = self.deadline { + self.stream.set_read_timeout(Some(remaining(deadline)?))?; + } + let result = self.stream.read(buffer); + self.finish_io(result) + } +} + +impl Write for ConnectionStream { + fn write(&mut self, buffer: &[u8]) -> io::Result { + if let Some(deadline) = self.deadline { + self.stream.set_write_timeout(Some(remaining(deadline)?))?; + } + let result = self.stream.write(buffer); + self.finish_io(result) + } + + fn flush(&mut self) -> io::Result<()> { + // TcpStream is unbuffered. + let result = self.stream.flush(); + self.finish_io(result) + } +} + +impl ConnectionStream { + fn finish_io(&self, result: io::Result) -> io::Result { + if let Some(deadline) = self.deadline { + remaining(deadline)?; + // Blocking sockets can report WouldBlock on Unix when SO_*TIMEO + // expires. Do not let tungstenite interpret it as resumable I/O. + if result + .as_ref() + .err() + .is_some_and(|e| e.kind() == io::ErrorKind::WouldBlock) + { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "CDP connection I/O timed out", + )); + } + } + result + } } fn establish_tunnel( - stream: &mut TcpStream, + stream: &mut ConnectionStream, destination: &Uri, proxy: &Intercept, - deadline: Instant, ) -> Result<()> { let host = destination .host() @@ -157,7 +302,6 @@ fn establish_tunnel( "proxy CONNECT response headers too large".into(), )); } - stream.set_read_timeout(Some(remaining(deadline)?))?; let mut byte = [0]; stream.read_exact(&mut byte)?; bytes.push(byte[0]); @@ -221,6 +365,150 @@ mod tests { } } + #[test] + fn stalled_tls_and_websocket_handshakes_time_out_on_both_routes() { + for proxied in [false, true] { + for secure in [false, true] { + let endpoint = listener(); + let address = endpoint.local_addr().unwrap(); + let matcher = if proxied { + Matcher::builder().all(format!("http://{address}")).build() + } else { + Matcher::builder().build() + }; + let scheme = if secure { "wss" } else { "ws" }; + let url = format!("{scheme}://{address}/?token=private-secret"); + let server = thread::spawn(move || { + let mut stream = accept_with_timeout(&endpoint); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + if proxied { + read_headers(&mut stream); + stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").unwrap(); + } + // Consume the client handshake but never reply. Client must + // close the socket on timeout rather than leaving work alive. + let mut bytes = Vec::new(); + stream.read_to_end(&mut bytes).unwrap(); + assert!(!bytes.is_empty()); + }); + let start = Instant::now(); + let error = connect_with_deadline( + &url, + &matcher, + Deadline::new(if secure { + Duration::from_secs(2) + } else { + Duration::from_millis(500) + }), + ) + .unwrap_err(); + assert!(matches!(error, Error::Timeout(ref text) if text.contains("handshake"))); + assert!(!error.to_string().contains("private-secret")); + assert!(start.elapsed() < Duration::from_secs(4)); + server.join().unwrap(); + } + } + } + + #[test] + fn trickling_connect_and_websocket_headers_do_not_reset_the_budget() { + for tunnel in [true, false] { + let endpoint = listener(); + let address = endpoint.local_addr().unwrap(); + let matcher = Matcher::builder().all(format!("http://{address}")).build(); + let server = thread::spawn(move || { + let mut stream = accept_with_timeout(&endpoint); + read_headers(&mut stream); + if !tunnel { + stream.write_all(b"HTTP/1.1 200 OK\r\n\r\n").unwrap(); + read_headers(&mut stream); + } + for byte in b"HTTP/1.1 200 OK\r\nX-Slow: padding padding padding\r\n\r\n" { + if stream.write_all(&[*byte]).is_err() { + return; + } + thread::sleep(Duration::from_millis(40)); + } + panic!("client allowed slow headers past deadline"); + }); + let start = Instant::now(); + let error = connect_with_deadline( + "ws://unresolved.invalid/", + &matcher, + Deadline::new(Duration::from_millis(200)), + ) + .unwrap_err(); + let expected = if tunnel { + "proxy_connect" + } else { + "websocket_handshake" + }; + assert!(matches!(error, Error::Timeout(ref message) if message.contains(expected))); + assert!(start.elapsed() < Duration::from_secs(1)); + server.join().unwrap(); + } + } + + #[test] + fn redirects_share_the_original_deadline_and_success_clears_it() { + let endpoint = listener(); + let address = endpoint.local_addr().unwrap(); + let url = format!("ws://{address}/"); + let location = url.clone(); + let server = thread::spawn(move || { + let mut stream = accept_with_timeout(&endpoint); + read_headers(&mut stream); + thread::sleep(Duration::from_millis(350)); + write!( + stream, + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\n\r\n" + ) + .unwrap(); + let mut stream = accept_with_timeout(&endpoint); + read_headers(&mut stream); + let mut rest = Vec::new(); + stream.read_to_end(&mut rest).unwrap(); + assert!(rest.is_empty()); + }); + let start = Instant::now(); + assert!(matches!( + connect_with_deadline( + &url, + &Matcher::builder().build(), + Deadline::new(Duration::from_millis(500)) + ), + Err(Error::Timeout(_)) + )); + assert!( + start.elapsed() < Duration::from_millis(800), + "redirect reset the budget" + ); + server.join().unwrap(); + + let endpoint = listener(); + let url = format!("ws://{}/", endpoint.local_addr().unwrap()); + let server = thread::spawn(move || { + let mut socket = tungstenite::accept(accept_with_timeout(&endpoint)).unwrap(); + thread::sleep(Duration::from_millis(600)); + socket + .send(Message::text("after connection deadline")) + .unwrap(); + }); + let mut socket = connect_with_deadline( + &url, + &Matcher::builder().build(), + Deadline::new(Duration::from_millis(500)), + ) + .unwrap(); + assert_eq!( + socket.read().unwrap().into_text().unwrap(), + "after connection deadline" + ); + server.join().unwrap(); + } + #[test] fn redirects_rematch_proxy_and_no_proxy_for_every_hop() { let proxy = listener(); diff --git a/src/cdp/proxy/resolver.rs b/src/cdp/proxy/resolver.rs new file mode 100644 index 0000000..6c96961 --- /dev/null +++ b/src/cdp/proxy/resolver.rs @@ -0,0 +1,209 @@ +use std::{ + io, + net::{IpAddr, SocketAddr, ToSocketAddrs}, + sync::{Arc, Mutex, OnceLock, mpsc}, + thread, +}; + +use super::Deadline; +use crate::{Error, Result}; + +type Addresses = io::Result>; + +struct Lookup { + host: String, + port: u16, + deadline: Deadline, + reply: mpsc::SyncSender, +} + +struct Resolver(mpsc::SyncSender); + +impl Resolver { + fn new(resolve: impl Fn(&str, u16) -> Addresses + Send + Sync + 'static) -> io::Result { + // ponytail: keep OS DNS (hosts/VPN support) with two workers and four + // queued lookups. Stuck OS calls cannot be cancelled; capacity stays + // bounded and saturation fails explicitly instead of spawning threads. + let (sender, receiver) = mpsc::sync_channel::(4); + let receiver = Arc::new(Mutex::new(receiver)); + let resolve = Arc::new(resolve); + for _ in 0..2 { + let receiver = receiver.clone(); + let resolve = resolve.clone(); + thread::Builder::new() + .name("cdp-dns".into()) + .spawn(move || { + loop { + let Ok(job) = receiver.lock().unwrap().recv() else { + break; + }; + if job.deadline.remaining("dns").is_ok() { + let result = resolve(&job.host, job.port); + // Only DNS happens here. A late result cannot open a socket. + let _ = job.reply.try_send(result); + } + } + })?; + } + Ok(Self(sender)) + } + + fn lookup( + &self, + host: &str, + port: u16, + deadline: Deadline, + stage: &str, + ) -> Result> { + deadline.remaining(stage)?; + let (reply, receiver) = mpsc::sync_channel(1); + self.0 + .try_send(Lookup { + host: host.into(), + port, + deadline, + reply, + }) + .map_err(|error| match error { + mpsc::TrySendError::Full(_) => Error::Cdp("DNS resolver busy; retry later".into()), + mpsc::TrySendError::Disconnected(_) => { + Error::Cdp("DNS resolver unavailable".into()) + } + })?; + let result = + receiver + .recv_timeout(deadline.remaining(stage)?) + .map_err(|error| match error { + mpsc::RecvTimeoutError::Timeout => deadline.timeout(stage), + mpsc::RecvTimeoutError::Disconnected if deadline.remaining(stage).is_err() => { + deadline.timeout(stage) + } + mpsc::RecvTimeoutError::Disconnected => { + Error::Cdp("DNS resolver unavailable".into()) + } + })?; + deadline.remaining(stage)?; + result.map_err(|error| deadline.normalize(Error::Io(error), stage)) + } +} + +pub(super) fn resolve( + host: &str, + port: u16, + deadline: Deadline, + stage: &str, +) -> Result> { + deadline.remaining(stage)?; + if let Ok(ip) = host.parse::() { + return Ok(vec![SocketAddr::new(ip, port)]); + } + static RESOLVER: OnceLock> = OnceLock::new(); + let resolver = RESOLVER + .get_or_init(|| { + Resolver::new(|host, port| { + (host, port) + .to_socket_addrs() + .map(|addresses| addresses.collect()) + }) + }) + .as_ref() + .map_err(|_| Error::Cdp("cannot start DNS resolver".into()))?; + resolver.lookup(host, port, deadline, stage) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{ + sync::{ + Condvar, + atomic::{AtomicUsize, Ordering}, + }, + time::{Duration, Instant}, + }; + + #[test] + fn slow_dns_is_bounded_expired_work_is_skipped_and_pool_recovers() { + let gate = Arc::new((Mutex::new(false), Condvar::new())); + let calls = Arc::new(AtomicUsize::new(0)); + let (started, running) = mpsc::channel(); + let resolver = Arc::new( + Resolver::new({ + let gate = gate.clone(); + let calls = calls.clone(); + move |_, port| { + calls.fetch_add(1, Ordering::SeqCst); + started.send(()).unwrap(); + let (lock, ready) = &*gate; + let _guard = ready + .wait_while(lock.lock().unwrap(), |open| !*open) + .unwrap(); + Ok(vec![SocketAddr::from(([127, 0, 0, 1], port))]) + } + }) + .unwrap(), + ); + let started_at = Instant::now(); + let mut callers = Vec::new(); + for _ in 0..2 { + let resolver = resolver.clone(); + callers.push(thread::spawn(move || { + resolver.lookup( + "slow.invalid", + 80, + Deadline::new(Duration::from_millis(100)), + "proxy_dns", + ) + })); + running.recv_timeout(Duration::from_secs(2)).unwrap(); + } + for caller in callers { + assert!( + matches!(caller.join().unwrap(), Err(Error::Timeout(message)) if message.contains("proxy_dns")) + ); + } + assert!(started_at.elapsed() < Duration::from_secs(2)); + // Both OS calls remain blocked. Four queued jobs fit, all later calls + // fail immediately; no additional worker is started. + let expired = Deadline::new(Duration::ZERO); + for _ in 0..4 { + let (reply, _) = mpsc::sync_channel(1); + resolver + .0 + .try_send(Lookup { + host: "expired.invalid".into(), + port: 80, + deadline: expired, + reply, + }) + .unwrap(); + } + for _ in 0..20 { + assert!( + matches!(resolver.lookup("extra.invalid", 80, Deadline::new(Duration::from_secs(1)), "dns"), + Err(Error::Cdp(message)) if message.contains("busy")) + ); + } + assert_eq!(calls.load(Ordering::SeqCst), 2); + // Numeric IPs do not depend on the resolver pool. + assert_eq!( + resolve("::1", 80, Deadline::new(Duration::from_secs(1)), "dns").unwrap()[0].ip(), + "::1".parse::().unwrap() + ); + *gate.0.lock().unwrap() = true; + gate.1.notify_all(); + let recovery = Deadline::new(Duration::from_secs(2)); + loop { + match resolver.lookup("recovered.invalid", 80, recovery, "dns") { + Ok(_) => break, + Err(Error::Cdp(message)) if message.contains("busy") => thread::yield_now(), + other => panic!("pool did not recover: {other:?}"), + } + } + assert_eq!( + calls.load(Ordering::SeqCst), + 3, + "expired queued jobs must not run" + ); + } +} diff --git a/src/client.rs b/src/client.rs index 22b9c28..3f0baf0 100644 --- a/src/client.rs +++ b/src/client.rs @@ -420,13 +420,15 @@ mod tests { }; fn client(server: &MockServer) -> Client { - Client::builder() + let mut client = Client::builder() .api_key("test-key") .project_id("test-project") .base_url(server.base_url()) .region("office-test") .build() - .unwrap() + .unwrap(); + client.http = HttpClient::builder().no_proxy().build().unwrap(); + client } #[test] diff --git a/tests/page_targets.rs b/tests/page_targets.rs index 3cc8cd3..1fbe0be 100644 --- a/tests/page_targets.rs +++ b/tests/page_targets.rs @@ -167,6 +167,12 @@ fn two_pages() -> Value { #[test] fn sdk_selects_the_requested_target_regardless_of_enumeration_order() { + if support::isolated_test( + "sdk_selects_the_requested_target_regardless_of_enumeration_order", + Duration::from_secs(20), + ) { + return; + } for reversed in [false, true] { let mut targets = two_pages(); if reversed { @@ -340,3 +346,61 @@ fn default_cli_still_uses_first_page_or_creates_blank_when_no_pages_exist() { assert_eq!(attached["params"]["targetId"], selected); } } + +#[test] +fn connection_timeout_reports_json_failure_without_sending_an_action() { + use std::io::Read; + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let url = format!( + "ws://{}/?token=private-secret", + listener.local_addr().unwrap() + ); + let server = api(&url); + let worker = thread::spawn(move || { + listener.set_nonblocking(true).unwrap(); + let until = Instant::now() + Duration::from_secs(5); + let mut stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(e) if e.kind() == ErrorKind::WouldBlock && Instant::now() < until => { + thread::sleep(Duration::from_millis(10)) + } + Err(e) => panic!("fixture accept: {e}"), + } + }; + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(20))) + .unwrap(); + let mut bytes = Vec::new(); + stream.read_to_end(&mut bytes).unwrap(); + // Only a WebSocket HTTP upgrade was sent; no CDP action frame followed. + assert!(bytes.starts_with(b"GET /")); + assert!(bytes.ends_with(b"\r\n\r\n")); + assert!(!String::from_utf8_lossy(&bytes).contains("Page.navigate")); + }); + let directory = tempfile::tempdir().unwrap(); + let start = Instant::now(); + let output = support::cli( + &server.base_url(), + directory.path(), + &[ + "action", + "open-url", + "--session-id", + "browser", + "--url", + "https://example.test/", + ], + ); + assert!(start.elapsed() < Duration::from_secs(19)); + assert_eq!(output.status.code(), Some(1)); + assert!(output.stdout.is_empty()); + let envelope: Value = serde_json::from_slice(&output.stderr).unwrap(); + assert_eq!( + envelope, + json!({"ok":false,"error":"timeout", + "message":"request timed out: CDP connection (stage: websocket_handshake, budget: 15s)"}) + ); + worker.join().unwrap(); +} diff --git a/tests/page_targets_browser.rs b/tests/page_targets_browser.rs index 5d84e35..24a680f 100644 --- a/tests/page_targets_browser.rs +++ b/tests/page_targets_browser.rs @@ -124,6 +124,12 @@ fn pages(cdp: &mut Cdp) -> Vec { #[test] #[ignore = "requires BROWSER_CLI_TEST_CHROME pointing to a Chrome/Chromium executable"] fn search_popup_can_be_selected_across_cli_invocations_and_closed_safely() { + if support::isolated_test( + "search_popup_can_be_selected_across_cli_invocations_and_closed_safely", + Duration::from_secs(120), + ) { + return; + } let chromium = std::env::var_os("BROWSER_CLI_TEST_CHROME") .expect("set BROWSER_CLI_TEST_CHROME to a local Chrome/Chromium executable"); let browser = Browser::start(Path::new(&chromium)); diff --git a/tests/support/mod.rs b/tests/support/mod.rs index 4b8276a..2641a96 100644 --- a/tests/support/mod.rs +++ b/tests/support/mod.rs @@ -20,24 +20,62 @@ pub fn cli(api: &str, directory: &Path, arguments: &[&str]) -> Output { "LEXMOUNT_BROWSER_CREDENTIALS_FILE", directory.join("missing-credentials.json"), ) - .env("NO_PROXY", "127.0.0.1,localhost") .env_remove("LEXMOUNT_REGION") - .env_remove("HTTP_PROXY") - .env_remove("HTTPS_PROXY") - .env_remove("ALL_PROXY") - .env_remove("http_proxy") - .env_remove("https_proxy") - .env_remove("all_proxy") .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); + clear_proxy_env(&mut command); + run(command, Duration::from_secs(20)) +} + +fn clear_proxy_env(command: &mut Command) { + for name in [ + "HTTP_PROXY", + "HTTPS_PROXY", + "ALL_PROXY", + "NO_PROXY", + "http_proxy", + "https_proxy", + "all_proxy", + "no_proxy", + ] { + command.env_remove(name); + } +} + +// Re-run only this SDK test in an isolated process, before starting fixtures. +// No process-global environment mutation or production localhost bypass. +#[allow(dead_code)] +pub fn isolated_test(name: &str, timeout: Duration) -> bool { + if std::env::var("BROWSER_CLI_ISOLATED_TEST").as_deref() == Ok(name) { + return false; + } + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", name, "--include-ignored", "--nocapture"]) + .env("BROWSER_CLI_ISOLATED_TEST", name) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + clear_proxy_env(&mut command); + let output = run(command, timeout); + assert!( + output.status.success(), + "isolated test failed: {}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + true +} + +fn run(mut command: Command, timeout: Duration) -> Output { #[cfg(windows)] { use std::os::windows::process::CommandExt; command.creation_flags(0x08000000); // CREATE_NO_WINDOW } let mut child = command.spawn().unwrap(); - let deadline = Instant::now() + Duration::from_secs(20); + let deadline = Instant::now() + timeout; while child.try_wait().unwrap().is_none() { if Instant::now() >= deadline { let _ = child.kill(); From a45b88277cb1b88efc836bac435b492708454712 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:17:02 +0800 Subject: [PATCH 6/7] fix: preserve downstream native TLS compatibility --- .github/workflows/ci.yml | 2 + src/cdp/proxy.rs | 40 +++++++---- tests/test_tls_backends.py | 136 +++++++++++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+), 15 deletions(-) create mode 100755 tests/test_tls_backends.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e2bcc2c..a14a107 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,8 @@ jobs: http_proxy=http://127.0.0.1:1 https_proxy=http://127.0.0.1:1 all_proxy=http://127.0.0.1:1 cargo test --all-targets --locked --offline - run: cargo clippy --all-targets --locked -- -D warnings + - name: Verify downstream TLS backend compatibility + run: ./tests/test_tls_backends.py - name: Verify release target contract run: | grep -q 'x86_64-unknown-linux-musl' .github/workflows/release.yml diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs index 7e645fa..41f93f0 100644 --- a/src/cdp/proxy.rs +++ b/src/cdp/proxy.rs @@ -2,6 +2,10 @@ use std::{ io::{self, Read, Write}, net::TcpStream, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, time::{Duration, Instant}, }; @@ -119,9 +123,14 @@ fn connect_once(url: &str, matcher: &Matcher, deadline: Deadline) -> Result Result Error::from(error), @@ -145,14 +154,9 @@ fn connect_once(url: &str, matcher: &Matcher, deadline: Deadline) -> Result stream, - MaybeTlsStream::Rustls(stream) => &mut stream.sock, - _ => return Err(Error::Config("unsupported CDP TLS backend".into())), - }; - stream.stream.set_read_timeout(None)?; - stream.stream.set_write_timeout(None)?; - stream.deadline = None; + timeout_handle.set_read_timeout(None)?; + timeout_handle.set_write_timeout(None)?; + connected.store(true, Ordering::Relaxed); Ok(socket) } @@ -212,17 +216,18 @@ fn remaining(deadline: Instant) -> io::Result { .ok_or_else(|| io::Error::new(io::ErrorKind::TimedOut, "CDP connection deadline exceeded")) } -// Each underlying I/O uses the remaining budget, including rustls' handshake +// Each underlying I/O uses the remaining budget, including TLS handshake // loops and peers that keep trickling bytes. Disabled after the WS upgrade. #[derive(Debug)] pub(super) struct ConnectionStream { stream: TcpStream, - deadline: Option, + deadline: Instant, + connected: Arc, } impl Read for ConnectionStream { fn read(&mut self, buffer: &mut [u8]) -> io::Result { - if let Some(deadline) = self.deadline { + if let Some(deadline) = self.active_deadline() { self.stream.set_read_timeout(Some(remaining(deadline)?))?; } let result = self.stream.read(buffer); @@ -232,7 +237,7 @@ impl Read for ConnectionStream { impl Write for ConnectionStream { fn write(&mut self, buffer: &[u8]) -> io::Result { - if let Some(deadline) = self.deadline { + if let Some(deadline) = self.active_deadline() { self.stream.set_write_timeout(Some(remaining(deadline)?))?; } let result = self.stream.write(buffer); @@ -247,8 +252,13 @@ impl Write for ConnectionStream { } impl ConnectionStream { + fn active_deadline(&self) -> Option { + // The flag changes only before the connected socket is returned. + (!self.connected.load(Ordering::Relaxed)).then_some(self.deadline) + } + fn finish_io(&self, result: io::Result) -> io::Result { - if let Some(deadline) = self.deadline { + if let Some(deadline) = self.active_deadline() { remaining(deadline)?; // Blocking sockets can report WouldBlock on Unix when SO_*TIMEO // expires. Do not let tungstenite interpret it as resumable I/O. diff --git a/tests/test_tls_backends.py b/tests/test_tls_backends.py new file mode 100755 index 0000000..6da899f --- /dev/null +++ b/tests/test_tls_backends.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Downstream SDK feature-unification regression; loopback only, test-only CA.""" +import base64 +import hashlib +import json +import os +from pathlib import Path +import shutil +import socket +import ssl +import subprocess +import tempfile +import threading + +ROOT = Path(__file__).resolve().parents[1] +CLIENT = r''' +fn main() { + let url = std::env::args().nth(1).unwrap(); + let (socket, _) = tungstenite::connect(url.as_str()).expect("control TLS/WS handshake"); + #[cfg(feature = "native-tls")] + assert!(matches!(socket.get_ref(), tungstenite::stream::MaybeTlsStream::NativeTls(_))); + #[cfg(not(feature = "native-tls"))] + assert!(matches!(socket.get_ref(), tungstenite::stream::MaybeTlsStream::Rustls(_))); + drop(socket); + let mut cdp = lexmount_browser::cdp::Cdp::connect(&url).expect("SDK connection"); + assert_eq!(cdp.evaluate("1 + 1").unwrap(), 2); +} +''' + + +def exact(stream, size): + data = b"" + while len(data) < size: + part = stream.recv(size - len(data)) + if not part: + raise EOFError("client closed") + data += part + return data + + +def headers(stream): + data = b"" + while not data.endswith(b"\r\n\r\n"): + data += exact(stream, 1) + assert len(data) < 16384 + return data + + +def serve(listener, context, errors): + try: + for control in (True, False): + raw, _ = listener.accept() + with raw: + raw.settimeout(20) + if raw.recv(1, socket.MSG_PEEK) == b"C": + assert headers(raw).startswith(b"CONNECT ") + raw.sendall(b"HTTP/1.1 200 OK\r\n\r\n") + with context.wrap_socket(raw, server_side=True) as stream: + request = headers(stream) + key = next(line.split(b":", 1)[1].strip() for line in request.split(b"\r\n") + if line.lower().startswith(b"sec-websocket-key:")) + accept = base64.b64encode(hashlib.sha1(key + b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11").digest()) + stream.sendall(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: " + accept + b"\r\n\r\n") + if control: + continue + for method, result in [ + ("Target.getTargets", {"targetInfos": [{"type": "page", "targetId": "page"}]}), + ("Target.attachToTarget", {"sessionId": "attached"}), + ("Page.enable", {}), ("Runtime.enable", {}), + ("Runtime.evaluate", {"result": {"value": 2}}), + ]: + opcode, length = exact(stream, 2) + assert opcode == 0x81 and length & 0x80 + length &= 0x7f + if length == 126: + length = int.from_bytes(exact(stream, 2), "big") + assert length < 4096 + mask = exact(stream, 4) + message = json.loads(bytes(value ^ mask[i % 4] for i, value in enumerate(exact(stream, length)))) + assert message["method"] == method + payload = json.dumps({"id": message["id"], "result": result}).encode() + assert len(payload) < 126 + stream.sendall(bytes([0x81, len(payload)]) + payload) + except Exception as error: + errors.append(error) + + +with tempfile.TemporaryDirectory(prefix="browser-cli-tls-") as directory: + temp = Path(directory) + (temp / "src").mkdir() + (temp / "src/main.rs").write_text(CLIENT) + (temp / "Cargo.toml").write_text(f'''[package] +name = "browser-cli-tls-regression" +version = "0.0.0" +edition = "2024" +[features] +native-tls = ["tungstenite/native-tls"] +[dependencies] +lexmount-browser = {{ path = {json.dumps(str(ROOT))} }} +tungstenite = {{ version = "0.27", default-features = false }} +''') + # Preserve the SDK's locked versions while adding downstream native-tls deps. + shutil.copyfile(ROOT / "Cargo.lock", temp / "Cargo.lock") + for args in [ + ["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", "-subj", "/CN=Test CA", "-keyout", "ca.key", "-out", "ca.pem", "-addext", "basicConstraints=critical,CA:TRUE"], + ["req", "-newkey", "rsa:2048", "-nodes", "-subj", "/CN=localhost", "-keyout", "key.pem", "-out", "leaf.csr"], + ["x509", "-req", "-in", "leaf.csr", "-CA", "ca.pem", "-CAkey", "ca.key", "-CAcreateserial", "-days", "1", "-out", "cert.pem", "-extfile", "leaf.ext"], + ]: + (temp / "leaf.ext").write_text("subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\n") + subprocess.run(["openssl", *args], cwd=temp, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(temp / "cert.pem", temp / "key.pem") + build_env = dict(os.environ) + build_env.setdefault("CARGO_TARGET_DIR", str(ROOT / "target/tls-backends")) + for backend in ("native-tls", "rustls"): + command = ["cargo", "build", "--manifest-path", str(temp / "Cargo.toml")] + if backend == "native-tls": + command += ["--features", "native-tls"] + subprocess.run(command, env=build_env, check=True) + for proxied in (False, True): + env = {k: v for k, v in build_env.items() if k.lower() not in ("http_proxy", "https_proxy", "all_proxy", "no_proxy")} + env["SSL_CERT_FILE"] = str(temp / "ca.pem") + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen() + listener.settimeout(20) + address = f"127.0.0.1:{listener.getsockname()[1]}" + if proxied: + env["HTTPS_PROXY"] = f"http://{address}" + errors = [] + worker = threading.Thread(target=serve, args=(listener, context, errors), daemon=True) + worker.start() + subprocess.run([str(Path(env["CARGO_TARGET_DIR"]) / "debug/browser-cli-tls-regression"), f"wss://{address}/"], env=env, check=True, timeout=25) + worker.join(timeout=5) + assert not worker.is_alive() and not errors, errors + print(f"{backend} {'proxy' if proxied else 'direct'}: TLS verified, SDK connected and evaluated", flush=True) From 248781f55697554349df835933bd613456fde869 Mon Sep 17 00:00:00 2001 From: TristanIsK <286724608+TristanIsK@users.noreply.github.com> Date: Sun, 20 Sep 2026 14:22:26 +0800 Subject: [PATCH 7/7] fix: clear connection timeouts on the same socket handle --- src/cdp/proxy.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/src/cdp/proxy.rs b/src/cdp/proxy.rs index 41f93f0..71af004 100644 --- a/src/cdp/proxy.rs +++ b/src/cdp/proxy.rs @@ -125,7 +125,10 @@ fn connect_once(url: &str, matcher: &Matcher, deadline: Deadline) -> Result io::Result { // loops and peers that keep trickling bytes. Disabled after the WS upgrade. #[derive(Debug)] pub(super) struct ConnectionStream { - stream: TcpStream, + stream: Arc, deadline: Instant, connected: Arc, } @@ -230,7 +233,7 @@ impl Read for ConnectionStream { if let Some(deadline) = self.active_deadline() { self.stream.set_read_timeout(Some(remaining(deadline)?))?; } - let result = self.stream.read(buffer); + let result = self.stream.as_ref().read(buffer); self.finish_io(result) } } @@ -240,13 +243,13 @@ impl Write for ConnectionStream { if let Some(deadline) = self.active_deadline() { self.stream.set_write_timeout(Some(remaining(deadline)?))?; } - let result = self.stream.write(buffer); + let result = self.stream.as_ref().write(buffer); self.finish_io(result) } fn flush(&mut self) -> io::Result<()> { // TcpStream is unbuffered. - let result = self.stream.flush(); + let result = self.stream.as_ref().flush(); self.finish_io(result) } }