// An HTTP/1.1 client over one RTCDataChannel: the channel is the connection (the guest agent bridges a // tag to its unix socket, e.g. shell to its MCP server; webrtc-peer bridges a tag to a TCP port). One request // at a time; keep-alive unless the response says close or is delimited by the close. No deps. const HEAD_MAX = 65536; // A response body over this is refused (the guest's tools cap their output well under it). export const BODY_MAX = 1 << 22; // The datachannel's message size every browser and pion take. const CHUNK = 16384; const enc = new TextEncoder(); const dec = new TextDecoder(); // Bytes from the channel, read as lines and exact counts. class Bytes { #parts = []; #size = 0; #wake; ended = null; push(chunk) { this.#parts.push(chunk); this.#size += chunk.length; this.#wake?.(); } end(error) { this.ended ??= error; this.#wake?.(); } #flat() { if (this.#parts.length !== 1) { const all = new Uint8Array(this.#size); let at = 0; for (const p of this.#parts) all.set(p, (at += p.length) - p.length); this.#parts = [all]; } return this.#parts[0]; } async #more() { if (this.ended) throw this.ended; await new Promise((resolve) => (this.#wake = resolve)); this.#wake = undefined; } async take(n) { while (this.#size < n) await this.#more(); const all = this.#flat(); this.#parts = [all.subarray(n)]; this.#size -= n; return all.subarray(0, n); } async line(max = HEAD_MAX) { for (;;) { const all = this.#flat(); for (let i = 0; i + 1 < all.length; i++) if (all[i] === 13 && all[i + 1] === 10) return dec.decode((await this.take(i + 2)).subarray(0, i)); if (this.#size > max) throw new Error("the HTTP response head is too long"); await this.#more(); } } // Everything until the channel closes (a response with neither length nor chunks). async rest(max) { while (!this.ended) { if (this.#size > max) throw new Error(`the HTTP response body is over ${max} bytes`); await new Promise((resolve) => (this.#wake = resolve)); } return this.take(this.#size); } } const tooBig = (n) => { if (n > BODY_MAX) throw new Error(`the HTTP response body is over ${BODY_MAX} bytes`); return n; }; async function readBody(bytes, status, headers) { if (status === 204 || status === 304) return [new Uint8Array(0), true]; if (/chunked/i.test(headers["transfer-encoding"] ?? "")) { const parts = []; let size = 0; for (;;) { const n = parseInt((await bytes.line()).split(";")[0], 16); if (!(n >= 0)) throw new Error("a bad HTTP chunk size"); if (n === 0) break; tooBig((size += n)); parts.push(await bytes.take(n)); await bytes.take(2); } while ((await bytes.line()) !== ""); // trailers const body = new Uint8Array(size); parts.reduce((at, p) => (body.set(p, at), at + p.length), 0); return [body, true]; } if (headers["content-length"] !== undefined) { const n = Number(headers["content-length"]); if (!Number.isSafeInteger(n) || n < 0) throw new Error("a bad HTTP content-length"); return [await bytes.take(tooBig(n)), true]; } return [await bytes.rest(BODY_MAX), false]; } /** * The client on an open datachannel (binaryType arraybuffer). It takes the channel's onmessage and onclose. * request({method, path, headers, body}) -> {status, headers (lowercase names), body: Uint8Array}; * reusable: false after a response that ends the connection (or once the channel closed). */ export function httpChannel(dc) { const bytes = new Bytes(); let busy = false; let keep = true; dc.binaryType = "arraybuffer"; dc.onmessage = ({ data }) => bytes.push(typeof data === "string" ? enc.encode(data) : new Uint8Array(data)); dc.onclose = () => bytes.end(new Error("the datachannel closed")); return { get reusable() { return keep && !busy && !bytes.ended && dc.readyState === "open"; }, close: () => ((keep = false), dc.close()), async request({ method = "POST", path = "/", headers = {}, body = "" }) { if (busy) throw new Error("one HTTP request at a time on a datachannel"); busy = true; try { const payload = typeof body === "string" ? enc.encode(body) : body; const head = [`${method} ${path} HTTP/1.1`, "host: guest", ...Object.entries({ ...headers, "content-length": payload.length }).map(([k, v]) => `${k}: ${v}`), "", ""].join("\r\n"); const out = new Uint8Array(enc.encode(head).length + payload.length); out.set(enc.encode(head)); out.set(payload, out.length - payload.length); for (let at = 0; at < out.length; at += CHUNK) dc.send(out.subarray(at, at + CHUNK)); const [version, code] = (await bytes.line()).split(" "); const status = Number(code); if (!/^HTTP\/1\.[01]$/.test(version) || !(status >= 100 && status < 600)) throw new Error(`not an HTTP response: ${version} ${code}`); const response = {}; for (let line; (line = await bytes.line()) !== ""; ) { const i = line.indexOf(":"); if (i > 0) response[line.slice(0, i).trim().toLowerCase()] = line.slice(i + 1).trim(); } const [data, framed] = await readBody(bytes, status, response); keep &&= framed && version === "HTTP/1.1" && !/close/i.test(response.connection ?? ""); return { status, headers: response, body: data }; } catch (error) { keep = false; throw error; } finally { busy = false; if (!keep) dc.close(); } }, }; }