Skip to main content

cdp_client/
net.rs

1use crate::connect::format_http_error;
2use std::io::{Read, Write};
3use std::net::SocketAddr;
4use std::time::Duration;
5use wasix_net::TcpStream;
6
7pub struct HttpResponse {
8    pub body: Vec<u8>,
9}
10
11pub struct Ws {
12    stream: Box<dyn ReadWrite>,
13    buf: Vec<u8>,
14}
15
16trait ReadWrite: Read + Write {
17    fn set_nonblocking(&mut self, on: bool) -> std::io::Result<()>;
18}
19
20impl ReadWrite for TcpStream {
21    fn set_nonblocking(&mut self, on: bool) -> std::io::Result<()> {
22        TcpStream::set_nonblocking(self, on)
23    }
24}
25
26#[cfg(not(target_os = "wasi"))]
27impl ReadWrite for native_tls::TlsStream<TcpStream> {
28    fn set_nonblocking(&mut self, on: bool) -> std::io::Result<()> {
29        self.get_mut().set_nonblocking(on)
30    }
31}
32
33struct Parts {
34    scheme: String,
35    host: String,
36    port: u16,
37    path: String,
38}
39
40pub fn http_get(url: &str) -> Result<HttpResponse, String> {
41    let parts = parse_url(url)?;
42    if parts.scheme != "http" && parts.scheme != "https" {
43        return Err(format!("discovery URL must be http or https: {url}\n"));
44    }
45    let mut stream = dial(&parts)?;
46    let host_header = host_header(&parts);
47    let request = format!(
48        "GET {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nAccept: application/json\r\n\r\n",
49        parts.path, host_header
50    );
51    stream
52        .write_all(request.as_bytes())
53        .map_err(|err| format!("CDP discovery failed for {url}: {err}\n"))?;
54    let (status, _headers, body, _rest) = read_http(stream.as_mut())?;
55    if !(200..300).contains(&status) {
56        return Err(format_http_error(status, &body));
57    }
58    let _ = status;
59    Ok(HttpResponse { body })
60}
61
62pub fn ws_connect(url: &str) -> Result<Ws, String> {
63    let parts = parse_url(url)?;
64    if parts.scheme != "ws" && parts.scheme != "wss" {
65        return Err(format!("WebSocket URL must be ws or wss: {url}\n"));
66    }
67    let mut stream = dial(&parts)?;
68    let key = ws_key();
69    let host_header = host_header(&parts);
70    let request = format!(
71        "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {}\r\nSec-WebSocket-Version: 13\r\n\r\n",
72        parts.path, host_header, key
73    );
74    stream
75        .write_all(request.as_bytes())
76        .map_err(|err| connect_error(&parts, &err.to_string()))?;
77    let (status, headers, body, rest) = read_http(stream.as_mut())?;
78    if status != 101 {
79        return Err(format_http_error(status, &body));
80    }
81    let accept = headers
82        .iter()
83        .find(|(name, _)| name.eq_ignore_ascii_case("upgrade"))
84        .map(|(_, value)| value.clone())
85        .unwrap_or_default();
86    if !accept.to_ascii_lowercase().contains("websocket") && !accept.is_empty() {
87        return Err(format_http_error(status, &body));
88    }
89    Ok(Ws {
90        stream,
91        buf: rest,
92    })
93}
94
95impl Ws {
96    pub fn send_text(&mut self, text: &str) -> Result<(), String> {
97        self.note_server_close()?;
98        let frame = encode_client_frame(0x1, text.as_bytes(), mask_key(text.len()));
99        self.stream
100            .write_all(&frame)
101            .map_err(|err| format!("websocket write failed: {err}\n"))?;
102        self.stream
103            .flush()
104            .map_err(|err| format!("websocket write failed: {err}\n"))?;
105        Ok(())
106    }
107
108    fn note_server_close(&mut self) -> Result<(), String> {
109        if let Some(message) = buffered_close(&self.buf) {
110            return Err(message);
111        }
112        let _ = self.stream.set_nonblocking(true);
113        let mut tmp = [0u8; 8192];
114        let mut failed = None;
115        loop {
116            match self.stream.read(&mut tmp) {
117                Ok(0) => {
118                    failed = Some(
119                        buffered_close(&self.buf)
120                            .unwrap_or_else(|| "websocket connection closed\n".to_string()),
121                    );
122                    break;
123                }
124                Ok(n) => self.buf.extend_from_slice(&tmp[..n]),
125                Err(err)
126                    if err.kind() == std::io::ErrorKind::WouldBlock
127                        || err.kind() == std::io::ErrorKind::TimedOut
128                        || err.kind() == std::io::ErrorKind::Interrupted =>
129                {
130                    break;
131                }
132                Err(err) => {
133                    failed = Some(
134                        buffered_close(&self.buf)
135                            .unwrap_or_else(|| format!("websocket read failed: {err}\n")),
136                    );
137                    break;
138                }
139            }
140        }
141        let _ = self.stream.set_nonblocking(false);
142        if let Some(message) = failed {
143            return Err(message);
144        }
145        if let Some(message) = buffered_close(&self.buf) {
146            return Err(message);
147        }
148        Ok(())
149    }
150
151    pub fn recv_text(&mut self, deadline: std::time::Instant) -> Result<Option<String>, String> {
152        let mut fragments: Vec<u8> = Vec::new();
153        let mut kind = 0u8;
154        loop {
155            if std::time::Instant::now() >= deadline {
156                return Ok(None);
157            }
158            let frame = match next_frame(&mut self.buf, self.stream.as_mut())? {
159                Some(frame) => frame,
160                None => continue,
161            };
162            match frame.opcode {
163                0x1 | 0x2 => {
164                    kind = frame.opcode;
165                    if frame.fin {
166                        if frame.opcode == 0x2 {
167                            return Err("unexpected binary websocket frame\n".to_string());
168                        }
169                        return Ok(Some(
170                            String::from_utf8(frame.payload)
171                                .map_err(|err| format!("websocket text: {err}\n"))?,
172                        ));
173                    }
174                    fragments = frame.payload;
175                }
176                0x0 => {
177                    fragments.extend_from_slice(&frame.payload);
178                    if frame.fin {
179                        if kind == 0x2 {
180                            return Err("unexpected binary websocket frame\n".to_string());
181                        }
182                        return Ok(Some(
183                            String::from_utf8(fragments)
184                                .map_err(|err| format!("websocket text: {err}\n"))?,
185                        ));
186                    }
187                }
188                0x8 => return Err(close_message(&frame.payload)),
189                0x9 => {
190                    self.note_server_close()?;
191                    let pong = encode_client_frame(0xA, &frame.payload, mask_key(frame.payload.len()));
192                    self.stream
193                        .write_all(&pong)
194                        .map_err(|err| format!("websocket pong failed: {err}\n"))?;
195                }
196                0xA => {}
197                other => return Err(format!("unsupported websocket opcode {other}\n")),
198            }
199        }
200    }
201}
202
203struct Frame {
204    opcode: u8,
205    fin: bool,
206    payload: Vec<u8>,
207}
208
209fn next_frame(buf: &mut Vec<u8>, stream: &mut dyn ReadWrite) -> Result<Option<Frame>, String> {
210    if let Some((consumed, frame)) = try_decode(buf)? {
211        buf.drain(..consumed);
212        return Ok(Some(frame));
213    }
214    let mut tmp = [0u8; 8192];
215    match stream.read(&mut tmp) {
216        Ok(0) => Err("websocket connection closed\n".to_string()),
217        Ok(n) => {
218            buf.extend_from_slice(&tmp[..n]);
219            if let Some((consumed, frame)) = try_decode(buf)? {
220                buf.drain(..consumed);
221                Ok(Some(frame))
222            } else {
223                Ok(None)
224            }
225        }
226        Err(err)
227            if err.kind() == std::io::ErrorKind::TimedOut
228                || err.kind() == std::io::ErrorKind::WouldBlock =>
229        {
230            Ok(None)
231        }
232        Err(err) => Err(format!("websocket read failed: {err}\n")),
233    }
234}
235
236fn try_decode(buf: &[u8]) -> Result<Option<(usize, Frame)>, String> {
237    if buf.len() < 2 {
238        return Ok(None);
239    }
240    let b0 = buf[0];
241    let b1 = buf[1];
242    let fin = b0 & 0x80 != 0;
243    let opcode = b0 & 0x0f;
244    let masked = b1 & 0x80 != 0;
245    let mut len = (b1 & 0x7f) as u64;
246    let mut offset = 2usize;
247    if len == 126 {
248        if buf.len() < 4 {
249            return Ok(None);
250        }
251        len = u16::from_be_bytes([buf[2], buf[3]]) as u64;
252        offset = 4;
253    } else if len == 127 {
254        if buf.len() < 10 {
255            return Ok(None);
256        }
257        len = u64::from_be_bytes(buf[2..10].try_into().unwrap());
258        offset = 10;
259    }
260    let mask = if masked {
261        if buf.len() < offset + 4 {
262            return Ok(None);
263        }
264        let mask = [buf[offset], buf[offset + 1], buf[offset + 2], buf[offset + 3]];
265        offset += 4;
266        Some(mask)
267    } else {
268        None
269    };
270    let len_usize = usize::try_from(len).map_err(|_| "websocket frame too large\n".to_string())?;
271    if buf.len() < offset + len_usize {
272        return Ok(None);
273    }
274    let mut payload = buf[offset..offset + len_usize].to_vec();
275    if let Some(mask) = mask {
276        for (i, byte) in payload.iter_mut().enumerate() {
277            *byte ^= mask[i % 4];
278        }
279    }
280    Ok(Some((
281        offset + len_usize,
282        Frame {
283            opcode,
284            fin,
285            payload,
286        },
287    )))
288}
289
290pub fn encode_client_frame(opcode: u8, payload: &[u8], mask: [u8; 4]) -> Vec<u8> {
291    let mut out = Vec::new();
292    out.push(0x80 | opcode);
293    let len = payload.len();
294    if len < 126 {
295        out.push(0x80 | len as u8);
296    } else if len <= u16::MAX as usize {
297        out.push(0x80 | 126);
298        out.extend_from_slice(&(len as u16).to_be_bytes());
299    } else {
300        out.push(0x80 | 127);
301        out.extend_from_slice(&(len as u64).to_be_bytes());
302    }
303    out.extend_from_slice(&mask);
304    for (i, byte) in payload.iter().enumerate() {
305        out.push(byte ^ mask[i % 4]);
306    }
307    out
308}
309
310fn buffered_close(buf: &[u8]) -> Option<String> {
311    let mut offset = 0;
312    while offset < buf.len() {
313        match try_decode(&buf[offset..]) {
314            Ok(Some((consumed, frame))) => {
315                if frame.opcode == 0x8 {
316                    return Some(close_message(&frame.payload));
317                }
318                offset += consumed;
319            }
320            _ => return None,
321        }
322    }
323    None
324}
325
326fn close_message(payload: &[u8]) -> String {
327    if payload.len() < 2 {
328        return "websocket closed\n".to_string();
329    }
330    let code = u16::from_be_bytes([payload[0], payload[1]]);
331    let reason = String::from_utf8_lossy(&payload[2..]);
332    if reason.is_empty() {
333        format!("websocket closed {code}\n")
334    } else {
335        format!("websocket closed {code} {reason}\n")
336    }
337}
338
339fn mask_key(seed: usize) -> [u8; 4] {
340    let n = seed as u32 ^ 0xA5A5_5A5A;
341    n.to_le_bytes()
342}
343
344fn ws_key() -> String {
345    let ticks = std::time::SystemTime::now()
346        .duration_since(std::time::UNIX_EPOCH)
347        .map(|d| d.as_nanos())
348        .unwrap_or(1);
349    let mut bytes = [0u8; 16];
350    let mut x = ticks ^ 0x9E37_79B9_7F4A_7C15;
351    for (i, byte) in bytes.iter_mut().enumerate() {
352        x = x
353            .wrapping_mul(0x6C62_272E_07BB_0142)
354            .wrapping_add(i as u128 + 1);
355        *byte = (x >> 11) as u8;
356    }
357    base64_encode(&bytes)
358}
359
360pub fn base64_encode(data: &[u8]) -> String {
361    const TABLE: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
362    let mut out = String::new();
363    let mut i = 0;
364    while i + 3 <= data.len() {
365        let n = ((data[i] as u32) << 16) | ((data[i + 1] as u32) << 8) | data[i + 2] as u32;
366        out.push(TABLE[((n >> 18) & 63) as usize] as char);
367        out.push(TABLE[((n >> 12) & 63) as usize] as char);
368        out.push(TABLE[((n >> 6) & 63) as usize] as char);
369        out.push(TABLE[(n & 63) as usize] as char);
370        i += 3;
371    }
372    if i < data.len() {
373        let rest = data.len() - i;
374        let mut n = (data[i] as u32) << 16;
375        if rest == 2 {
376            n |= (data[i + 1] as u32) << 8;
377        }
378        out.push(TABLE[((n >> 18) & 63) as usize] as char);
379        out.push(TABLE[((n >> 12) & 63) as usize] as char);
380        if rest == 2 {
381            out.push(TABLE[((n >> 6) & 63) as usize] as char);
382            out.push('=');
383        } else {
384            out.push('=');
385            out.push('=');
386        }
387    }
388    out
389}
390
391pub fn base64_decode(data: &str) -> Result<Vec<u8>, String> {
392    fn val(byte: u8) -> Result<u8, String> {
393        match byte {
394            b'A'..=b'Z' => Ok(byte - b'A'),
395            b'a'..=b'z' => Ok(byte - b'a' + 26),
396            b'0'..=b'9' => Ok(byte - b'0' + 52),
397            b'+' => Ok(62),
398            b'/' => Ok(63),
399            _ => Err("invalid base64\n".to_string()),
400        }
401    }
402    let bytes: Vec<u8> = data
403        .bytes()
404        .filter(|b| !b.is_ascii_whitespace())
405        .collect();
406    if bytes.len() % 4 != 0 {
407        return Err("invalid base64\n".to_string());
408    }
409    let mut out = Vec::new();
410    for chunk in bytes.chunks(4) {
411        let pad = chunk.iter().filter(|b| **b == b'=').count();
412        let a = val(chunk[0])?;
413        let b = val(chunk[1])?;
414        let c = if chunk[2] == b'=' { 0 } else { val(chunk[2])? };
415        let d = if chunk[3] == b'=' { 0 } else { val(chunk[3])? };
416        let n = ((a as u32) << 18) | ((b as u32) << 12) | ((c as u32) << 6) | d as u32;
417        out.push((n >> 16) as u8);
418        if pad < 2 {
419            out.push((n >> 8) as u8);
420        }
421        if pad < 1 {
422            out.push(n as u8);
423        }
424    }
425    Ok(out)
426}
427
428fn parse_url(url: &str) -> Result<Parts, String> {
429    let (scheme, rest) = url
430        .split_once("://")
431        .ok_or_else(|| format!("unsupported CDP URL \"{url}\"\n"))?;
432    let (authority, path) = match rest.find('/') {
433        Some(idx) => (&rest[..idx], &rest[idx..]),
434        None => (rest, "/"),
435    };
436    let (host, port_text) = if let Some(host) = authority.strip_prefix('[') {
437        let end = host
438            .find(']')
439            .ok_or_else(|| format!("bad CDP URL \"{url}\"\n"))?;
440        let host = &host[..end];
441        let after = &authority[end + 2..];
442        let port = after.strip_prefix(':');
443        (host.to_string(), port.map(str::to_string))
444    } else if let Some((host, port)) = authority.rsplit_once(':') {
445        if host.contains(':') {
446            (authority.to_string(), None)
447        } else {
448            (host.to_string(), Some(port.to_string()))
449        }
450    } else {
451        (authority.to_string(), None)
452    };
453    let default_port = match scheme {
454        "https" | "wss" => 443,
455        _ => 80,
456    };
457    let port = match port_text {
458        Some(text) => text
459            .parse::<u16>()
460            .map_err(|_| format!("bad port in CDP URL \"{url}\"\n"))?,
461        None => default_port,
462    };
463    if host.is_empty() {
464        return Err(format!("bad CDP URL \"{url}\"\n"));
465    }
466    let path = if path.is_empty() { "/".to_string() } else { path.to_string() };
467    Ok(Parts {
468        scheme: scheme.to_string(),
469        host,
470        port,
471        path,
472    })
473}
474
475fn host_header(parts: &Parts) -> String {
476    let default = matches!(
477        (parts.scheme.as_str(), parts.port),
478        ("http" | "ws", 80) | ("https" | "wss", 443)
479    );
480    if default {
481        parts.host.clone()
482    } else if parts.host.contains(':') {
483        format!("[{}]:{}", parts.host, parts.port)
484    } else {
485        format!("{}:{}", parts.host, parts.port)
486    }
487}
488
489fn connect_error(parts: &Parts, err: &str) -> String {
490    format!(
491        "CDP connection to {}:{} failed: {err}. This CLI does not launch Chrome. Pass --cdp or set SLICC_CDP_URL.\n",
492        parts.host, parts.port
493    )
494}
495
496fn dial(parts: &Parts) -> Result<Box<dyn ReadWrite>, String> {
497    let addr = resolve(&parts.host, parts.port)?;
498    let tcp = TcpStream::connect_timeout(&addr, Duration::from_secs(10)).map_err(|err| {
499        connect_error(parts, &err.to_string())
500    })?;
501    tcp.set_read_timeout(Some(Duration::from_millis(200)))
502        .map_err(|err| connect_error(parts, &err.to_string()))?;
503    tcp.set_nodelay(true).ok();
504    if parts.scheme == "https" || parts.scheme == "wss" {
505        wrap_tls(&parts.host, tcp)
506    } else {
507        Ok(Box::new(tcp))
508    }
509}
510
511fn resolve(host: &str, port: u16) -> Result<SocketAddr, String> {
512    wasix_net::resolve(host, port)
513        .map_err(|err| format!("could not resolve {host}: {err}\n"))?
514        .into_iter()
515        .next()
516        .ok_or_else(|| format!("could not resolve {host}\n"))
517}
518
519#[cfg(not(target_os = "wasi"))]
520fn wrap_tls(host: &str, tcp: TcpStream) -> Result<Box<dyn ReadWrite>, String> {
521    let connector = native_tls::TlsConnector::new()
522        .map_err(|err| format!("TLS setup failed: {err}\n"))?;
523    let stream = connector
524        .connect(host, tcp)
525        .map_err(|err| format!("TLS handshake with {host} failed: {err}\n"))?;
526    Ok(Box::new(stream))
527}
528
529#[cfg(target_os = "wasi")]
530fn wrap_tls(_host: &str, _tcp: TcpStream) -> Result<Box<dyn ReadWrite>, String> {
531    Err("wss and https are not available in this build\n".to_string())
532}
533
534fn read_http(stream: &mut dyn ReadWrite) -> Result<(u16, Vec<(String, String)>, Vec<u8>, Vec<u8>), String> {
535    let mut buf = Vec::new();
536    let header_end = loop {
537        if let Some(pos) = find_header_end(&buf) {
538            break pos;
539        }
540        let mut tmp = [0u8; 4096];
541        match stream.read(&mut tmp) {
542            Ok(0) => {
543                return Err("HTTP connection closed before headers\n".to_string());
544            }
545            Ok(n) => buf.extend_from_slice(&tmp[..n]),
546            Err(err)
547                if err.kind() == std::io::ErrorKind::TimedOut
548                    || err.kind() == std::io::ErrorKind::WouldBlock =>
549            {
550                if buf.is_empty() {
551                    continue;
552                }
553                return Err("timed out reading HTTP headers\n".to_string());
554            }
555            Err(err) => return Err(format!("HTTP read failed: {err}\n")),
556        }
557        if buf.len() > 1024 * 1024 {
558            return Err("HTTP headers too large\n".to_string());
559        }
560    };
561    let header_bytes = buf[..header_end].to_vec();
562    let mut rest = buf[header_end + 4..].to_vec();
563    let header_text = String::from_utf8_lossy(&header_bytes);
564    let mut lines = header_text.split("\r\n");
565    let status_line = lines.next().unwrap_or("");
566    let status = status_line
567        .split_whitespace()
568        .nth(1)
569        .and_then(|text| text.parse::<u16>().ok())
570        .ok_or_else(|| format!("bad HTTP status line: {status_line}\n"))?;
571    let mut headers = Vec::new();
572    for line in lines {
573        if let Some((name, value)) = line.split_once(':') {
574            headers.push((name.trim().to_string(), value.trim().to_string()));
575        }
576    }
577    let content_length = headers
578        .iter()
579        .find(|(name, _)| name.eq_ignore_ascii_case("content-length"))
580        .and_then(|(_, value)| value.parse::<usize>().ok());
581    let chunked = headers.iter().any(|(name, value)| {
582        name.eq_ignore_ascii_case("transfer-encoding") && value.to_ascii_lowercase().contains("chunked")
583    });
584    let close = headers.iter().any(|(name, value)| {
585        name.eq_ignore_ascii_case("connection") && value.eq_ignore_ascii_case("close")
586    });
587    if status == 101 && !chunked && content_length.is_none() {
588        return Ok((status, headers, Vec::new(), rest));
589    }
590    let body = if chunked {
591        read_chunked(&mut rest, stream)?
592    } else if let Some(len) = content_length {
593        while rest.len() < len {
594            let mut tmp = [0u8; 4096];
595            match stream.read(&mut tmp) {
596                Ok(0) => break,
597                Ok(n) => rest.extend_from_slice(&tmp[..n]),
598                Err(err)
599                    if err.kind() == std::io::ErrorKind::TimedOut
600                        || err.kind() == std::io::ErrorKind::WouldBlock =>
601                {
602                    continue;
603                }
604                Err(err) => return Err(format!("HTTP body read failed: {err}\n")),
605            }
606        }
607        let body = rest[..len.min(rest.len())].to_vec();
608        let extra = if status == 101 {
609            rest[len.min(rest.len())..].to_vec()
610        } else {
611            Vec::new()
612        };
613        return Ok((status, headers, body, extra));
614    } else {
615        loop {
616            let mut tmp = [0u8; 4096];
617            match stream.read(&mut tmp) {
618                Ok(0) => break,
619                Ok(n) => rest.extend_from_slice(&tmp[..n]),
620                Err(err)
621                    if err.kind() == std::io::ErrorKind::TimedOut
622                        || err.kind() == std::io::ErrorKind::WouldBlock =>
623                {
624                    break;
625                }
626                Err(err) => return Err(format!("HTTP body read failed: {err}\n")),
627            }
628            if rest.len() > 8 * 1024 * 1024 {
629                break;
630            }
631        }
632        rest
633    };
634    let _ = close;
635    Ok((status, headers, body, Vec::new()))
636}
637
638fn read_chunked(rest: &mut Vec<u8>, stream: &mut dyn ReadWrite) -> Result<Vec<u8>, String> {
639    let mut body = Vec::new();
640    loop {
641        let line_end = loop {
642            if let Some(pos) = rest.windows(2).position(|w| w == b"\r\n") {
643                break pos;
644            }
645            let mut tmp = [0u8; 2048];
646            match stream.read(&mut tmp) {
647                Ok(0) => return Err("chunked body ended early\n".to_string()),
648                Ok(n) => rest.extend_from_slice(&tmp[..n]),
649                Err(err)
650                    if err.kind() == std::io::ErrorKind::TimedOut
651                        || err.kind() == std::io::ErrorKind::WouldBlock =>
652                {
653                    continue;
654                }
655                Err(err) => return Err(format!("chunked read failed: {err}\n")),
656            }
657        };
658        let line = String::from_utf8_lossy(&rest[..line_end]).to_string();
659        rest.drain(..line_end + 2);
660        let size_text = line.split(';').next().unwrap_or("").trim();
661        let size = usize::from_str_radix(size_text, 16)
662            .map_err(|_| format!("bad chunk size {size_text}\n"))?;
663        if size == 0 {
664            return Ok(body);
665        }
666        while rest.len() < size + 2 {
667            let mut tmp = [0u8; 4096];
668            match stream.read(&mut tmp) {
669                Ok(0) => return Err("chunked body ended early\n".to_string()),
670                Ok(n) => rest.extend_from_slice(&tmp[..n]),
671                Err(err)
672                    if err.kind() == std::io::ErrorKind::TimedOut
673                        || err.kind() == std::io::ErrorKind::WouldBlock =>
674                {
675                    continue;
676                }
677                Err(err) => return Err(format!("chunked read failed: {err}\n")),
678            }
679        }
680        body.extend_from_slice(&rest[..size]);
681        rest.drain(..size + 2);
682    }
683}
684
685fn find_header_end(buf: &[u8]) -> Option<usize> {
686    buf.windows(4).position(|w| w == b"\r\n\r\n")
687}
688
689#[cfg(test)]
690mod tests {
691    use super::*;
692
693    #[test]
694    fn decodes_unmasked_text_frame() {
695        let payload = b"hello";
696        let mut frame = vec![0x81, payload.len() as u8];
697        frame.extend_from_slice(payload);
698        let parsed = try_decode(&frame).unwrap().unwrap();
699        assert_eq!(parsed.1.payload, b"hello");
700        assert!(parsed.1.fin);
701        assert_eq!(parsed.1.opcode, 0x1);
702    }
703
704    #[test]
705    fn client_frame_round_trip_mask() {
706        let encoded = encode_client_frame(0x1, b"{}", [1, 2, 3, 4]);
707        assert_eq!(encoded[0], 0x81);
708        assert_eq!(encoded[1] & 0x80, 0x80);
709        let parsed = try_decode(&encoded).unwrap().unwrap();
710        assert_eq!(parsed.1.payload, b"{}");
711    }
712
713    #[test]
714    fn base64_round_trip() {
715        assert_eq!(base64_decode(&base64_encode(b"hi")).unwrap(), b"hi");
716        assert_eq!(base64_decode(&base64_encode(&[0, 1, 2, 3, 4])).unwrap(), [0, 1, 2, 3, 4]);
717    }
718
719    #[test]
720    fn close_frame_reports_code_and_reason() {
721        assert_eq!(
722            close_message(b"\x03\xf3the CDP host is gone"),
723            "websocket closed 1011 the CDP host is gone\n"
724        );
725        assert_eq!(
726            close_message(b"\x03\xf1a message is over 268435456 bytes"),
727            "websocket closed 1009 a message is over 268435456 bytes\n"
728        );
729        assert_eq!(close_message(&[]), "websocket closed\n");
730        assert_eq!(close_message(&[0x03, 0xf3]), "websocket closed 1011\n");
731    }
732
733    #[test]
734    fn parses_opaque_browser_path() {
735        let parts = parse_url("ws://127.0.0.1:9222/devtools/browser/abc-def?x=1").unwrap();
736        assert_eq!(parts.host, "127.0.0.1");
737        assert_eq!(parts.port, 9222);
738        assert_eq!(parts.path, "/devtools/browser/abc-def?x=1");
739    }
740
741    #[cfg(not(target_os = "wasi"))]
742    fn accept_headers(listener: std::net::TcpListener) -> std::net::TcpStream {
743        let (mut sock, _) = listener.accept().unwrap();
744        let mut buf = Vec::new();
745        let mut tmp = [0u8; 1024];
746        loop {
747            let n = sock.read(&mut tmp).unwrap();
748            buf.extend_from_slice(&tmp[..n]);
749            if buf.windows(4).any(|w| w == b"\r\n\r\n") {
750                break;
751            }
752        }
753        sock
754    }
755
756    #[cfg(not(target_os = "wasi"))]
757    #[test]
758    fn http_get_uses_the_stream() {
759        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
760        let addr = listener.local_addr().unwrap();
761        let server = std::thread::spawn(move || {
762            let mut sock = accept_headers(listener);
763            sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}")
764                .unwrap();
765        });
766        let body = http_get(&format!("http://{addr}/json/version")).unwrap();
767        assert_eq!(body.body, b"{}");
768        server.join().unwrap();
769    }
770
771    #[cfg(not(target_os = "wasi"))]
772    #[test]
773    fn websocket_uses_the_stream() {
774        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
775        let addr = listener.local_addr().unwrap();
776        let server = std::thread::spawn(move || {
777            let mut sock = accept_headers(listener);
778            sock.write_all(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n")
779                .unwrap();
780            sock.write_all(&[0x81, 5, b'h', b'e', b'l', b'l', b'o']).unwrap();
781        });
782        let mut ws = ws_connect(&format!("ws://{addr}/devtools/browser/abc")).unwrap();
783        let text = ws
784            .recv_text(std::time::Instant::now() + Duration::from_secs(2))
785            .unwrap();
786        assert_eq!(text.as_deref(), Some("hello"));
787        server.join().unwrap();
788    }
789
790    #[cfg(not(target_os = "wasi"))]
791    #[test]
792    fn server_close_is_reported_and_not_answered() {
793        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
794        let addr = listener.local_addr().unwrap();
795        let server = std::thread::spawn(move || {
796            let mut sock = accept_headers(listener);
797            let reason = b"the CDP host is gone";
798            let mut message = b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n".to_vec();
799            message.extend_from_slice(&[0x81, 5, b'h', b'e', b'l', b'l', b'o']);
800            message.extend_from_slice(&[0x88, (2 + reason.len()) as u8, 0x03, 0xf3]);
801            message.extend_from_slice(reason);
802            sock.write_all(&message).unwrap();
803            sock.set_read_timeout(Some(Duration::from_millis(500))).unwrap();
804            let mut tmp = [0u8; 64];
805            match sock.read(&mut tmp) {
806                Ok(n) => n,
807                Err(err)
808                    if err.kind() == std::io::ErrorKind::WouldBlock
809                        || err.kind() == std::io::ErrorKind::TimedOut =>
810                {
811                    0
812                }
813                Err(_) => 0,
814            }
815        });
816        let mut ws = ws_connect(&format!("ws://{addr}/devtools/browser/abc")).unwrap();
817        let text = ws
818            .recv_text(std::time::Instant::now() + Duration::from_secs(2))
819            .unwrap();
820        assert_eq!(text.as_deref(), Some("hello"));
821        let err = ws.send_text("{}").unwrap_err();
822        assert_eq!(err, "websocket closed 1011 the CDP host is gone\n");
823        assert_eq!(server.join().unwrap(), 0);
824    }
825}