gatewayd: ws, the WebSocket error type and the opening handshake

Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)

crates/gatewayd/src/lib.rs

@@ -5,3 +5,4 @@ pub mod config;
 pub mod http;
 pub mod net;
 pub mod secrets;
+pub mod ws;

crates/gatewayd/src/ws/handshake.rs

@@ -0,0 +1,113 @@
+//! The opening handshake (RFC 6455, section 4.1) and the base64 it needs.
+
+use std::io::{Read, Write};
+
+use crate::http::{Head, read_head};
+use crate::ws::WsError;
+
+/// RFC 6455's magic string, appended to the key before hashing.
+pub const GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+
+/// Standard base64 with padding (RFC 4648, section 4).
+pub fn base64(bytes: &[u8]) -> String {
+    // Standard alphabet (ALPHABET), "=" padding: each 3 bytes become 4 characters; a last group of
+    // 1 or 2 bytes becomes 2 or 3 characters and 2 or 1 "=". No indexing that can go out of bounds.
+    let mut out = String::new();
+    for chunk in bytes.chunks(3) {
+        let b0 = chunk.first().copied().unwrap_or(0);
+        let b1 = chunk.get(1).copied().unwrap_or(0);
+        let b2 = chunk.get(2).copied().unwrap_or(0);
+        let n = ((b0 as u32) << 16) | ((b1 as u32) << 8) | (b2 as u32);
+        out.push(ALPHABET[((n >> 18) & 0x3f) as usize] as char);
+        out.push(ALPHABET[((n >> 12) & 0x3f) as usize] as char);
+        out.push(if chunk.len() > 1 {
+            ALPHABET[((n >> 6) & 0x3f) as usize] as char
+        } else {
+            '='
+        });
+        out.push(if chunk.len() > 2 {
+            ALPHABET[(n & 0x3f) as usize] as char
+        } else {
+            '='
+        });
+    }
+    out
+}
+
+/// The `Sec-WebSocket-Accept` a server must send for `key`.
+pub fn accept_for(key: &str) -> String {
+    // base64(sha1(key + GUID)), with `proto::sha1::sha1`.
+    let mut data = Vec::with_capacity(key.len() + GUID.len());
+    data.extend_from_slice(key.as_bytes());
+    data.extend_from_slice(GUID.as_bytes());
+    base64(&proto::sha1::sha1(&data))
+}
+
+/// A fresh key: 16 bytes from `random` (in `gatewayd`, `/dev/urandom`), in base64.
+pub fn new_key(random: &mut dyn Read) -> std::io::Result<String> {
+    // 16 bytes read from `random` with read_exact, then base64.
+    let mut key = [0u8; 16];
+    random.read_exact(&mut key)?;
+    Ok(base64(&key))
+}
+
+/// The request, exactly.
+pub fn request_text(host: &str, path: &str, key: &str, token: &str) -> String {
+    format!(
+        "GET {path} HTTP/1.1\r\nHost: {host}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
+         Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nAuthorization: Bearer {token}\r\n\r\n"
+    )
+}
+
+/// Is `head` a server's acceptance of `key`? Status 101, `Upgrade: websocket`, a `Connection`
+/// holding the token `upgrade`, and the right `Sec-WebSocket-Accept` (case matters there).
+pub fn check_response(head: &Head, key: &str) -> Result<(), WsError> {
+    // In this order, each a Handshake error: status is not 101 ("status <n>"); no Upgrade header
+    // equal to "websocket" ignoring case; no Connection header with a comma-separated token equal
+    // to "upgrade" ignoring case; Sec-WebSocket-Accept missing, or not exactly `accept_for(key)`.
+    if head.status != 101 {
+        return Err(WsError::Handshake(format!("status {}", head.status)));
+    }
+    if !head
+        .header("Upgrade")
+        .is_some_and(|v| v.eq_ignore_ascii_case("websocket"))
+    {
+        return Err(WsError::Handshake(
+            "the Upgrade header is not websocket".to_string(),
+        ));
+    }
+    let upgraded = head.header("Connection").is_some_and(|v| {
+        v.split(',')
+            .any(|tok| tok.trim().eq_ignore_ascii_case("upgrade"))
+    });
+    if !upgraded {
+        return Err(WsError::Handshake(
+            "the Connection header has no upgrade token".to_string(),
+        ));
+    }
+    match head.header("Sec-WebSocket-Accept") {
+        Some(got) if got == accept_for(key) => Ok(()),
+        _ => Err(WsError::Handshake(
+            "the Sec-WebSocket-Accept is missing or wrong".to_string(),
+        )),
+    }
+}
+
+/// The whole handshake on `stream`. Nothing after the server's head is read.
+pub fn handshake(
+    stream: &mut (impl Read + Write),
+    host: &str,
+    path: &str,
+    token: &str,
+    random: &mut dyn Read,
+) -> Result<(), WsError> {
+    // A new key; write `request_text` and flush; `read_head` (its error is a Handshake error); then
+    // `check_response`. Read nothing after the head.
+    let key = new_key(random)?;
+    stream.write_all(request_text(host, path, &key, token).as_bytes())?;
+    stream.flush()?;
+    let head = read_head(stream).map_err(|e| WsError::Handshake(e.to_string()))?;
+    check_response(&head, &key)
+}

