browser_control/cdp/
mod.rs

1//! Minimal Chrome DevTools Protocol (CDP) WebSocket client.
2//!
3//! The socket / reader-task / writer-task / pending-correlation / timeout
4//! machinery lives in the shared [`crate::transport`] (`WsRpc`); this module
5//! supplies only the CDP-specific framing/typing via the [`CdpProtocol`]
6//! adapter and the convenience methods.
7
8use anyhow::{anyhow, Result};
9use serde_json::{json, Value};
10use tokio::sync::broadcast;
11
12pub mod protocol;
13use protocol::{CdpError, Request, Response};
14
15use crate::errors::{is_cdp_target_gone, SessionError, TargetKind};
16use crate::transport::{Decoded, Protocol, RequestError, WsRpc, CONNECT_TIMEOUT, REQUEST_TIMEOUT};
17
18#[derive(Debug, Clone)]
19pub struct CdpEvent {
20    pub method: String,
21    pub params: Value,
22    pub session_id: Option<String>,
23}
24
25/// CDP framing/typing adapter for the shared transport.
26pub struct CdpProtocol;
27
28impl Protocol for CdpProtocol {
29    type ProtoError = CdpError;
30    type Event = CdpEvent;
31
32    fn encode_request(
33        id: u64,
34        method: &str,
35        params: Value,
36        session_id: Option<&str>,
37    ) -> Result<String> {
38        let req = Request {
39            id,
40            method,
41            params,
42            session_id: session_id.map(|s| s.to_string()),
43        };
44        Ok(serde_json::to_string(&req)?)
45    }
46
47    fn decode_frame(text: &str) -> Decoded<CdpError, CdpEvent> {
48        let resp: Response = match serde_json::from_str(text) {
49            Ok(r) => r,
50            Err(_) => return Decoded::Ignore,
51        };
52        if let Some(id) = resp.id {
53            let result = if let Some(err) = resp.error {
54                Err(err)
55            } else {
56                Ok(resp.result)
57            };
58            Decoded::Reply { id, result }
59        } else if let Some(method) = resp.method {
60            Decoded::Event(CdpEvent {
61                method,
62                params: resp.params,
63                session_id: resp.session_id,
64            })
65        } else {
66            Decoded::Ignore
67        }
68    }
69
70    fn closed_error() -> CdpError {
71        CdpError {
72            code: -1,
73            message: "connection closed".into(),
74        }
75    }
76}
77
78pub struct CdpClient {
79    rpc: WsRpc<CdpProtocol>,
80}
81
82impl CdpClient {
83    /// Connect by full WebSocket URL (ws:// or wss://).
84    pub async fn connect(ws_url: &str) -> Result<Self> {
85        Ok(Self {
86            rpc: WsRpc::connect(ws_url, "CDP").await?,
87        })
88    }
89
90    /// Connect by HTTP base URL (e.g. http://127.0.0.1:9222). Fetches /json/version to discover the WS URL.
91    pub async fn connect_http(base_url: &str) -> Result<Self> {
92        let base = base_url.trim_end_matches('/');
93        let url = format!("{base}/json/version");
94        let client = reqwest::Client::builder()
95            .timeout(CONNECT_TIMEOUT)
96            .build()
97            .map_err(|e| anyhow!("building reqwest client: {e}"))?;
98        // Check the HTTP status before parsing: a non-2xx (wrong port / a
99        // non-CDP server answering) otherwise surfaces as a confusing serde
100        // error or "webSocketDebuggerUrl missing" instead of the real cause.
101        let http_resp = client
102            .get(&url)
103            .send()
104            .await?
105            .error_for_status()
106            .map_err(|e| anyhow!("fetching {url}: {e}"))?;
107        let resp: Value = http_resp.json().await?;
108        let ws_url = resp
109            .get("webSocketDebuggerUrl")
110            .and_then(|v| v.as_str())
111            .ok_or_else(|| anyhow!("webSocketDebuggerUrl missing from {url}"))?
112            .to_string();
113        Self::connect(&ws_url).await
114    }
115
116    /// Send a method on the root browser-level session.
117    pub async fn send(&self, method: &str, params: Value) -> Result<Value> {
118        self.send_with_session(method, params, None).await
119    }
120
121    pub async fn send_with_session(
122        &self,
123        method: &str,
124        params: Value,
125        session_id: Option<&str>,
126    ) -> Result<Value> {
127        match self.rpc.request(method, params, session_id).await {
128            Ok(v) => Ok(v),
129            Err(RequestError::Protocol(e)) => Err(classify_cdp_error(e, session_id.is_some())),
130            Err(RequestError::Timeout) => match session_id {
131                Some(sid) => Err(anyhow!(
132                    "CDP request {method} (session {sid}) timed out after {:?}",
133                    REQUEST_TIMEOUT
134                )),
135                None => Err(anyhow!(
136                    "CDP request {method} timed out after {:?}",
137                    REQUEST_TIMEOUT
138                )),
139            },
140            Err(RequestError::Transport(e)) => Err(e),
141        }
142    }
143
144    /// Subscribe to all events. Drop the receiver to unsubscribe.
145    pub fn subscribe(&self) -> broadcast::Receiver<CdpEvent> {
146        self.rpc.subscribe()
147    }
148
149    /// Attach to a target via Target.attachToTarget(flatten=true) and return the session id.
150    pub async fn attach_to_target(&self, target_id: &str) -> Result<String> {
151        let v = self
152            .send(
153                "Target.attachToTarget",
154                json!({ "targetId": target_id, "flatten": true }),
155            )
156            .await?;
157        v.get("sessionId")
158            .and_then(|v| v.as_str())
159            .map(|s| s.to_string())
160            .ok_or_else(|| anyhow!("sessionId missing from Target.attachToTarget response"))
161    }
162
163    /// Convenience: list targets via Target.getTargets.
164    pub async fn list_targets(&self) -> Result<Vec<Value>> {
165        let v = self.send("Target.getTargets", Value::Null).await?;
166        match v.get("targetInfos") {
167            Some(Value::Array(a)) => Ok(a.clone()),
168            _ => Ok(vec![]),
169        }
170    }
171
172    /// Gracefully shut down. Dropping the client also aborts the tasks via
173    /// `WsRpc`'s `Drop`, so this is the explicit-flush path.
174    pub async fn close(self) {
175        self.rpc.close().await;
176    }
177}
178
179/// Convert a `CdpError` reply into a typed `SessionError::TargetGone` if
180/// its message matches a known "gone" indicator, otherwise pass through as
181/// the generic CDP error. `attached` is true when the call carried a
182/// `sessionId` — only attached-session failures are classified, because
183/// browser-session errors typically mean "bad request," not "target gone."
184fn classify_cdp_error(err: CdpError, attached: bool) -> anyhow::Error {
185    if attached && is_cdp_target_gone(&err.message) {
186        return SessionError::TargetGone {
187            kind: TargetKind::Cdp,
188            details: format!("CDP error {}: {}", err.code, err.message),
189        }
190        .into();
191    }
192    anyhow!(err)
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198    use futures_util::{SinkExt, StreamExt};
199    use std::time::Duration;
200    use tokio::sync::oneshot;
201    use tokio_tungstenite::tungstenite::Message;
202
203    #[tokio::test]
204    async fn round_trip_request_response() {
205        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
206        let addr = listener.local_addr().unwrap();
207        tokio::spawn(async move {
208            let (stream, _) = listener.accept().await.unwrap();
209            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
210            while let Some(Ok(msg)) = ws.next().await {
211                if let Message::Text(t) = msg {
212                    let req: Value = serde_json::from_str(&t).unwrap();
213                    let id = req["id"].as_u64().unwrap();
214                    let resp = json!({"id": id, "result": {"ok": true, "echo": req["method"]}});
215                    ws.send(Message::Text(resp.to_string())).await.unwrap();
216                }
217            }
218        });
219        let url = format!("ws://{}", addr);
220        let client = CdpClient::connect(&url).await.unwrap();
221        let v = client
222            .send("Page.navigate", json!({"url": "about:blank"}))
223            .await
224            .unwrap();
225        assert_eq!(v["ok"], true);
226        assert_eq!(v["echo"], "Page.navigate");
227        client.close().await;
228    }
229
230    #[tokio::test]
231    async fn broadcast_event_to_subscriber() {
232        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
233        let addr = listener.local_addr().unwrap();
234        let (ready_tx, ready_rx) = oneshot::channel::<()>();
235        tokio::spawn(async move {
236            let (stream, _) = listener.accept().await.unwrap();
237            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
238            // Wait until the test confirms it has subscribed before pushing event.
239            let _ = ready_rx.await;
240            let evt = json!({
241                "method": "Target.targetCreated",
242                "params": {"targetInfo": {"targetId": "abc"}},
243                "sessionId": "S1"
244            });
245            ws.send(Message::Text(evt.to_string())).await.unwrap();
246            // Keep socket alive briefly.
247            while let Some(Ok(_)) = ws.next().await {}
248        });
249
250        let url = format!("ws://{}", addr);
251        let client = CdpClient::connect(&url).await.unwrap();
252        let mut rx = client.subscribe();
253        ready_tx.send(()).unwrap();
254
255        let evt = tokio::time::timeout(Duration::from_secs(5), rx.recv())
256            .await
257            .expect("event timeout")
258            .expect("event recv");
259        assert_eq!(evt.method, "Target.targetCreated");
260        assert_eq!(evt.session_id.as_deref(), Some("S1"));
261        assert_eq!(evt.params["targetInfo"]["targetId"], "abc");
262        client.close().await;
263    }
264
265    /// Attached-session CDP error matching a "target gone" indicator
266    /// surfaces as a typed `SessionError::TargetGone`. The recovery
267    /// wrappers rely on the typed variant to skip substring matching.
268    #[tokio::test]
269    async fn send_with_session_classifies_target_gone() {
270        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
271        let addr = listener.local_addr().unwrap();
272        tokio::spawn(async move {
273            let (stream, _) = listener.accept().await.unwrap();
274            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
275            while let Some(Ok(Message::Text(t))) = ws.next().await {
276                let req: Value = serde_json::from_str(&t).unwrap();
277                let id = req["id"].as_u64().unwrap();
278                let resp = json!({
279                    "id": id,
280                    "error": {"code": -32000, "message": "No target with given id found: T42"}
281                });
282                ws.send(Message::Text(resp.to_string())).await.unwrap();
283            }
284        });
285        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
286        let err = client
287            .send_with_session("Runtime.evaluate", json!({}), Some("S1"))
288            .await
289            .expect_err("must error");
290        let typed = err
291            .downcast_ref::<crate::errors::SessionError>()
292            .expect("typed SessionError");
293        match typed {
294            crate::errors::SessionError::TargetGone { kind, details } => {
295                assert_eq!(*kind, crate::errors::TargetKind::Cdp);
296                assert!(details.contains("No target with given id"));
297            }
298            other => panic!("expected TargetGone, got {other:?}"),
299        }
300        client.close().await;
301    }
302
303    /// Browser-session CDP errors (no `sessionId`) are NOT classified —
304    /// they pass through as generic anyhow errors. `Target.attachToTarget`
305    /// failing with "no such target" is a routing problem at the browser
306    /// level, not a wedged renderer, and shouldn't trigger tab recovery.
307    #[tokio::test]
308    async fn send_root_does_not_classify_target_gone() {
309        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
310        let addr = listener.local_addr().unwrap();
311        tokio::spawn(async move {
312            let (stream, _) = listener.accept().await.unwrap();
313            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
314            while let Some(Ok(Message::Text(t))) = ws.next().await {
315                let req: Value = serde_json::from_str(&t).unwrap();
316                let id = req["id"].as_u64().unwrap();
317                let resp = json!({
318                    "id": id,
319                    "error": {"code": -32000, "message": "No target with given id found: T42"}
320                });
321                ws.send(Message::Text(resp.to_string())).await.unwrap();
322            }
323        });
324        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
325        let err = client
326            .send("Target.attachToTarget", json!({}))
327            .await
328            .expect_err("must error");
329        assert!(
330            err.downcast_ref::<crate::errors::SessionError>().is_none(),
331            "root-session error must NOT classify as TargetGone"
332        );
333        client.close().await;
334    }
335
336    /// Test #15: connect-side timeout fires when the WS upgrade hangs.
337    ///
338    /// The TCP listener accepts the connection but never writes the HTTP
339    /// upgrade response, so `tokio_tungstenite::connect_async` would wait
340    /// indefinitely without the bound.
341    #[tokio::test]
342    async fn connect_times_out_when_upgrade_hangs() {
343        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
344        let addr = listener.local_addr().unwrap();
345        // Hold the accepted connection forever (no HTTP response).
346        tokio::spawn(async move {
347            let (_stream, _) = listener.accept().await.unwrap();
348            std::future::pending::<()>().await;
349        });
350
351        let url = format!("ws://{addr}");
352        let start = std::time::Instant::now();
353        let err = match CdpClient::connect(&url).await {
354            Ok(_) => panic!("connect must fail when upgrade hangs"),
355            Err(e) => e,
356        };
357        let elapsed = start.elapsed();
358
359        // Must fail within the bound + a generous slack for CI variance.
360        assert!(
361            elapsed < CONNECT_TIMEOUT + Duration::from_secs(2),
362            "connect did not honour the 5s bound (took {elapsed:?})"
363        );
364        let msg = format!("{err:#}");
365        assert!(
366            msg.contains("timed out"),
367            "error should mention timeout, got: {msg}"
368        );
369    }
370}
browser_control/cdp/mod.rs

browser_control/cdp/
mod.rs
