fix: load balancer panics, stealth flag, shadow DOM polyfill
Fixes #1: replace unwrap() panics in run_multi_worker_serve with match on the Result. Fixes #2: route /json round-robin via worker_port instead of hardcoded port+1. Fixes #3: return 502 Bad Gateway when a worker is unreachable instead of silently dropping the connection. Fixes #5: thread --stealth through start_with_options -> CdpContext -> BrowserContext::with_options so the flag actually has an effect in serve mode (was let _ before). Log message now reflects the actual capability based on the stealth cargo feature being compiled in. Fixes #15: try/catch the parentNode assignment in the attachShadow polyfill so it doesn't throw a TypeError against the read-only Node.parentNode getter (Cloudflare Turnstile triggers this path).
This commit is contained in:
1 parent
74cf88d0f4
commit
99e75f1e62
4 files changed
+106
-24
No files matched your search
@@ -31,11 +31,19 @@ pub struct CdpContext {
|
||||
|
||||
impl CdpContext {
|
||||
pub fn new() -> Self {
|
||||
Self::new_with_proxy(None)
|
||||
Self::new_with_options(None, false)
|
||||
}
|
||||
|
||||
pub fn new_with_proxy(proxy: Option<String>) -> Self {
|
||||
let default_context = Arc::new(BrowserContext::with_proxy("default".to_string(), proxy));
|
||||
Self::new_with_options(proxy, false)
|
||||
}
|
||||
|
||||
pub fn new_with_options(proxy: Option<String>, stealth: bool) -> Self {
|
||||
let default_context = Arc::new(BrowserContext::with_options(
|
||||
"default".to_string(),
|
||||
proxy,
|
||||
stealth,
|
||||
));
|
||||
CdpContext {
|
||||
pages: Vec::new(),
|
||||
sessions: HashMap::new(),
|
||||
|
||||
@@ -24,10 +24,14 @@ enum ServerMessage {
|
||||
}
|
||||
|
||||
pub async fn start(port: u16) -> anyhow::Result<()> {
|
||||
start_with_options(port, None).await
|
||||
start_with_options(port, None, false).await
|
||||
}
|
||||
|
||||
pub async fn start_with_options(port: u16, proxy: Option<String>) -> anyhow::Result<()> {
|
||||
pub async fn start_with_options(
|
||||
port: u16,
|
||||
proxy: Option<String>,
|
||||
stealth: bool,
|
||||
) -> anyhow::Result<()> {
|
||||
let addr = SocketAddr::from(([127, 0, 0, 1], port));
|
||||
let listener = TcpListener::bind(&addr).await?;
|
||||
|
||||
@@ -42,7 +46,7 @@ pub async fn start_with_options(port: u16, proxy: Option<String>) -> anyhow::Res
|
||||
.run_until(async {
|
||||
let (msg_tx, msg_rx) = mpsc::unbounded_channel::<ServerMessage>();
|
||||
|
||||
let processor_handle = tokio::task::spawn_local(cdp_processor(msg_rx, proxy));
|
||||
let processor_handle = tokio::task::spawn_local(cdp_processor(msg_rx, proxy, stealth));
|
||||
|
||||
loop {
|
||||
match listener.accept().await {
|
||||
@@ -67,8 +71,9 @@ pub async fn start_with_options(port: u16, proxy: Option<String>) -> anyhow::Res
|
||||
async fn cdp_processor(
|
||||
mut rx: mpsc::UnboundedReceiver<ServerMessage>,
|
||||
proxy: Option<String>,
|
||||
stealth: bool,
|
||||
) {
|
||||
let mut ctx = CdpContext::new_with_proxy(proxy);
|
||||
let mut ctx = CdpContext::new_with_options(proxy, stealth);
|
||||
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);
|
||||
@@ -632,4 +637,4 @@ async fn handle_http_json(stream: TcpStream, port: u16, endpoint: &str) -> anyho
|
||||
stream.write_all(resp.as_bytes()).await?;
|
||||
stream.flush().await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -139,15 +139,19 @@ async fn main() -> anyhow::Result<()> {
|
||||
tracing::info!("User-Agent: {}", ua);
|
||||
}
|
||||
if stealth {
|
||||
tracing::info!("Stealth mode enabled (TLS fingerprint spoofing)");
|
||||
#[cfg(feature = "stealth")]
|
||||
tracing::info!(
|
||||
"Stealth mode enabled (TLS fingerprint impersonation + tracker blocking)"
|
||||
);
|
||||
#[cfg(not(feature = "stealth"))]
|
||||
tracing::info!("Stealth mode enabled (tracker blocking)");
|
||||
}
|
||||
let _ = stealth;
|
||||
|
||||
if workers > 1 {
|
||||
tracing::info!("{} worker processes", workers);
|
||||
run_multi_worker_serve(port, workers, proxy, stealth).await?;
|
||||
} else {
|
||||
obscura_cdp::start_with_options(port, proxy).await?;
|
||||
obscura_cdp::start_with_options(port, proxy, stealth).await?;
|
||||
}
|
||||
}
|
||||
Some(Command::Fetch { url, dump, selector, wait, wait_until, user_agent, stealth, eval, quiet }) => {
|
||||
@@ -161,7 +165,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
if let Some(ref proxy) = args.proxy {
|
||||
tracing::info!("Using proxy: {}", proxy);
|
||||
}
|
||||
obscura_cdp::start_with_options(args.port, args.proxy).await?;
|
||||
obscura_cdp::start_with_options(args.port, args.proxy, false).await?;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -222,11 +226,49 @@ async fn run_multi_worker_serve(
|
||||
let request_line = String::from_utf8_lossy(&full_peek[..n]);
|
||||
|
||||
if request_line.contains("/json") {
|
||||
let worker_addr = format!("127.0.0.1:{}", port + 1);
|
||||
if let Ok(mut worker_stream) = tokio::net::TcpStream::connect(&worker_addr).await {
|
||||
tokio::spawn(async move {
|
||||
let _ = tokio::io::copy_bidirectional(&mut tokio::net::TcpStream::from_std(client_stream.into_std().unwrap()).unwrap(), &mut worker_stream).await;
|
||||
});
|
||||
let worker_addr = format!("127.0.0.1:{}", worker_port);
|
||||
match tokio::net::TcpStream::connect(&worker_addr).await {
|
||||
Ok(mut worker_stream) => {
|
||||
tokio::spawn(async move {
|
||||
let std_stream = match client_stream.into_std() {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
"/json: failed to convert client to std stream: {}",
|
||||
e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let mut client = match tokio::net::TcpStream::from_std(std_stream) {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
tracing::error!(
|
||||
"/json: failed to recreate tokio TcpStream: {}",
|
||||
e
|
||||
);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let _ = tokio::io::copy_bidirectional(
|
||||
&mut client,
|
||||
&mut worker_stream,
|
||||
)
|
||||
.await;
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("/json worker {} unreachable: {}", worker_addr, e);
|
||||
tokio::spawn(async move {
|
||||
let mut s = client_stream;
|
||||
let _ = s
|
||||
.write_all(
|
||||
b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n",
|
||||
)
|
||||
.await;
|
||||
let _ = s.shutdown().await;
|
||||
});
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
@@ -234,9 +276,20 @@ async fn run_multi_worker_serve(
|
||||
|
||||
let worker_addr = format!("127.0.0.1:{}", worker_port);
|
||||
tokio::spawn(async move {
|
||||
if let Ok(mut worker_stream) = tokio::net::TcpStream::connect(&worker_addr).await {
|
||||
let mut client = client_stream;
|
||||
let _ = tokio::io::copy_bidirectional(&mut client, &mut worker_stream).await;
|
||||
match tokio::net::TcpStream::connect(&worker_addr).await {
|
||||
Ok(mut worker_stream) => {
|
||||
let mut client = client_stream;
|
||||
let _ =
|
||||
tokio::io::copy_bidirectional(&mut client, &mut worker_stream).await;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("worker {} unreachable: {}", worker_addr, e);
|
||||
let mut s = client_stream;
|
||||
let _ = s
|
||||
.write_all(b"HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n")
|
||||
.await;
|
||||
let _ = s.shutdown().await;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -654,4 +707,4 @@ fn dump_links(page: &Page) {
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
Vendored
+20
-4
@@ -2382,17 +2382,33 @@ Element.prototype.attachShadow = function attachShadow(opts) {
|
||||
get lastChild() { return children[children.length - 1] || null; },
|
||||
get firstElementChild() { return children.find(c => c.nodeType === 1) || null; },
|
||||
get children() { return children.filter(c => c.nodeType === 1); },
|
||||
appendChild(c) { if (c) { children.push(c); c.parentNode = shadow; } return c; },
|
||||
appendChild(c) {
|
||||
if (c) {
|
||||
children.push(c);
|
||||
try { c.parentNode = shadow; } catch (_) { /* parentNode is getter-only on Node, ignore */ }
|
||||
}
|
||||
return c;
|
||||
},
|
||||
insertBefore(n, ref) {
|
||||
if (!n) return n;
|
||||
if (!ref) { shadow.appendChild(n); return n; }
|
||||
const idx = children.indexOf(ref);
|
||||
if (idx >= 0) { children.splice(idx, 0, n); n.parentNode = shadow; }
|
||||
if (idx >= 0) {
|
||||
children.splice(idx, 0, n);
|
||||
try { n.parentNode = shadow; } catch (_) {}
|
||||
}
|
||||
else shadow.appendChild(n);
|
||||
return n;
|
||||
},
|
||||
removeChild(c) { const idx = children.indexOf(c); if (idx >= 0) children.splice(idx, 1); return c; },
|
||||
replaceChild(n, o) { const idx = children.indexOf(o); if (idx >= 0) { children[idx] = n; n.parentNode = shadow; } return o; },
|
||||
replaceChild(n, o) {
|
||||
const idx = children.indexOf(o);
|
||||
if (idx >= 0) {
|
||||
children[idx] = n;
|
||||
try { n.parentNode = shadow; } catch (_) {}
|
||||
}
|
||||
return o;
|
||||
},
|
||||
querySelector(s) {
|
||||
for (const c of children) {
|
||||
if (c.matches && c.matches(s)) return c;
|
||||
@@ -2922,4 +2938,4 @@ globalThis.__obscura_init = function() {
|
||||
}
|
||||
}
|
||||
delete globalThis.__obscura_init;
|
||||
};
|
||||
};
|
||||
Reference in new issue
Block a user