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",

flake.nix

@@ -30,11 +30,16 @@
           pkgs.pkg-config
           pkgs.glib
           pkgs.gtk3
+          pkgs.stdenv.cc.cc.lib
         ];
       in
       {
         devShells.default = pkgs.mkShell {
           packages = tools;
+
+          shellHook = ''
+            export LD_LIBRARY_PATH=${pkgs.lib.makeLibraryPath [pkgs.stdenv.cc.cc.lib]}:$LD_LIBRARY_PATH
+          '';
         };
       }
     );

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,7 @@
-use bytes::{Buf, BufMut, Bytes};
+use std::sync::Arc;
+
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 use crate::{VarInt, VarIntUnexpectedEnd};
 
@@ -68,6 +71,22 @@ impl Capsule {
         }
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, CapsuleError> {
+        let mut buf = Vec::new();
+        loop {
+            if stream.read_buf(&mut buf).await? == 0 {
+                return 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 +120,13 @@ 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?;
+        Ok(())
+    }
 }
 
 fn is_grease(val: u64) -> bool {
@@ -113,7 +139,7 @@ fn is_grease(val: u64) -> bool {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[derive(Debug, Clone, thiserror::Error)]
 pub enum CapsuleError {
     #[error("unexpected end of buffer")]
     UnexpectedEnd,
@@ -129,6 +155,15 @@ pub enum CapsuleError {
 
     #[error("varint decode error: {0:?}")]
     VarInt(#[from] VarIntUnexpectedEnd),
+
+    #[error("io error: {0}")]
+    Io(Arc<std::io::Error>),
+}
+
+impl From<std::io::Error> for CapsuleError {
+    fn from(err: std::io::Error) -> Self {
+        CapsuleError::Io(Arc::new(err))
+    }
 }
 
 #[cfg(test)]

web-transport-proto/src/connect.rs

@@ -1,6 +1,7 @@
-use std::str::FromStr;
+use std::{str::FromStr, sync::Arc};
 
-use bytes::{Buf, BufMut};
+use bytes::{Buf, BufMut, BytesMut};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use url::Url;
 
 use super::{qpack, Frame, VarInt};
@@ -48,6 +49,15 @@ pub enum ConnectError {
 
     #[error("non-200 status: {0:?}")]
     ErrorStatus(http::StatusCode),
+
+    #[error("io error: {0}")]
+    Io(Arc<std::io::Error>),
+}
+
+impl From<std::io::Error> for ConnectError {
+    fn from(err: std::io::Error) -> Self {
+        ConnectError::Io(Arc::new(err))
+    }
 }
 
 #[derive(Debug)]
@@ -97,6 +107,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 {
+            if stream.read_buf(&mut buf).await? == 0 {
+                return 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 +144,13 @@ 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?;
+        Ok(())
+    }
 }
 
 #[derive(Debug)]
@@ -148,6 +181,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 {
+            if stream.read_buf(&mut buf).await? == 0 {
+                return 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 +211,11 @@ 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?;
+        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

@@ -2,11 +2,13 @@ use std::{
     collections::HashMap,
     fmt::Debug,
     ops::{Deref, DerefMut},
+    sync::Arc,
 };
 
-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 +98,15 @@ pub enum SettingsError {
 
     #[error("invalid size")]
     InvalidSize,
+
+    #[error("io error: {0}")]
+    Io(Arc<std::io::Error>),
+}
+
+impl From<std::io::Error> for SettingsError {
+    fn from(err: std::io::Error) -> Self {
+        SettingsError::Io(Arc::new(err))
+    }
 }
 
 // A map of settings to values.
@@ -128,11 +139,31 @@ impl Settings {
         Ok(settings)
     }
 
+    pub async fn read<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Self, SettingsError> {
+        let mut buf = Vec::new();
+
+        loop {
+            if stream.read_buf(&mut buf).await? == 0 {
+                return 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 +174,14 @@ 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?;
+        Ok(())
+    }
+
     pub fn enable_webtransport(&mut self, max_sessions: u32) {
         let max = VarInt::from_u32(max_sessions);