Ben Younes
10810e250c
fix(cdp): unblock HTTP control plane during synchronous V8 execution Fixes #62. Previously, the TCP accept loop and CDP processor shared a single-threaded tokio LocalSet. Long synchronous JS evaluations starved the accept loop, causing incoming connections to hang in the kernel backlog and effectively killing the HTTP control plane. This moves the TCP listener and `/json/*` endpoints to a dedicated OS thread using `std::net::TcpListener`. HTTP requests are now handled directly via blocking I/O on this thread, completely bypassing V8. WebSocket connections are handed off to the LocalSet via a bounded mpsc channel (capacity 128). If V8 stalls and the queue fills, new WS connections are dropped (ECONNRESET) rather than buffering unboundedly to prevent OOMs. Includes a new integration test to verify the control plane stays responsive during a 5s JS loop, plus inline documentation clarifying thread lifetime semantics and a minor test TOCTOU race.
902 lines
39 KiB
Rust
902 lines
39 KiB
Rust
use std::collections::HashMap;
|
|
use std::net::SocketAddr;
|
|
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use serde_json::json;
|
|
use tokio::net::TcpStream;
|
|
use tokio::sync::mpsc;
|
|
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<String>,
|
|
}
|
|
|
|
enum ServerMessage {
|
|
Cdp(CdpMessage),
|
|
NewConnection {
|
|
reply_tx: mpsc::UnboundedSender<String>,
|
|
},
|
|
}
|
|
|
|
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<String>,
|
|
stealth: bool,
|
|
) -> anyhow::Result<()> {
|
|
start_with_full_options(port, proxy, stealth, None).await
|
|
}
|
|
|
|
pub async fn start_with_full_options(
|
|
port: u16,
|
|
proxy: Option<String>,
|
|
stealth: bool,
|
|
user_agent: Option<String>,
|
|
) -> anyhow::Result<()> {
|
|
start_with_host(port, "127.0.0.1", proxy, stealth, user_agent).await
|
|
}
|
|
|
|
pub async fn start_with_host(
|
|
port: u16,
|
|
host: &str,
|
|
proxy: Option<String>,
|
|
stealth: bool,
|
|
user_agent: Option<String>,
|
|
) -> anyhow::Result<()> {
|
|
start_with_host_and_security(port, host, proxy, stealth, user_agent, false).await
|
|
}
|
|
|
|
pub async fn start_with_host_and_security(
|
|
port: u16,
|
|
host: &str,
|
|
proxy: Option<String>,
|
|
stealth: bool,
|
|
user_agent: Option<String>,
|
|
allow_file_access: 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::<std::net::TcpStream>(MAX_PENDING_WS_HANDOFFS);
|
|
|
|
// 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.
|
|
//
|
|
// Lifecycle note: this thread is spawned detached (no `join` handle).
|
|
// It is intended to run for the entire process lifetime — the same
|
|
// contract Chromium DevTools / Playwright clients expect from a CDP
|
|
// server. When `start_with_*` returns (whether by Ok or panic in the
|
|
// LocalSet), `ws_rx` drops; the next `ws_tx.blocking_send` then
|
|
// returns `SendError`, which `accept_dispatch` surfaces as
|
|
// "accept channel closed" and the loop logs+continues. The listener
|
|
// FD stays bound until the process exits. If we ever need to support
|
|
// graceful shutdown for embedded/library use, add an
|
|
// `Arc<AtomicBool>` shutdown flag checked between `accept()`s and
|
|
// switch to a non-blocking `set_nonblocking(true)` + poll loop.
|
|
// For the standalone `obscura serve` binary the detached lifetime is
|
|
// correct.
|
|
std::thread::Builder::new()
|
|
.name("obscura-cdp-accept".into())
|
|
.spawn(move || {
|
|
for stream in std_listener.incoming() {
|
|
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),
|
|
}
|
|
}
|
|
})?;
|
|
|
|
let local = tokio::task::LocalSet::new();
|
|
local
|
|
.run_until(async {
|
|
let (msg_tx, msg_rx) = mpsc::unbounded_channel::<ServerMessage>();
|
|
|
|
let _processor_handle = tokio::task::spawn_local(cdp_processor(
|
|
msg_rx, proxy, stealth, user_agent, allow_file_access,
|
|
));
|
|
|
|
while let Some(stream) = ws_rx.recv().await {
|
|
// Convert std TcpStream → tokio TcpStream inside the LocalSet
|
|
// where the tokio runtime is active.
|
|
stream
|
|
.set_nonblocking(true)
|
|
.map_err(|e| error!("set_nonblocking on WS stream: {}", e))
|
|
.ok();
|
|
let tokio_stream = match TcpStream::from_std(stream) {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
error!("TcpStream::from_std failed: {}", e);
|
|
continue;
|
|
}
|
|
};
|
|
let tx = msg_tx.clone();
|
|
tokio::task::spawn_local(async move {
|
|
if let Err(e) = handle_connection_ws(tokio_stream, tx).await {
|
|
error!("WebSocket connection error: {}", e);
|
|
}
|
|
});
|
|
}
|
|
})
|
|
.await;
|
|
|
|
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<std::net::TcpStream>,
|
|
) -> 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": "Obscura/0.1.0",
|
|
"Protocol-Version": "1.3",
|
|
"User-Agent": "Obscura/0.1.0 (Headless Browser)",
|
|
"V8-Version": "N/A",
|
|
"WebKit-Version": "N/A",
|
|
"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(())
|
|
}
|
|
|
|
async fn cdp_processor(
|
|
mut rx: mpsc::UnboundedReceiver<ServerMessage>,
|
|
proxy: Option<String>,
|
|
stealth: bool,
|
|
user_agent: Option<String>,
|
|
allow_file_access: bool,
|
|
) {
|
|
let mut ctx = CdpContext::new_with_security(proxy, stealth, user_agent, allow_file_access);
|
|
let (itx, irx) = mpsc::unbounded_channel::<obscura_js::ops::InterceptedRequest>();
|
|
ctx.intercept_tx = Some(itx);
|
|
let mut intercept_rx: Option<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>> = Some(irx);
|
|
let mut intercepted_paused: HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>> = 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<ServerMessage> =
|
|
std::collections::VecDeque::new();
|
|
|
|
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 dispatch's v8_lock can claim
|
|
// the only Isolate cleanly.
|
|
let msg = if let Some(d) = deferred.pop_front() {
|
|
d
|
|
} else {
|
|
match rx.recv().await {
|
|
Some(m) => m,
|
|
None => break,
|
|
}
|
|
};
|
|
|
|
match msg {
|
|
ServerMessage::NewConnection { reply_tx } => {
|
|
let _ = reply_tx.send(
|
|
json!({"__init": true})
|
|
.to_string(),
|
|
);
|
|
}
|
|
ServerMessage::Cdp(cdp_msg) => {
|
|
let is_navigation = cdp_msg.text.contains("Page.navigate");
|
|
let has_interception = ctx.fetch_intercept.enabled;
|
|
|
|
if is_navigation && has_interception {
|
|
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;
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
fn handle_fetch_resolution(
|
|
text: &str,
|
|
_ctx: &mut CdpContext,
|
|
reply_tx: &mpsc::UnboundedSender<String>,
|
|
intercepted_paused: &mut HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>>,
|
|
) {
|
|
if let Ok(req) = serde_json::from_str::<CdpRequest>(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 {
|
|
url: None, method: None, headers: None, body: None,
|
|
},
|
|
"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<String>,
|
|
rx: &mut mpsc::UnboundedReceiver<ServerMessage>,
|
|
intercept_rx: &mut Option<mpsc::UnboundedReceiver<obscura_js::ops::InterceptedRequest>>,
|
|
intercepted_paused: &mut HashMap<String, tokio::sync::oneshot::Sender<obscura_js::ops::InterceptResolution>>,
|
|
deferred: &mut std::collections::VecDeque<ServerMessage>,
|
|
) {
|
|
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 = req.params.get("waitUntil")
|
|
.and_then(|v| {
|
|
if let Some(s) = v.as_str() {
|
|
Some(obscura_browser::WaitUntil::from_str(s))
|
|
} else if let Some(arr) = v.as_array() {
|
|
arr.iter()
|
|
.filter_map(|item| item.as_str())
|
|
.map(obscura_browser::WaitUntil::from_str)
|
|
.max_by_key(|w| match w {
|
|
obscura_browser::WaitUntil::DomContentLoaded => 0,
|
|
obscura_browser::WaitUntil::Load => 1,
|
|
obscura_browser::WaitUntil::NetworkIdle2 => 2,
|
|
obscura_browser::WaitUntil::NetworkIdle0 => 3,
|
|
})
|
|
} else {
|
|
None
|
|
}
|
|
})
|
|
.unwrap_or(obscura_browser::WaitUntil::Load);
|
|
|
|
let preload_scripts: Vec<String> = 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();
|
|
|
|
tokio::task::spawn_local(async move {
|
|
// Issue #19: serialize V8 work across pages. The interception path
|
|
// spawns navigation here while the parent task continues to pump
|
|
// CDP messages via `dispatch` (which also acquires this lock); both
|
|
// sides must coordinate or V8 aborts the process at concurrency >= 5.
|
|
let _v8_guard = obscura_js::v8_lock::global().lock().await;
|
|
let result = page.navigate_with_wait(&url_owned, wait_until).await.map_err(|e| e.to_string());
|
|
for source in &preload_scripts {
|
|
if let Err(e) = page.execute_preload_script(source) {
|
|
tracing::debug!("Preload script error: {}", e);
|
|
}
|
|
}
|
|
drop(_v8_guard);
|
|
let _ = nav_done_tx.send((page, result)).await;
|
|
});
|
|
|
|
let navigate_result: Result<(), String>;
|
|
let page_back: Option<obscura_browser::Page>;
|
|
|
|
// 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`.
|
|
//
|
|
// The `obscura_js::v8_lock` mutex 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: {}", &event_str[..event_str.len().min(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::<CdpRequest>(&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");
|
|
|
|
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);
|
|
}
|
|
|
|
let ts = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs_f64();
|
|
let es = session_for_events;
|
|
|
|
for event in [
|
|
crate::types::CdpEvent { method: "Page.lifecycleEvent".into(), params: json!({"frameId": frame_id, "loaderId": loader_id, "name": "init", "timestamp": ts}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Runtime.executionContextsCleared".into(), params: json!({}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Page.frameNavigated".into(), params: json!({"frame": {"id": frame_id, "loaderId": loader_id, "url": page_url, "domainAndRegistry": "", "securityOrigin": page_url, "mimeType": "text/html", "adFrameStatus": {"adFrameType": "none"}}, "type": "Navigation"}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Runtime.executionContextCreated".into(), params: json!({"context": {"id": 2, "origin": page_url, "name": "", "uniqueId": format!("ctx-nav-{}", page_id_for_events), "auxData": {"isDefault": true, "type": "default", "frameId": frame_id}}}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Page.lifecycleEvent".into(), params: json!({"frameId": frame_id, "loaderId": loader_id, "name": "commit", "timestamp": ts}), session_id: es.clone() },
|
|
] {
|
|
if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); }
|
|
}
|
|
|
|
for net_event in &network_events {
|
|
for event in [
|
|
crate::types::CdpEvent { method: "Network.requestWillBeSent".into(), params: json!({"requestId": net_event.request_id, "loaderId": loader_id, "documentURL": page_url, "request": {"url": net_event.url, "method": net_event.method, "headers": net_event.headers}, "timestamp": net_event.timestamp, "wallTime": net_event.timestamp, "initiator": {"type": "other"}, "type": net_event.resource_type, "frameId": frame_id}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Network.responseReceived".into(), params: json!({"requestId": net_event.request_id, "loaderId": loader_id, "timestamp": net_event.timestamp, "type": net_event.resource_type, "response": {"url": net_event.url, "status": net_event.status, "statusText": "", "headers": &*net_event.response_headers, "mimeType": ""}, "frameId": frame_id}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Network.loadingFinished".into(), params: json!({"requestId": net_event.request_id, "timestamp": net_event.timestamp, "encodedDataLength": net_event.body_size}), session_id: es.clone() },
|
|
] {
|
|
if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); }
|
|
}
|
|
}
|
|
|
|
for event in [
|
|
crate::types::CdpEvent { method: "Page.lifecycleEvent".into(), params: json!({"frameId": frame_id, "loaderId": loader_id, "name": "DOMContentLoaded", "timestamp": ts}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Page.domContentEventFired".into(), params: json!({"timestamp": ts}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Page.lifecycleEvent".into(), params: json!({"frameId": frame_id, "loaderId": loader_id, "name": "load", "timestamp": ts}), session_id: es.clone() },
|
|
crate::types::CdpEvent { method: "Page.loadEventFired".into(), params: json!({"timestamp": ts}), session_id: es.clone() },
|
|
] {
|
|
if let Ok(json) = serde_json::to_string(&event) { let _ = reply_tx.send(json); }
|
|
}
|
|
if reached_network_idle {
|
|
let idle_event = crate::types::CdpEvent { method: "Page.lifecycleEvent".into(), params: json!({"frameId": frame_id, "loaderId": loader_id, "name": "networkIdle", "timestamp": ts}), session_id: es.clone() };
|
|
if let Ok(json) = serde_json::to_string(&idle_event) { let _ = reply_tx.send(json); }
|
|
}
|
|
let stop_event = crate::types::CdpEvent { method: "Page.frameStoppedLoading".into(), params: json!({"frameId": frame_id}), session_id: es };
|
|
if let Ok(json) = serde_json::to_string(&stop_event) { let _ = reply_tx.send(json); }
|
|
}
|
|
|
|
async fn process_cdp_message(
|
|
text: &str,
|
|
ctx: &mut CdpContext,
|
|
reply_tx: &mpsc::UnboundedSender<String>,
|
|
) {
|
|
let req: CdpRequest = match serde_json::from_str(text) {
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
warn!("Invalid CDP: {}: {}", e, &text[..text.len().min(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<u8> {
|
|
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<u8> = 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<String> {
|
|
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": "Obscura/0.1.0",
|
|
"revision": "0",
|
|
"userAgent": "Obscura/0.1.0",
|
|
"jsVersion": "V8",
|
|
}))
|
|
}
|
|
"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<String>) -> 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<ServerMessage>,
|
|
) -> anyhow::Result<()> {
|
|
let ws_stream = tokio_tungstenite::accept_async(stream).await?;
|
|
info!("WebSocket connected");
|
|
let (mut ws_sender, mut ws_receiver) = ws_stream.split();
|
|
|
|
let (reply_tx, mut reply_rx) = mpsc::unbounded_channel::<String>();
|
|
|
|
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::<CdpRequest>(&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(())
|
|
}
|