use std::collections::HashMap; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use futures_util::{SinkExt, StreamExt}; use serde_json::json; use tokio::net::TcpStream; use tokio::sync::mpsc; use tokio::sync::Notify; use tokio_tungstenite::tungstenite::Message; use tracing::{error, info, warn}; use crate::dispatch::{self, CdpContext}; // PR #36 comment 4341743194: the deferral queue in `process_with_interception` // must be bounded so a stalled navigation cannot OOM the process. When the cap // is reached we return an explicit error response rather than silently dropping. const MAX_DEFERRED_MESSAGES: usize = 256; // The WS-stream forwarding channel must also be bounded: if the LocalSet // (CDP processor + nav tasks) stalls, the accept thread keeps pushing // `std::net::TcpStream`s into the queue. An unbounded channel would let // that queue grow without limit and OOM the process. With a bounded // capacity, when the LocalSet is saturated the accept thread closes the // new connection on the spot instead of buffering it — the kernel TCP // backlog still absorbs short-term spikes, but a long-term stall now // fails loudly at accept time rather than silently piling up FDs. const MAX_PENDING_WS_HANDOFFS: usize = 128; use crate::types::CdpRequest; struct CdpMessage { text: String, reply_tx: mpsc::UnboundedSender, } enum ServerMessage { Cdp(CdpMessage), NewConnection { reply_tx: mpsc::UnboundedSender, }, } pub async fn start(port: u16) -> anyhow::Result<()> { start_with_options(port, None, false).await } pub async fn start_with_options( port: u16, proxy: Option, stealth: bool, ) -> anyhow::Result<()> { start_with_full_options(port, proxy, stealth, None, None).await } pub async fn start_with_full_options( port: u16, proxy: Option, stealth: bool, user_agent: Option, storage_dir: Option, ) -> anyhow::Result<()> { start_with_host(port, "127.0.0.1", proxy, stealth, user_agent, storage_dir).await } pub async fn start_with_host( port: u16, host: &str, proxy: Option, stealth: bool, user_agent: Option, storage_dir: Option, ) -> anyhow::Result<()> { start_with_host_and_security(port, host, proxy, stealth, user_agent, false, storage_dir).await } pub async fn start_with_host_and_security( port: u16, host: &str, proxy: Option, stealth: bool, user_agent: Option, allow_file_access: bool, storage_dir: Option, ) -> anyhow::Result<()> { start_with_full_serve_options( port, host, proxy, stealth, user_agent, allow_file_access, storage_dir, false, ) .await } pub async fn start_with_host_security_and_storage( port: u16, host: &str, proxy: Option, stealth: bool, user_agent: Option, allow_file_access: bool, storage_dir: Option, ) -> anyhow::Result<()> { start_with_full_serve_options( port, host, proxy, stealth, user_agent, allow_file_access, storage_dir, false, ) .await } /// Full serve entry point that also accepts `allow_private_network` (issue /// #33). Older entry points default it to `false` so existing callers and /// public API consumers are unaffected. pub async fn start_with_full_serve_options( port: u16, host: &str, proxy: Option, stealth: bool, user_agent: Option, allow_file_access: bool, storage_dir: Option, allow_private_network: bool, ) -> anyhow::Result<()> { let ip: std::net::IpAddr = host .parse() .map_err(|e| anyhow::anyhow!("invalid --host '{}': {}", host, e))?; let addr = SocketAddr::new(ip, port); // Issue #62: the HTTP control plane (/json/version, /json) must remain // reachable even while V8 JS evaluation blocks the tokio LocalSet thread. // // We use a dedicated OS thread with a blocking std::net::TcpListener so // the kernel's accept backlog is always drained promptly. HTTP endpoints // are served directly via blocking I/O; WebSocket connections are // forwarded to the existing LocalSet for CDP processing. let std_listener = std::net::TcpListener::bind(addr) .map_err(|e| anyhow::anyhow!("bind {}:{}: {}", host, port, e))?; std_listener .set_nonblocking(false) .map_err(|e| anyhow::anyhow!("set_nonblocking: {}", e))?; info!("Obscura CDP server listening on ws://{}:{}", host, port); info!( "DevTools endpoint: ws://{}:{}/devtools/browser", host, port ); if allow_file_access { info!("file:// navigation enabled (--allow-file-access). Do not expose this port to untrusted networks."); } let (ws_tx, mut ws_rx) = mpsc::channel::(MAX_PENDING_WS_HANDOFFS); // Ctrl-C / graceful shutdown coordination. let shutdown_flag = Arc::new(AtomicBool::new(false)); let shutdown_notify = Arc::new(Notify::new()); // Dedicated accept thread: drains the kernel backlog immediately and // handles HTTP endpoints (/json/version, /json, /json/protocol) with // blocking I/O so they never contend with the LocalSet's V8 work. let accept_flag = shutdown_flag.clone(); std::thread::Builder::new() .name("obscura-cdp-accept".into()) .spawn(move || { for stream in std_listener.incoming() { if accept_flag.load(Ordering::Relaxed) { break; } match stream { Ok(stream) => { if let Err(e) = accept_dispatch(stream, port, &ws_tx) { if !format!("{}", e).contains("close") { error!("Accept dispatch error: {}", e); } } } Err(e) => error!("Accept error: {}", e), } } })?; // Build the shared browser context once. Every connection's processor // reuses it (one cookie jar, one HTTP client) while running on its own OS // thread, so each page's V8 isolate is confined to a single thread and the // #430 cross-page abort cannot happen (V8's TryGetCurrent check is // per-thread). This is the thread-per-runtime fix for #430. let mut bctx = obscura_browser::BrowserContext::with_storage_and_network( "default".to_string(), proxy, stealth, user_agent, storage_dir, allow_private_network, ); bctx.allow_file_access = allow_file_access; let shared_ctx = Arc::new(bctx); // One graceful-shutdown watcher for the whole server. It flips the accept // flag (stopping the accept thread) and wakes every connection processor via // `notify_waiters()`. On its own thread so it needs no LocalSet and cannot be // starved by a connection's V8 work. Watches SIGTERM as well as Ctrl-C so // `docker stop` / `kill` also flush cookies (issue #333). { let sf = shutdown_flag.clone(); let sn = shutdown_notify.clone(); std::thread::Builder::new() .name("obscura-cdp-signal".into()) .spawn(move || { if let Ok(rt) = tokio::runtime::Builder::new_current_thread() .enable_all() .build() { rt.block_on(async { #[cfg(unix)] { use tokio::signal::unix::{signal, SignalKind}; match signal(SignalKind::terminate()) { Ok(mut term) => { tokio::select! { _ = tokio::signal::ctrl_c() => {} _ = term.recv() => {} } } Err(_) => { let _ = tokio::signal::ctrl_c().await; } } } #[cfg(not(unix))] { let _ = tokio::signal::ctrl_c().await; } }); } sf.store(true, Ordering::Relaxed); sn.notify_waiters(); }) .ok(); } // Force V8 and its process-global isolate tables (the leaptiering // JSDispatchTable / external-pointer tables) to initialize once on this main // thread before any connection thread creates an isolate. Creating the very // first isolate off the main thread segfaults inside // InitializeBuiltinJSDispatchTable (#430 thread-per-connection). Building and // dropping one runtime here does the one-time setup single-threaded. drop(obscura_js::runtime::ObscuraJsRuntime::new()); // Accept loop: hand each WebSocket connection to its own OS thread so its // pages' isolates live on a dedicated thread. loop { let stream = tokio::select! { stream = ws_rx.recv() => stream, _ = shutdown_notify.notified() => None, }; let stream = match stream { Some(s) => s, None => break, }; // Nagle off + nonblocking on the std socket before it moves to the // connection thread. CDP exchanges many small (~100-byte) frames during // newPage()/navigate; with Nagle on, each small write waits on an ACK or // the 40ms delayed-ACK timer (~90ms on newPage, ~30ms on goto). stream .set_nonblocking(true) .map_err(|e| error!("set_nonblocking on WS stream: {}", e)) .ok(); stream .set_nodelay(true) .map_err(|e| error!("set_nodelay on WS stream: {}", e)) .ok(); run_connection(stream, shared_ctx.clone(), shutdown_notify.clone()); } // Server is shutting down: persist the shared cookie jar once. shared_ctx.save_cookies(); Ok(()) } /// Run one WebSocket connection on its own OS thread: a `current_thread` tokio /// runtime + `LocalSet` hosting this connection's `cdp_processor` (with its own /// `CdpContext` and pages) and its frame reader. Confining a connection's pages /// to one thread is what removes the #430 abort; the interception handshake and /// the nav `spawn_local` all stay on this one thread, so no cross-thread V8 /// plumbing is needed. fn run_connection( std_stream: std::net::TcpStream, default_context: Arc, shutdown_notify: Arc, ) { std::thread::Builder::new() .name("obscura-cdp-conn".into()) .spawn(move || { let rt = match tokio::runtime::Builder::new_current_thread() .enable_all() .build() { Ok(r) => r, Err(e) => { error!("connection runtime build failed: {}", e); return; } }; let local = tokio::task::LocalSet::new(); local.block_on(&rt, async move { let tokio_stream = match TcpStream::from_std(std_stream) { Ok(s) => s, Err(e) => { error!("TcpStream::from_std failed: {}", e); return; } }; let (msg_tx, msg_rx) = mpsc::unbounded_channel::(); let processor = tokio::task::spawn_local(cdp_processor( msg_rx, default_context, shutdown_notify, )); if let Err(e) = handle_connection_ws(tokio_stream, msg_tx).await { error!("WebSocket connection error: {}", e); } // Connection closed (or shutting down): stop this connection's // processor so the thread can exit. processor.abort(); }); }) .ok(); } const HTTP_PEEK_BUF: usize = 4096; const WS_PEEK_BUF: usize = 4; /// Dispatch a freshly-accepted TCP connection on the dedicated accept thread. /// /// Peek at the first bytes to decide HTTP vs WebSocket: /// - HTTP (`GET /json/*`): serve synchronously via blocking I/O so the /// response is never stalled by the LocalSet. /// - WebSocket: set non-blocking, convert to tokio `TcpStream`, and forward /// to the LocalSet for CDP processing. fn accept_dispatch( stream: std::net::TcpStream, port: u16, ws_tx: &mpsc::Sender, ) -> anyhow::Result<()> { let mut buf = [0u8; WS_PEEK_BUF]; let n = stream.peek(&mut buf)?; if n >= 4 && &buf == b"GET " { let mut peek_buf = [0u8; HTTP_PEEK_BUF]; let n = stream.peek(&mut peek_buf)?; let line = String::from_utf8_lossy(&peek_buf[..n]); let endpoint = if line.contains("/json/version") { Some("version") } else if line.contains("/json/list") || line.contains("/json\r\n") || line.contains("/json HTTP") { Some("list") } else if line.contains("/json/protocol") { Some("protocol") } else { None }; if let Some(ep) = endpoint { return handle_http_json_blocking(stream, port, ep); } // Fall through: GET request that isn't a /json endpoint → treat as // WebSocket upgrade (Chromium DevTools clients issue GET with // Upgrade: websocket). } // Try to hand off the WS stream to the LocalSet. If the bounded channel // is full the LocalSet is saturated — drop the connection cleanly // rather than blocking the accept thread (which would freeze the HTTP // control plane that this whole rework exists to keep alive). The // dropped `stream` closes itself; the client will see ECONNRESET and // can retry. ws_tx .try_send(stream) .map_err(|e| match e { mpsc::error::TrySendError::Full(_) => { warn!("WS handoff channel full ({}); dropping new WebSocket connection", MAX_PENDING_WS_HANDOFFS); anyhow::anyhow!("ws handoff channel full") } mpsc::error::TrySendError::Closed(_) => anyhow::anyhow!("accept channel closed"), }) } /// Serve an HTTP `/json/*` endpoint with blocking I/O on the accept thread. fn handle_http_json_blocking( mut stream: std::net::TcpStream, port: u16, endpoint: &str, ) -> anyhow::Result<()> { use std::io::{Read, Write}; let mut buf = vec![0u8; 4096]; let _ = stream.read(&mut buf)?; let body = match endpoint { "version" => serde_json::to_string_pretty(&json!({ "Browser": "Chrome/145.0.0.0", "Protocol-Version": "1.3", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36", "V8-Version": "14.5.0.0", "WebKit-Version": "537.36", "webSocketDebuggerUrl": format!("ws://127.0.0.1:{}/devtools/browser", port), }))?, "list" => serde_json::to_string_pretty(&json!([{ "description": "", "devtoolsFrontendUrl": "", "id": "page-1", "title": "", "type": "page", "url": "about:blank", "webSocketDebuggerUrl": format!("ws://127.0.0.1:{}/devtools/page/page-1", port), }]))?, "protocol" => { serde_json::to_string_pretty(&json!({ "version": { "major": "1", "minor": "3" } }))? } _ => "{}".to_string(), }; let resp = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", body.len(), body, ); stream.write_all(resp.as_bytes())?; stream.flush()?; Ok(()) } /// Per-connection CDP processor. Each connection runs its own processor (with /// its own `CdpContext` and pages) on its own OS thread, so every page's V8 /// isolate is confined to a single thread. This removes the #430 abort by /// construction: V8's `heap->isolate() == Isolate::TryGetCurrent()` invariant is /// per-thread, so two connections' isolates can never collide. All processors /// share one `Arc` (one cookie jar, one HTTP client). The shared /// context's cookies are persisted once by the accept side on shutdown. async fn cdp_processor( mut rx: mpsc::UnboundedReceiver, default_context: Arc, shutdown_notify: Arc, ) { let mut ctx = CdpContext::new_with_shared_context(default_context); let (itx, irx) = mpsc::unbounded_channel::(); ctx.intercept_tx = Some(itx); let mut intercept_rx: Option> = Some(irx); let mut intercepted_paused: HashMap> = HashMap::new(); // Issue #19 follow-up: messages deferred from inside // `process_with_interception` because routing them through // `process_cdp_message → dispatch` while a nav was in flight would have // tripped V8's TryGetCurrent invariant. Drained at the top of each // outer iteration so they get processed sequentially with no other nav // in flight. let mut deferred: std::collections::VecDeque = std::collections::VecDeque::new(); // Graceful shutdown: one signal watcher on the accept side flips the flag // and calls `notify_waiters()`. Polled once here (via the select! below) it // registers and stays registered across iterations, so a later // `notify_waiters()` wakes this processor even while it is mid-dispatch. let mut shutdown = Box::pin(shutdown_notify.notified()); loop { // Drain any deferred messages from the previous interception window // before pulling new ones off the wire. Each is processed with no // nav-task spawn_local in flight, so this connection's only entered // Isolate is the one dispatch is about to touch. let msg = if let Some(d) = deferred.pop_front() { d } else { tokio::select! { msg = rx.recv() => match msg { Some(m) => m, None => break, }, _ = &mut shutdown => { tracing::info!("Shutdown signal received (connection processor)"); break; } } }; match msg { ServerMessage::NewConnection { reply_tx } => { let _ = reply_tx.send( json!({"__init": true}) .to_string(), ); } ServerMessage::Cdp(cdp_msg) => { // Route every Page.navigate through the spawn-and-defer path, // not just intercepted ones. Holding the V8 lock across a // multi-second navigate inside the regular dispatch wedges the // entire processor (40-site sweep: 39/40 timeouts). Spawning // navigation lets `cdp_processor` keep multiplexing other CDP // messages via the `process_with_interception` select loop; // unrelated requests get deferred only briefly and are drained // as soon as the nav settles. let is_navigation = is_navigate_method(&cdp_msg.text); if is_navigation { process_with_interception( &cdp_msg.text, &mut ctx, &cdp_msg.reply_tx, &mut rx, &mut intercept_rx, &mut intercepted_paused, &mut deferred, ).await; } else { if cdp_msg.text.contains("Fetch.") { handle_fetch_resolution(&cdp_msg.text, &mut ctx, &cdp_msg.reply_tx, &mut intercepted_paused); } process_cdp_message(&cdp_msg.text, &mut ctx, &cdp_msg.reply_tx).await; } } } } // Cookies live in the shared BrowserContext and are persisted once by the // accept side when the whole server shuts down, so a single connection // closing (or its processor being aborted) does not need to save here. let _ = &ctx; } // Whether a raw CDP frame is exactly a `Page.navigate` call, and so should take // the spawn-and-defer navigation path. Matching on the parsed method rather than // a `contains("Page.navigate")` substring avoids catching // `Page.navigateToHistoryEntry` (goBack / goForward), which has no `url` param // and belongs to its own handler, or any other frame that merely embeds the // literal text (e.g. a `Runtime.evaluate` expression). See issue #363. fn is_navigate_method(text: &str) -> bool { serde_json::from_str::(text) .map(|req| req.method == "Page.navigate") .unwrap_or(false) } // Parse a CDP header list (`[{"name":..,"value":..}, ..]`, as used by // Fetch.continueRequest / fulfillRequest) into a map. Returns None when the // `headers` field is absent, so the caller can leave the request's headers // untouched rather than clearing them. fn parse_cdp_headers(params: &serde_json::Value) -> Option> { let arr = params.get("headers")?.as_array()?; Some( arr.iter() .filter_map(|h| { Some(( h.get("name")?.as_str()?.to_string(), h.get("value")?.as_str()?.to_string(), )) }) .collect(), ) } fn handle_fetch_resolution( text: &str, _ctx: &mut CdpContext, reply_tx: &mpsc::UnboundedSender, intercepted_paused: &mut HashMap>, ) { if let Ok(req) = serde_json::from_str::(text) { let method = req.method.as_str(); let request_id = req.params.get("requestId").and_then(|v| v.as_str()).unwrap_or(""); tracing::info!("INTERCEPTION resolution: {} for {}, paused_count={}", method, request_id, intercepted_paused.len()); if let Some(resolver) = intercepted_paused.remove(request_id) { tracing::info!("INTERCEPTION resolved: {}", request_id); let resolution = match method { "Fetch.continueRequest" => obscura_js::ops::InterceptResolution::Continue { // Honor the client's overrides (Playwright route.continue, // Puppeteer request.continue). op_fetch_url applies each and // re-validates a rewritten URL through the SSRF gate. Leaving // these None silently sent the request unmodified (issue #365). url: req.params.get("url").and_then(|v| v.as_str()).map(|s| s.to_string()), method: req.params.get("method").and_then(|v| v.as_str()).map(|s| s.to_string()), headers: parse_cdp_headers(&req.params), body: req.params.get("postData").and_then(|v| v.as_str()).map(|s| s.to_string()), }, "Fetch.fulfillRequest" => { let status = req.params.get("responseCode").and_then(|v| v.as_u64()).unwrap_or(200) as u16; let raw_body = req.params.get("body").and_then(|v| v.as_str()).unwrap_or(""); let body = decode_base64(raw_body); let headers = req.params.get("responseHeaders") .and_then(|v| v.as_array()) .map(|arr| arr.iter().filter_map(|h| { Some((h.get("name")?.as_str()?.to_string(), h.get("value")?.as_str()?.to_string())) }).collect()) .unwrap_or_default(); obscura_js::ops::InterceptResolution::Fulfill { status, headers, body } } "Fetch.failRequest" => { let reason = req.params.get("errorReason").and_then(|v| v.as_str()).unwrap_or("Failed").to_string(); obscura_js::ops::InterceptResolution::Fail { reason } } _ => return, }; let _ = resolver.send(resolution); let resp = crate::types::CdpResponse::success(req.id, json!({}), req.session_id); if let Ok(json) = serde_json::to_string(&resp) { let _ = reply_tx.send(json); } } } } async fn process_with_interception( text: &str, ctx: &mut CdpContext, reply_tx: &mpsc::UnboundedSender, rx: &mut mpsc::UnboundedReceiver, intercept_rx: &mut Option>, intercepted_paused: &mut HashMap>, deferred: &mut std::collections::VecDeque, ) { let req: CdpRequest = match serde_json::from_str(text) { Ok(r) => r, Err(e) => { warn!("Invalid CDP: {}", e); return; } }; tracing::info!("INTERCEPTION navigate: {} (id={})", req.method, req.id); let session_id = &req.session_id; let page_id = session_id .as_ref() .and_then(|sid| ctx.sessions.get(sid)) .cloned(); let page_id = match page_id { Some(id) => id, None => { process_cdp_message(text, ctx, reply_tx).await; return; } }; let page_index = ctx.pages.iter().position(|p| p.id == page_id); let mut page = match page_index { Some(idx) => ctx.pages.remove(idx), None => { process_cdp_message(text, ctx, reply_tx).await; return; } }; // Issue #19 follow-up: V8 only allows ONE entered Isolate per OS thread. // The regular dispatch path enforces this via `get_session_page_mut` // (which `suspend_js`'es every other page before letting the target // page run JS). The interception path here bypasses that — it removes // the target page and spawns a nav task — so we have to enforce the // same invariant explicitly. Otherwise nav-2's `init_js` constructs // Isolate-2 while page-1's Isolate-1 is still alive in ctx.pages, and // the next V8 scope unwind aborts the process via `Context::Exit`'s // `heap->isolate() == Isolate::TryGetCurrent()` check. for other in ctx.pages.iter_mut() { if other.has_js() { other.suspend_js(); } } let url = req.params.get("url").and_then(|v| v.as_str()).unwrap_or(""); let wait_until = crate::domains::page::parse_wait_until(&req.params); let nav_method = req.params.get("__method").and_then(|v| v.as_str()).unwrap_or("GET").to_string(); let nav_body = req.params.get("__body").and_then(|v| v.as_str()).unwrap_or("").to_string(); let preload_scripts: Vec = ctx.preload_scripts.iter().map(|(_, s)| s.clone()).collect(); if let Some(tx) = &ctx.intercept_tx { page.set_intercept_tx(tx.clone()); } let session_for_events = req.session_id.clone(); let frame_id = page.frame_id.clone(); let loader_id = format!("loader-{}", uuid::Uuid::new_v4()); let (nav_done_tx, mut nav_done_rx) = mpsc::channel::<(obscura_browser::Page, Result<(), String>)>(1); let url_owned = url.to_string(); let nav_v8_lock = ctx.v8_lock.clone(); tokio::task::spawn_local(async move { // Issue #19: serialize this connection's V8 work across its pages. This // nav task runs while the connection's processor keeps pumping other CDP // messages via `dispatch` (which takes the same per-connection lock), so // both sides coordinate on one page's isolate at a time on this thread. // The lock is per-connection, so other connections are unaffected (#430). let _v8_guard = nav_v8_lock.lock_owned().await; // Preloads (addBinding shims, addScriptToEvaluateOnNewDocument sources) // must run BEFORE the page's own scripts (CDP contract). Hand them // to the page so navigate_single can inject them at the right point. page.set_preload_scripts(preload_scripts); let result = if nav_method == "POST" && !nav_body.is_empty() { page.navigate_with_wait_post(&url_owned, wait_until, &nav_method, &nav_body).await } else { page.navigate_with_wait(&url_owned, wait_until).await } .map_err(|e| e.to_string()); drop(_v8_guard); let _ = nav_done_tx.send((page, result)).await; }); let navigate_result: Result<(), String>; let page_back: Option; // Issue #19 follow-up (PR #36 maintainer's fetch-intercept repro): // While the spawned nav task is executing V8 (potentially parked on // `op_fetch_url`'s `resolve_rx.await` *with Isolate-N still entered*), // we must NOT let the parent's `select!` route foreign Cdp messages // through `process_cdp_message → dispatch → page handlers`, because // those handlers call `get_session_page_mut` which `suspend_js`'es // OTHER pages (drops their `JsRuntime`, which calls // `JsRealmInner::destroy`). That trips V8's // `heap->isolate() == Isolate::TryGetCurrent()` invariant and aborts // the process via `V8_Fatal`. // // This connection's `ctx.v8_lock` doesn't save us here: it's a // `tokio::sync::Mutex` that is released around `.await`s inside V8 // ops, so it doesn't actually keep the V8 enter/exit pair contiguous // on the thread. // // Park foreign Cdp messages into the outer deferred queue so the // outer `cdp_processor` loop processes them after this nav fully // completes (and its JsRuntime is no longer in flight on the // LocalSet). loop { let has_irx = intercept_rx.is_some(); tokio::select! { Some((returned_page, result)) = nav_done_rx.recv() => { page_back = Some(returned_page); navigate_result = result; break; } Some(intercepted) = async { if let Some(ref mut irx) = intercept_rx { irx.recv().await } else { std::future::pending().await } }, if has_irx => { tracing::info!("INTERCEPTION: requestPaused for {} {} (sending to client)", intercepted.method, intercepted.url); let rws_event = json!({ "method": "Network.requestWillBeSent", "params": { "requestId": intercepted.request_id, "loaderId": "", "documentURL": "", "request": { "url": intercepted.url, "method": intercepted.method, "headers": intercepted.headers, "initialPriority": "High", "referrerPolicy": "strict-origin-when-cross-origin", }, "timestamp": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs_f64(), "wallTime": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs_f64(), "initiator": {"type": "script"}, "type": intercepted.resource_type, "frameId": frame_id, }, "sessionId": session_for_events, }); let _ = reply_tx.send(rws_event.to_string()); let event_json = json!({ "method": "Fetch.requestPaused", "params": { "requestId": intercepted.request_id, "request": { "url": intercepted.url, "method": intercepted.method, "headers": intercepted.headers, "initialPriority": "High", "referrerPolicy": "strict-origin-when-cross-origin", }, "frameId": frame_id, "resourceType": intercepted.resource_type, "networkId": intercepted.request_id, "responseErrorReason": null, "responseStatusCode": null, "responseHeaders": null, }, "sessionId": session_for_events, }); let event_str = event_json.to_string(); tracing::info!("INTERCEPTION event JSON: {}", crate::util::truncate_on_char_boundary(&event_str, 300)); let _ = reply_tx.send(event_str); intercepted_paused.insert(intercepted.request_id.clone(), intercepted.resolver); tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; } Some(msg) = rx.recv() => { tracing::info!("INTERCEPTION select: received CDP message during navigation"); match msg { ServerMessage::NewConnection { reply_tx: new_tx } => { // Safe: no V8 enter, just bookkeeping. let pid = ctx.create_page(); let sid = format!("{}-session", pid); ctx.sessions.insert(sid.clone(), pid.clone()); let _ = new_tx.send(json!({"__init": true, "pageId": pid, "sessionId": sid}).to_string()); } ServerMessage::Cdp(msg) => { if msg.text.contains("Fetch.continueRequest") || msg.text.contains("Fetch.fulfillRequest") || msg.text.contains("Fetch.failRequest") { // Safe: only flips a oneshot to resume the parked // op inside the spawned nav task. No V8 enter on // this side; the actual V8 work happens back on // the nav task's thread. handle_fetch_resolution(&msg.text, ctx, &msg.reply_tx, intercepted_paused); } else { // UNSAFE during nav: would route through dispatch, // which can `suspend_js` other pages and trip the // V8 invariant. Defer until nav completes — // pushed to the outer `cdp_processor` queue so // it's processed sequentially with no nav task // in flight. if deferred.len() >= MAX_DEFERRED_MESSAGES { tracing::warn!("INTERCEPTION: deferred queue full ({}), returning error to client", MAX_DEFERRED_MESSAGES); if let Ok(req) = serde_json::from_str::(&msg.text) { let resp = crate::types::CdpResponse::error( req.id, -32000, "Server busy: navigation in progress, try again later".to_string(), req.session_id, ); if let Ok(json) = serde_json::to_string(&resp) { let _ = msg.reply_tx.send(json); } } } else { tracing::info!("INTERCEPTION: deferring CDP message until nav completes"); deferred.push_back(ServerMessage::Cdp(msg)); } } } } } } } // Deferred messages are handled by the outer `cdp_processor` loop // (it drains `deferred` before pulling the next message off `rx`). let mut page = page_back.expect("navigation task should return the page"); // Fold in network events for script-initiated requests (fetch/XHR/dynamic // resource) so they emit as Network.requestWillBeSent / responseReceived // alongside the static navigation subresources (#406). page.sync_js_network_events(); let network_events: Vec<_> = page.network_events.drain(..).collect(); let page_url = page.url_string(); let page_id_for_events = page.id.clone(); let reached_network_idle = page.lifecycle.is_network_idle(); ctx.pages.push(page); let response = match navigate_result { Ok(()) => crate::types::CdpResponse::success( req.id, json!({"frameId": frame_id, "loaderId": loader_id}), req.session_id.clone(), ), Err(e) => crate::types::CdpResponse::error(req.id, -32000, e, req.session_id.clone()), }; if let Ok(json) = serde_json::to_string(&response) { let _ = reply_tx.send(json); } // Shared event emission: includes the post-#190 Network.requestWillBeSent // -before-frameNavigated ordering, the #189 requestId=loaderId trick that // makes `page.goto()` resolve to a Response, and the #192 per-isolated- // world fresh context ids. Pushes to `ctx.pending_events`; we then drain // to the WS reply channel. crate::domains::page::emit_navigation_events( ctx, &session_for_events, &frame_id, &loader_id, &page_url, &page_id_for_events, &network_events, wait_until, reached_network_idle, ); for event in ctx.pending_events.drain(..) { if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); } } } async fn process_cdp_message( text: &str, ctx: &mut CdpContext, reply_tx: &mpsc::UnboundedSender, ) { let req: CdpRequest = match serde_json::from_str(text) { Ok(r) => r, Err(e) => { warn!("Invalid CDP: {}: {}", e, crate::util::truncate_on_char_boundary(text, 200)); return; } }; tracing::debug!("CDP: {} (id={}, s={:?})", req.method, req.id, req.session_id); let response = dispatch::dispatch(&req, ctx).await; // Chromium CDP semantics: events emitted as a side-effect of a command // (e.g. Target.targetCreated + Target.attachedToTarget from // Target.createTarget) MUST arrive BEFORE the command's response. // Playwright awaits the response and immediately reads state wired up // by those events; if the response lands first, accessing // Target._page errors with "Cannot read properties of undefined". for event in ctx.pending_events.drain(..) { if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); } } if let Ok(json) = serde_json::to_string(&response) { let _ = reply_tx.send(json); } if let Some((nav_url, nav_method, nav_body)) = check_pending_navigation(ctx, &req.session_id) { tracing::info!("JS-triggered nav: {} {} (body: {} bytes)", nav_method, nav_url, nav_body.len()); let nav_req = CdpRequest { id: 0, method: "Page.navigate".to_string(), params: json!({"url": nav_url, "__method": nav_method, "__body": nav_body}), session_id: req.session_id.clone(), }; let _ = dispatch::dispatch(&nav_req, ctx).await; for event in ctx.pending_events.drain(..) { if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); } } } } fn decode_base64(input: &str) -> String { fn val(c: u8) -> Option { match c { b'A'..=b'Z' => Some(c - b'A'), b'a'..=b'z' => Some(c - b'a' + 26), b'0'..=b'9' => Some(c - b'0' + 52), b'+' => Some(62), b'/' => Some(63), _ => None, } } let bytes: Vec = input.bytes().filter_map(val).collect(); let mut out = Vec::with_capacity(bytes.len() * 3 / 4); for chunk in bytes.chunks(4) { let b = [ chunk.first().copied().unwrap_or(0), chunk.get(1).copied().unwrap_or(0), chunk.get(2).copied().unwrap_or(0), chunk.get(3).copied().unwrap_or(0), ]; out.push((b[0] << 2) | (b[1] >> 4)); if chunk.len() > 2 { out.push((b[1] << 4) | (b[2] >> 2)); } if chunk.len() > 3 { out.push((b[2] << 6) | b[3]); } } String::from_utf8_lossy(&out).to_string() } fn fast_path_response(text: &str) -> Option { let req: CdpRequest = serde_json::from_str(text).ok()?; let result = match req.method.as_str() { "Network.enable" | "Network.setCacheDisabled" | "Network.setRequestInterception" | "Page.enable" | "Page.setLifecycleEventsEnabled" | "Page.setInterceptFileChooserDialog" | "Runtime.runIfWaitingForDebugger" | "Runtime.discardConsoleEntries" | "Performance.enable" | "Log.enable" | "Security.enable" | "Emulation.setDeviceMetricsOverride" | "Emulation.setTouchEmulationEnabled" | "CSS.enable" | "Accessibility.enable" | "ServiceWorker.enable" | "Inspector.enable" | "Debugger.enable" | "Profiler.enable" | "HeapProfiler.enable" | "Overlay.enable" | "Storage.enable" | "Target.setAutoAttach" => { Some(json!({})) } "Browser.getVersion" => { Some(json!({ "protocolVersion": "1.3", "product": "Chrome/145.0.0.0", "revision": "@0000000000000000000000000000000000000000", "userAgent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36", "jsVersion": "14.5.0.0", })) } "Browser.setDownloadBehavior" | "Browser.getWindowBounds" => { Some(json!({})) } // Critical: Puppeteer calls this as the *first* CDP command on connect // (`BrowserConnector._connectToCdpBrowser`). If another client or a long // `Page.navigate` / interception holds the single `cdp_processor` task, // queued Target commands starve and Puppeteer hits protocolTimeout on // `Target.getBrowserContexts`. Fast-path bypasses the queue — same payload // as `domains::target::handle` when default context id is `"default"`. "Target.getBrowserContexts" => { Some(json!({ "browserContextIds": ["default"] })) } _ => None, }; if let Some(value) = result { let resp = crate::types::CdpResponse::success(req.id, value, req.session_id); serde_json::to_string(&resp).ok() } else { None } } fn check_pending_navigation(ctx: &CdpContext, session_id: &Option) -> Option<(String, String, String)> { let page_id = session_id .as_ref() .and_then(|sid| ctx.sessions.get(sid))?; let page = ctx.pages.iter().find(|p| &p.id == page_id)?; page.take_pending_navigation() } async fn handle_connection_ws( stream: TcpStream, msg_tx: mpsc::UnboundedSender, ) -> anyhow::Result<()> { // tokio_tungstenite wraps the stream in a 128 KiB write BufWriter by // default. CDP traffic is many small (~100-byte) frames, and that buffer // adds extra latency per frame. write_buffer_size=0 makes every WS write // hit the socket directly. Combined with set_nodelay(true) above, gets // per-frame latency on localhost down toward ideal. use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; let mut cfg = WebSocketConfig::default(); cfg.write_buffer_size = 0; cfg.max_write_buffer_size = 64 << 20; let ws_stream = tokio_tungstenite::accept_async_with_config(stream, Some(cfg)).await?; info!("WebSocket connected"); let (mut ws_sender, mut ws_receiver) = ws_stream.split(); let (reply_tx, mut reply_rx) = mpsc::unbounded_channel::(); let _ = msg_tx.send(ServerMessage::NewConnection { reply_tx: reply_tx.clone(), }); if let Some(init_msg) = reply_rx.recv().await { tracing::debug!("Connection init: {}", &init_msg[..init_msg.len().min(100)]); } let send_task = tokio::task::spawn_local(async move { while let Some(msg) = reply_rx.recv().await { if msg.contains("\"__init\"") { continue; } if ws_sender.send(Message::Text(msg.into())).await.is_err() { break; } } }); while let Some(msg) = ws_receiver.next().await { let msg = match msg { Ok(m) => m, Err(e) => { warn!("WS read error: {}", e); break; } }; match msg { Message::Text(text) => { if text.contains("\"Browser.close\"") { if let Ok(req) = serde_json::from_str::(&text) { let resp = crate::types::CdpResponse::success(req.id, json!({}), None); if let Ok(json) = serde_json::to_string(&resp) { let _ = reply_tx.send(json); } } break; } if let Some(resp) = fast_path_response(&text) { let _ = reply_tx.send(resp); } else { let _ = msg_tx.send(ServerMessage::Cdp(CdpMessage { text: text.to_string(), reply_tx: reply_tx.clone(), })); } } Message::Close(_) => { info!("WS closed by client"); break; } _ => {} } } send_task.abort(); Ok(()) } #[cfg(test)] mod tests { use super::{is_navigate_method, parse_cdp_headers}; use serde_json::json; // Issue #363: only an exact Page.navigate may take the spawn-and-defer // navigation path. A substring match also caught Page.navigateToHistoryEntry // (goBack / goForward), which has no `url` param, so it was misrouted into // the raw-navigate path and failed with "Invalid URL" instead of reaching // its real handler. #[test] fn only_exact_page_navigate_routes_as_navigation() { assert!(is_navigate_method( r#"{"id":1,"method":"Page.navigate","params":{"url":"https://example.com"}}"# )); assert!(!is_navigate_method( r#"{"id":2,"method":"Page.navigateToHistoryEntry","params":{"entryId":0}}"# )); } // A Runtime.evaluate whose expression merely contains the literal // "Page.navigate" must not be misrouted, and malformed input is not a // navigation. #[test] fn unrelated_methods_do_not_route_as_navigation() { assert!(!is_navigate_method( r#"{"id":3,"method":"Runtime.evaluate","params":{"expression":"'Page.navigate'"}}"# )); assert!(!is_navigate_method("not json")); } // Issue #365: Fetch.continueRequest header overrides must be parsed from the // CDP `[{name, value}]` list so they can be applied to the outgoing request. #[test] fn parse_cdp_headers_reads_name_value_pairs() { let params = json!({ "headers": [ {"name": "X-A", "value": "1"}, {"name": "X-B", "value": "2"}, ] }); let headers = parse_cdp_headers(¶ms).expect("headers present"); assert_eq!(headers.get("X-A").map(String::as_str), Some("1")); assert_eq!(headers.get("X-B").map(String::as_str), Some("2")); } // No `headers` field means "leave the request's headers untouched", which is // None, not an empty map that would clear them. #[test] fn parse_cdp_headers_absent_is_none() { assert!(parse_cdp_headers(&json!({"url": "https://example.com"})).is_none()); } }