ferroui_devtools/
server.rs

1use crate::Browser;
2use ferroui_base::threading::{Dispatcher, DispatcherPriority, DispatcherTimer};
3use serde_json::{json, Value};
4use std::cell::RefCell;
5use std::collections::HashMap;
6use std::io::{self, Cursor, Read, Write};
7use std::net::{Ipv4Addr, Shutdown, SocketAddr, TcpListener, TcpStream};
8use std::rc::{Rc, Weak};
9use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
10use std::sync::{mpsc, Arc, Mutex};
11use std::thread::{self, JoinHandle};
12use std::time::{Duration, Instant};
13use tungstenite::protocol::WebSocketConfig;
14use tungstenite::{Message, WebSocket};
15
16const MAX_CLIENTS: usize = 16;
17const MAX_MESSAGE: usize = 1024 * 1024;
18const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
19enum Operation {
20    List,
21    Connect(Option<String>),
22    Message(u64, String),
23    Poll(u64),
24    Disconnect(u64),
25}
26struct Request {
27    operation: Operation,
28    reply: mpsc::SyncSender<Value>,
29    deadline: Instant,
30}
31/// Hands a request to the UI thread; `false` when it cannot be queued.
32type Submit = Arc<dyn Fn(Request) -> bool + Send + Sync>;
33#[derive(Clone)]
34struct Transport {
35    submit: Submit,
36    stop: Arc<AtomicBool>,
37}
38impl Transport {
39    fn request(&self, operation: Operation) -> io::Result<Value> {
40        let (reply, receiver) = mpsc::sync_channel(1);
41        let deadline = Instant::now() + REQUEST_TIMEOUT;
42        if !(self.submit)(Request {
43            operation,
44            reply,
45            deadline,
46        }) {
47            return Err(io::Error::other("UI dispatcher unavailable"));
48        }
49        loop {
50            if self.stop.load(Ordering::Relaxed) || Instant::now() >= deadline {
51                return Err(io::Error::new(
52                    io::ErrorKind::TimedOut,
53                    "UI dispatcher unavailable",
54                ));
55            }
56            match receiver.recv_timeout(Duration::from_millis(50)) {
57                Ok(v) => return Ok(v),
58                Err(mpsc::RecvTimeoutError::Timeout) => {}
59                Err(_) => return Err(io::Error::other("UI dispatcher closed")),
60            }
61        }
62    }
63}
64
65ferroui_base::ferro_thread_local! {
66    /// The browsers of the servers of this UI thread, by server.
67    static BROWSERS: RefCell<HashMap<u64, Weak<RefCell<Browser>>>> = RefCell::new(HashMap::new());
68}
69static NEXT_SERVER: AtomicU64 = AtomicU64::new(1);
70
71/// Runs a request on the UI thread, where the transport posted it. A command that runs a nested
72/// dispatcher frame (a modal drag loop, a synchronous wait) still holds the browser: a request
73/// that arrives inside it waits for a timer tick and tries again.
74fn run(server: u64, port: u16, request: Request) {
75    if Instant::now() >= request.deadline {
76        return;
77    }
78    let Some(browser) = BROWSERS.with(|b| b.borrow().get(&server).and_then(Weak::upgrade)) else {
79        return;
80    };
81    let Ok(mut browser) = browser.try_borrow_mut() else {
82        let request = RefCell::new(Some(request));
83        DispatcherTimer::run_once(
84            move || {
85                if let Some(request) = request.borrow_mut().take() {
86                    run(server, port, request);
87                }
88            },
89            Duration::from_millis(10),
90            DispatcherPriority::BACKGROUND,
91        );
92        return;
93    };
94    let value = match request.operation {
95        Operation::List => browser.discovery(port),
96        Operation::Connect(target) => match browser.connect(target.as_deref()) {
97            Ok(id) => json!({"connection":id}),
98            Err(e) => json!({"error":e}),
99        },
100        Operation::Message(id, message) => json!(browser.handle(id, &message)),
101        Operation::Poll(id) => json!(browser.poll(id)),
102        Operation::Disconnect(id) => {
103            browser.disconnect(id);
104            Value::Null
105        }
106    };
107    let _ = request.reply.send(value);
108}
109
110/// An explicitly started localhost listener. Keep the handle on the UI thread for the lifetime
111/// of the endpoint. Dropping it stops the listener, closes clients and removes highlights/logging.
112/// The UI thread does work only for the requests of clients: each is posted to its dispatcher.
113pub struct DevToolsServer {
114    id: u64,
115    address: SocketAddr,
116    browser: Rc<RefCell<Browser>>,
117    stop: Arc<AtomicBool>,
118    sockets: Arc<Mutex<HashMap<u64, TcpStream>>>,
119    worker: Option<JoinHandle<()>>,
120}
121impl DevToolsServer {
122    /// Start on 127.0.0.1. Port zero requests an available port. Must run on the UI thread after
123    /// platform setup; windows opened later are discovered automatically.
124    pub fn start(port: u16) -> io::Result<Self> {
125        Self::with_browser(port, Browser::new())
126    }
127    pub fn with_browser(port: u16, browser: Browser) -> io::Result<Self> {
128        let dispatcher = Dispatcher::ui_thread();
129        dispatcher.verify_access();
130        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, port))?;
131        listener.set_nonblocking(true)?;
132        let address = listener.local_addr()?;
133        let browser = Rc::new(RefCell::new(browser));
134        let id = NEXT_SERVER.fetch_add(1, Ordering::Relaxed);
135        BROWSERS.with(|b| b.borrow_mut().insert(id, Rc::downgrade(&browser)));
136        let stop = Arc::new(AtomicBool::new(false));
137        let port = address.port();
138        let transport = Transport {
139            submit: Arc::new(move |request| {
140                dispatcher.post(
141                    move || run(id, port, request),
142                    DispatcherPriority::BACKGROUND,
143                );
144                true
145            }),
146            stop: stop.clone(),
147        };
148        let sockets = Arc::new(Mutex::new(HashMap::new()));
149        let active = sockets.clone();
150        let worker = thread::Builder::new()
151            .name("ferroui-cdp".into())
152            .spawn(move || {
153                let mut workers: Vec<JoinHandle<()>> = Vec::new();
154                let mut next = 1;
155                while !transport.stop.load(Ordering::Relaxed) {
156                    let mut index = 0;
157                    while index < workers.len() {
158                        if workers[index].is_finished() {
159                            let _ = workers.swap_remove(index).join();
160                        } else {
161                            index += 1;
162                        }
163                    }
164                    match listener.accept() {
165                        Ok((stream, _)) => {
166                            if workers.len() >= MAX_CLIENTS {
167                                let _ = stream.shutdown(Shutdown::Both);
168                                continue;
169                            }
170                            let id = next;
171                            next += 1;
172                            if let Ok(clone) = stream.try_clone() {
173                                active.lock().unwrap().insert(id, clone);
174                            }
175                            let active = active.clone();
176                            let transport = transport.clone();
177                            workers.push(thread::spawn(move || {
178                                if let Err(error) = serve(stream, address.port(), &transport) {
179                                    if std::env::var_os("FERROUI_CDP_TRACE").is_some() {
180                                        eprintln!("CDP connection: {error}");
181                                    }
182                                }
183                                active.lock().unwrap().remove(&id);
184                            }));
185                        }
186                        // A connection reset before it was accepted, or descriptors running out,
187                        // fails this accept only; the listener stays.
188                        Err(_) => thread::sleep(Duration::from_millis(10)),
189                    }
190                }
191                for worker in workers {
192                    let _ = worker.join();
193                }
194            })?;
195        Ok(Self {
196            id,
197            address,
198            browser,
199            stop,
200            sockets,
201            worker: Some(worker),
202        })
203    }
204    pub fn local_addr(&self) -> SocketAddr {
205        self.address
206    }
207    /// Register a window from a custom lifetime, on the UI thread.
208    pub fn register(&self, root: &ferroui_base::Ref<ferroui_controls::TopLevel>) -> String {
209        self.browser.borrow_mut().register(root)
210    }
211}
212impl Drop for DevToolsServer {
213    fn drop(&mut self) {
214        self.stop.store(true, Ordering::Relaxed);
215        BROWSERS.with(|b| b.borrow_mut().remove(&self.id));
216        for socket in self.sockets.lock().unwrap().values() {
217            let _ = socket.shutdown(Shutdown::Both);
218        }
219        if let Some(worker) = self.worker.take() {
220            let _ = worker.join();
221        }
222    }
223}
224
225struct PrefixedStream {
226    prefix: Cursor<Vec<u8>>,
227    stream: TcpStream,
228}
229impl Read for PrefixedStream {
230    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
231        if self.prefix.position() < (self.prefix.get_ref().len() as u64) {
232            self.prefix.read(buffer)
233        } else {
234            self.stream.read(buffer)
235        }
236    }
237}
238impl Write for PrefixedStream {
239    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
240        self.stream.write(buffer)
241    }
242    fn flush(&mut self) -> io::Result<()> {
243        self.stream.flush()
244    }
245}
246
247fn serve(mut stream: TcpStream, port: u16, transport: &Transport) -> io::Result<()> {
248    stream.set_nonblocking(false)?;
249    stream.set_read_timeout(Some(Duration::from_secs(2)))?;
250    stream.set_write_timeout(Some(Duration::from_secs(2)))?;
251    let mut header = Vec::new();
252    let mut byte = [0];
253    // The whole head within a deadline, not only each byte, so slow clients cannot hold the slots.
254    let started = Instant::now();
255    while !header.ends_with(b"\r\n\r\n") {
256        if header.len() >= 16 * 1024 {
257            return http(&mut stream, 431, json!({"error":"Headers too large"}));
258        }
259        if started.elapsed() > Duration::from_secs(5) {
260            return Err(io::Error::new(
261                io::ErrorKind::TimedOut,
262                "Request head too slow",
263            ));
264        }
265        stream.read_exact(&mut byte)?;
266        header.push(byte[0]);
267    }
268    let text = std::str::from_utf8(&header).map_err(io::Error::other)?;
269    let mut lines = text.split("\r\n");
270    let mut start = lines.next().unwrap_or("").split_whitespace();
271    let method = start.next().unwrap_or("");
272    let path = start.next().unwrap_or("").trim_end_matches('/');
273    if method != "GET" {
274        return http(&mut stream, 405, json!({"error":"GET required"}));
275    }
276    let mut headers = HashMap::new();
277    for line in lines.filter(|l| !l.is_empty()) {
278        let (name, value) = line
279            .split_once(':')
280            .ok_or_else(|| io::Error::other("Malformed header"))?;
281        if headers
282            .insert(name.trim().to_ascii_lowercase(), value.trim().to_owned())
283            .is_some()
284        {
285            return http(&mut stream, 400, json!({"error":"Duplicate header"}));
286        }
287    }
288    // Reject DNS rebinding and browser pages attempting to operate a local UI.
289    let allowed_hosts = [format!("127.0.0.1:{port}"), format!("localhost:{port}")];
290    if !headers
291        .get("host")
292        .is_some_and(|h| allowed_hosts.contains(h))
293    {
294        return http(&mut stream, 403, json!({"error":"Invalid Host"}));
295    }
296    if headers
297        .get("origin")
298        .is_some_and(|o| o != "devtools://devtools" && o != "chrome-devtools://devtools")
299    {
300        return http(&mut stream, 403, json!({"error":"Origin is not allowed"}));
301    }
302    if headers
303        .get("upgrade")
304        .is_some_and(|v| v.eq_ignore_ascii_case("websocket"))
305    {
306        let target = if path == "/devtools/browser" {
307            None
308        } else if let Some(id) = path
309            .strip_prefix("/devtools/page/")
310            .filter(|s| !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()))
311        {
312            Some(id.to_owned())
313        } else {
314            return http(
315                &mut stream,
316                404,
317                json!({"error":"Unknown WebSocket target"}),
318            );
319        };
320        let result = transport.request(Operation::Connect(target))?;
321        let Some(connection) = result["connection"].as_u64() else {
322            return http(&mut stream, 404, result);
323        };
324        let stream = PrefixedStream {
325            prefix: Cursor::new(header),
326            stream,
327        };
328        let config = WebSocketConfig::default()
329            .max_message_size(Some(MAX_MESSAGE))
330            .max_frame_size(Some(MAX_MESSAGE));
331        let result = match tungstenite::accept_with_config(stream, Some(config)) {
332            Ok(mut socket) => {
333                socket
334                    .get_mut()
335                    .stream
336                    .set_read_timeout(Some(Duration::from_millis(100)))?;
337                websocket(&mut socket, connection, transport)
338            }
339            Err(e) => Err(io::Error::other(e.to_string())),
340        };
341        // Always release UI-side session state, even after an unsuccessful handshake.
342        let _ = transport.request(Operation::Disconnect(connection));
343        result
344    } else {
345        match path {
346            "/json" | "/json/" | "/json/list" => match transport.request(Operation::List) {
347                Ok(list) => http(&mut stream, 200, list),
348                Err(_) => http(
349                    &mut stream,
350                    503,
351                    json!({"error":"The UI thread did not answer"}),
352                ),
353            },
354            "/json/version" => http(&mut stream, 200, Browser::version(port)),
355            "/json/protocol" => http(
356                &mut stream,
357                200,
358                serde_json::from_str(include_str!("protocol.json"))
359                    .expect("embedded protocol is valid"),
360            ),
361            _ => http(&mut stream, 404, json!({"error":"Unknown endpoint"})),
362        }
363    }
364}
365fn http(stream: &mut TcpStream, status: u16, body: Value) -> io::Result<()> {
366    let reason = match status {
367        200 => "OK",
368        400 => "Bad Request",
369        403 => "Forbidden",
370        404 => "Not Found",
371        405 => "Method Not Allowed",
372        431 => "Request Header Fields Too Large",
373        503 => "Service Unavailable",
374        _ => "Error",
375    };
376    let body = body.to_string();
377    write!(stream,"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json; charset=UTF-8\r\nContent-Length: {}\r\nConnection: close\r\nCache-Control: no-store\r\n\r\n{body}",body.len())?;
378    stream.flush()
379}
380fn send(socket: &mut WebSocket<PrefixedStream>, messages: Value) -> io::Result<()> {
381    if let Some(messages) = messages.as_array() {
382        for message in messages {
383            socket
384                .send(Message::Text(message.to_string().into()))
385                .map_err(|e| io::Error::other(e.to_string()))?;
386        }
387    }
388    Ok(())
389}
390fn websocket(
391    socket: &mut WebSocket<PrefixedStream>,
392    connection: u64,
393    transport: &Transport,
394) -> io::Result<()> {
395    let mut last_poll = Instant::now();
396    while !transport.stop.load(Ordering::Relaxed) {
397        match socket.read() {
398            Ok(Message::Text(message)) => send(
399                socket,
400                transport.request(Operation::Message(connection, message.to_string()))?,
401            )?,
402            Ok(Message::Close(_)) => {
403                let _ = socket.flush();
404                break;
405            }
406            Ok(Message::Ping(_)) => {
407                socket
408                    .flush()
409                    .map_err(|e| io::Error::other(e.to_string()))?;
410            }
411            Ok(Message::Binary(_)) => return Err(io::Error::other("CDP expects text frames")),
412            Ok(_) => {}
413            Err(tungstenite::Error::Io(e))
414                if matches!(
415                    e.kind(),
416                    io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
417                ) => {}
418            Err(e) => return Err(io::Error::other(e.to_string())),
419        }
420        if last_poll.elapsed() >= Duration::from_millis(250) {
421            send(socket, transport.request(Operation::Poll(connection))?)?;
422            last_poll = Instant::now();
423        }
424    }
425    Ok(())
426}
427
428#[cfg(test)]
429mod tests {
430    use super::*;
431    fn endpoint() -> (
432        SocketAddr,
433        JoinHandle<()>,
434        Arc<AtomicBool>,
435        mpsc::Receiver<Request>,
436    ) {
437        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
438        let address = listener.local_addr().unwrap();
439        let (sender, receiver) = mpsc::sync_channel(128);
440        let stop = Arc::new(AtomicBool::new(false));
441        let transport = Transport {
442            submit: Arc::new(move |request| sender.try_send(request).is_ok()),
443            stop: stop.clone(),
444        };
445        let worker = thread::spawn(move || {
446            let (stream, _) = listener.accept().unwrap();
447            let _ = serve(stream, address.port(), &transport);
448        });
449        (address, worker, stop, receiver)
450    }
451    fn get(path: &str, headers: &str) -> String {
452        let (address, worker, _, _receiver) = endpoint();
453        let mut stream = TcpStream::connect(address).unwrap();
454        write!(
455            stream,
456            "GET {path} HTTP/1.1\r\nHost: {address}\r\n{headers}\r\n"
457        )
458        .unwrap();
459        let mut response = String::new();
460        stream.read_to_string(&mut response).unwrap();
461        worker.join().unwrap();
462        response
463    }
464    #[test]
465    fn http_discovery_protocol_and_origin_rejection() {
466        let version = get("/json/version", "");
467        assert!(version.starts_with("HTTP/1.1 200"));
468        assert!(version.contains("webSocketDebuggerUrl"));
469        let schema = get("/json/protocol", "");
470        assert!(schema.contains("getDocument"));
471        assert!(schema.contains("dispatchMouseEvent"));
472        assert!(get("/missing", "").starts_with("HTTP/1.1 404"));
473        assert!(get("/json/version", "Origin: https://example.org\r\n").starts_with("HTTP/1.1 403"));
474        let (address, worker, _, _receiver) = endpoint();
475        let mut stream = TcpStream::connect(address).unwrap();
476        write!(
477            stream,
478            "GET /json/version HTTP/1.1\r\nHost: attacker.example:{}\r\n\r\n",
479            address.port()
480        )
481        .unwrap();
482        let mut response = String::new();
483        stream.read_to_string(&mut response).unwrap();
484        assert!(response.starts_with("HTTP/1.1 403"));
485        worker.join().unwrap();
486    }
487    #[test]
488    fn websocket_round_trip_and_disconnect_release_ui_session() {
489        let (address, worker, stop, receiver) = endpoint();
490        let ui = thread::spawn(move || {
491            let request = receiver.recv().unwrap();
492            assert!(matches!(request.operation, Operation::Connect(None)));
493            request.reply.send(json!({"connection":7})).unwrap();
494            let request = receiver.recv().unwrap();
495            let Operation::Message(id, message) = request.operation else {
496                panic!("expected command")
497            };
498            assert_eq!(id, 7);
499            assert_eq!(
500                serde_json::from_str::<Value>(&message).unwrap()["method"],
501                "DOM.getDocument"
502            );
503            request
504                .reply
505                .send(json!([{"id":1,"result":{"root":{"nodeId":1}}}]))
506                .unwrap();
507            loop {
508                let request = receiver.recv().unwrap();
509                match request.operation {
510                    Operation::Disconnect(7) => {
511                        request.reply.send(Value::Null).unwrap();
512                        break;
513                    }
514                    Operation::Poll(7) => request.reply.send(json!([])).unwrap(),
515                    _ => panic!("unexpected operation"),
516                }
517            }
518        });
519        let (mut ws, _) = tungstenite::connect(format!("ws://{address}/devtools/browser")).unwrap();
520        ws.send(Message::Text(
521            r#"{"id":1,"method":"DOM.getDocument"}"#.into(),
522        ))
523        .unwrap();
524        let response = ws.read().unwrap();
525        assert_eq!(
526            serde_json::from_str::<Value>(response.to_text().unwrap()).unwrap()["result"]["root"]
527                ["nodeId"],
528            1
529        );
530        ws.close(None).unwrap();
531        let _ = ws.read();
532        worker.join().unwrap();
533        ui.join().unwrap();
534        stop.store(true, Ordering::Relaxed);
535    }
536}
ferroui_devtools/server.rs

ferroui_devtools/
server.rs
