fix: m0 interop audit

doc/concept/transport.md

@@ -51,7 +51,8 @@ Native clients dial the relay directly with the MoQ ALPN, skipping the HTTP/3
 handshake. The URL path and `?jwt=` token travel in the MoQ `SETUP` message
 instead of a `CONNECT` request, so the server sees the same thing either way.
 `moq-tokio` dials WebTransport for `https://` URLs; use `moql://` or `moqt://`
-for raw QUIC.
+for raw QUIC. A browser cannot dial raw QUIC, so `@moq/net` reaches a `moqt://`
+or `moql://` URL over WebTransport at the same host, port, and path.
 
 ## TCP and Unix sockets (local workers)
 

js/net/src/connection/accept.ts

@@ -1,4 +1,5 @@
 import * as Ietf from "../ietf/index.ts";
+import * as Message from "../ietf/message.ts";
 import * as Lite from "../lite/index.ts";
 import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../origin.ts";
 import { Stream } from "../stream.ts";
@@ -159,8 +160,6 @@ async function acceptSetup(
 
 	const client = await Ietf.ClientSetup.decode(stream.reader, version);
 
-	await stream.writer.u53(Lite.StreamId.ServerCompat);
-
 	const encoder = new TextEncoder();
 	const params = new Ietf.SetupOptions();
 	params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true));
