browser_commander/browser/
cdp_endpoint.rs

1//! Waiting for a spawned browser's DevTools endpoint and proving it is ours
2//! (issue #101).
3//!
4//! A reserved port is released before Chrome binds it, so another process can
5//! take it in between. The endpoint is only trusted once the browser's own
6//! stderr (or its `DevToolsActivePort` file) says it listens on that port, and
7//! the `/json/version` served there names the same browser WebSocket URL the
8//! browser announced; anything else is a [`PortRaceError`].
9
10use std::future::Future;
11use std::io;
12use std::path::Path;
13use std::time::{Duration, Instant};
14
15use anyhow::{anyhow, Result};
16use serde_json::Value;
17use tokio::io::{AsyncReadExt, AsyncWriteExt};
18use tokio::net::TcpStream;
19
20use crate::browser::browser_process::BrowserProcess;
21use crate::browser::debugging_port::{
22    assert_fixed_debugging_port, classify_dev_tools_ownership, DevToolsOutputWatcher,
23    DevToolsOwnership, PortRaceError, LOOPBACK_HOST,
24};
25
26const POLL_INTERVAL: Duration = Duration::from_millis(100);
27const PROBE_TIMEOUT: Duration = Duration::from_millis(500);
28const RESPONSE_LIMIT: usize = 1024 * 1024;
29
30/// What [`wait_for_cdp_endpoint`] waits for.
31#[derive(Debug, Clone, Copy)]
32pub struct CdpEndpointRequest<'a> {
33    /// The fixed port the browser was started with.
34    pub remote_debugging_port: u16,
35    /// The browser's profile, where it may write `DevToolsActivePort`.
36    pub user_data_dir: &'a Path,
37    /// The spawned browser.
38    pub browser_process: &'a BrowserProcess,
39    /// The browser's stderr, when it is captured. Without it ownership is
40    /// only confirmed through `DevToolsActivePort`.
41    pub dev_tools_output: Option<&'a DevToolsOutputWatcher>,
42    /// How long to wait overall.
43    pub timeout: Duration,
44}
45
46fn web_socket_debugger_url(response: &[u8]) -> Option<String> {
47    if !(response.starts_with(b"HTTP/1.1 200") || response.starts_with(b"HTTP/1.0 200")) {
48        return None;
49    }
50    let header_end = response
51        .windows(4)
52        .position(|window| window == b"\r\n\r\n")?;
53    serde_json::from_slice::<Value>(&response[header_end + 4..])
54        .ok()?
55        .get("webSocketDebuggerUrl")?
56        .as_str()
57        .map(str::to_owned)
58}
59
60/// Fetch `/json/version` from the loopback DevTools server on `port` and
61/// return its `webSocketDebuggerUrl`.
62///
63/// The response is parsed as soon as it is complete, without waiting for the
64/// server to close the connection.
65pub async fn fetch_cdp_version(port: u16, timeout: Duration) -> Option<String> {
66    let request = format!(
67        "GET /json/version HTTP/1.1\r\nHost: {LOOPBACK_HOST}:{port}\r\nConnection: close\r\n\r\n"
68    );
69    let request_future = async {
70        let mut stream = TcpStream::connect((LOOPBACK_HOST, port)).await?;
71        stream.write_all(request.as_bytes()).await?;
72        let mut response = Vec::new();
73        let mut chunk = [0_u8; 4096];
74        loop {
75            let bytes_read = stream.read(&mut chunk).await?;
76            if bytes_read == 0 {
77                break;
78            }
79            response.extend_from_slice(&chunk[..bytes_read]);
80            if let Some(url) = web_socket_debugger_url(&response) {
81                return Ok::<Option<String>, io::Error>(Some(url));
82            }
83            if response.len() > RESPONSE_LIMIT {
84                return Ok(None);
85            }
86        }
87        Ok(web_socket_debugger_url(&response))
88    };
89    tokio::time::timeout(timeout, request_future)
90        .await
91        .ok()?
92        .ok()?
93}
94
95/// Read the port from a profile's `DevToolsActivePort` file.
96pub fn read_dev_tools_active_port(user_data_dir: &Path) -> Option<u16> {
97    std::fs::read_to_string(user_data_dir.join("DevToolsActivePort"))
98        .ok()?
99        .lines()
100        .next()?
101        .trim()
102        .parse()
103        .ok()
104}
105
106/// Wait until the browser's DevTools endpoint is ready and proven to be ours.
107///
108/// Returns `http://127.0.0.1:<port>`. Fails with a [`PortRaceError`] (inside
109/// the [`anyhow::Error`]) when another process owns the port, and with a plain
110/// error when the browser exits or `timeout` passes.
111pub async fn wait_for_cdp_endpoint(request: CdpEndpointRequest<'_>) -> Result<String> {
112    wait_for_cdp_endpoint_with(request, fetch_cdp_version).await
113}
114
115pub(crate) async fn wait_for_cdp_endpoint_with<P, F>(
116    request: CdpEndpointRequest<'_>,
117    probe: P,
118) -> Result<String>
119where
120    P: Fn(u16, Duration) -> F,
121    F: Future<Output = Option<String>>,
122{
123    let port = assert_fixed_debugging_port(request.remote_debugging_port)?;
124    let endpoint = format!("http://{LOOPBACK_HOST}:{port}");
125    let started = Instant::now();
126    while started.elapsed() < request.timeout {
127        let output = request
128            .dev_tools_output
129            .map(DevToolsOutputWatcher::state)
130            .unwrap_or_default();
131        let mut owned = match request.dev_tools_output {
132            Some(_) => match classify_dev_tools_ownership(&output, port) {
133                DevToolsOwnership::Race => {
134                    let detail = output.listening.as_ref().map_or_else(
135                        || "bind failed".to_owned(),
136                        |listening| listening.url.clone(),
137                    );
138                    return Err(PortRaceError::new(port, detail).into());
139                }
140                DevToolsOwnership::Owned => true,
141                DevToolsOwnership::Pending => false,
142            },
143            None => false,
144        };
145        if let Some(code) = request.browser_process.exit_code() {
146            return Err(anyhow!(
147                "Browser exited before its DevTools endpoint was ready (exit {code})"
148            ));
149        }
150        if !owned && read_dev_tools_active_port(request.user_data_dir) == Some(port) {
151            owned = true;
152        }
153
154        if owned {
155            let remaining = request.timeout.saturating_sub(started.elapsed());
156            // The HTTP handler can lag the listening line by a moment, so an
157            // unanswered probe is simply retried.
158            if let Some(served) = probe(port, remaining.min(PROBE_TIMEOUT)).await {
159                if let Some(announced) = output.listening.as_ref().map(|listening| &listening.url) {
160                    if &served != announced {
161                        return Err(PortRaceError::new(
162                            port,
163                            format!("port serves {served}, browser announced {announced}"),
164                        )
165                        .into());
166                    }
167                }
168                return Ok(endpoint);
169            }
170        }
171        tokio::time::sleep(POLL_INTERVAL).await;
172    }
173    Err(anyhow!(
174        "Timed out after {}ms waiting for the DevTools endpoint on port {port}",
175        request.timeout.as_millis()
176    ))
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use crate::browser::browser_process::fake::FakeProcess;
183    use std::sync::atomic::{AtomicUsize, Ordering};
184    use std::sync::Arc;
185
186    const ANNOUNCED: &str = "ws://127.0.0.1:40001/devtools/browser/ours";
187
188    fn request<'a>(
189        process: &'a BrowserProcess,
190        watcher: &'a DevToolsOutputWatcher,
191        timeout: Duration,
192    ) -> CdpEndpointRequest<'a> {
193        CdpEndpointRequest {
194            remote_debugging_port: 40001,
195            user_data_dir: Path::new("/nonexistent/browser-commander-profile"),
196            browser_process: process,
197            dev_tools_output: Some(watcher),
198            timeout,
199        }
200    }
201
202    #[tokio::test]
203    async fn returns_the_endpoint_once_ownership_is_confirmed() {
204        let process = FakeProcess::new().handle();
205        let watcher = DevToolsOutputWatcher::new();
206        watcher.push(format!("\nDevTools listening on {ANNOUNCED}\n").as_bytes());
207
208        let endpoint = wait_for_cdp_endpoint_with(
209            request(&process, &watcher, Duration::from_secs(2)),
210            |_, _| async { Some(ANNOUNCED.to_owned()) },
211        )
212        .await
213        .unwrap();
214
215        assert_eq!(endpoint, "http://127.0.0.1:40001");
216    }
217
218    #[tokio::test]
219    async fn reports_a_race_when_the_port_serves_another_browser() {
220        let process = FakeProcess::new().handle();
221        let watcher = DevToolsOutputWatcher::new();
222        watcher.push(format!("DevTools listening on {ANNOUNCED}\n").as_bytes());
223
224        let error = wait_for_cdp_endpoint_with(
225            request(&process, &watcher, Duration::from_secs(2)),
226            |_, _| async { Some("ws://127.0.0.1:40001/devtools/browser/theirs".to_owned()) },
227        )
228        .await
229        .unwrap_err();
230
231        let race = error
232            .downcast_ref::<PortRaceError>()
233            .expect("a PortRaceError");
234        assert_eq!(race.port, 40001);
235        assert!(error.to_string().contains("browser announced"), "{error}");
236    }
237
238    #[tokio::test]
239    async fn reports_a_race_on_the_ipv6_fallback_without_probing() {
240        let process = FakeProcess::new().handle();
241        let watcher = DevToolsOutputWatcher::new();
242        watcher.push(b"DevTools listening on ws://[::1]:40001/devtools/browser/ours\n");
243        let probes = AtomicUsize::new(0);
244
245        let error = wait_for_cdp_endpoint_with(
246            request(&process, &watcher, Duration::from_secs(2)),
247            |_, _| {
248                probes.fetch_add(1, Ordering::SeqCst);
249                async { None }
250            },
251        )
252        .await
253        .unwrap_err();
254
255        assert!(error.downcast_ref::<PortRaceError>().is_some(), "{error}");
256        assert_eq!(probes.load(Ordering::SeqCst), 0);
257    }
258
259    #[tokio::test]
260    async fn does_not_probe_before_the_browser_claims_the_port() {
261        let process = FakeProcess::new().handle();
262        let watcher = DevToolsOutputWatcher::new();
263        let probes = Arc::new(AtomicUsize::new(0));
264        let counter = Arc::clone(&probes);
265
266        let error = wait_for_cdp_endpoint_with(
267            request(&process, &watcher, Duration::from_millis(250)),
268            move |_, _| {
269                counter.fetch_add(1, Ordering::SeqCst);
270                async { Some(ANNOUNCED.to_owned()) }
271            },
272        )
273        .await
274        .unwrap_err();
275
276        assert!(error.to_string().starts_with("Timed out"), "{error}");
277        assert_eq!(probes.load(Ordering::SeqCst), 0);
278    }
279
280    #[tokio::test]
281    async fn reports_an_early_exit() {
282        let fake = FakeProcess::new();
283        fake.exit(21);
284        let process = fake.handle();
285        let watcher = DevToolsOutputWatcher::new();
286
287        let error = wait_for_cdp_endpoint_with(
288            request(&process, &watcher, Duration::from_secs(2)),
289            |_, _| async { None },
290        )
291        .await
292        .unwrap_err();
293
294        assert!(error.to_string().contains("(exit 21)"), "{error}");
295    }
296
297    #[tokio::test]
298    async fn confirms_ownership_through_dev_tools_active_port() {
299        let profile =
300            crate::browser::profile_directory::create_temporary_user_data_dir(None).unwrap();
301        std::fs::write(
302            profile.join("DevToolsActivePort"),
303            "40001\n/devtools/browser/x",
304        )
305        .unwrap();
306        let process = FakeProcess::new().handle();
307
308        let endpoint = wait_for_cdp_endpoint_with(
309            CdpEndpointRequest {
310                remote_debugging_port: 40001,
311                user_data_dir: &profile,
312                browser_process: &process,
313                dev_tools_output: None,
314                timeout: Duration::from_secs(2),
315            },
316            |_, _| async { Some(ANNOUNCED.to_owned()) },
317        )
318        .await
319        .unwrap();
320
321        assert_eq!(endpoint, "http://127.0.0.1:40001");
322        std::fs::remove_dir_all(profile).unwrap();
323    }
324
325    #[tokio::test]
326    async fn cdp_probe_does_not_wait_for_the_server_to_close_the_connection() {
327        let listener = tokio::net::TcpListener::bind((LOOPBACK_HOST, 0))
328            .await
329            .unwrap();
330        let port = listener.local_addr().unwrap().port();
331        let response_body = r#"{"webSocketDebuggerUrl":"ws://127.0.0.1/devtools/browser/id"}"#;
332        let response = format!(
333            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{response_body}",
334            response_body.len()
335        );
336        let server = tokio::spawn(async move {
337            let (mut stream, _) = listener.accept().await.unwrap();
338            let mut request = [0_u8; 1024];
339            let _ = stream.read(&mut request).await.unwrap();
340            stream.write_all(response.as_bytes()).await.unwrap();
341            tokio::time::sleep(Duration::from_secs(1)).await;
342        });
343
344        assert_eq!(
345            fetch_cdp_version(port, Duration::from_millis(200)).await,
346            Some("ws://127.0.0.1/devtools/browser/id".to_owned())
347        );
348        server.abort();
349    }
350}
browser_commander/browser/cdp_endpoint.rs

browser_commander/browser/
cdp_endpoint.rs
