tor-browser

create-websocket-stream.test.js (16779B)

---

'use strict';

const assert = require('assert');
const EventEmitter = require('events');
const { createServer } = require('http');
const { Duplex } = require('stream');
const { randomBytes } = require('crypto');

const createWebSocketStream = require('../lib/stream');
const Sender = require('../lib/sender');
const WebSocket = require('..');
const { EMPTY_BUFFER } = require('../lib/constants');

describe('createWebSocketStream', () => {
 it('is exposed as a property of the `WebSocket` class', () => {
   assert.strictEqual(WebSocket.createWebSocketStream, createWebSocketStream);
 });

 it('returns a `Duplex` stream', () => {
   const duplex = createWebSocketStream(new EventEmitter());

   assert.ok(duplex instanceof Duplex);
 });

 it('passes the options object to the `Duplex` constructor', (done) => {
   const wss = new WebSocket.Server({ port: 0 }, () => {
     const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
     const duplex = createWebSocketStream(ws, {
       allowHalfOpen: false,
       encoding: 'utf8'
     });

     duplex.on('data', (chunk) => {
       assert.strictEqual(chunk, 'hi');

       duplex.on('close', () => {
         wss.close(done);
       });
     });
   });

   wss.on('connection', (ws) => {
     ws.send(Buffer.from('hi'));
     ws.close();
   });
 });

 describe('The returned stream', () => {
   it('buffers writes if `readyState` is `CONNECTING`', (done) => {
     const chunk = randomBytes(1024);
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);

       assert.strictEqual(ws.readyState, WebSocket.CONNECTING);

       const duplex = createWebSocketStream(ws);

       duplex.write(chunk);
     });

     wss.on('connection', (ws) => {
       ws.on('message', (message, isBinary) => {
         ws.on('close', (code, reason) => {
           assert.deepStrictEqual(message, chunk);
           assert.ok(isBinary);
           assert.strictEqual(code, 1005);
           assert.strictEqual(reason, EMPTY_BUFFER);
           wss.close(done);
         });
       });

       ws.close();
     });
   });

   it('errors if a write occurs when `readyState` is `CLOSING`', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('error', (err) => {
         assert.ok(duplex.destroyed);
         assert.ok(err instanceof Error);
         assert.strictEqual(
           err.message,
           'WebSocket is not open: readyState 2 (CLOSING)'
         );

         duplex.on('close', () => {
           wss.close(done);
         });
       });

       ws.on('open', () => {
         ws._receiver.on('conclude', () => {
           duplex.write('hi');
         });
       });
     });

     wss.on('connection', (ws) => {
       ws.close();
     });
   });

   it('errors if a write occurs when `readyState` is `CLOSED`', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('error', (err) => {
         assert.ok(duplex.destroyed);
         assert.ok(err instanceof Error);
         assert.strictEqual(
           err.message,
           'WebSocket is not open: readyState 3 (CLOSED)'
         );

         duplex.on('close', () => {
           wss.close(done);
         });
       });

       ws.on('close', () => {
         duplex.write('hi');
       });
     });

     wss.on('connection', (ws) => {
       ws.close();
     });
   });

   it('does not error if `_final()` is called while connecting', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);

       assert.strictEqual(ws.readyState, WebSocket.CONNECTING);

       const duplex = createWebSocketStream(ws);

       duplex.on('close', () => {
         wss.close(done);
       });

       duplex.resume();
       duplex.end();
     });
   });

   it('makes `_final()` a noop if no socket is assigned', (done) => {
     const server = createServer();

     server.on('upgrade', (request, socket) => {
       socket.on('end', socket.end);

       const headers = [
         'HTTP/1.1 101 Switching Protocols',
         'Upgrade: websocket',
         'Connection: Upgrade',
         'Sec-WebSocket-Accept: foo'
       ];

       socket.write(headers.concat('\r\n').join('\r\n'));
     });

     server.listen(() => {
       const called = [];
       const ws = new WebSocket(`ws://localhost:${server.address().port}`);
       const duplex = WebSocket.createWebSocketStream(ws);
       const final = duplex._final;

       duplex._final = (callback) => {
         called.push('final');
         assert.strictEqual(ws.readyState, WebSocket.CLOSING);
         assert.strictEqual(ws._socket, null);

         final(callback);
       };

       duplex.on('error', (err) => {
         called.push('error');
         assert.ok(err instanceof Error);
         assert.strictEqual(
           err.message,
           'Invalid Sec-WebSocket-Accept header'
         );
       });

       duplex.on('finish', () => {
         called.push('finish');
       });

       duplex.on('close', () => {
         assert.deepStrictEqual(called, ['final', 'error']);
         server.close(done);
       });

       ws.on('upgrade', () => {
         process.nextTick(() => {
           duplex.end();
         });
       });
     });
   });

   it('reemits errors', (done) => {
     let duplexCloseEventEmitted = false;
     let serverClientCloseEventEmitted = false;

     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('error', (err) => {
         assert.ok(err instanceof RangeError);
         assert.strictEqual(err.code, 'WS_ERR_INVALID_OPCODE');
         assert.strictEqual(
           err.message,
           'Invalid WebSocket frame: invalid opcode 5'
         );

         duplex.on('close', () => {
           duplexCloseEventEmitted = true;
           if (serverClientCloseEventEmitted) wss.close(done);
         });
       });
     });

     wss.on('connection', (ws) => {
       ws._socket.write(Buffer.from([0x85, 0x00]));
       ws.on('close', (code, reason) => {
         assert.strictEqual(code, 1002);
         assert.deepStrictEqual(reason, EMPTY_BUFFER);

         serverClientCloseEventEmitted = true;
         if (duplexCloseEventEmitted) wss.close(done);
       });
     });
   });

   it('does not swallow errors that may occur while destroying', (done) => {
     const frame = Buffer.concat(
       Sender.frame(Buffer.from([0x22, 0xfa, 0xec, 0x78]), {
         fin: true,
         rsv1: true,
         opcode: 0x02,
         mask: false,
         readOnly: false
       })
     );

     const wss = new WebSocket.Server(
       {
         perMessageDeflate: true,
         port: 0
       },
       () => {
         const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
         const duplex = createWebSocketStream(ws);

         duplex.on('error', (err) => {
           assert.ok(err instanceof Error);
           assert.strictEqual(err.code, 'Z_DATA_ERROR');
           assert.strictEqual(err.errno, -3);

           duplex.on('close', () => {
             wss.close(done);
           });
         });

         let bytesRead = 0;

         ws.on('open', () => {
           ws._socket.on('data', (chunk) => {
             bytesRead += chunk.length;
             if (bytesRead === frame.length) duplex.destroy();
           });
         });
       }
     );

     wss.on('connection', (ws) => {
       ws._socket.write(frame);
     });
   });

   it("does not suppress the throwing behavior of 'error' events", (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       createWebSocketStream(ws);
     });

     wss.on('connection', (ws) => {
       ws._socket.write(Buffer.from([0x85, 0x00]));
     });

     assert.strictEqual(process.listenerCount('uncaughtException'), 1);

     const [listener] = process.listeners('uncaughtException');

     process.removeAllListeners('uncaughtException');
     process.once('uncaughtException', (err) => {
       assert.ok(err instanceof Error);
       assert.strictEqual(
         err.message,
         'Invalid WebSocket frame: invalid opcode 5'
       );

       process.on('uncaughtException', listener);
       wss.close(done);
     });
   });

   it("is destroyed after 'end' and 'finish' are emitted (1/2)", (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const events = [];
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('end', () => {
         events.push('end');
         assert.ok(duplex.destroyed);
       });

       duplex.on('close', () => {
         assert.deepStrictEqual(events, ['finish', 'end']);
         wss.close(done);
       });

       duplex.on('finish', () => {
         events.push('finish');
         assert.ok(!duplex.destroyed);
         assert.ok(duplex.readable);

         duplex.resume();
       });

       ws.on('close', () => {
         duplex.end();
       });
     });

     wss.on('connection', (ws) => {
       ws.send('foo');
       ws.close();
     });
   });

   it("is destroyed after 'end' and 'finish' are emitted (2/2)", (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const events = [];
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('end', () => {
         events.push('end');
         assert.ok(!duplex.destroyed);
         assert.ok(duplex.writable);

         duplex.end();
       });

       duplex.on('close', () => {
         assert.deepStrictEqual(events, ['end', 'finish']);
         wss.close(done);
       });

       duplex.on('finish', () => {
         events.push('finish');
       });

       duplex.resume();
     });

     wss.on('connection', (ws) => {
       ws.close();
     });
   });

   it('handles backpressure (1/3)', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       // eslint-disable-next-line no-unused-vars
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
     });

     wss.on('connection', (ws) => {
       const duplex = createWebSocketStream(ws);

       duplex.resume();

       duplex.on('drain', () => {
         duplex.on('close', () => {
           wss.close(done);
         });

         duplex.end();
       });

       const chunk = randomBytes(1024);
       let ret;

       do {
         ret = duplex.write(chunk);
       } while (ret !== false);
     });
   });

   it('handles backpressure (2/3)', (done) => {
     const wss = new WebSocket.Server(
       { port: 0, perMessageDeflate: true },
       () => {
         const called = [];
         const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
         const duplex = createWebSocketStream(ws);
         const read = duplex._read;

         duplex._read = () => {
           duplex._read = read;
           called.push('read');
           assert.ok(ws._receiver._writableState.needDrain);
           read();
           assert.ok(ws._socket.isPaused());
         };

         ws.on('open', () => {
           ws._socket.on('pause', () => {
             duplex.resume();
           });

           ws._receiver.on('drain', () => {
             called.push('drain');
             assert.ok(!ws._socket.isPaused());
             duplex.end();
           });

           const opts = {
             fin: true,
             opcode: 0x02,
             mask: false,
             readOnly: false
           };

           const list = [
             ...Sender.frame(randomBytes(16 * 1024), { rsv1: false, ...opts }),
             ...Sender.frame(Buffer.alloc(1), { rsv1: true, ...opts })
           ];

           // This hack is used because there is no guarantee that more than
           // 16 KiB will be sent as a single TCP packet.
           ws._socket.push(Buffer.concat(list));
         });

         duplex.on('close', () => {
           assert.deepStrictEqual(called, ['read', 'drain']);
           wss.close(done);
         });
       }
     );
   });

   it('handles backpressure (3/3)', (done) => {
     const wss = new WebSocket.Server(
       { port: 0, perMessageDeflate: true },
       () => {
         const called = [];
         const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
         const duplex = createWebSocketStream(ws);
         const read = duplex._read;

         duplex._read = () => {
           called.push('read');
           assert.ok(!ws._receiver._writableState.needDrain);
           read();
           assert.ok(!ws._socket.isPaused());
           duplex.end();
         };

         ws.on('open', () => {
           ws._receiver.on('drain', () => {
             called.push('drain');
             assert.ok(ws._socket.isPaused());
             duplex.resume();
           });

           const opts = {
             fin: true,
             opcode: 0x02,
             mask: false,
             readOnly: false
           };

           const list = [
             ...Sender.frame(randomBytes(16 * 1024), { rsv1: false, ...opts }),
             ...Sender.frame(Buffer.alloc(1), { rsv1: true, ...opts })
           ];

           ws._socket.push(Buffer.concat(list));
         });

         duplex.on('close', () => {
           assert.deepStrictEqual(called, ['drain', 'read']);
           wss.close(done);
         });
       }
     );
   });

   it('can be destroyed (1/2)', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const error = new Error('Oops');
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('error', (err) => {
         assert.strictEqual(err, error);

         duplex.on('close', () => {
           wss.close(done);
         });
       });

       ws.on('open', () => {
         duplex.destroy(error);
       });
     });
   });

   it('can be destroyed (2/2)', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       duplex.on('close', () => {
         wss.close(done);
       });

       ws.on('open', () => {
         duplex.destroy();
       });
     });
   });

   it('converts text messages to strings in readable object mode', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const events = [];
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws, { readableObjectMode: true });

       duplex.on('data', (data) => {
         events.push('data');
         assert.strictEqual(data, 'foo');
       });

       duplex.on('end', () => {
         events.push('end');
         duplex.end();
       });

       duplex.on('close', () => {
         assert.deepStrictEqual(events, ['data', 'end']);
         wss.close(done);
       });
     });

     wss.on('connection', (ws) => {
       ws.send('foo');
       ws.close();
     });
   });

   it('resumes the socket if `readyState` is `CLOSING`', (done) => {
     const wss = new WebSocket.Server({ port: 0 }, () => {
       const ws = new WebSocket(`ws://localhost:${wss.address().port}`);
       const duplex = createWebSocketStream(ws);

       ws.on('message', () => {
         assert.ok(ws._socket.isPaused());

         duplex.on('close', () => {
           wss.close(done);
         });

         duplex.end();

         process.nextTick(() => {
           assert.strictEqual(ws.readyState, WebSocket.CLOSING);
           duplex.resume();
         });
       });
     });

     wss.on('connection', (ws) => {
       ws.send(randomBytes(16 * 1024));
     });
   });
 });
});