1
  2/**
  3 * Module dependencies.
  4 */
  5
  6var qs = require('querystring');
  7var parse = require('url').parse;
  8var base64id = require('base64id');
  9var transports = require('./transports');
 10var EventEmitter = require('events').EventEmitter;
 11var Socket = require('./socket');
 12var util = require('util');
 13var debug = require('debug')('engine');
 14var cookieMod = require('cookie');
 15
 16/**
 17 * Module exports.
 18 */
 19
 20module.exports = Server;
 21
 22/**
 23 * Server constructor.
 24 *
 25 * @param {Object} options
 26 * @api public
 27 */
 28
 29function Server (opts) {
 30  if (!(this instanceof Server)) {
 31    return new Server(opts);
 32  }
 33
 34  this.clients = {};
 35  this.clientsCount = 0;
 36
 37  opts = opts || {};
 38
 39  this.wsEngine = opts.wsEngine || process.env.EIO_WS_ENGINE;
 40  this.pingTimeout = opts.pingTimeout || 60000;
 41  this.pingInterval = opts.pingInterval || 25000;
 42  this.upgradeTimeout = opts.upgradeTimeout || 10000;
 43  this.maxHttpBufferSize = opts.maxHttpBufferSize || 10E7;
 44  this.transports = opts.transports || Object.keys(transports);
 45  this.allowUpgrades = false !== opts.allowUpgrades;
 46  this.allowRequest = opts.allowRequest;
 47  this.cookie = false !== opts.cookie ? (opts.cookie || 'io') : false;
 48  this.cookiePath = false !== opts.cookiePath ? (opts.cookiePath || '/') : false;
 49  this.cookieHttpOnly = false !== opts.cookieHttpOnly;
 50  this.perMessageDeflate = false !== opts.perMessageDeflate ? (opts.perMessageDeflate || true) : false;
 51  this.httpCompression = false !== opts.httpCompression ? (opts.httpCompression || {}) : false;
 52
 53  var self = this;
 54
 55  // initialize compression options
 56  ['perMessageDeflate', 'httpCompression'].forEach(function (type) {
 57    var compression = self[type];
 58    if (true === compression) self[type] = compression = {};
 59    if (compression && null == compression.threshold) {
 60      compression.threshold = 1024;
 61    }
 62  });
 63
 64  // initialize websocket server
 65  if (~this.transports.indexOf('websocket')) {
 66    // keep require('ws') as separate expression for packers (browserify, etc)
 67    var WebSocketServer = (this.wsEngine ? require(this.wsEngine) : require('ws')).Server;
 68    this.ws = new WebSocketServer({
 69      noServer: true,
 70      clientTracking: false,
 71      perMessageDeflate: this.perMessageDeflate,
 72      maxPayload: this.maxHttpBufferSize
 73    });
 74  }
 75}
 76
 77/**
 78 * Protocol errors mappings.
 79 */
 80
 81Server.errors = {
 82  UNKNOWN_TRANSPORT: 0,
 83  UNKNOWN_SID: 1,
 84  BAD_HANDSHAKE_METHOD: 2,
 85  BAD_REQUEST: 3
 86};
 87
 88Server.errorMessages = {
 89  0: 'Transport unknown',
 90  1: 'Session ID unknown',
 91  2: 'Bad handshake method',
 92  3: 'Bad request'
 93};
 94
 95/**
 96 * Inherits from EventEmitter.
 97 */
 98
 99util.inherits(Server, EventEmitter);
