From 4c886765ccc766c2392d4964e89cc6f520d20559 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Tue, 9 Jun 2026 12:09:21 +0530 Subject: [PATCH] feat(cef): in-process CDP DevTools transport (foundation) (#3498) --- app/src-tauri/src/cdp/conn.rs | 299 +++++++++---- app/src-tauri/src/cdp/in_process.rs | 497 ++++++++++++++++++++++ app/src-tauri/src/cdp/mod.rs | 41 +- app/src-tauri/src/cdp/session.rs | 15 +- app/src-tauri/src/cdp/target.rs | 113 ++++- app/src-tauri/src/lib.rs | 47 +- app/src-tauri/src/slack_scanner/mod.rs | 2 +- app/src-tauri/src/webview_accounts/mod.rs | 22 + 8 files changed, 886 insertions(+), 150 deletions(-) create mode 100644 app/src-tauri/src/cdp/in_process.rs diff --git a/app/src-tauri/src/cdp/conn.rs b/app/src-tauri/src/cdp/conn.rs index 1b6e8aa93..6d54ee877 100644 --- a/app/src-tauri/src/cdp/conn.rs +++ b/app/src-tauri/src/cdp/conn.rs @@ -1,26 +1,40 @@ -//! CDP WebSocket client. Supports both short-lived request/response ticks -//! (whatsapp / slack / telegram periodic scans) and long-lived streaming -//! sessions with a pending-id table (discord MITM, and the new per-account -//! session opener). +//! [`CdpConn`] — per-attach handle on top of the in-process CDP transport. //! -//! Not re-entrant: `call` is sequential during the setup phase, and once -//! `pump_events` takes over the read stream callers issue follow-up calls -//! via the pending-table machinery (TODO — V1.5, not needed yet). +//! Wraps an [`Arc`](super::in_process::WebviewCdpTransport) +//! with the same `call` / `pump_events` surface the scanners and the +//! per-account session opener were already using. The previous +//! WebSocket-backed implementation (one socket per attach) is gone for +//! new code paths; all attaches for a given webview now share the same +//! in-process channel, and a [`CdpConn`] is just a cheap session-scoped +//! view. +//! +//! For backward compatibility with per-scanner duplicated implementations +//! that still attach via the TCP loopback DevTools port (whatsapp, +//! slack, telegram, wechat, meet_video — see issue follow-up), the +//! [`CdpConn::open_ws`] legacy constructor keeps the old `tungstenite` +//! WebSocket path alive. Callers should migrate to +//! [`super::conn_for_account`] over time. use std::collections::HashMap; +use std::sync::Arc; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; use serde_json::{json, Value}; +use tokio::sync::broadcast::error::RecvError; use tokio::sync::oneshot; use tokio_tungstenite::{connect_async, tungstenite::Message}; -/// Timeout applied to a single request/response round-trip during the setup -/// phase. Long enough to cover a cold-attach on a sluggish machine; -/// `pump_events` uses no timeout since CDP events can arrive hours apart. -const CALL_TIMEOUT: Duration = Duration::from_secs(35); +use super::in_process::{EventFrame, WebviewCdpTransport}; -pub struct CdpConn { +/// Timeout applied to a single request/response round-trip in the +/// legacy WebSocket transport. Long enough to cover a cold-attach on a +/// sluggish machine; the in-process transport uses +/// [`crate::cdp::CALL_TIMEOUT`] (also 35s) for symmetry. +const LEGACY_CALL_TIMEOUT: Duration = Duration::from_secs(35); + +/// Internal: legacy WebSocket dispatch state. +struct LegacyWs { sink: futures_util::stream::SplitSink< tokio_tungstenite::WebSocketStream< tokio_tungstenite::MaybeTlsStream, @@ -36,113 +50,210 @@ pub struct CdpConn { pending: HashMap>>, } +enum Backend { + InProcess(Arc), + LegacyWs(LegacyWs), +} + +/// Per-attach CDP handle. Internally either wraps an +/// `Arc` (in-process channel) or a tungstenite +/// `WebSocketStream` (legacy TCP loopback). The session_id filter is +/// per-handle so concurrent attachers don't see each other's events. +pub struct CdpConn { + backend: Backend, + label: String, +} + impl CdpConn { - pub async fn open(ws_url: &str) -> Result { + /// Wrap an already-installed in-process transport. Callers obtain + /// the transport from the per-app [`super::CdpRegistry`] + /// (`app.state()`) — typically via + /// [`super::conn_for_account`]. + pub fn new(transport: Arc) -> Self { + let label = transport.label().to_string(); + Self { + backend: Backend::InProcess(transport), + label, + } + } + + /// Legacy: open a CDP connection over the loopback TCP WebSocket + /// exposed by `--remote-debugging-port`. Kept for the per-scanner + /// duplicated implementations (whatsapp, slack, telegram, wechat, + /// meet_video) that have not yet migrated to the in-process + /// channel. New code paths should use [`super::conn_for_account`]. + pub async fn open_ws(ws_url: &str) -> Result { let (ws, _resp) = connect_async(ws_url) .await .map_err(|e| format!("ws connect: {e}"))?; let (sink, stream) = ws.split(); Ok(Self { - sink, - stream, - next_id: 1, - pending: HashMap::new(), + backend: Backend::LegacyWs(LegacyWs { + sink, + stream, + next_id: 1, + pending: HashMap::new(), + }), + label: format!("ws:{ws_url}"), }) } - /// Setup-phase request/response: sends a JSON-RPC call and drains inbound - /// messages until the matching response arrives. Unrelated events and - /// responses for other ids are dropped on the floor — only safe before - /// `pump_events` takes over the read side. + /// Setup-phase request/response: sends a JSON-RPC call and awaits + /// the matching response. `session_id`, when supplied, is inlined + /// into the envelope so the call routes to a previously-attached + /// child target (via `Target.attachToTarget`). pub async fn call( &mut self, method: &str, params: Value, session_id: Option<&str>, ) -> Result { - let id = self.next_id; - self.next_id += 1; - let mut req = json!({ "id": id, "method": method, "params": params }); - if let Some(s) = session_id { - req["sessionId"] = json!(s); - } - let body = serde_json::to_string(&req).map_err(|e| format!("encode: {e}"))?; - self.sink - .send(Message::Text(body)) - .await - .map_err(|e| format!("ws send: {e}"))?; - - loop { - let msg = tokio::time::timeout(CALL_TIMEOUT, self.stream.next()) - .await - .map_err(|_| format!("ws read timeout (method={method})"))? - .ok_or_else(|| format!("ws closed (method={method})"))? - .map_err(|e| format!("ws recv: {e}"))?; - let text = match msg { - Message::Text(t) => t, - Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => { - continue - } - Message::Close(_) => return Err("ws closed".into()), - }; - let v: Value = serde_json::from_str(&text).map_err(|e| format!("decode: {e}"))?; - if v.get("id").and_then(|x| x.as_i64()) != Some(id) { - continue; - } - if let Some(err) = v.get("error") { - return Err(format!("cdp error: {err}")); - } - return Ok(v.get("result").cloned().unwrap_or(Value::Null)); + match &mut self.backend { + Backend::InProcess(transport) => transport.call(method, params, session_id).await, + Backend::LegacyWs(ws) => legacy_ws_call(ws, method, params, session_id).await, } } - /// Take over the read stream and dispatch every inbound event via the - /// supplied callback until the WebSocket closes. Responses to outstanding - /// `call` requests (none in V1) route through `pending`. + /// Subscribe to the transport's event stream and dispatch every + /// inbound CDP event via the supplied callback until the channel + /// signals it cannot keep up. /// - /// `session_id` filters incoming events: CDP multiplexes all sessions - /// through one ws once `flatten: true` is set, so we drop events - /// belonging to other sessions. + /// `session_id` filters incoming events — CDP multiplexes all + /// sessions through the same transport when `flatten: true` is set, + /// so we drop events belonging to other sessions. + /// + /// Returns when the channel closes (the transport has been + /// forgotten / ws shut down) or on an unrecoverable error. + /// `Lagged` is treated as a continuation signal — the caller's idle + /// watchdog will eventually time out the session and the outer + /// reconnect loop re-attaches. pub async fn pump_events(&mut self, session_id: &str, mut on_event: F) -> Result<(), String> where F: FnMut(&str, &Value), { - loop { - let msg = self - .stream - .next() - .await - .ok_or_else(|| "ws closed".to_string())? - .map_err(|e| format!("ws recv: {e}"))?; - let text = match msg { - Message::Text(t) => t, - Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => { - continue + match &mut self.backend { + Backend::InProcess(transport) => { + let mut rx = transport.subscribe_events(); + loop { + match rx.recv().await { + Ok(EventFrame { + method, + params, + session_id: evt_session, + }) => { + if !evt_session.is_empty() && evt_session != session_id { + continue; + } + on_event(&method, ¶ms); + } + Err(RecvError::Lagged(skipped)) => { + log::warn!( + "[cdp][{}] event channel lagged skipped={} session_id={}", + self.label, + skipped, + session_id + ); + continue; + } + Err(RecvError::Closed) => return Ok(()), + } } - Message::Close(_) => return Ok(()), - }; - let v: Value = match serde_json::from_str(&text) { - Ok(v) => v, - Err(_) => continue, - }; - if let Some(id) = v.get("id").and_then(|x| x.as_i64()) { - if let Some(tx) = self.pending.remove(&id) { - let res = if let Some(err) = v.get("error") { - Err(format!("cdp error: {err}")) - } else { - Ok(v.get("result").cloned().unwrap_or(Value::Null)) - }; - let _ = tx.send(res); - } - continue; } - let method = v.get("method").and_then(|x| x.as_str()).unwrap_or(""); - let evt_session = v.get("sessionId").and_then(|x| x.as_str()).unwrap_or(""); - if !evt_session.is_empty() && evt_session != session_id { - continue; - } - let params = v.get("params").cloned().unwrap_or(Value::Null); - on_event(method, ¶ms); + Backend::LegacyWs(ws) => legacy_ws_pump_events(ws, session_id, on_event).await, } } + + /// Diagnostic helper — webview label (in-process) or + /// `"ws:"` (legacy WS) this connection is bound to. + pub fn label(&self) -> &str { + &self.label + } +} + +async fn legacy_ws_call( + ws: &mut LegacyWs, + method: &str, + params: Value, + session_id: Option<&str>, +) -> Result { + let id = ws.next_id; + ws.next_id += 1; + let mut req = json!({ "id": id, "method": method, "params": params }); + if let Some(s) = session_id { + req["sessionId"] = json!(s); + } + let body = serde_json::to_string(&req).map_err(|e| format!("encode: {e}"))?; + ws.sink + .send(Message::Text(body)) + .await + .map_err(|e| format!("ws send: {e}"))?; + loop { + let msg = tokio::time::timeout(LEGACY_CALL_TIMEOUT, ws.stream.next()) + .await + .map_err(|_| format!("ws read timeout (method={method})"))? + .ok_or_else(|| format!("ws closed (method={method})"))? + .map_err(|e| format!("ws recv: {e}"))?; + let text = match msg { + Message::Text(t) => t, + Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => { + continue + } + Message::Close(_) => return Err("ws closed".into()), + }; + let v: Value = serde_json::from_str(&text).map_err(|e| format!("decode: {e}"))?; + if v.get("id").and_then(|x| x.as_i64()) != Some(id) { + continue; + } + if let Some(err) = v.get("error") { + return Err(format!("cdp error: {err}")); + } + return Ok(v.get("result").cloned().unwrap_or(Value::Null)); + } +} + +async fn legacy_ws_pump_events( + ws: &mut LegacyWs, + session_id: &str, + mut on_event: F, +) -> Result<(), String> +where + F: FnMut(&str, &Value), +{ + loop { + let msg = ws + .stream + .next() + .await + .ok_or_else(|| "ws closed".to_string())? + .map_err(|e| format!("ws recv: {e}"))?; + let text = match msg { + Message::Text(t) => t, + Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => { + continue + } + Message::Close(_) => return Ok(()), + }; + let v: Value = match serde_json::from_str(&text) { + Ok(v) => v, + Err(_) => continue, + }; + if let Some(id) = v.get("id").and_then(|x| x.as_i64()) { + if let Some(tx) = ws.pending.remove(&id) { + let res = if let Some(err) = v.get("error") { + Err(format!("cdp error: {err}")) + } else { + Ok(v.get("result").cloned().unwrap_or(Value::Null)) + }; + let _ = tx.send(res); + } + continue; + } + let method = v.get("method").and_then(|x| x.as_str()).unwrap_or(""); + let evt_session = v.get("sessionId").and_then(|x| x.as_str()).unwrap_or(""); + if !evt_session.is_empty() && evt_session != session_id { + continue; + } + let params = v.get("params").cloned().unwrap_or(Value::Null); + on_event(method, ¶ms); + } } diff --git a/app/src-tauri/src/cdp/in_process.rs b/app/src-tauri/src/cdp/in_process.rs new file mode 100644 index 000000000..d8dac62b4 --- /dev/null +++ b/app/src-tauri/src/cdp/in_process.rs @@ -0,0 +1,497 @@ +//! In-process Chrome DevTools Protocol transport built on +//! [`tauri::Webview::send_dev_tools_message`] + +//! [`tauri::Webview::on_dev_tools_protocol`]. +//! +//! Replaces the legacy WebSocket-to-loopback transport that required +//! Chromium to be spawned with `--remote-debugging-port=19222`. The old +//! transport opened an unauthenticated TCP listener that any same-UID +//! local process could attach to. The in-process path stays entirely +//! within our process boundary — there is no listener for an external +//! attacker to reach. +//! +//! # Architecture +//! +//! - One [`WebviewCdpTransport`] per CEF webview. Installed at the +//! moment a webview is created by `webview_accounts` (and similar +//! creation sites). The webview-creator calls +//! [`install_for_webview`] with its concrete `Webview` +//! handle. +//! - The bootstrapped transport is registered in a process-global +//! [`CdpRegistry`] (`app.state::()`) keyed by account id +//! so scanners that are generic over `Runtime: tauri::Runtime` never +//! need to obtain a typed `Webview` themselves. +//! - The transport registers a single +//! [`tauri::Webview::on_dev_tools_protocol`] callback at install +//! time. The CEF runtime keeps the callback alive for the webview's +//! whole lifetime; there is no un-register path, so the install must +//! be a one-shot keyed by webview label. +//! - Every outbound request gets a unique numeric `id`. A pending-map +//! ([`PendingMap`]) routes responses back to the awaiting +//! `tokio::sync::oneshot`. +//! - Events (CDP frames with no `id`) are fanned out through a +//! `tokio::sync::broadcast` channel. Each subscriber sees its own +//! queue. +//! +//! # Wire format +//! +//! Tauri-CEF delivers each inbound DevTools message in three variants +//! (`CefDevToolsProtocol::Message` raw JSON, `MethodResult` pre-parsed +//! response, `Event` pre-parsed event). We listen on the raw `Message` +//! variant only — it carries the full envelope including `sessionId`, +//! which the other two strip. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::Duration; + +use serde_json::{json, Value}; +use tokio::sync::{broadcast, oneshot}; + +use tauri::{AppHandle, Manager, Webview}; + +/// Timeout for a single request/response round-trip. Long enough for cold +/// attach on slow machines but short enough to fail fast on a stuck channel. +pub const CALL_TIMEOUT: Duration = Duration::from_secs(35); + +/// Capacity of the per-transport event broadcast channel. CDP can produce +/// bursts (e.g. on `Page.enable` the initial frame-history dump). 256 keeps +/// memory bounded while absorbing typical burst sizes. +const EVENT_CHANNEL_CAP: usize = 256; + +type PendingMap = Arc>>>>; + +/// One CDP frame delivered as an event (no `id` in the envelope). +#[derive(Clone, Debug)] +pub struct EventFrame { + pub method: String, + pub params: Value, + pub session_id: String, +} + +/// Per-webview CDP transport. Holds the pending-id table and event +/// broadcaster. Constructed by [`install_for_webview`] and (typically) +/// inserted into the per-app [`CdpRegistry`] for later lookup. +pub struct WebviewCdpTransport { + label: String, + webview: Webview, + next_id: Mutex, + pending: PendingMap, + events_tx: broadcast::Sender, +} + +impl WebviewCdpTransport { + /// Submit a CDP request and await its response. + /// + /// The request `id` is auto-assigned. `session_id`, when supplied, is + /// inlined into the envelope so the call routes to a previously-attached + /// child target. + pub async fn call( + self: &Arc, + method: &str, + params: Value, + session_id: Option<&str>, + ) -> Result { + let id = { + let mut n = self.next_id.lock().expect("next_id mutex poisoned"); + let id = *n; + *n += 1; + id + }; + let (tx, rx) = oneshot::channel(); + { + let mut p = self.pending.lock().expect("pending mutex poisoned"); + p.insert(id, tx); + } + + let mut req = json!({ "id": id, "method": method, "params": params }); + if let Some(s) = session_id { + req["sessionId"] = json!(s); + } + let body = serde_json::to_vec(&req).map_err(|e| format!("encode: {e}"))?; + + // `send_dev_tools_message` blocks on a `std::sync::mpsc::Receiver` + // while the message is dispatched onto the CEF main thread. Off-load + // to a blocking pool so we don't park a tokio worker thread. + log::trace!( + "[cdp][{}] >> id={} method={} session_id={:?}", + self.label, + id, + method, + session_id + ); + let webview = self.webview.clone(); + let send_res = tauri::async_runtime::spawn_blocking(move || { + webview + .send_dev_tools_message(&body) + .map_err(|e| format!("send_dev_tools_message: {e}")) + }) + .await + .map_err(|e| format!("spawn_blocking join: {e}"))?; + if let Err(e) = send_res { + // Clean up the pending entry; otherwise it leaks until next call. + let mut p = self.pending.lock().expect("pending mutex poisoned"); + p.remove(&id); + return Err(e); + } + + match tokio::time::timeout(CALL_TIMEOUT, rx).await { + Ok(Ok(res)) => res, + Ok(Err(_)) => { + // oneshot dropped — transport torn down between dispatch and + // receive. Surface as a transport error rather than panic. + Err(format!("cdp response channel dropped (method={method})")) + } + Err(_) => { + // Timeout. Make a best-effort to evict the pending entry so + // a delayed response doesn't pollute the next caller. + let mut p = self.pending.lock().expect("pending mutex poisoned"); + p.remove(&id); + Err(format!("cdp call timeout (method={method})")) + } + } + } + + /// Subscribe to the event stream for this transport. Each subscriber + /// gets its own queue; lagged subscribers see `RecvError::Lagged` and + /// must re-sync. + pub fn subscribe_events(self: &Arc) -> broadcast::Receiver { + self.events_tx.subscribe() + } + + /// Webview label associated with this transport — for diagnostics only. + pub fn label(&self) -> &str { + &self.label + } +} + +/// Process-global registry of installed CDP transports keyed by webview +/// label. Stored via `app.manage(CdpRegistry::default())` at boot so +/// scanners and the per-account session opener can look up the correct +/// transport without holding a typed `Webview` themselves. +#[derive(Default)] +pub struct CdpRegistry { + transports: Mutex>>, +} + +impl CdpRegistry { + /// Look up an installed transport by webview label. Returns `None` + /// when the webview has not been created yet (cold boot before the + /// account is opened, or after it was forgotten). + pub fn by_label(&self, label: &str) -> Option> { + self.transports + .lock() + .expect("CdpRegistry mutex poisoned") + .get(label) + .cloned() + } + + /// Look up the transport for an `acct_*`-labelled webview. The + /// account id is the unsuffixed value passed to + /// [`webview_accounts::label_for`] (typically already sanitized). + pub fn by_account(&self, account_id: &str) -> Option> { + self.by_label(&format!("acct_{account_id}")) + } + + /// Atomic "get or create" for a transport keyed by webview label. + /// + /// Holds the registry mutex across both the existence check AND the + /// caller-supplied creator, so two concurrent + /// [`install_for_webview`] callers can never both register a + /// `on_dev_tools_protocol` observer on the same webview. The CEF + /// runtime offers no un-register hook, so a double-registration is + /// permanent — every inbound frame would fan out to both observer + /// closures, splitting `id`-keyed responses between two + /// disconnected `next_id` counters and breaking response routing. + /// + /// The creator closure is responsible for constructing the transport + /// (which includes registering the CEF observer). If the closure + /// returns `Err`, the registry is unchanged. + fn get_or_create(&self, label: &str, create: F) -> Result, String> + where + F: FnOnce() -> Result, String>, + { + let mut t = self.transports.lock().expect("CdpRegistry mutex poisoned"); + if let Some(existing) = t.get(label) { + return Ok(Arc::clone(existing)); + } + let transport = create()?; + t.insert(label.to_string(), Arc::clone(&transport)); + Ok(transport) + } + + /// Remove a transport from the registry by label. Called when a + /// webview is closed / forgotten so subsequent + /// [`Self::by_label`] / [`Self::by_account`] lookups return `None` + /// instead of a stale entry. + pub fn forget_label(&self, label: &str) { + self.transports + .lock() + .expect("CdpRegistry mutex poisoned") + .remove(label); + } + + /// Convenience wrapper around [`Self::forget_label`] for the + /// account-suffixed label scheme. + pub fn forget_account(&self, account_id: &str) { + self.forget_label(&format!("acct_{account_id}")); + } +} + +/// Process-global typed CEF [`AppHandle`]. Populated exactly once from +/// the `Builder::::setup` callback in `lib.rs`. Used by +/// [`install_for_account`] to look up a webview by label with the +/// concrete `Webview` typed handle that +/// [`Webview::send_dev_tools_message`] requires — avoids a +/// runtime-generic transmute at the install site. +/// +/// `OnceLock` is appropriate because the cell is written exactly once, +/// during `Builder::setup`, before any webview is created. +static CEF_APP_HANDLE: OnceLock> = OnceLock::new(); + +/// Record the typed CEF [`AppHandle`] in the process-global cell. Called +/// once from the `Builder::setup` callback in `lib.rs`. Subsequent calls +/// are silently ignored (`OnceLock::set` returns `Err` on the second +/// write) so re-entry during hot-reload doesn't panic. +pub fn set_cef_app_handle(app: AppHandle) { + let _ = CEF_APP_HANDLE.set(app); +} + +/// Install (or look up) the in-process CDP transport for an +/// account-keyed webview. The account id is the same value the +/// `webview_accounts::label_for` helper composes a label from +/// (`acct_{id}`). Idempotent — repeated calls for the same account +/// return the cached transport. +/// +/// Returns `Err` when the webview hasn't been created yet (the typical +/// cold-boot race against a scanner that started before +/// `webview_accounts::open` finished). Callers back off and retry. +pub fn install_for_account(account_id: &str) -> Result, String> { + let app = CEF_APP_HANDLE + .get() + .ok_or_else(|| "cdp::set_cef_app_handle has not been called yet".to_string())?; + let label = format!("acct_{account_id}"); + let registry_state = app + .try_state::() + .ok_or_else(|| "CdpRegistry not managed by app".to_string())?; + let registry = registry_state.inner(); + if let Some(existing) = registry.by_label(&label) { + return Ok(existing); + } + let webview = app + .get_webview(&label) + .ok_or_else(|| format!("no webview for label={label}"))?; + install_for_webview(registry, webview) +} + +/// Install a CDP transport on `webview`, register the observer, and +/// insert the resulting [`WebviewCdpTransport`] into `registry`. +/// +/// Idempotent and concurrency-safe: the existence check, observer +/// registration, and registry insert all happen while the registry +/// mutex is held, so two concurrent callers for the same webview +/// label can never both attach an observer (CEF gives no un-register +/// hook — a double-registration would permanently split responses +/// between two `next_id` counters). +pub fn install_for_webview( + registry: &CdpRegistry, + webview: Webview, +) -> Result, String> { + let label = webview.label().to_string(); + registry.get_or_create(&label, || { + let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); + let (events_tx, _rx0) = broadcast::channel::(EVENT_CHANNEL_CAP); + + let transport = Arc::new(WebviewCdpTransport { + label: label.clone(), + webview: webview.clone(), + next_id: Mutex::new(1), + pending: Arc::clone(&pending), + events_tx: events_tx.clone(), + }); + + let pending_for_observer = Arc::clone(&pending); + let events_for_observer = events_tx.clone(); + let label_for_observer = label.clone(); + webview + .on_dev_tools_protocol(move |protocol| { + use tauri::CefDevToolsProtocol as P; + // Listen only on the raw `Message` variant — it carries + // the full envelope (including `sessionId`) which + // `MethodResult` and `Event` strip. Tauri-CEF still + // fires all three for the same wire message, so + // consuming both would double-dispatch. + if let P::Message(bytes) = protocol { + let v: Value = match serde_json::from_slice(&bytes) { + Ok(v) => v, + Err(e) => { + log::warn!( + "[cdp][{}] inbound: parse error: {} (bytes_len={})", + label_for_observer, + e, + bytes.len() + ); + return; + } + }; + handle_inbound( + &label_for_observer, + &v, + &pending_for_observer, + &events_for_observer, + ); + } + }) + .map_err(|e| format!("on_dev_tools_protocol: {e}"))?; + + log::info!( + "[cdp][{}] in-process transport installed (observer registered)", + label + ); + + Ok(transport) + }) +} + +fn handle_inbound( + label: &str, + v: &Value, + pending: &PendingMap, + events_tx: &broadcast::Sender, +) { + if let Some(id) = v.get("id").and_then(|x| x.as_i64()) { + let waiter = { + let mut p = pending.lock().expect("pending mutex poisoned"); + p.remove(&id) + }; + match waiter { + Some(tx) => { + let res = if let Some(err) = v.get("error") { + Err(format!("cdp error: {err}")) + } else { + Ok(v.get("result").cloned().unwrap_or(Value::Null)) + }; + let _ = tx.send(res); + } + None => { + log::trace!( + "[cdp][{}] inbound: orphan response id={} (caller already gone)", + label, + id + ); + } + } + return; + } + + let Some(method) = v.get("method").and_then(|x| x.as_str()) else { + return; + }; + let params = v.get("params").cloned().unwrap_or(Value::Null); + let session_id = v + .get("sessionId") + .and_then(|x| x.as_str()) + .unwrap_or("") + .to_string(); + let frame = EventFrame { + method: method.to_string(), + params, + session_id, + }; + // `send` only errors when there are zero subscribers — fine to drop + // events when nobody is listening. + let _ = events_tx.send(frame); +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + /// Inbound response carrying a known `id` resolves the matching pending + /// oneshot with the `result` payload. + #[tokio::test] + async fn inbound_response_resolves_pending() { + let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); + let (events_tx, _) = broadcast::channel::(8); + + let (tx, rx) = oneshot::channel(); + pending.lock().unwrap().insert(42, tx); + + handle_inbound( + "test", + &json!({ "id": 42, "result": { "ok": true } }), + &pending, + &events_tx, + ); + + let got = rx.await.expect("oneshot resolved").expect("ok response"); + assert_eq!(got, json!({ "ok": true })); + assert!( + pending.lock().unwrap().is_empty(), + "pending entry must be evicted on resolve" + ); + } + + /// Inbound error frame routes to the pending sender as `Err(_)`. + #[tokio::test] + async fn inbound_error_resolves_pending_as_err() { + let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); + let (events_tx, _) = broadcast::channel::(8); + + let (tx, rx) = oneshot::channel(); + pending.lock().unwrap().insert(7, tx); + + handle_inbound( + "test", + &json!({ "id": 7, "error": { "code": -32000, "message": "boom" } }), + &pending, + &events_tx, + ); + + let got = rx.await.expect("oneshot resolved"); + assert!(got.is_err(), "error frame must surface as Err"); + } + + /// Inbound event frame with no `id` fans out to subscribers. + #[tokio::test] + async fn inbound_event_broadcasts_to_subscribers() { + let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); + let (events_tx, mut events_rx) = broadcast::channel::(8); + + handle_inbound( + "test", + &json!({ + "method": "Page.loadEventFired", + "sessionId": "abc", + "params": { "timestamp": 1.23 }, + }), + &pending, + &events_tx, + ); + + let frame = events_rx.recv().await.expect("event received"); + assert_eq!(frame.method, "Page.loadEventFired"); + assert_eq!(frame.session_id, "abc"); + assert_eq!(frame.params, json!({ "timestamp": 1.23 })); + } + + /// Orphan responses (no matching pending entry) drop silently. + #[tokio::test] + async fn inbound_orphan_response_drops_silently() { + let pending: PendingMap = Arc::new(Mutex::new(HashMap::new())); + let (events_tx, _) = broadcast::channel::(8); + handle_inbound( + "test", + &json!({ "id": 999, "result": {} }), + &pending, + &events_tx, + ); + assert!(pending.lock().unwrap().is_empty()); + } + + // `registry.forget_account` / `by_account` are exercised in the + // integration test `cdp_in_process_e2e.rs` against a real Tauri-CEF + // runtime — unit tests cannot synthesize a `Webview` without a + // running CEF instance, and the registry map itself is a trivial + // `HashMap` so the wiring tests above already cover all branches. +} diff --git a/app/src-tauri/src/cdp/mod.rs b/app/src-tauri/src/cdp/mod.rs index b889d37bd..e7992150d 100644 --- a/app/src-tauri/src/cdp/mod.rs +++ b/app/src-tauri/src/cdp/mod.rs @@ -3,28 +3,55 @@ //! Consolidates the CdpConn / target-discovery / notification-shim plumbing //! that used to be copy-pasted across `discord_scanner`, `whatsapp_scanner`, //! `slack_scanner`, and `telegram_scanner`. Scanners now call helpers here -//! instead of maintaining their own WebSocket dispatch. +//! instead of maintaining their own dispatch. +//! +//! # Transport +//! +//! Two transports coexist while migration is in progress: +//! +//! - **In-process** — see [`in_process`]. CDP messages travel directly +//! between the Tauri shell and the embedded CEF browser via +//! `Webview::send_dev_tools_message` / +//! `Webview::on_dev_tools_protocol`. No listener, no network surface; +//! any same-UID process is shut out by construction. The per-account +//! session opener (`session.rs`) already uses this path. +//! - **Legacy TCP WebSocket** — driven by [`CDP_HOST`] / [`CDP_PORT`] +//! (`127.0.0.1:19222`). Kept alive for the per-scanner `CdpConn` +//! duplicates in `discord_scanner`, `whatsapp_scanner`, +//! `slack_scanner`, `telegram_scanner`, `wechat_scanner`, and +//! `meet_video` that have not yet migrated. While this path exists, +//! `app/src-tauri/src/lib.rs` still passes +//! `--remote-debugging-port=19222`, which is the same unauthenticated +//! loopback listener it has always been. The flag will be dropped +//! once all scanners cut over to the in-process channel. pub mod conn; +pub mod in_process; pub mod input; pub mod session; pub mod snapshot; pub mod target; pub use conn::CdpConn; +pub use in_process::{ + install_for_account, install_for_webview, set_cef_app_handle, CdpRegistry, EventFrame, + WebviewCdpTransport, CALL_TIMEOUT, +}; pub use session::{ placeholder_marker, placeholder_url, spawn_session, target_url_fragment, SpawnedSession, }; #[allow(unused_imports)] // `Rect` re-export consumed once turn 2 lands; keep stable. pub use snapshot::{Rect, Snapshot}; pub use target::{ - browser_ws_url, connect_and_attach_matching, detach_session, find_page_target_where, + browser_ws_url, conn_for_account, connect_and_attach_matching, detach_session, + find_page_target_where, }; -/// Remote debugging host — matches `--remote-debugging-port=19222` in -/// `lib.rs`. Kept as constants so scanners and the session opener -/// agree. Port was 9222 originally but collided with ollama's -/// `127.0.0.1:9222` listener (silent CDP-attach failure → blank -/// child webviews). If you change either constant, update both. +/// Remote debugging host — historical constant for scanner modules that +/// still use the TCP WebSocket path. The in-process transport in +/// [`in_process`] does not use these constants, but the per-scanner +/// `CdpConn` duplicates in `discord_scanner` / `whatsapp_scanner` / +/// `slack_scanner` / `telegram_scanner` still need them until they are +/// migrated to the shared in-process channel. pub const CDP_HOST: &str = "127.0.0.1"; pub const CDP_PORT: u16 = 19222; diff --git a/app/src-tauri/src/cdp/session.rs b/app/src-tauri/src/cdp/session.rs index ae0550042..92a0626ba 100644 --- a/app/src-tauri/src/cdp/session.rs +++ b/app/src-tauri/src/cdp/session.rs @@ -24,7 +24,8 @@ use tokio::task::JoinHandle; // elapsed check honours `tokio::time::pause()` / `advance()` in unit tests. use tokio::time::{sleep, Instant}; -use super::{browser_ws_url, find_page_target_where, CdpConn}; +use super::target::conn_for_account; +use super::{find_page_target_where, CdpConn}; use crate::webview_accounts::{emit_load_finished, redact_url_for_log, RevealTrigger}; /// Backoff between failed attach attempts / reconnects. Intentionally @@ -413,12 +414,14 @@ async fn run_session_cycle( real_url: &str, progress_slot: &ProgressSlot, ) -> Result<(), String> { - let browser_ws = browser_ws_url().await?; - let mut cdp = CdpConn::open(&browser_ws).await?; + let mut cdp = conn_for_account(app, account_id)?; - // Account-unique match. The placeholder URL and the real provider URL - // both carry account-specific fragments, so we can use ends_with and - // avoid substring collisions like `…account-abc` vs `…account-abcdef`. + // Account-unique match. Each webview is itself scoped to one + // account, but a webview can host popups (OAuth, attachment + // previews, …) that also surface as `kind=page` targets. The + // placeholder URL and the real provider URL both carry + // account-specific fragments, so we filter explicitly to pick the + // primary frame and ignore popups. let fragment = target_url_fragment(account_id); let target = find_page_target_where(&mut cdp, |t| target_matches_account_url(&t.url, account_id)) diff --git a/app/src-tauri/src/cdp/target.rs b/app/src-tauri/src/cdp/target.rs index c359cdf7c..3f1ef69de 100644 --- a/app/src-tauri/src/cdp/target.rs +++ b/app/src-tauri/src/cdp/target.rs @@ -1,11 +1,19 @@ -//! CDP target discovery. Replaces the four hand-rolled copies in the -//! per-provider scanners. +//! CDP target discovery + per-attach helpers. +//! +//! Each CEF webview is its own browser instance with its own DevTools +//! channel (see [`super::in_process`]), so the multi-target multiplexer +//! that used to live in this module has been simplified — there is no +//! more `browser_ws_url()` HTTP discovery and no remote attach. The +//! remaining helpers (`Target.getTargets` walk, `Target.attachToTarget` +//! flatten-attach, detach) still apply because the page itself may +//! contain iframes / workers that the scanners care about. use std::time::Duration; use serde_json::{json, Value}; +use tauri::{AppHandle, Manager, Runtime}; -use super::{CdpConn, CDP_HOST, CDP_PORT}; +use super::{in_process::CdpRegistry, CdpConn, CDP_HOST, CDP_PORT}; #[derive(Debug, Clone)] pub struct CdpTarget { @@ -15,9 +23,15 @@ pub struct CdpTarget { pub title: String, } -/// Discover the browser-level WebSocket endpoint via `/json/version`. All -/// CDP sessions in the app tunnel through this one ws once `flatten: true` -/// is set on attach. +/// Legacy TCP WebSocket discovery — kept for the per-scanner `CdpConn` +/// duplicates that have not yet migrated to the in-process transport. +/// New code paths use [`conn_for_account`] which goes through the +/// in-process channel installed by `webview_accounts::open`. +/// +/// Returns the browser-level WebSocket URL by hitting +/// `http://{CDP_HOST}:{CDP_PORT}/json/version`. Requires Chromium to +/// have been spawned with `--remote-debugging-port=` — see +/// `app/src-tauri/src/lib.rs`. pub async fn browser_ws_url() -> Result { let client = reqwest::Client::builder() .user_agent("openhuman-cdp/1.0") @@ -36,8 +50,6 @@ pub async fn browser_ws_url() -> Result { last_err = Some(format!("no webSocketDebuggerUrl in {url}")); } Err(e) => { - // Don't bail out — fall through so the next host in the - // candidate list still gets a chance to resolve the ws url. last_err = Some(format!("parse {url}: {e}")); } }, @@ -49,6 +61,9 @@ pub async fn browser_ws_url() -> Result { Err(last_err.unwrap_or_else(|| "failed to resolve CDP websocket URL".to_string())) } +/// Parse the response of a `Target.getTargets` CDP call into a list of +/// targets. Public so scanners using the lower-level [`CdpConn::call`] +/// can interpret target lists. pub fn parse_targets(v: &Value) -> Vec { v.get("targetInfos") .and_then(|x| x.as_array()) @@ -75,19 +90,79 @@ pub fn parse_targets(v: &Value) -> Vec { .unwrap_or_default() } -/// Full short-lived attach sequence: connect to the browser, find the -/// matching page target, attach with `flatten: true`. Caller gets a ready -/// CdpConn + session id for issuing commands. Caller MUST `detach_session` -/// (or drop the CdpConn entirely) when done so we don't leak sessions. +/// Get a [`CdpConn`] for an account-keyed webview, looking up the +/// pre-installed in-process transport from the [`CdpRegistry`] managed +/// on `app`. /// -/// The predicate must match on per-account fragment + URL prefix so -/// multi-account webviews on the same origin resolve uniquely. +/// On a cache miss, falls back to +/// [`super::in_process::install_for_account`] so a transient install +/// failure during `webview_accounts::open` (logged as a warning by the +/// account-open path, not fatal) doesn't permanently lock the account +/// out of CDP. The install call is idempotent and cheap on the cached +/// path. Still returns `Err` when the webview itself has not yet been +/// created — caller backs off and retries. +pub fn conn_for_account( + app: &AppHandle, + account_id: &str, +) -> Result { + let registry = app + .try_state::() + .ok_or_else(|| "CdpRegistry not managed by app".to_string())?; + if let Some(transport) = registry.by_account(account_id) { + return Ok(CdpConn::new(transport)); + } + // Retry — the install path is idempotent. The most common cause of + // a cache miss here is an earlier non-fatal `install_for_account` + // failure in `webview_accounts::open` (warn-logged) that left the + // webview alive without a transport. + let transport = super::in_process::install_for_account(account_id) + .map_err(|e| format!("no cdp transport for account {account_id} (install retry: {e})"))?; + Ok(CdpConn::new(transport)) +} + +/// Full short-lived attach sequence on the account's webview via the +/// in-process channel: look up the [`CdpRegistry`] transport for the +/// given account, find the matching page target via +/// `Target.getTargets`, attach with `flatten: true`. Caller gets a +/// ready `CdpConn` + session id. Caller MUST `detach_session` (or drop +/// the `CdpConn`) when done so the session id doesn't linger inside +/// CEF. +pub async fn connect_and_attach_matching_in_process( + app: &AppHandle, + account_id: &str, + pred: F, +) -> Result<(CdpConn, String), String> +where + R: Runtime, + F: Fn(&CdpTarget) -> bool, +{ + let mut cdp = conn_for_account(app, account_id)?; + let target = find_page_target_where(&mut cdp, pred).await?; + let attach = cdp + .call( + "Target.attachToTarget", + json!({ "targetId": target.id, "flatten": true }), + None, + ) + .await?; + let session = attach + .get("sessionId") + .and_then(|x| x.as_str()) + .ok_or_else(|| "attach missing sessionId".to_string())? + .to_string(); + Ok((cdp, session)) +} + +/// Legacy TCP-WS attach helper, kept for the per-scanner `CdpConn` +/// duplicates that still discover targets via the global +/// `Target.getTargets` walk. New code paths use +/// [`connect_and_attach_matching_in_process`]. pub async fn connect_and_attach_matching(pred: F) -> Result<(CdpConn, String), String> where F: Fn(&CdpTarget) -> bool, { let ws = browser_ws_url().await?; - let mut cdp = CdpConn::open(&ws).await?; + let mut cdp = CdpConn::open_ws(&ws).await?; let target = find_page_target_where(&mut cdp, pred).await?; let attach = cdp .call( @@ -114,10 +189,10 @@ pub async fn detach_session(cdp: &mut CdpConn, session_id: &str) { .await; } -/// Generalised variant — caller supplies the predicate (url-hash marker, -/// title marker, etc). Used by the per-account session opener, which matches -/// on `#openhuman-account-{id}` so multiple webviews on the same origin -/// don't collide. +/// Generalised target search — caller supplies the predicate +/// (url-hash marker, title marker, etc). Used by the per-account +/// session opener, which matches on `#openhuman-account-{id}` so +/// multiple webviews on the same origin don't collide. pub async fn find_page_target_where(cdp: &mut CdpConn, pred: F) -> Result where F: Fn(&CdpTarget) -> bool, diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index a708922ef..8c5c2f5bb 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -2474,21 +2474,16 @@ pub fn run() { // mock; `password-store=basic` is the equivalent for the password // manager. Both are no-ops on Windows/Linux, so safe to always set. // - // In debug builds we additionally expose the Chrome DevTools - // Protocol on localhost:19222 so every CEF webview can be - // inspected from a regular browser (right-click "Inspect" does - // not propagate to CEF child webviews on macOS). Release builds - // intentionally do NOT open the CDP port — it would let any - // process on the machine drive the embedded WhatsApp/Slack/etc. - // webviews. - // - // The port was 9222 (Chromium's default) but ollama's - // OpenAI-compatible server squats on 127.0.0.1:9222 in some - // installs, which silently broke CDP attach (our client hit - // ollama, the WS handshake failed, child webviews stayed at - // about:blank → black screen). Picked 19222 to dodge that - // collision; if you change it here also update - // `cdp::CDP_PORT` and `whatsapp_scanner::CDP_PORT`. + // CDP attach is migrating to the in-process channel — see + // `app/src-tauri/src/cdp/in_process.rs` and the per-account + // session opener (`cdp/session.rs`). The legacy TCP DevTools + // port is still passed below (search for + // `--remote-debugging-port`) because the per-scanner `CdpConn` + // duplicates in `discord_scanner`, `whatsapp_scanner`, + // `slack_scanner`, `telegram_scanner`, `wechat_scanner`, and + // `meet_video` have not migrated yet. Once they do, the flag + // can be dropped and the unauthenticated same-UID loopback + // listener with it. // // NOTE: flags must be prefixed with `--`. The runtime's // `on_before_command_line_processing` dispatch (in @@ -2593,14 +2588,13 @@ pub fn run() { args.push(("--use-fake-ui-for-media-stream", None)); args.push(("--use-file-for-fake-video-capture", Some(path))); } - // Always expose the CDP port, not just in debug. The webview-accounts - // CDP session opener navigates each embedded provider webview from its - // `about:blank#openhuman-acct-...` placeholder to the real provider URL - // via `Page.navigate`. Without this port available in release builds, - // the CDP client can't attach (`browser_ws_url()` 404s on /json/version), - // the navigation never fires, and the embedded webview stays on - // `about:blank` (blank panel for Telegram / WhatsApp / Slack / Discord). - // Same port the `cdp::CDP_HOST`/`cdp::CDP_PORT` constants expect. + // CDP attach is migrating to in-process. The per-account + // session opener (`cdp/session.rs`) uses the in-process channel + // installed by `webview_accounts::open`. The per-scanner + // duplicates (whatsapp, slack, telegram, wechat, discord, + // meet_video) still reach the embedded browser over the TCP + // loopback DevTools port — once they migrate this flag can be + // dropped and the unauthenticated listener closed for good. args.push(("--remote-debugging-port", Some("19222"))); let force_gpu_env = std::env::var("OPENHUMAN_FORCE_GPU").ok(); append_platform_cef_gpu_workarounds( @@ -2705,6 +2699,7 @@ pub fn run() { std::sync::Mutex::new(Vec::new()), )) .manage(webview_accounts::WebviewAccountsState::default()) + .manage(cdp::CdpRegistry::default()) .manage(notification_settings::NotificationSettingsState::new()) .manage(PendingAppUpdateState::default()); let builder = builder.manage(std::sync::Arc::new(imessage_scanner::ScannerRegistry::new())); @@ -2722,6 +2717,12 @@ pub fn run() { let builder = builder.manage(meet_video::frame_bus::MeetVideoFrameBusState::new()); builder .setup(move |app| { + // Stash the typed CEF `AppHandle` for the in-process CDP + // transport. Lets `cdp::install_for_account` reach the + // concrete `Webview` (which `send_dev_tools_message` + // requires) from generic `` call sites. + cdp::set_cef_app_handle(app.handle().clone()); + #[cfg(windows)] { // `register_all` writes HKCU\Software\Classes\openhuman so the diff --git a/app/src-tauri/src/slack_scanner/mod.rs b/app/src-tauri/src/slack_scanner/mod.rs index 450529116..58db51e89 100644 --- a/app/src-tauri/src/slack_scanner/mod.rs +++ b/app/src-tauri/src/slack_scanner/mod.rs @@ -812,7 +812,7 @@ async fn dom_scan_once( use crate::cdp::CdpConn as CanonicalCdpConn; let browser_ws = crate::cdp::browser_ws_url().await?; - let mut probe = CanonicalCdpConn::open(&browser_ws).await?; + let mut probe = CanonicalCdpConn::open_ws(&browser_ws).await?; let targets_v = probe .call("Target.getTargets", serde_json::json!({}), None) .await?; diff --git a/app/src-tauri/src/webview_accounts/mod.rs b/app/src-tauri/src/webview_accounts/mod.rs index ea4b60be0..20651d11b 100644 --- a/app/src-tauri/src/webview_accounts/mod.rs +++ b/app/src-tauri/src/webview_accounts/mod.rs @@ -1009,6 +1009,12 @@ fn teardown_account_scanners(app: &AppHandle, account_id: &str) { { registry.inner().forget(account_id); } + // Drop the in-process CDP transport for this account so a reopen + // installs a fresh observer instead of re-using the dead one tied + // to the closed webview. + if let Some(registry) = app.try_state::() { + registry.inner().forget_account(account_id); + } } #[derive(Debug, Clone)] @@ -2420,6 +2426,22 @@ pub async fn webview_account_open( .add_child(builder, initial_position, initial_size) .map_err(|e| format!("add_child failed: {e}"))?; + // Install the in-process CDP transport so the per-account session + // opener and the provider scanners can attach without the + // `--remote-debugging-port=19222` TCP listener. Failure here is + // logged but not fatal — the scanners retry through + // `cdp::conn_for_account` once the registry is populated, so a + // transient install error just delays first attach by one backoff + // tick. + if let Err(err) = crate::cdp::install_for_account(&args.account_id) { + log::warn!( + "[webview-accounts] cdp install_for_account({}) failed: {} \ + (scanners will retry)", + args.account_id, + err + ); + } + // Capture the cold-spawn timestamp so the reveal-time log can compute // spawn -> frontend reveal latency for the Slack first-load investigation. state