75a4e7e3f2
joule-moq: grant_proof (the WRP1 record, check_proof, the SETUP options and helpers, grant_digest_and_principal, base64url) and a strict Ed25519 check shared with the receipts. The relay sends its SETUP first with a fresh GRANT_CHALLENGE, checks every grant its rules mark proof against the client's GRANT_PROOF before serving any request, and closes a session that presents one without a proof that passes with UNAUTHORIZED; the transports give it the target the client addressed. Its test client presents grants with proofs; tests/grant_proof.rs. The web client presents its grant with a proof signed in wasm, the per-participant secret gone. Not yet built or run. Co-Authored-By: Claude Opus 5.5 <[email protected]>
415 lines
16 KiB
JavaScript
415 lines
16 KiB
JavaScript
// The two ways a page reaches the relay, behind one interface:
|
|
//
|
|
// - WebTransport over HTTP/3 (draft-ietf-webtrans-http3), offering the protocol
|
|
// `moqt-22` (`WT-Available-Protocols`), and, for a relay with a development
|
|
// certificate, `serverCertificateHashes` (ECDSA P-256, valid at most fourteen days);
|
|
// - the relay's WebSocket binding: a WebSocket with the subprotocol `moqt-22`, each
|
|
// message one capsule of draft-ietf-webtrans-http2-15 §6, streams numbered as QUIC
|
|
// numbers them, stream-count credit by WT_MAX_STREAMS, no data flow control (TCP's
|
|
// backpressure bounds the sender).
|
|
//
|
|
// Each gives:
|
|
// kind "webtransport" | "websocket"
|
|
// protocol the negotiated protocol, or "" where none was signalled
|
|
// openUni({sendOrder}) a Writer
|
|
// openBidi() { writer, reader }
|
|
// acceptUni() the next Reader the relay opened, or null once closed
|
|
// acceptBidi() the next { writer, reader }, or null once closed
|
|
// close(code, reason)
|
|
// closed a promise of { code, reason }
|
|
// A Writer has write(bytes), close() (a FIN) and reset(code); a Reader has read(), which
|
|
// gives { value, done } or throws a StreamReset with the code, and cancel(code).
|
|
|
|
import * as cap from "./capsule.js";
|
|
|
|
export class StreamReset extends Error {
|
|
constructor(code, what) {
|
|
super(`${what || "a stream"} was reset with code ${code}`);
|
|
this.code = code;
|
|
}
|
|
}
|
|
|
|
export const PROTOCOL = "moqt-22";
|
|
|
|
// A queue of things handed to whoever waits for them.
|
|
class Queue {
|
|
constructor() { this.items = []; this.waiters = []; this.closed = false; }
|
|
push(x) {
|
|
if (this.closed) return;
|
|
const w = this.waiters.shift();
|
|
if (w) w(x); else this.items.push(x);
|
|
}
|
|
close() {
|
|
this.closed = true;
|
|
for (const w of this.waiters.splice(0)) w(null);
|
|
}
|
|
next() {
|
|
if (this.items.length) return Promise.resolve(this.items.shift());
|
|
if (this.closed) return Promise.resolve(null);
|
|
return new Promise((r) => this.waiters.push(r));
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------
|
|
// WebTransport
|
|
// ---------------------------------------------------------------------------------------
|
|
|
|
function wtReader(readable) {
|
|
const r = readable.getReader();
|
|
return {
|
|
async read() {
|
|
try {
|
|
const { value, done } = await r.read();
|
|
return { value, done };
|
|
} catch (e) {
|
|
throw new StreamReset(e && e.streamErrorCode != null ? e.streamErrorCode : -1, "a WebTransport stream");
|
|
}
|
|
},
|
|
cancel(code) {
|
|
try { r.cancel(new WebTransportError({ streamErrorCode: code })); } catch { /* already closed */ }
|
|
},
|
|
};
|
|
}
|
|
|
|
function wtWriter(writable) {
|
|
const w = writable.getWriter();
|
|
return {
|
|
async write(b) { await w.write(b); },
|
|
async close() { try { await w.close(); } catch { /* reset or session closed */ } },
|
|
reset(code) { try { w.abort(new WebTransportError({ streamErrorCode: code })); } catch { /* closed */ } },
|
|
};
|
|
}
|
|
|
|
export async function connectWebTransport(url, { hashes = [], timeoutMs = 10000 } = {}) {
|
|
if (typeof WebTransport !== "function") throw new Error("this browser has no WebTransport");
|
|
const opts = { protocols: [PROTOCOL] };
|
|
if (hashes.length) opts.serverCertificateHashes = hashes.map((value) => ({ algorithm: "sha-256", value }));
|
|
const wt = new WebTransport(url, opts);
|
|
let timer;
|
|
try {
|
|
await Promise.race([wt.ready, new Promise((_, no) => { timer = setTimeout(() => no(new Error(`WebTransport not ready in ${timeoutMs} ms`)), timeoutMs); })]);
|
|
} catch (e) {
|
|
try { wt.close(); } catch { /* never opened */ }
|
|
throw e;
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
const uni = wt.incomingUnidirectionalStreams.getReader();
|
|
const bidi = wt.incomingBidirectionalStreams.getReader();
|
|
const closed = wt.closed.then(
|
|
(i) => ({ code: i ? i.closeCode : 0, reason: i ? i.reason : "" }),
|
|
(e) => ({ code: e && e.closeCode != null ? e.closeCode : -1, reason: String(e && e.message || e) }),
|
|
);
|
|
return {
|
|
kind: "webtransport",
|
|
// The URL the session opened: its authority and path are what a grant's proof names
|
|
// (WAI live-rooms §10.5.1).
|
|
url,
|
|
// `protocol` is the server's answer to WT-Available-Protocols, where the browser has it.
|
|
protocol: typeof wt.protocol === "string" ? wt.protocol : "",
|
|
async openUni({ sendOrder } = {}) {
|
|
const s = await wt.createUnidirectionalStream(sendOrder != null ? { sendOrder } : {});
|
|
return wtWriter(s);
|
|
},
|
|
async openBidi() {
|
|
const s = await wt.createBidirectionalStream();
|
|
return { writer: wtWriter(s.writable), reader: wtReader(s.readable) };
|
|
},
|
|
async acceptUni() {
|
|
try { const { value, done } = await uni.read(); return done ? null : wtReader(value); } catch { return null; }
|
|
},
|
|
async acceptBidi() {
|
|
try {
|
|
const { value, done } = await bidi.read();
|
|
return done ? null : { writer: wtWriter(value.writable), reader: wtReader(value.readable) };
|
|
} catch { return null; }
|
|
},
|
|
close(code = 0, reason = "") {
|
|
try { wt.close({ closeCode: code, reason }); } catch { /* closed */ }
|
|
},
|
|
closed,
|
|
stats: () => ({}),
|
|
};
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------------------
|
|
// The WebSocket binding
|
|
// ---------------------------------------------------------------------------------------
|
|
|
|
// What this end grants the relay: bidirectional streams (requests the relay opens to it)
|
|
// and unidirectional streams (the relay's data streams), cumulative as in QUIC, given
|
|
// back in batches of an eighth as streams finish.
|
|
const LOCAL_MAX_BIDI = 64;
|
|
const LOCAL_MAX_UNI = 1024;
|
|
// Bytes queued on the WebSocket before a writer waits.
|
|
const SEND_HIGH_WATER = 1 << 20;
|
|
|
|
class WsSession {
|
|
constructor(ws) {
|
|
this.ws = ws;
|
|
this.kind = "websocket";
|
|
this.protocol = ws.protocol;
|
|
this.nextUni = 0;
|
|
this.nextBidi = 0;
|
|
this.peerMax = { bidi: null, uni: null };
|
|
this.local = { bidi: { max: LOCAL_MAX_BIDI, pending: 0, initial: LOCAL_MAX_BIDI }, uni: { max: LOCAL_MAX_UNI, pending: 0, initial: LOCAL_MAX_UNI } };
|
|
this.recv = new Map(); // stream id -> inbox
|
|
this.halves = new Map(); // peer-initiated bidi id -> halves still open
|
|
this.stopped = new Map(); // stream id -> STOP_SENDING code
|
|
this.seen = { bidi: new Set(), uni: new Set() };
|
|
this.creditWaiters = [];
|
|
this.incomingUni = new Queue();
|
|
this.incomingBidi = new Queue();
|
|
this.counters = { capsules_in: 0, capsules_out: 0, bytes_in: 0, bytes_out: 0, credit_granted: 0, blocked_sent: 0 };
|
|
this.isClosed = null;
|
|
this.closed = new Promise((r) => { this._closed = r; });
|
|
ws.onmessage = (ev) => this.onMessage(ev.data);
|
|
ws.onclose = (ev) => this.finish({ code: this.closeCode != null ? this.closeCode : 0, reason: this.closeReason || ev.reason || "the WebSocket closed" });
|
|
ws.onerror = () => {};
|
|
// Each end first sends its WT_MAX_STREAMS for both kinds.
|
|
this.push(cap.maxStreamsCapsule(true, LOCAL_MAX_BIDI));
|
|
this.push(cap.maxStreamsCapsule(false, LOCAL_MAX_UNI));
|
|
}
|
|
|
|
push(b) {
|
|
if (this.isClosed) return;
|
|
this.counters.capsules_out++;
|
|
this.counters.bytes_out += b.length;
|
|
this.ws.send(b);
|
|
}
|
|
|
|
async room() {
|
|
while (this.ws.bufferedAmount > SEND_HIGH_WATER && !this.isClosed) await new Promise((r) => setTimeout(r, 5));
|
|
}
|
|
|
|
finish(info) {
|
|
if (this.isClosed) return;
|
|
this.isClosed = info;
|
|
for (const ib of this.recv.values()) ib.fail(-1);
|
|
this.recv.clear();
|
|
this.incomingUni.close();
|
|
this.incomingBidi.close();
|
|
for (const w of this.creditWaiters.splice(0)) w();
|
|
this._closed(info);
|
|
}
|
|
|
|
fatal(code, reason) {
|
|
this.push(cap.closeSessionCapsule(code, reason));
|
|
this.closeCode = code;
|
|
this.closeReason = reason;
|
|
try { this.ws.close(1000); } catch { /* closed */ }
|
|
this.finish({ code, reason });
|
|
}
|
|
|
|
// A stream's credit comes back once both of its halves are done.
|
|
returnCredit(bidi) {
|
|
const l = bidi ? this.local.bidi : this.local.uni;
|
|
l.pending++;
|
|
if (l.pending >= Math.max(1, Math.floor(l.initial / 8))) {
|
|
l.max += l.pending;
|
|
l.pending = 0;
|
|
this.counters.credit_granted++;
|
|
this.push(cap.maxStreamsCapsule(bidi, l.max));
|
|
}
|
|
}
|
|
|
|
halfDone(id) {
|
|
if (cap.streamId.clientInitiated(id)) return; // credit for the streams this end opens is the relay's to give
|
|
const bidi = cap.streamId.bidi(id);
|
|
if (!bidi) return this.returnCredit(false);
|
|
const left = (this.halves.get(id) || 2) - 1;
|
|
if (left <= 0) {
|
|
this.halves.delete(id);
|
|
this.returnCredit(true);
|
|
} else this.halves.set(id, left);
|
|
}
|
|
|
|
inbox(id) {
|
|
const self = this;
|
|
const ib = {
|
|
chunks: [], done: false, code: null, waiter: null, finished: false,
|
|
deliver() { if (this.waiter) { const w = this.waiter; this.waiter = null; w(); } },
|
|
data(d) { this.chunks.push(d); this.deliver(); },
|
|
end() { this.done = true; this.deliver(); },
|
|
fail(code) { this.code = code; this.deliver(); },
|
|
settle() { if (!this.finished) { this.finished = true; self.halfDone(id); } },
|
|
};
|
|
const reader = {
|
|
async read() {
|
|
for (;;) {
|
|
if (ib.chunks.length) return { value: ib.chunks.shift(), done: false };
|
|
if (ib.code != null) { ib.settle(); throw new StreamReset(ib.code, "a WebSocket-binding stream"); }
|
|
if (ib.done) { ib.settle(); return { value: undefined, done: true }; }
|
|
await new Promise((r) => { ib.waiter = r; });
|
|
}
|
|
},
|
|
cancel(code) {
|
|
if (ib.done || ib.code != null) return;
|
|
self.push(cap.stopSendingCapsule(id, code));
|
|
ib.fail(code);
|
|
ib.settle();
|
|
self.recv.delete(id);
|
|
},
|
|
};
|
|
return { ib, reader };
|
|
}
|
|
|
|
writer(id) {
|
|
const self = this;
|
|
let ended = false;
|
|
const end = () => { if (!ended) { ended = true; self.halfDone(id); } };
|
|
return {
|
|
async write(b) {
|
|
if (self.isClosed) throw new Error("the session is closed");
|
|
if (self.stopped.has(id)) throw new StreamReset(self.stopped.get(id), "a stopped stream");
|
|
for (let at = 0; at < b.length; at += cap.CHUNK) {
|
|
await self.room();
|
|
self.push(cap.streamCapsule(id, b.subarray(at, Math.min(b.length, at + cap.CHUNK)), false));
|
|
}
|
|
},
|
|
async close() {
|
|
if (ended) return;
|
|
self.push(cap.streamCapsule(id, new Uint8Array(0), true));
|
|
end();
|
|
},
|
|
reset(code) {
|
|
if (ended) return;
|
|
self.push(cap.resetStreamCapsule(id, code));
|
|
end();
|
|
},
|
|
};
|
|
}
|
|
|
|
onMessage(data) {
|
|
if (typeof data === "string") return; // a text message carries nothing
|
|
const msg = new Uint8Array(data);
|
|
this.counters.capsules_in++;
|
|
this.counters.bytes_in += msg.length;
|
|
let c;
|
|
try { c = cap.decodeCapsule(msg); } catch (e) { return this.fatal(0x1, String(e.message)); }
|
|
switch (c.kind) {
|
|
case "stream": return this.onStream(c.id, c.data, c.fin, null);
|
|
case "reset": return this.onStream(c.id, null, false, c.code);
|
|
case "stop": this.stopped.set(c.id, c.code); return;
|
|
case "max_streams": {
|
|
const k = c.bidi ? "bidi" : "uni";
|
|
if (this.peerMax[k] != null && c.max < this.peerMax[k]) return this.fatal(cap.FLOW_CONTROL_ERROR, "a WT_MAX_STREAMS that lowers the limit");
|
|
this.peerMax[k] = c.max;
|
|
for (const w of this.creditWaiters.splice(0)) w();
|
|
return;
|
|
}
|
|
case "close":
|
|
this.closeCode = c.code;
|
|
this.closeReason = c.message;
|
|
this.finish({ code: c.code, reason: c.message });
|
|
try { this.ws.close(1000); } catch { /* closed */ }
|
|
return;
|
|
default:
|
|
return; // DATAGRAM, DRAIN, BLOCKED and unknown capsules: read and ignored
|
|
}
|
|
}
|
|
|
|
onStream(id, data, fin, reset) {
|
|
let ib = this.recv.get(id);
|
|
if (!ib) {
|
|
const peerInitiated = !cap.streamId.clientInitiated(id);
|
|
if (!peerInitiated) return; // a half this end stopped reading, or one already ended
|
|
const bidi = cap.streamId.bidi(id);
|
|
const k = bidi ? "bidi" : "uni";
|
|
const idx = cap.streamId.index(id);
|
|
if (idx >= this.local[k].max) return this.fatal(cap.FLOW_CONTROL_ERROR, "a stream past WT_MAX_STREAMS");
|
|
if (this.seen[k].has(idx)) return;
|
|
this.seen[k].add(idx);
|
|
const { ib: inbox, reader } = this.inbox(id);
|
|
ib = inbox;
|
|
this.recv.set(id, ib);
|
|
if (bidi) {
|
|
this.halves.set(id, 2);
|
|
this.incomingBidi.push({ writer: this.writer(id), reader });
|
|
} else this.incomingUni.push(reader);
|
|
}
|
|
if (data && data.length) ib.data(data.slice());
|
|
if (reset != null) { ib.fail(reset); this.recv.delete(id); } else if (fin) { ib.end(); this.recv.delete(id); }
|
|
}
|
|
|
|
async open(bidi) {
|
|
const k = bidi ? "bidi" : "uni";
|
|
for (;;) {
|
|
if (this.isClosed) throw new Error("the session is closed");
|
|
const next = bidi ? this.nextBidi : this.nextUni;
|
|
const max = this.peerMax[k];
|
|
if (max != null && next < max) {
|
|
if (bidi) this.nextBidi++; else this.nextUni++;
|
|
return cap.streamId.nth(next, true, bidi);
|
|
}
|
|
if (max != null && this.blockedAt !== `${k}:${max}`) {
|
|
this.blockedAt = `${k}:${max}`;
|
|
this.counters.blocked_sent++;
|
|
this.push(cap.streamsBlockedCapsule(bidi, max));
|
|
}
|
|
await new Promise((r) => this.creditWaiters.push(r));
|
|
}
|
|
}
|
|
|
|
async openUni() {
|
|
return this.writer(await this.open(false));
|
|
}
|
|
|
|
async openBidi() {
|
|
const id = await this.open(true);
|
|
const { ib, reader } = this.inbox(id);
|
|
this.recv.set(id, ib);
|
|
return { writer: this.writer(id), reader };
|
|
}
|
|
|
|
acceptUni() { return this.incomingUni.next(); }
|
|
acceptBidi() { return this.incomingBidi.next(); }
|
|
|
|
close(code = 0, reason = "") {
|
|
if (this.isClosed) return;
|
|
this.push(cap.closeSessionCapsule(code, reason));
|
|
this.closeCode = code;
|
|
this.closeReason = reason;
|
|
try { this.ws.close(1000); } catch { /* closed */ }
|
|
this.finish({ code, reason });
|
|
}
|
|
|
|
stats() { return { ...this.counters, peer_max: { ...this.peerMax }, local_max: { bidi: this.local.bidi.max, uni: this.local.uni.max } }; }
|
|
}
|
|
|
|
export async function connectWebSocket(url, { timeoutMs = 10000 } = {}) {
|
|
const ws = new WebSocket(url, PROTOCOL);
|
|
ws.binaryType = "arraybuffer";
|
|
await new Promise((ok, no) => {
|
|
const t = setTimeout(() => no(new Error(`WebSocket not open in ${timeoutMs} ms`)), timeoutMs);
|
|
ws.onopen = () => { clearTimeout(t); ok(); };
|
|
ws.onerror = () => { clearTimeout(t); no(new Error("the WebSocket did not open")); };
|
|
ws.onclose = (e) => { clearTimeout(t); no(new Error(`the WebSocket closed: ${e.code} ${e.reason}`)); };
|
|
});
|
|
if (ws.protocol !== PROTOCOL) {
|
|
ws.close();
|
|
throw new Error(`the relay answered the subprotocol "${ws.protocol}"`);
|
|
}
|
|
const s = new WsSession(ws);
|
|
s.url = url;
|
|
// Until the relay's credit arrives this end opens nothing; it comes first.
|
|
return s;
|
|
}
|
|
|
|
// Connect by WebTransport, falling back to the WebSocket binding when the browser has no
|
|
// WebTransport or the session does not open; `force` = "websocket" skips WebTransport.
|
|
export async function connect({ webtransport, websocket, hashes, force, log = () => {} }) {
|
|
if (force !== "websocket" && webtransport) {
|
|
try {
|
|
const t = await connectWebTransport(webtransport, { hashes });
|
|
log(`transport: WebTransport ${webtransport} (protocol "${t.protocol}")`);
|
|
return t;
|
|
} catch (e) {
|
|
if (!websocket || force === "webtransport") throw e;
|
|
log(`transport: WebTransport failed (${e.message}); falling back to the WebSocket binding`);
|
|
}
|
|
}
|
|
const t = await connectWebSocket(websocket);
|
|
log(`transport: WebSocket binding ${websocket} (subprotocol "${t.protocol}")`);
|
|
return t;
|
|
}
|