@@ -171,7 +170,7 @@ async function acceptSetup(
 	Ietf.activeCountIntoSetup(params, version);
 
 	const server = new Ietf.ServerSetup({ version, parameters: params });
-	await server.encode(stream.writer, version);
+	await Message.write(stream.writer, Lite.StreamId.ServerCompat, server, version);
 
 	const maxRequestId = client.parameters.getVarint(Ietf.SetupOption.MaxRequestId) ?? 0n;
 
@@ -225,8 +224,6 @@ async function acceptNegotiated(
 		selectedVersion = match;
 	}
 
-	await stream.writer.u53(Lite.StreamId.ServerCompat);
-
 	const encoder = new TextEncoder();
 	const params = new Ietf.SetupOptions();
 	params.setVarint(Ietf.SetupOption.MaxRequestId, Ietf.initialMaxRequestId(true));
@@ -237,7 +234,7 @@ async function acceptNegotiated(
 	Ietf.activeCountIntoSetup(params, setupVersion);
 
 	const server = new Ietf.ServerSetup({ version: selectedVersion, parameters: params });
-	await server.encode(stream.writer, setupVersion);
+	await Message.write(stream.writer, Lite.StreamId.ServerCompat, server, setupVersion);
 
 	if (Object.values(Lite.Version).includes(selectedVersion as Lite.Version)) {
 		return new Lite.Connection({

js/net/src/connection/connect.test.ts

@@ -91,6 +91,43 @@ test("WebTransport negotiates lite-07-wip only when explicitly offered", async (
 	expect(version).toBe("moq-lite-07-wip");
 });
 
+test("a raw QUIC URL dials WebTransport over https", async () => {
+	const pair = createMockTransportPair(ALPN_06);
+	const original = globalThis.WebTransport;
+	let dialed: string | undefined;
+
+	function StubWebTransport(this: unknown, target: string | URL) {
+		dialed = target.toString();
+		return pair.client;
+	}
+	globalThis.WebTransport = StubWebTransport as unknown as typeof WebTransport;
+
+	try {
+		const connection = await connect(new URL("moqt://example.com:4443/anon?jwt=x"), {
+			websocket: { enabled: false },
+		});
+		connection.abort();
+	} finally {
+		globalThis.WebTransport = original;
+	}
+
+	expect(dialed).toBe("https://example.com:4443/anon?jwt=x");
+});
+
+test("a failed connect names why the transport failed", async () => {
+	const original = globalThis.WebTransport;
+	function FailingWebTransport(this: unknown): never {
+		throw new Error("handshake refused");
+	}
+	globalThis.WebTransport = FailingWebTransport as unknown as typeof WebTransport;
+
+	try {
+		await expect(connect(url, { websocket: { enabled: false } })).rejects.toThrow("handshake refused");
+	} finally {
+		globalThis.WebTransport = original;
+	}
+});
+
 test("connect logs the relay URL without its credentials", async () => {
 	const pair = createMockTransportPair(ALPN_05);
 	const restoreTransport = stubWebTransport(pair.client);

js/net/src/connection/connect.ts

@@ -2,6 +2,7 @@ import { Milli, Nano, sleep } from "@moq/clock";
 import Session, { type Version as QmuxVersion } from "@moq/qmux";
 import { error, fromClose, ProtocolViolation, sessionCause } from "../error.ts";
 import * as Ietf from "../ietf/index.ts";
+import * as Message from "../ietf/message.ts";
 import * as Lite from "../lite/index.ts";
 import type { Consumer as OriginConsumer, Producer as OriginProducer } from "../origin.ts";
 import { Stream } from "../stream.ts";
@@ -261,7 +262,19 @@ export async function connect({ url, ...props }: ConnectProps): Promise<Establis
 	}
 }
 
-async function connectInner(url: URL, props: Omit<ConnectProps, "url">, abort: Promise<void>): Promise<Established> {
+/**
+ * The URL a browser transport dials. A raw QUIC `moqt:` or `moql:` URL is reached over
+ * WebTransport with its scheme swapped for https (draft-ietf-moq-transport-18 §3.1.3); the
+ * CONNECT carries the scheme, so a server refuses the raw one. Rebuilt rather than assigned,
+ * since the URL setter refuses a change from a non-special scheme to a special one.
+ */
+function dialable(url: URL): URL {
+	if (url.protocol !== "moqt:" && url.protocol !== "moql:") return url;
+	return new URL(`https:${url.href.slice(url.protocol.length)}`);
+}
+
+async function connectInner(raw: URL, props: Omit<ConnectProps, "url">, abort: Promise<void>): Promise<Established> {
+	const url = dialable(raw);
 	const wiring: SessionProps = {
 		publish: props.publish,
 		client: true,
@@ -306,7 +319,12 @@ async function connectInner(url: URL, props: Omit<ConnectProps, "url">, abort: P
 				? [websocket, webtransport]
 				: [webtransport]
 			: [websocket],
-	);
+	).catch((err: unknown) => {
+		// Promise.any rejects with an empty message; name each transport's failure instead.
+		if (!(err instanceof AggregateError)) throw err;
+		const reasons = err.errors.map((e: unknown) => (e instanceof Error ? e.message : String(e)));
+		throw new Error(`no transport connected: ${reasons.join("; ")}`, { cause: err });
+	});
 	done();
 
 	if (!session) throw new Error("no transport available");
@@ -422,7 +440,6 @@ async function negotiate(
 	}
 
 	const stream = await Stream.open(session, { version: setupVersion });
-	await stream.writer.u53(Lite.StreamId.ClientCompat);
 
 	const encoder = new TextEncoder();
 
@@ -443,7 +460,7 @@ async function negotiate(
 					: setup,
 		parameters: params,
 	});
-	await client.encode(stream.writer, setupVersion);
+	await Message.write(stream.writer, Lite.StreamId.ClientCompat, client, setupVersion);
 
 	const serverCompat = await stream.reader.u53();
 	if (serverCompat !== Lite.StreamId.ServerCompat) {

js/net/src/connection/handshake.ts

@@ -1,6 +1,7 @@
 import { closeReason, ProtocolViolation, SessionCode, StreamError } from "../error.ts";
 import { type Hop, randomHop } from "../hop.ts";
 import * as Ietf from "../ietf/index.ts";
+import * as Message from "../ietf/message.ts";
 import { Reader, Stream, UnexpectedEnd, Writer } from "../stream.ts";
 import * as Extensions from "./extensions.ts";
 
@@ -82,8 +83,7 @@ async function sendSetup(transport: WebTransport, version: Ietf.IetfVersion, set
 	// Via Writer.open for its deadline: a peer can advertise a stream limit of zero and
 	// never raise it, which would otherwise hang the handshake rather than failing it.
 	const writer = await Writer.open(transport, { version });
-	await writer.u53(Ietf.Setup.id); // 0x2F00 stream type
-	await setupMsg.encode(writer, version);
+	await Message.write(writer, Ietf.Setup.id, setupMsg, version); // 0x2F00 stream type
 	return writer;
 }
 

js/net/src/ietf/adapter.ts

@@ -6,6 +6,7 @@ import { type OpenOptions, Reader, Stream, type Writer } from "../stream.ts";
 import { untilAborted } from "../util/abort.ts";
 import * as Varint from "../varint.ts";
 import { GoAway } from "./goaway.ts";
+import * as Message from "./message.ts";
 import * as Namespace from "./namespace.ts";
 import { MaxRequestId } from "./request.ts";
 import { refuseUpdate } from "./request_stream.ts";
@@ -759,8 +760,8 @@ export class ControlStreamAdapter implements Session {
 				const max = this.#peerMax;
 				await this.#writeMutex.runExclusive(async () => {
 					if (this.#closed) return;
-					await this.#writer.u53(MaxRequestId.id);
-					await new MaxRequestId({ requestId: max }).encode(this.#writer, this.version);
+					const grant = new MaxRequestId({ requestId: max });
+					await Message.write(this.#writer, MaxRequestId.id, grant, this.version);
 				});
 			}
 		} catch {

js/net/src/ietf/filter.ts

@@ -5,7 +5,7 @@
  */
 
 import { ProtocolViolation } from "../error.ts";
-import type { Reader } from "../stream.ts";
+import { concat, type Reader } from "../stream.ts";
 import * as Varint from "../varint.ts";
 import { type IetfVersion, Version } from "./version.ts";
 
@@ -89,17 +89,6 @@ function unvarint(buf: Uint8Array, version: IetfVersion): [bigint, Uint8Array] {
 	return usesLeadingOnes(version) ? Varint.decodeLeadingOnes(buf) : Varint.decodeBigInt(buf);
 }
 
-function concat(parts: Uint8Array[]): Uint8Array {
-	const total = parts.reduce((n, p) => n + p.length, 0);
-	const out = new Uint8Array(total);
-	let offset = 0;
-	for (const part of parts) {
-		out.set(part, offset);
-		offset += part.length;
-	}
-	return out;
-}
-
 function endDelta(startGroup: bigint, endGroup: bigint): bigint {
 	if (endGroup < startGroup) {
 		throw new Error(`filter range runs backwards: ${startGroup} > ${endGroup}`);

js/net/src/ietf/message.test.ts

@@ -0,0 +1,38 @@
+import { expect, test } from "bun:test";
+import { Reader, Writer } from "../stream.ts";
+import * as Message from "./message.ts";
+import { RequestError } from "./request.ts";
+import { type IetfVersion, Version } from "./version.ts";
+
+/** A writer that keeps every chunk its stream is handed. */
+function recording(version: IetfVersion): { writer: Writer; chunks: Uint8Array[] } {
+	const chunks: Uint8Array[] = [];
+	const writer = new Writer(
+		new WritableStream<Uint8Array>({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }),
+		version,
+	);
+	return { writer, chunks };
+}
+
+test("a control message goes out in one write, and one that fails to encode writes nothing", async () => {
+	for (const version of [Version.DRAFT_14, Version.DRAFT_17]) {
+		const { writer, chunks } = recording(version);
+		const err = new RequestError({ requestId: 3n, errorCode: 2, reasonPhrase: "nope" });
+		await Message.write(writer, RequestError.id, err, version);
+		expect(chunks.length).toBe(1);
+
+		const reader = new Reader(undefined, chunks[0], version);
+		expect(await reader.u53()).toBe(RequestError.id);
+		const decoded = await RequestError.decode(reader, version);
+		expect(decoded.reasonPhrase).toBe("nope");
+		expect(await reader.done()).toBe(true);
+
+		// A message written without a type, as on a stream whose header named it, is one write too.
+		await err.encode(writer, version);
+		expect(chunks.length).toBe(2);
+
+		const huge = new RequestError({ requestId: 3n, errorCode: 2, reasonPhrase: "x".repeat(65_536) });
+		await expect(Message.write(writer, RequestError.id, huge, version)).rejects.toThrow("Message too large");
+		expect(chunks.length).toBe(2);
+	}
+});

js/net/src/ietf/message.ts

@@ -1,56 +1,30 @@
-import { Reader, Writer } from "../stream.ts";
+import { concat, encodeVarint, Reader, Writer } from "../stream.ts";
 import { MalformedAuthToken } from "./token.ts";
+import type { IetfVersion } from "./version.ts";
 
-// Encodes a message with a u16 (16-bit) size prefix as per draft-14. A type `id` is written
-// only once the body fits, so an oversized message leaves nothing on the stream.
+// Encodes a message with a u16 (16-bit) size prefix as per draft-14, after its type `id` if
+// given. The whole message goes out in one write, so an oversized one leaves nothing on the stream.
 export async function encode(writer: Writer, f: (w: Writer) => Promise<void>, id?: number) {
-	let scratch = new Uint8Array();
-
-	const temp = new Writer(
-		new WritableStream<Uint8Array>({
-			write(chunk: Uint8Array) {
-				const needed = scratch.byteLength + chunk.byteLength;
-				if (needed > scratch.buffer.byteLength) {
-					// Resize the buffer to the needed size.
-					const capacity = Math.max(needed, scratch.buffer.byteLength * 2);
-					const newBuffer = new ArrayBuffer(capacity);
-					const newScratch = new Uint8Array(newBuffer, 0, needed);
-
-					// Copy the old data into the new buffer.
-					newScratch.set(scratch);
-
-					// Copy the new chunk into the new buffer.
-					newScratch.set(chunk, scratch.byteLength);
-
-					scratch = newScratch;
-				} else {
-					// Copy chunk data into buffer
-					scratch = new Uint8Array(scratch.buffer, 0, needed);
-					scratch.set(chunk, needed - chunk.byteLength);
-				}
-			},
-		}),
-		writer.version,
-	);
-
-	try {
-		await f(temp);
-	} finally {
-		temp.close();
+	const body = await Writer.collect(writer.version, f);
+	if (body.byteLength > 65535) {
+		throw new Error(`Message too large: ${body.byteLength} bytes (max 65535)`);
 	}
 
-	await temp.closed;
-
-	// Check that message fits in u16
-	if (scratch.byteLength > 65535) {
-		throw new Error(`Message too large: ${scratch.byteLength} bytes (max 65535)`);
-	}
+	const size = new Uint8Array([body.byteLength >> 8, body.byteLength & 0xff]);
+	const parts = id === undefined ? [size, body] : [encodeVarint(id, writer.version), size, body];
+	await writer.write(concat(parts));
+}
 
-	if (id !== undefined) await writer.u53(id);
+/** A message that encodes everything but its type. */
+export interface Encode {
+	encode(w: Writer, version: IetfVersion): Promise<void>;
+}
 
-	// Write u16 size (2 bytes, big-endian)
-	await writer.u16(scratch.byteLength);
-	await writer.write(scratch);
+// Writes `msg` after the type `id` the caller picked for it, in one write, so a message that
+// fails to encode leaves nothing on the stream.
+export async function write(writer: Writer, id: number, msg: Encode, version: IetfVersion) {
+	const body = await Writer.collect(writer.version, (w) => msg.encode(w, version));
+	await writer.write(concat([encodeVarint(id, writer.version), body]));
 }
 
 // Reads a message with a u16 size prefix.

js/net/src/ietf/object.ts

@@ -78,27 +78,7 @@ export async function encodeObjectExtensions(
 		return new Uint8Array();
 	}
 
-	const chunks: Uint8Array[] = [];
-	const writer = new Writer(
-		new WritableStream<Uint8Array>({
-			write(chunk) {
-				chunks.push(new Uint8Array(chunk));
-			},
-		}),
-		version,
-	);
-	await encodeObjectTime(writer, timestamp, timescale, version);
-	writer.close();
-	await writer.closed;
-
-	const size = chunks.reduce((total, chunk) => total + chunk.byteLength, 0);
-	const result = new Uint8Array(size);
-	let offset = 0;
-	for (const chunk of chunks) {
-		result.set(chunk, offset);
-		offset += chunk.byteLength;
-	}
-	return result;
+	return Writer.collect(version, (w) => encodeObjectTime(w, timestamp, timescale, version));
 }
 
 /** Decode the Timestamp an object's Properties block carries, consuming the whole block. */