100
101/**
102 * Hash of open clients.
103 *
104 * @api public
105 */
106
107Server.prototype.clients;
108
109/**
110 * Returns a list of available transports for upgrade given a certain transport.
111 *
112 * @return {Array}
113 * @api public
114 */
115
116Server.prototype.upgrades = function (transport) {
117  if (!this.allowUpgrades) return [];
118  return transports[transport].upgradesTo || [];
119};
120
121/**
122 * Verifies a request.
123 *
124 * @param {http.IncomingMessage}
125 * @return {Boolean} whether the request is valid
126 * @api private
127 */
128
129Server.prototype.verify = function (req, upgrade, fn) {
130  // transport check
131  var transport = req._query.transport;
132  if (!~this.transports.indexOf(transport)) {
133    debug('unknown transport "%s"', transport);
134    return fn(Server.errors.UNKNOWN_TRANSPORT, false);
135  }
136
137  // sid check
138  var sid = req._query.sid;
139  if (sid) {
140    if (!this.clients.hasOwnProperty(sid)) {
141      return fn(Server.errors.UNKNOWN_SID, false);
142    }
143    if (!upgrade && this.clients[sid].transport.name !== transport) {
144      debug('bad request: unexpected transport without upgrade');
145      return fn(Server.errors.BAD_REQUEST, false);
146    }
147  } else {
148    // handshake is GET only
149    if ('GET' !== req.method) return fn(Server.errors.BAD_HANDSHAKE_METHOD, false);
150    if (!this.allowRequest) return fn(null, true);
151    return this.allowRequest(req, fn);
152  }
153
154  fn(null, true);
155};
156
157/**
158 * Prepares a request by processing the query string.
159 *
160 * @api private
161 */
162
163Server.prototype.prepare = function (req) {
164  // try to leverage pre-existing `req._query` (e.g: from connect)
165  if (!req._query) {
166    req._query = ~req.url.indexOf('?') ? qs.parse(parse(req.url).query) : {};
167  }
168};
169
170/**
171 * Closes all clients.
172 *
173 * @api public
174 */
175
176Server.prototype.close = function () {
177  debug('closing all open clients');
178  for (var i in this.clients) {
179    if (this.clients.hasOwnProperty(i)) {
180      this.clients[i].close(true);
181    }
182  }
183  if (this.ws) {
184    debug('closing webSocketServer');
185    this.ws.close();
186    // don't delete this.ws because it can be used again if the http server starts listening again
187  }
188  return this;
189};
190
191/**
192 * Handles an Engine.IO HTTP request.
193 *
194 * @param {http.IncomingMessage} request
195 * @param {http.ServerResponse|http.OutgoingMessage} response
196 * @api public
197 */
198
199Server.prototype.handleRequest = function (req, res) {
200  debug('handling "%s" http request "%s"', req.method, req.url);
201  this.prepare(req);
202  req.res = res;
203
204  var self = this;
205  this.verify(req, false, function (err, success) {
206    if (!success) {
207      sendErrorMessage(req, res, err);
208      return;
209    }
210
211    if (req._query.sid) {
212      debug('setting new request for existing client');
213      self.clients[req._query.sid].transport.onRequest(req);
214    } else {
215      self.handshake(req._query.transport, req);
216    }
217  });
218};
219
220/**
221 * Sends an Engine.IO Error Message
222 *
223 * @param {http.ServerResponse} response
224 * @param {code} error code
225 * @api private
226 */
227
228function sendErrorMessage (req, res, code) {
229  var headers = { 'Content-Type': 'application/json' };
230
231  if (req.headers.origin) {
232    headers['Access-Control-Allow-Credentials'] = 'true';
233    headers['Access-Control-Allow-Origin'] = req.headers.origin;
234  } else {
235    headers['Access-Control-Allow-Origin'] = '*';
236  }
237  res.writeHead(400, headers);
238  res.end(JSON.stringify({
239    code: code,
240    message: Server.errorMessages[code]
241  }));
242}
243
244/**
245 * generate a socket id.
246 * Overwrite this method to generate your custom socket id
247 *
248 * @param {Object} request object
249 * @api public
250 */
251
252Server.prototype.generateId = function (req) {
253  return base64id.generateId();
254};
255
256/**
257 * Handshakes a new client.
258 *
259 * @param {String} transport name
260 * @param {Object} request object
261 * @api private
262 */
263
264Server.prototype.handshake = function (transportName, req) {
265  var id = this.generateId(req);
266
267  debug('handshaking client "%s"', id);
268
269  try {
270    var transport = new transports[transportName](req);
271    if ('polling' === transportName) {
272      transport.maxHttpBufferSize = this.maxHttpBufferSize;
273      transport.httpCompression = this.httpCompression;
274    } else if ('websocket' === transportName) {
275      transport.perMessageDeflate = this.perMessageDeflate;
276    }
277
278    if (req._query && req._query.b64) {
279      transport.supportsBinary = false;
280    } else {
281      transport.supportsBinary = true;
282    }
283  } catch (e) {
284    sendErrorMessage(req, req.res, Server.errors.BAD_REQUEST);
285    return;
286  }
287  var socket = new Socket(id, this, transport, req);
288  var self = this;
289
290  if (false !== this.cookie) {
291    transport.on('headers', function (headers) {
292      headers['Set-Cookie'] = cookieMod.serialize(self.cookie, id,
293        {
294          path: self.cookiePath,
295          httpOnly: self.cookiePath ? self.cookieHttpOnly : false
296        });
297    });
298  }
299
300  transport.onRequest(req);
301
302  this.clients[id] = socket;
303  this.clientsCount++;
304
305  socket.once('close', function () {
306    delete self.clients[id];
307    self.clientsCount--;
308  });
309
310  this.emit('connection', socket);
311};
312
313/**
314 * Handles an Engine.IO HTTP Upgrade.
315 *
316 * @api public
317 */
318
319Server.prototype.handleUpgrade = function (req, socket, upgradeHead) {
320  this.prepare(req);
321
322  var self = this;
323  this.verify(req, true, function (err, success) {
324    if (!success) {
325      abortConnection(socket, err);
326      return;
327    }
328
329    var head = new Buffer(upgradeHead.length);
330    upgradeHead.copy(head);
331    upgradeHead = null;
332
333    // delegate to ws
334    self.ws.handleUpgrade(req, socket, head, function (conn) {
335      self.onWebSocket(req, conn);
336    });
337  });
338};
339
340/**
341 * Called upon a ws.io connection.
342 *
343 * @param {ws.Socket} websocket
344 * @api private
345 */
346
347Server.prototype.onWebSocket = function (req, socket) {
348  socket.on('error', onUpgradeError);
349
350  if (!transports[req._query.transport].prototype.handlesUpgrades) {
351    debug('transport doesnt handle upgraded requests');
352    socket.close();
353    return;
354  }
355
356  // get client id
357  var id = req._query.sid;
358
359  // keep a reference to the ws.Socket
360  req.websocket = socket;
361
362  if (id) {
363    var client = this.clients[id];
364    if (!client) {
365      debug('upgrade attempt for closed client');
366      socket.close();
367    } else if (client.upgrading) {
368      debug('transport has already been trying to upgrade');
369      socket.close();
370    } else if (client.upgraded) {
371      debug('transport had already been upgraded');
372      socket.close();
373    } else {
374      debug('upgrading existing transport');
375
376      // transport error handling takes over
377      socket.removeListener('error', onUpgradeError);
378
379      var transport = new transports[req._query.transport](req);
380      if (req._query && req._query.b64) {
381        transport.supportsBinary = false;
382      } else {
383        transport.supportsBinary = true;
384      }
385      transport.perMessageDeflate = this.perMessageDeflate;
386      client.maybeUpgrade(transport);
387    }
388  } else {
389    // transport error handling takes over
390    socket.removeListener('error', onUpgradeError);
391
392    this.handshake(req._query.transport, req);
393  }
394
395  function onUpgradeError () {
396    debug('websocket error before upgrade');
397    // socket.close() not needed
398  }
399};
400
401/**
402 * Captures upgrade requests for a http.Server.
403 *
404 * @param {http.Server} server
405 * @param {Object} options
406 * @api public
407 */
408
409Server.prototype.attach = function (server, options) {
410  var self = this;
411  options = options || {};
412  var path = (options.path || '/engine.io').replace(/\/$/, '');
413
414  var destroyUpgradeTimeout = options.destroyUpgradeTimeout || 1000;
415
416  // normalize path
417  path += '/';
418
419  function check (req) {
420    return path === req.url.substr(0, path.length);
421  }
422
423  // cache and clean up listeners
424  var listeners = server.listeners('request').slice(0);
425  server.removeAllListeners('request');
426  server.on('close', self.close.bind(self));
427
428  // add request handler
429  server.on('request', function (req, res) {
430    if (check(req)) {
431      debug('intercepting request for path "%s"', path);
432      self.handleRequest(req, res);
433    } else {
434      for (var i = 0, l = listeners.length; i < l; i++) {
435        listeners[i].call(server, req, res);
436      }
437    }
438  });
439
440  if (~self.transports.indexOf('websocket')) {
441    server.on('upgrade', function (req, socket, head) {
442      if (check(req)) {
443        self.handleUpgrade(req, socket, head);
444      } else if (false !== options.destroyUpgrade) {
445        // default node behavior is to disconnect when no handlers
446        // but by adding a handler, we prevent that
447        // and if no eio thing handles the upgrade
448        // then the socket needs to die!
449        setTimeout(function () {
450          if (socket.writable && socket.bytesWritten <= 0) {
451            return socket.end();
452          }
453        }, destroyUpgradeTimeout);
454      }
455    });
456  }
457};
458
459/**
460 * Closes the connection
461 *
462 * @param {net.Socket} socket
463 * @param {code} error code
464 * @api private
465 */
466
467function abortConnection (socket, code) {
468  if (socket.writable) {
469    var message = Server.errorMessages.hasOwnProperty(code) ? Server.errorMessages[code] : code;
470    var length = Buffer.byteLength(message);
471    socket.write(
472      'HTTP/1.1 400 Bad Request\r\n' +
473      'Connection: close\r\n' +
474      'Content-type: text/html\r\n' +
475      'Content-Length: ' + length + '\r\n' +
476      '\r\n' +
477      message
478    );
479  }
480  socket.destroy();
481}