crates/gatewayd/src/ws/mod.rs

@@ -0,0 +1,42 @@
+//! The WebSocket client (RFC 6455; M4a spec, section 6): the handshake, the frame codec, and the
+//! connection that uses them.
+
+pub mod handshake;
+
+/// Why a WebSocket ended or could not start. Every one of these ends the connection; `gatewayd`
+/// then reconnects.
+#[derive(Debug)]
+pub enum WsError {
+    /// The server's answer to the handshake was not an upgrade to a WebSocket.
+    Handshake(String),
+    /// A frame broke the protocol.
+    Protocol(String),
+    /// A message over `MAX_MESSAGE`, refused from its length fields.
+    TooLarge,
+    /// The server closed the connection (a close frame, or the end of the stream).
+    Closed,
+    /// Nothing was heard for the dead-after limit.
+    Dead,
+    Io(std::io::Error),
+}
+
+impl std::fmt::Display for WsError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            WsError::Handshake(why) => write!(f, "WebSocket handshake failed: {why}"),
+            WsError::Protocol(why) => write!(f, "WebSocket protocol error: {why}"),
+            WsError::TooLarge => write!(f, "WebSocket message too large"),
+            WsError::Closed => write!(f, "WebSocket closed"),
+            WsError::Dead => write!(f, "WebSocket silent for too long"),
+            WsError::Io(e) => write!(f, "WebSocket I/O: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for WsError {}
+
+impl From<std::io::Error> for WsError {
+    fn from(e: std::io::Error) -> Self {
+        WsError::Io(e)
+    }
+}

crates/gatewayd/tests/ws_handshake.rs

@@ -0,0 +1,200 @@
+//! Base64 and the WebSocket opening handshake (RFC 4648; RFC 6455, section 4). Do not edit.
+
+use std::io::{Cursor, Read, Write};
+
+use gatewayd::http::Head;
+use gatewayd::ws::WsError;
+use gatewayd::ws::handshake::{
+    accept_for, base64, check_response, handshake, new_key, request_text,
+};
+
+#[test]
+fn base64_vectors() {
+    for (input, want) in [
+        ("", ""),
+        ("f", "Zg=="),
+        ("fo", "Zm8="),
+        ("foo", "Zm9v"),
+        ("foob", "Zm9vYg=="),
+        ("fooba", "Zm9vYmE="),
+        ("foobar", "Zm9vYmFy"),
+    ] {
+        assert_eq!(base64(input.as_bytes()), want, "{input:?}");
+    }
+    assert_eq!(base64(&[0xff, 0xfe, 0xfd, 0x00, 0x3f]), "//79AD8=");
+    assert_eq!(
+        base64(&(0u8..=15).collect::<Vec<_>>()),
+        "AAECAwQFBgcICQoLDA0ODw=="
+    );
+}
+
+#[test]
+fn the_rfc_example_accept() {
+    assert_eq!(
+        accept_for("dGhlIHNhbXBsZSBub25jZQ=="),
+        "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
+    );
+}
+
+#[test]
+fn a_key_is_sixteen_random_bytes() {
+    let mut random = Cursor::new((0u8..=15).collect::<Vec<_>>());
+    assert_eq!(new_key(&mut random).unwrap(), "AAECAwQFBgcICQoLDA0ODw==");
+    let mut short = Cursor::new(vec![1u8; 15]);
+    assert!(
+        new_key(&mut short).is_err(),
+        "too few random bytes is an error, not a weak key"
+    );
+}
+
+#[test]
+fn the_request_is_exactly_this() {
+    assert_eq!(
+        request_text("a.example", "/api/v4/websocket", "KEY==", "TOKEN"),
+        "GET /api/v4/websocket HTTP/1.1\r\nHost: a.example\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
+         Sec-WebSocket-Key: KEY==\r\nSec-WebSocket-Version: 13\r\nAuthorization: Bearer TOKEN\r\n\r\n"
+    );
+}
+
+fn head(status: u16, headers: &[(&str, &str)]) -> Head {
+    Head {
+        status,
+        headers: headers
+            .iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect(),
+    }
+}
+
+const KEY: &str = "dGhlIHNhbXBsZSBub25jZQ==";
+const ACCEPT: &str = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=";
+
+#[test]
+fn only_a_proper_upgrade_is_accepted() {
+    let good = [
+        ("Upgrade", "websocket"),
+        ("Connection", "Upgrade"),
+        ("Sec-WebSocket-Accept", ACCEPT),
+    ];
+    assert!(check_response(&head(101, &good), KEY).is_ok());
+    let loose = [
+        ("upgrade", "WebSocket"),
+        ("connection", "keep-alive, Upgrade"),
+        ("sec-websocket-accept", ACCEPT),
+    ];
+    assert!(
+        check_response(&head(101, &loose), KEY).is_ok(),
+        "names and tokens are case-insensitive"
+    );
+    let cases: [(u16, &[(&str, &str)]); 7] = [
+        (200, &good),
+        (401, &good),
+        (
+            101,
+            &[("Connection", "Upgrade"), ("Sec-WebSocket-Accept", ACCEPT)],
+        ),
+        (
+            101,
+            &[
+                ("Upgrade", "h2c"),
+                ("Connection", "Upgrade"),
+                ("Sec-WebSocket-Accept", ACCEPT),
+            ],
+        ),
+        (
+            101,
+            &[("Upgrade", "websocket"), ("Sec-WebSocket-Accept", ACCEPT)],
+        ),
+        (101, &[("Upgrade", "websocket"), ("Connection", "Upgrade")]),
+        (
+            101,
+            &[
+                ("Upgrade", "websocket"),
+                ("Connection", "Upgrade"),
+                ("Sec-WebSocket-Accept", "s3pplmbitxaq9kygzzhzrbk+xoo="),
+            ],
+        ),
+    ];
+    for (status, headers) in cases {
+        assert!(
+            matches!(
+                check_response(&head(status, headers), KEY),
+                Err(WsError::Handshake(_))
+            ),
+            "{status} {headers:?}"
+        );
+    }
+}
+
+/// A server side scripted as bytes; records what the client wrote.
+struct Scripted {
+    input: Cursor<Vec<u8>>,
+    output: Vec<u8>,
+}
+
+impl Read for Scripted {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.input.read(buf)
+    }
+}
+
+impl Write for Scripted {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.output.write(buf)
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+#[test]
+fn a_whole_handshake_leaves_the_first_frame_unread() {
+    let key_bytes: Vec<u8> = (0u8..=15).collect();
+    let key = base64(&key_bytes);
+    let reply = format!(
+        "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {}\r\n\r\n",
+        accept_for(&key)
+    );
+    let mut bytes = reply.into_bytes();
+    bytes.extend_from_slice(&[0x81, 0x02, b'h', b'i']);
+    let mut s = Scripted {
+        input: Cursor::new(bytes),
+        output: Vec::new(),
+    };
+    handshake(
+        &mut s,
+        "a.example",
+        "/api/v4/websocket",
+        "TOKEN",
+        &mut Cursor::new(key_bytes),
+    )
+    .unwrap();
+    assert_eq!(
+        String::from_utf8(s.output).unwrap(),
+        request_text("a.example", "/api/v4/websocket", &key, "TOKEN")
+    );
+    let mut rest = Vec::new();
+    s.input.read_to_end(&mut rest).unwrap();
+    assert_eq!(rest, [0x81, 0x02, b'h', b'i']);
+}
+
+#[test]
+fn a_refused_or_broken_handshake_is_a_handshake_error() {
+    for reply in [
+        &b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n"[..],
+        b"HTTP/1.1 101 Switching",
+        b"not http at all\r\n\r\n",
+        b"",
+    ] {
+        let mut s = Scripted {
+            input: Cursor::new(reply.to_vec()),
+            output: Vec::new(),
+        };
+        let got = handshake(&mut s, "h", "/p", "t", &mut Cursor::new(vec![7u8; 16]));
+        assert!(
+            matches!(got, Err(WsError::Handshake(_))),
+            "{:?}: {got:?}",
+            String::from_utf8_lossy(reply)
+        );
+    }
+}