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    /// Fetch the browser's full cookie jar across Chromium versions.
173    ///
174    /// Modern Chromium exposes browser-wide cookie export as
175    /// `Storage.getCookies` on the browser endpoint. Older builds and some
176    /// protocol surfaces only exposed the deprecated Network method, either
177    /// on the browser endpoint or through an attached page session.
178    pub async fn get_all_cookies(&self) -> Result<Value> {
179        match self.send("Storage.getCookies", json!({})).await {
180            Ok(v) => return Ok(v),
181            Err(e) if !is_cdp_method_not_found(&e) => return Err(e),
182            Err(_) => {}
183        }
184
185        match self.send("Network.getAllCookies", Value::Null).await {
186            Ok(v) => return Ok(v),
187            Err(e) if !is_cdp_method_not_found(&e) => return Err(e),
188            Err(_) => {}
189        }
190
191        self.get_all_cookies_via_page_session().await
192    }
193
194    async fn get_all_cookies_via_page_session(&self) -> Result<Value> {
195        let mut created_target = None::<String>;
196        let target_id = match self
197            .list_targets()
198            .await?
199            .into_iter()
200            .find(|t| t.get("type").and_then(Value::as_str) == Some("page"))
201            .and_then(|t| {
202                t.get("targetId")
203                    .and_then(Value::as_str)
204                    .map(str::to_string)
205            }) {
206            Some(id) => id,
207            None => {
208                let v = self
209                    .send(
210                        "Target.createTarget",
211                        json!({ "url": "about:blank", "background": true }),
212                    )
213                    .await?;
214                let id = v
215                    .get("targetId")
216                    .and_then(Value::as_str)
217                    .ok_or_else(|| anyhow!("Target.createTarget returned no targetId"))?
218                    .to_string();
219                created_target = Some(id.clone());
220                id
221            }
222        };
223
224        let session_id = match self.attach_to_target(&target_id).await {
225            Ok(session_id) => session_id,
226            Err(e) => {
227                if let Some(target_id) = created_target {
228                    let _ = self
229                        .send("Target.closeTarget", json!({ "targetId": target_id }))
230                        .await;
231                }
232                return Err(e);
233            }
234        };
235        let result = self
236            .send_with_session("Network.getAllCookies", Value::Null, Some(&session_id))
237            .await;
238        let _ = self
239            .send(
240                "Target.detachFromTarget",
241                json!({ "sessionId": session_id }),
242            )
243            .await;
244        if let Some(target_id) = created_target {
245            let _ = self
246                .send("Target.closeTarget", json!({ "targetId": target_id }))
247                .await;
248        }
249        result
250    }
251
252    /// Gracefully shut down. Dropping the client also aborts the tasks via
253    /// `WsRpc`'s `Drop`, so this is the explicit-flush path.
254    pub async fn close(self) {
255        self.rpc.close().await;
256    }
257}
258
259pub fn is_cdp_method_not_found(err: &anyhow::Error) -> bool {
260    err.downcast_ref::<CdpError>()
261        .is_some_and(|e| e.code == -32601)
262}
263
264/// Convert a `CdpError` reply into a typed `SessionError::TargetGone` if
265/// its message matches a known "gone" indicator, otherwise pass through as
266/// the generic CDP error. `attached` is true when the call carried a
267/// `sessionId` — only attached-session failures are classified, because
268/// browser-session errors typically mean "bad request," not "target gone."
269fn classify_cdp_error(err: CdpError, attached: bool) -> anyhow::Error {
270    if attached && is_cdp_target_gone(&err.message) {
271        return SessionError::TargetGone {
272            kind: TargetKind::Cdp,
273            details: format!("CDP error {}: {}", err.code, err.message),
274        }
275        .into();
276    }
277    anyhow!(err)
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283    use futures_util::{SinkExt, StreamExt};
284    use std::sync::Arc;
285    use std::time::Duration;
286    use tokio::sync::{oneshot, Mutex};
287    use tokio_tungstenite::tungstenite::Message;
288
289    #[tokio::test]
290    async fn round_trip_request_response() {
291        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
292        let addr = listener.local_addr().unwrap();
293        tokio::spawn(async move {
294            let (stream, _) = listener.accept().await.unwrap();
295            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
296            while let Some(Ok(msg)) = ws.next().await {
297                if let Message::Text(t) = msg {
298                    let req: Value = serde_json::from_str(&t).unwrap();
299                    let id = req["id"].as_u64().unwrap();
300                    let resp = json!({"id": id, "result": {"ok": true, "echo": req["method"]}});
301                    ws.send(Message::Text(resp.to_string())).await.unwrap();
302                }
303            }
304        });
305        let url = format!("ws://{}", addr);
306        let client = CdpClient::connect(&url).await.unwrap();
307        let v = client
308            .send("Page.navigate", json!({"url": "about:blank"}))
309            .await
310            .unwrap();
311        assert_eq!(v["ok"], true);
312        assert_eq!(v["echo"], "Page.navigate");
313        client.close().await;
314    }
315
316    #[tokio::test]
317    async fn broadcast_event_to_subscriber() {
318        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
319        let addr = listener.local_addr().unwrap();
320        let (ready_tx, ready_rx) = oneshot::channel::<()>();
321        tokio::spawn(async move {
322            let (stream, _) = listener.accept().await.unwrap();
323            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
324            // Wait until the test confirms it has subscribed before pushing event.
325            let _ = ready_rx.await;
326            let evt = json!({
327                "method": "Target.targetCreated",
328                "params": {"targetInfo": {"targetId": "abc"}},
329                "sessionId": "S1"
330            });
331            ws.send(Message::Text(evt.to_string())).await.unwrap();
332            // Keep socket alive briefly.
333            while let Some(Ok(_)) = ws.next().await {}
334        });
335
336        let url = format!("ws://{}", addr);
337        let client = CdpClient::connect(&url).await.unwrap();
338        let mut rx = client.subscribe();
339        ready_tx.send(()).unwrap();
340
341        let evt = tokio::time::timeout(Duration::from_secs(5), rx.recv())
342            .await
343            .expect("event timeout")
344            .expect("event recv");
345        assert_eq!(evt.method, "Target.targetCreated");
346        assert_eq!(evt.session_id.as_deref(), Some("S1"));
347        assert_eq!(evt.params["targetInfo"]["targetId"], "abc");
348        client.close().await;
349    }
350
351    /// Attached-session CDP error matching a "target gone" indicator
352    /// surfaces as a typed `SessionError::TargetGone`. The recovery
353    /// wrappers rely on the typed variant to skip substring matching.
354    #[tokio::test]
355    async fn send_with_session_classifies_target_gone() {
356        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
357        let addr = listener.local_addr().unwrap();
358        tokio::spawn(async move {
359            let (stream, _) = listener.accept().await.unwrap();
360            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
361            while let Some(Ok(Message::Text(t))) = ws.next().await {
362                let req: Value = serde_json::from_str(&t).unwrap();
363                let id = req["id"].as_u64().unwrap();
364                let resp = json!({
365                    "id": id,
366                    "error": {"code": -32000, "message": "No target with given id found: T42"}
367                });
368                ws.send(Message::Text(resp.to_string())).await.unwrap();
369            }
370        });
371        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
372        let err = client
373            .send_with_session("Runtime.evaluate", json!({}), Some("S1"))
374            .await
375            .expect_err("must error");
376        let typed = err
377            .downcast_ref::<crate::errors::SessionError>()
378            .expect("typed SessionError");
379        match typed {
380            crate::errors::SessionError::TargetGone { kind, details } => {
381                assert_eq!(*kind, crate::errors::TargetKind::Cdp);
382                assert!(details.contains("No target with given id"));
383            }
384            other => panic!("expected TargetGone, got {other:?}"),
385        }
386        client.close().await;
387    }
388
389    /// Browser-session CDP errors (no `sessionId`) are NOT classified —
390    /// they pass through as generic anyhow errors. `Target.attachToTarget`
391    /// failing with "no such target" is a routing problem at the browser
392    /// level, not a wedged renderer, and shouldn't trigger tab recovery.
393    #[tokio::test]
394    async fn send_root_does_not_classify_target_gone() {
395        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
396        let addr = listener.local_addr().unwrap();
397        tokio::spawn(async move {
398            let (stream, _) = listener.accept().await.unwrap();
399            let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
400            while let Some(Ok(Message::Text(t))) = ws.next().await {
401                let req: Value = serde_json::from_str(&t).unwrap();
402                let id = req["id"].as_u64().unwrap();
403                let resp = json!({
404                    "id": id,
405                    "error": {"code": -32000, "message": "No target with given id found: T42"}
406                });
407                ws.send(Message::Text(resp.to_string())).await.unwrap();
408            }
409        });
410        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
411        let err = client
412            .send("Target.attachToTarget", json!({}))
413            .await
414            .expect_err("must error");
415        assert!(
416            err.downcast_ref::<crate::errors::SessionError>().is_none(),
417            "root-session error must NOT classify as TargetGone"
418        );
419        client.close().await;
420    }
421
422    #[tokio::test]
423    async fn get_all_cookies_prefers_storage_get_cookies() {
424        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
425        let addr = listener.local_addr().unwrap();
426        let seen = Arc::new(Mutex::new(Vec::<String>::new()));
427        tokio::spawn({
428            let seen = seen.clone();
429            async move {
430                let (stream, _) = listener.accept().await.unwrap();
431                let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
432                while let Some(Ok(Message::Text(t))) = ws.next().await {
433                    let req: Value = serde_json::from_str(&t).unwrap();
434                    let id = req["id"].as_u64().unwrap();
435                    let method = req["method"].as_str().unwrap_or("").to_string();
436                    seen.lock().await.push(method.clone());
437                    let result = match method.as_str() {
438                        "Storage.getCookies" => {
439                            json!({"cookies": [{"name": "sid", "value": "modern"}]})
440                        }
441                        _ => json!({}),
442                    };
443                    let resp = json!({"id": id, "result": result});
444                    ws.send(Message::Text(resp.to_string())).await.unwrap();
445                }
446            }
447        });
448
449        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
450        let v = client.get_all_cookies().await.unwrap();
451        assert_eq!(v["cookies"][0]["value"], "modern");
452        assert_eq!(*seen.lock().await, vec!["Storage.getCookies".to_string()]);
453        client.close().await;
454    }
455
456    #[tokio::test]
457    async fn get_all_cookies_falls_back_to_root_network_method() {
458        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
459        let addr = listener.local_addr().unwrap();
460        let seen = Arc::new(Mutex::new(Vec::<String>::new()));
461        tokio::spawn({
462            let seen = seen.clone();
463            async move {
464                let (stream, _) = listener.accept().await.unwrap();
465                let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
466                while let Some(Ok(Message::Text(t))) = ws.next().await {
467                    let req: Value = serde_json::from_str(&t).unwrap();
468                    let id = req["id"].as_u64().unwrap();
469                    let method = req["method"].as_str().unwrap_or("").to_string();
470                    seen.lock().await.push(method.clone());
471                    let resp = match method.as_str() {
472                        "Storage.getCookies" => json!({
473                            "id": id,
474                            "error": {
475                                "code": -32601,
476                                "message": "'Storage.getCookies' wasn't found"
477                            }
478                        }),
479                        "Network.getAllCookies" => json!({
480                            "id": id,
481                            "result": {
482                                "cookies": [{"name": "sid", "value": "root-legacy"}]
483                            }
484                        }),
485                        _ => json!({"id": id, "result": {}}),
486                    };
487                    ws.send(Message::Text(resp.to_string())).await.unwrap();
488                }
489            }
490        });
491
492        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
493        let v = client.get_all_cookies().await.unwrap();
494        assert_eq!(v["cookies"][0]["value"], "root-legacy");
495        assert_eq!(
496            *seen.lock().await,
497            vec![
498                "Storage.getCookies".to_string(),
499                "Network.getAllCookies".to_string()
500            ]
501        );
502        client.close().await;
503    }
504
505    #[tokio::test]
506    async fn get_all_cookies_falls_back_to_attached_network_method() {
507        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
508        let addr = listener.local_addr().unwrap();
509        let seen = Arc::new(Mutex::new(Vec::<(String, Option<String>)>::new()));
510        tokio::spawn({
511            let seen = seen.clone();
512            async move {
513                let (stream, _) = listener.accept().await.unwrap();
514                let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
515                while let Some(Ok(Message::Text(t))) = ws.next().await {
516                    let req: Value = serde_json::from_str(&t).unwrap();
517                    let id = req["id"].as_u64().unwrap();
518                    let method = req["method"].as_str().unwrap_or("").to_string();
519                    let session = req
520                        .get("sessionId")
521                        .and_then(Value::as_str)
522                        .map(str::to_string);
523                    seen.lock().await.push((method.clone(), session));
524                    let resp = match method.as_str() {
525                        "Storage.getCookies" | "Network.getAllCookies"
526                            if req.get("sessionId").is_none() =>
527                        {
528                            json!({
529                                "id": id,
530                                "error": {
531                                    "code": -32601,
532                                    "message": format!("'{method}' wasn't found")
533                                }
534                            })
535                        }
536                        "Target.getTargets" => json!({
537                            "id": id,
538                            "result": {
539                                "targetInfos": [{
540                                    "targetId": "T1",
541                                    "type": "page",
542                                    "url": "about:blank"
543                                }]
544                            }
545                        }),
546                        "Target.attachToTarget" => {
547                            json!({"id": id, "result": {"sessionId": "S1"}})
548                        }
549                        "Network.getAllCookies" => json!({
550                            "id": id,
551                            "result": {
552                                "cookies": [{"name": "sid", "value": "legacy"}]
553                            }
554                        }),
555                        "Target.detachFromTarget" => json!({"id": id, "result": {}}),
556                        _ => json!({"id": id, "result": {}}),
557                    };
558                    ws.send(Message::Text(resp.to_string())).await.unwrap();
559                }
560            }
561        });
562
563        let client = CdpClient::connect(&format!("ws://{addr}")).await.unwrap();
564        let v = client.get_all_cookies().await.unwrap();
565        assert_eq!(v["cookies"][0]["value"], "legacy");
566        assert_eq!(
567            *seen.lock().await,
568            vec![
569                ("Storage.getCookies".to_string(), None),
570                ("Network.getAllCookies".to_string(), None),
571                ("Target.getTargets".to_string(), None),
572                ("Target.attachToTarget".to_string(), None),
573                ("Network.getAllCookies".to_string(), Some("S1".to_string())),
574                ("Target.detachFromTarget".to_string(), None),
575            ]
576        );
577        client.close().await;
578    }
579
580    /// Test #15: connect-side timeout fires when the WS upgrade hangs.
581    ///
582    /// The TCP listener accepts the connection but never writes the HTTP
583    /// upgrade response, so `tokio_tungstenite::connect_async` would wait
584    /// indefinitely without the bound.
585    #[tokio::test]
586    async fn connect_times_out_when_upgrade_hangs() {
587        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
588        let addr = listener.local_addr().unwrap();
589        // Hold the accepted connection forever (no HTTP response).
590        tokio::spawn(async move {
591            let (_stream, _) = listener.accept().await.unwrap();
592            std::future::pending::<()>().await;
593        });
594
595        let url = format!("ws://{addr}");
596        let start = std::time::Instant::now();
597        let err = match CdpClient::connect(&url).await {
598            Ok(_) => panic!("connect must fail when upgrade hangs"),
599            Err(e) => e,
600        };
601        let elapsed = start.elapsed();
602
603        // Must fail within the bound + a generous slack for CI variance.
604        assert!(
605            elapsed < CONNECT_TIMEOUT + Duration::from_secs(2),
606            "connect did not honour the 5s bound (took {elapsed:?})"
607        );
608        let msg = format!("{err:#}");
609        assert!(
610            msg.contains("timed out"),
611            "error should mention timeout, got: {msg}"
612        );
613    }
614}
browser_control/cdp/mod.rs

browser_control/cdp/
mod.rs
