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