Initial web-transport-quiche support

Cargo.toml

@@ -2,6 +2,7 @@
 members = [
     "web-transport",
     "web-transport-proto",
+    "web-transport-quiche",
     "web-transport-quinn",
     "web-transport-trait",
     "web-transport-wasm",

web-demo/client.js

@@ -0,0 +1,75 @@
+// @ts-expect-error embed the certificate fingerprint using bundler
+import fingerprintHex from "bundle-text:../dev/localhost.hex";
+
+// Convert the hex to binary.
+const fingerprint = [];
+for (let c = 0; c < fingerprintHex.length - 1; c += 2) {
+	fingerprint.push(parseInt(fingerprintHex.substring(c, c + 2), 16));
+}
+
+const params = new URLSearchParams(window.location.search);
+
+const url = params.get("url") || "https://localhost:4443";
+const datagram = params.get("datagram") || false;
+
+function log(msg) {
+	const element = document.createElement("div");
+	element.innerText = msg;
+
+	document.body.appendChild(element);
+}
+
+async function run() {
+	// Connect using the hex fingerprint in the cert folder.
+	const transport = new WebTransport(url, {
+		serverCertificateHashes: [
+			{
+				algorithm: "sha-256",
+				value: new Uint8Array(fingerprint),
+			},
+		],
+	});
+	await transport.ready;
+
+	log("connected");
+
+	let writer;
+	let reader;
+
+	if (!datagram) {
+		// Create a bidirectional stream
+		const stream = await transport.createBidirectionalStream();
+		log("created stream");
+
+		writer = stream.writable.getWriter();
+		reader = stream.readable.getReader();
+	} else {
+		log("using datagram");
+
+		// Create a datagram
+		writer = transport.datagrams.writable.getWriter();
+		reader = transport.datagrams.readable.getReader();
+	}
+
+	// Create a message
+	const msg = "Hello, world!";
+	const encoded = new TextEncoder().encode(msg);
+
+	await writer.write(encoded);
+	await writer.close();
+	writer.releaseLock();
+
+	log("send: " + msg);
+
+	// Read a message from it
+	// TODO handle partial reads
+	const { value } = await reader.read();
+
+	const recv = new TextDecoder().decode(value);
+	log("recv: " + recv);
+
+	transport.close();
+	log("closed");
+}
+
+run();

web-transport-proto/Cargo.toml

@@ -16,4 +16,7 @@ categories = ["network-programming", "web-programming"]
 bytes = "1"
 http = "1"
 thiserror = "2"
+
+# Just for AsyncRead and AsyncWrite traits
+tokio = { version = "1", default-features = false }
 url = "2"

web-transport-proto/src/capsule.rs

@@ -1,4 +1,5 @@
-use bytes::{Buf, BufMut, Bytes};
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 use crate::{VarInt, VarIntUnexpectedEnd};
 
@@ -68,6 +69,22 @@ impl Capsule {
         }
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, CapsuleError> {
+        let mut buf = Vec::new();
+        loop {
+            stream
+                .read_buf(&mut buf)
+                .await
+                .map_err(|_| CapsuleError::UnexpectedEnd)?;
+            let mut limit = std::io::Cursor::new(&buf);
+            match Self::decode(&mut limit) {
+                Ok(capsule) => return Ok(capsule),
+                Err(CapsuleError::UnexpectedEnd) => continue,
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
     pub fn encode<B: BufMut>(&self, buf: &mut B) {
         match self {
             Self::CloseWebTransportSession {
@@ -101,6 +118,16 @@ impl Capsule {
             }
         }
     }
+
+    pub async fn write<S: AsyncWrite + Unpin>(&self, stream: &mut S) -> Result<(), CapsuleError> {
+        let mut buf = BytesMut::new();
+        self.encode(&mut buf);
+        stream
+            .write_all_buf(&mut buf)
+            .await
+            .map_err(|_| CapsuleError::UnexpectedEnd)?;
+        Ok(())
+    }
 }
 
 fn is_grease(val: u64) -> bool {

web-transport-proto/src/connect.rs

@@ -1,6 +1,7 @@
 use std::str::FromStr;
 
-use bytes::{Buf, BufMut};
+use bytes::{Buf, BufMut, BytesMut};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use url::Url;
 
 use super::{qpack, Frame, VarInt};
@@ -97,6 +98,22 @@ impl ConnectRequest {
         Ok(Self { url })
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, ConnectError> {
+        let mut buf = Vec::new();
+        loop {
+            stream
+                .read_buf(&mut buf)
+                .await
+                .map_err(|_| ConnectError::UnexpectedEnd)?;
+            let mut limit = std::io::Cursor::new(&buf);
+            match Self::decode(&mut limit) {
+                Ok(request) => return Ok(request),
+                Err(ConnectError::UnexpectedEnd) => continue,
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
     pub fn encode<B: BufMut>(&self, buf: &mut B) {
         let mut headers = qpack::Headers::default();
         headers.set(":method", "CONNECT");
@@ -118,6 +135,16 @@ impl ConnectRequest {
         size.encode(buf);
         buf.put_slice(&tmp);
     }
+
+    pub async fn write<S: AsyncWrite + Unpin>(&self, stream: &mut S) -> Result<(), ConnectError> {
+        let mut buf = BytesMut::new();
+        self.encode(&mut buf);
+        stream
+            .write_all_buf(&mut buf)
+            .await
+            .map_err(|_| ConnectError::UnexpectedEnd)?;
+        Ok(())
+    }
 }
 
 #[derive(Debug)]
@@ -148,6 +175,22 @@ impl ConnectResponse {
         Ok(Self { status })
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, ConnectError> {
+        let mut buf = Vec::new();
+        loop {
+            stream
+                .read_buf(&mut buf)
+                .await
+                .map_err(|_| ConnectError::UnexpectedEnd)?;
+            let mut limit = std::io::Cursor::new(&buf);
+            match Self::decode(&mut limit) {
+                Ok(response) => return Ok(response),
+                Err(ConnectError::UnexpectedEnd) => continue,
+                Err(e) => return Err(e),
+            }
+        }
+    }
+
     pub fn encode<B: BufMut>(&self, buf: &mut B) {
         let mut headers = qpack::Headers::default();
         headers.set(":status", self.status.as_str());
@@ -162,4 +205,14 @@ impl ConnectResponse {
         size.encode(buf);
         buf.put_slice(&tmp);
     }
+
+    pub async fn write<S: AsyncWrite + Unpin>(&self, stream: &mut S) -> Result<(), ConnectError> {
+        let mut buf = BytesMut::new();
+        self.encode(&mut buf);
+        stream
+            .write_all_buf(&mut buf)
+            .await
+            .map_err(|_| ConnectError::UnexpectedEnd)?;
+        Ok(())
+    }
 }

web-transport-proto/src/error.rs

@@ -2,17 +2,17 @@
 const ERROR_FIRST: u64 = 0x52e4a40fa8db;
 const ERROR_LAST: u64 = 0x52e5ac983162;
 
-pub fn error_from_http3(code: u64) -> Option<u32> {
-    if !(ERROR_FIRST..=ERROR_LAST).contains(&code) {
+pub const fn error_from_http3(code: u64) -> Option<u32> {
+    if code < ERROR_FIRST || code > ERROR_LAST {
         return None;
     }
 
     let code = code - ERROR_FIRST;
-    let code = code / 0x1f;
+    let code = code - code / 0x1f;
 
-    Some(code.try_into().unwrap())
+    Some(code as u32)
 }
 
-pub fn error_to_http3(code: u32) -> u64 {
+pub const fn error_to_http3(code: u32) -> u64 {
     ERROR_FIRST + code as u64 + code as u64 / 0x1e
 }

web-transport-proto/src/settings.rs

@@ -4,9 +4,10 @@ use std::{
     ops::{Deref, DerefMut},
 };
 
-use bytes::{Buf, BufMut};
+use bytes::{Buf, BufMut, BytesMut};
 
 use thiserror::Error;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 use super::{Frame, StreamUni, VarInt, VarIntUnexpectedEnd};
 
@@ -96,6 +97,9 @@ pub enum SettingsError {
 
     #[error("invalid size")]
     InvalidSize,
+
+    #[error("unsupported")]
+    Unsupported,
 }
 
 // A map of settings to values.
@@ -128,11 +132,32 @@ impl Settings {
         Ok(settings)
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, SettingsError> {
+        let mut buf = Vec::new();
+
+        loop {
+            stream
+                .read_buf(&mut buf)
+                .await
+                .map_err(|_| SettingsError::UnexpectedEnd)?;
+
+            // Look at the buffer we've already read.
+            let mut limit = std::io::Cursor::new(&buf);
+
+            match Settings::decode(&mut limit) {
+                Ok(settings) => return Ok(settings),
+                Err(SettingsError::UnexpectedEnd) => continue, // More data needed.
+                Err(e) => return Err(e),
+            };
+        }
+    }
+
     pub fn encode<B: BufMut>(&self, buf: &mut B) {
         StreamUni::CONTROL.encode(buf);
         Frame::SETTINGS.encode(buf);
 
         // Encode to a temporary buffer so we can learn the length.
+        // TODO avoid doing this, just use a fixed size varint.
         let mut tmp = Vec::new();
         for (id, value) in &self.0 {
             id.encode(&mut tmp);
@@ -143,6 +168,17 @@ impl Settings {
         buf.put_slice(&tmp);
     }
 
+    pub async fn write<S: AsyncWrite + Unpin>(&self, stream: &mut S) -> Result<(), SettingsError> {
+        // TODO avoid allocating to the heap
+        let mut buf = BytesMut::new();
+        self.encode(&mut buf);
+        stream
+            .write_all_buf(&mut buf)
+            .await
+            .map_err(|_| SettingsError::UnexpectedEnd)?;
+        Ok(())
+    }
+
     pub fn enable_webtransport(&mut self, max_sessions: u32) {
         let max = VarInt::from_u32(max_sessions);
 

web-transport-proto/src/varint.rs

@@ -1,10 +1,11 @@
 // Based on Quinn: https://github.com/quinn-rs/quinn/tree/main/quinn-proto/src
 // Licensed under Apache-2.0 OR MIT
 
-use std::{convert::TryInto, fmt};
+use std::{convert::TryInto, fmt, io::Cursor};
 
 use bytes::{Buf, BufMut};
 use thiserror::Error;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 /// An integer less than 2^62
 ///
@@ -162,6 +163,31 @@ impl VarInt {
         Ok(Self(x))
     }
 
+    // Read a varint from the stream.
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, VarIntUnexpectedEnd> {
+        // 8 bytes is the max size of a varint
+        let mut buf = [0; 8];
+
+        // Read the first byte because it includes the length.
+        stream
+            .read_exact(&mut buf[0..1])
+            .await
+            .map_err(|_| VarIntUnexpectedEnd)?;
+
+        // 0b00 = 1, 0b01 = 2, 0b10 = 4, 0b11 = 8
+        let size = 1 << (buf[0] >> 6);
+        stream
+            .read_exact(&mut buf[1..size])
+            .await
+            .map_err(|_| VarIntUnexpectedEnd)?;
+
+        // Use a cursor to read the varint on the stack.
+        let mut cursor = Cursor::new(&buf[..size]);
+        let v = VarInt::decode(&mut cursor).unwrap();
+
+        Ok(v)
+    }
+
     pub fn encode<B: BufMut>(&self, w: &mut B) {
         let x = self.0;
         if x < 2u64.pow(6) {
@@ -176,6 +202,25 @@ impl VarInt {
             unreachable!("malformed VarInt")
         }
     }
+
+    pub async fn write<S: AsyncWrite + Unpin>(
+        &self,
+        stream: &mut S,
+    ) -> Result<(), VarIntUnexpectedEnd> {
+        // Super jaink but keeps everything on the stack.
+        let mut buf = [0u8; 8];
+        let mut cursor: &mut [u8] = &mut buf;
+        self.encode(&mut cursor);
+        let size = 8 - cursor.len();
+
+        let mut cursor = &buf[..size];
+        stream
+            .write_all_buf(&mut cursor)
+            .await
+            .map_err(|_| VarIntUnexpectedEnd)?;
+
+        Ok(())
+    }
 }
 
 /// Error returned when constructing a `VarInt` from a value >= 2^62