deno.land / std@0.91.0 / ws / mod.ts
نووسراو ببینە
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.import { hasOwnProperty } from "../_util/has_own_property.ts";import { BufReader, BufWriter } from "../io/bufio.ts";import { readLong, readShort, sliceLongToBytes } from "../io/ioutil.ts";import { Sha1 } from "../hash/sha1.ts";import { writeResponse } from "../http/_io.ts";import { TextProtoReader } from "../textproto/mod.ts";import { Deferred, deferred } from "../async/deferred.ts";import { assert } from "../_util/assert.ts";import { concat } from "../bytes/mod.ts";
export enum OpCode { Continue = 0x0, TextFrame = 0x1, BinaryFrame = 0x2, Close = 0x8, Ping = 0x9, Pong = 0xa,}
export type WebSocketEvent = | string | Uint8Array | WebSocketCloseEvent // Received after closing connection finished. | WebSocketPingEvent // Received after pong frame responded. | WebSocketPongEvent;
export interface WebSocketCloseEvent { code: number; reason?: string;}
/** Returns true if input value is a WebSocketCloseEvent, false otherwise. */export function isWebSocketCloseEvent( a: WebSocketEvent,): a is WebSocketCloseEvent { return hasOwnProperty(a, "code");}
export type WebSocketPingEvent = ["ping", Uint8Array];
/** Returns true if input value is a WebSocketPingEvent, false otherwise. */export function isWebSocketPingEvent( a: WebSocketEvent,): a is WebSocketPingEvent { return Array.isArray(a) && a[0] === "ping" && a[1] instanceof Uint8Array;}
export type WebSocketPongEvent = ["pong", Uint8Array];
/** Returns true if input value is a WebSocketPongEvent, false otherwise. */export function isWebSocketPongEvent( a: WebSocketEvent,): a is WebSocketPongEvent { return Array.isArray(a) && a[0] === "pong" && a[1] instanceof Uint8Array;}
export type WebSocketMessage = string | Uint8Array;
export interface WebSocketFrame { isLastFrame: boolean; opcode: OpCode; mask?: Uint8Array; payload: Uint8Array;}
export interface WebSocket extends AsyncIterable<WebSocketEvent> { readonly conn: Deno.Conn; readonly isClosed: boolean;
[Symbol.asyncIterator](): AsyncIterableIterator<WebSocketEvent>;
/** * @throws `Deno.errors.ConnectionReset` */ send(data: WebSocketMessage): Promise<void>;
/** * @param data * @throws `Deno.errors.ConnectionReset` */ ping(data?: WebSocketMessage): Promise<void>;
/** Close connection after sending close frame to peer. * This is canonical way of disconnection but it may hang because of peer's response delay. * Default close code is 1000 (Normal Closure) * @throws `Deno.errors.ConnectionReset` */ close(): Promise<void>; close(code: number): Promise<void>; close(code: number, reason: string): Promise<void>;
/** Close connection forcely without sending close frame to peer. * This is basically undesirable way of disconnection. Use carefully. */ closeForce(): void;}
/** Unmask masked websocket payload */export function unmask(payload: Uint8Array, mask?: Uint8Array): void { if (mask) { for (let i = 0, len = payload.length; i < len; i++) { payload[i] ^= mask[i & 3]; } }}
/** Write WebSocket frame to inputted writer. */export async function writeFrame( frame: WebSocketFrame, writer: Deno.Writer,): Promise<void> { const payloadLength = frame.payload.byteLength; let header: Uint8Array; const hasMask = frame.mask ? 0x80 : 0; if (frame.mask && frame.mask.byteLength !== 4) { throw new Error( "invalid mask. mask must be 4 bytes: length=" + frame.mask.byteLength, ); } if (payloadLength < 126) { header = new Uint8Array([0x80 | frame.opcode, hasMask | payloadLength]); } else if (payloadLength < 0xffff) { header = new Uint8Array([ 0x80 | frame.opcode, hasMask | 0b01111110, payloadLength >>> 8, payloadLength & 0x00ff, ]); } else { header = new Uint8Array([ 0x80 | frame.opcode, hasMask | 0b01111111, ...sliceLongToBytes(payloadLength), ]); } if (frame.mask) { header = concat(header, frame.mask); } unmask(frame.payload, frame.mask); header = concat(header, frame.payload); const w = BufWriter.create(writer); await w.write(header); await w.flush();}
/** Read websocket frame from given BufReader * @throws `Deno.errors.UnexpectedEof` When peer closed connection without close frame * @throws `Error` Frame is invalid */export async function readFrame(buf: BufReader): Promise<WebSocketFrame> { let b = await buf.readByte(); assert(b !== null); let isLastFrame = false; switch (b >>> 4) { case 0b1000: isLastFrame = true; break; case 0b0000: isLastFrame = false; break; default: throw new Error("invalid signature"); } const opcode = b & 0x0f; // has_mask & payload b = await buf.readByte(); assert(b !== null); const hasMask = b >>> 7; let payloadLength = b & 0b01111111; if (payloadLength === 126) { const l = await readShort(buf); assert(l !== null); payloadLength = l; } else if (payloadLength === 127) { const l = await readLong(buf); assert(l !== null); payloadLength = Number(l); } // mask let mask: Uint8Array | undefined; if (hasMask) { mask = new Uint8Array(4); assert((await buf.readFull(mask)) !== null); } // payload const payload = new Uint8Array(payloadLength); assert((await buf.readFull(payload)) !== null); return { isLastFrame, opcode, mask, payload, };}
class WebSocketImpl implements WebSocket { readonly conn: Deno.Conn; private readonly mask?: Uint8Array; private readonly bufReader: BufReader; private readonly bufWriter: BufWriter; private sendQueue: Array<{ frame: WebSocketFrame; d: Deferred<void>; }> = [];
constructor({ conn, bufReader, bufWriter, mask, }: { conn: Deno.Conn; bufReader?: BufReader; bufWriter?: BufWriter; mask?: Uint8Array; }) { this.conn = conn; this.mask = mask; this.bufReader = bufReader || new BufReader(conn); this.bufWriter = bufWriter || new BufWriter(conn); }
async *[Symbol.asyncIterator](): AsyncIterableIterator<WebSocketEvent> { const decoder = new TextDecoder(); let frames: WebSocketFrame[] = []; let payloadsLength = 0; while (!this._isClosed) { let frame: WebSocketFrame; try { frame = await readFrame(this.bufReader); } catch { this.ensureSocketClosed(); break; } unmask(frame.payload, frame.mask); switch (frame.opcode) { case OpCode.TextFrame: case OpCode.BinaryFrame: case OpCode.Continue: frames.push(frame); payloadsLength += frame.payload.length; if (frame.isLastFrame) { const concat = new Uint8Array(payloadsLength); let offs = 0; for (const frame of frames) { concat.set(frame.payload, offs); offs += frame.payload.length; } if (frames[0].opcode === OpCode.TextFrame) { // text yield decoder.decode(concat); } else { // binary yield concat; } frames = []; payloadsLength = 0; } break; case OpCode.Close: { // [0x12, 0x34] -> 0x1234 const code = (frame.payload[0] << 8) | frame.payload[1]; const reason = decoder.decode( frame.payload.subarray(2, frame.payload.length), ); await this.close(code, reason); yield { code, reason }; return; } case OpCode.Ping: await this.enqueue({ opcode: OpCode.Pong, payload: frame.payload, isLastFrame: true, }); yield ["ping", frame.payload] as WebSocketPingEvent; break; case OpCode.Pong: yield ["pong", frame.payload] as WebSocketPongEvent; break; default: } } }
private dequeue(): void { const [entry] = this.sendQueue; if (!entry) return; if (this._isClosed) return; const { d, frame } = entry; writeFrame(frame, this.bufWriter) .then(() => d.resolve()) .catch((e) => d.reject(e)) .finally(() => { this.sendQueue.shift(); this.dequeue(); }); }
private enqueue(frame: WebSocketFrame): Promise<void> { if (this._isClosed) { throw new Deno.errors.ConnectionReset("Socket has already been closed"); } const d = deferred<void>(); this.sendQueue.push({ d, frame }); if (this.sendQueue.length === 1) { this.dequeue(); } return d; }
send(data: WebSocketMessage): Promise<void> { const opcode = typeof data === "string" ? OpCode.TextFrame : OpCode.BinaryFrame; const payload = typeof data === "string" ? new TextEncoder().encode(data) : data; const isLastFrame = true; const frame = { isLastFrame, opcode, payload, mask: this.mask, }; return this.enqueue(frame); }
ping(data: WebSocketMessage = ""): Promise<void> { const payload = typeof data === "string" ? new TextEncoder().encode(data) : data; const frame = { isLastFrame: true, opcode: OpCode.Ping, mask: this.mask, payload, }; return this.enqueue(frame); }
private _isClosed = false; get isClosed(): boolean { return this._isClosed; }
async close(code = 1000, reason?: string): Promise<void> { try { const header = [code >>> 8, code & 0x00ff]; let payload: Uint8Array; if (reason) { const reasonBytes = new TextEncoder().encode(reason); payload = new Uint8Array(2 + reasonBytes.byteLength); payload.set(header); payload.set(reasonBytes, 2); } else { payload = new Uint8Array(header); } await this.enqueue({ isLastFrame: true, opcode: OpCode.Close, mask: this.mask, payload, }); } catch (e) { throw e; } finally { this.ensureSocketClosed(); } }
closeForce(): void { this.ensureSocketClosed(); }
private ensureSocketClosed(): void { if (this.isClosed) return; try { this.conn.close(); } catch (e) { console.error(e); } finally { this._isClosed = true; const rest = this.sendQueue; this.sendQueue = []; rest.forEach((e) => e.d.reject( new Deno.errors.ConnectionReset("Socket has already been closed"), ) ); } }}
/** Returns true if input headers are usable for WebSocket, otherwise false. */export function acceptable(req: { headers: Headers }): boolean { const upgrade = req.headers.get("upgrade"); if (!upgrade || upgrade.toLowerCase() !== "websocket") { return false; } const secKey = req.headers.get("sec-websocket-key"); return ( req.headers.has("sec-websocket-key") && typeof secKey === "string" && secKey.length > 0 );}
const kGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
/** Create value of Sec-WebSocket-Accept header from inputted nonce. */export function createSecAccept(nonce: string): string { const sha1 = new Sha1(); sha1.update(nonce + kGUID); const bytes = sha1.digest(); return btoa(String.fromCharCode(...bytes));}
/** Upgrade inputted TCP connection into WebSocket connection. */export async function acceptWebSocket(req: { conn: Deno.Conn; bufWriter: BufWriter; bufReader: BufReader; headers: Headers;}): Promise<WebSocket> { const { conn, headers, bufReader, bufWriter } = req; if (acceptable(req)) { const sock = new WebSocketImpl({ conn, bufReader, bufWriter }); const secKey = headers.get("sec-websocket-key"); if (typeof secKey !== "string") { throw new Error("sec-websocket-key is not provided"); } const secAccept = createSecAccept(secKey); const newHeaders = new Headers({ Upgrade: "websocket", Connection: "Upgrade", "Sec-WebSocket-Accept": secAccept, }); const secProtocol = headers.get("sec-websocket-protocol"); if (typeof secProtocol === "string") { newHeaders.set("Sec-WebSocket-Protocol", secProtocol); } const secVersion = headers.get("sec-websocket-version"); if (typeof secVersion === "string") { newHeaders.set("Sec-WebSocket-Version", secVersion); } await writeResponse(bufWriter, { status: 101, headers: newHeaders, }); return sock; } throw new Error("request is not acceptable");}
const kSecChars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-.~_";
/** Returns base64 encoded 16 bytes string for Sec-WebSocket-Key header. */export function createSecKey(): string { let key = ""; for (let i = 0; i < 16; i++) { const j = Math.floor(Math.random() * kSecChars.length); key += kSecChars[j]; } return btoa(key);}
export async function handshake( url: URL, headers: Headers, bufReader: BufReader, bufWriter: BufWriter,): Promise<void> { const { hostname, pathname, search } = url; const key = createSecKey();
if (!headers.has("host")) { headers.set("host", hostname); } headers.set("upgrade", "websocket"); headers.set("connection", "upgrade"); headers.set("sec-websocket-key", key); headers.set("sec-websocket-version", "13");
let headerStr = `GET ${pathname}${search} HTTP/1.1\r\n`; for (const [key, value] of headers) { headerStr += `${key}: ${value}\r\n`; } headerStr += "\r\n";
await bufWriter.write(new TextEncoder().encode(headerStr)); await bufWriter.flush();
const tpReader = new TextProtoReader(bufReader); const statusLine = await tpReader.readLine(); if (statusLine === null) { throw new Deno.errors.UnexpectedEof(); } const m = statusLine.match(/^(?<version>\S+) (?<statusCode>\S+) /); if (!m) { throw new Error("ws: invalid status line: " + statusLine); }
assert(m.groups); const { version, statusCode } = m.groups; if (version !== "HTTP/1.1" || statusCode !== "101") { throw new Error( `ws: server didn't accept handshake: ` + `version=${version}, statusCode=${statusCode}`, ); }
const responseHeaders = await tpReader.readMIMEHeader(); if (responseHeaders === null) { throw new Deno.errors.UnexpectedEof(); }
const expectedSecAccept = createSecAccept(key); const secAccept = responseHeaders.get("sec-websocket-accept"); if (secAccept !== expectedSecAccept) { throw new Error( `ws: unexpected sec-websocket-accept header: ` + `expected=${expectedSecAccept}, actual=${secAccept}`, ); }}
export function createWebSocket(params: { conn: Deno.Conn; bufWriter?: BufWriter; bufReader?: BufReader; mask?: Uint8Array;}): WebSocket { return new WebSocketImpl(params);}
std
Version Info
3 years ago
External Dependencies
No external dependencies 🎉
