bittorrent-tracker/server.js

module.exports = Server

var Buffer = require('safe-buffer').Buffer
var bencode = require('bencode')
var debug = require('debug')('bittorrent-tracker:server')
var dgram = require('dgram')
var EventEmitter = require('events').EventEmitter
var http = require('http')
var inherits = require('inherits')
var peerid = require('bittorrent-peerid')
var series = require('run-series')
var string2compact = require('string2compact')
var WebSocketServer = require('ws').Server

var common = require('./lib/common')
var Swarm = require('./lib/server/swarm')
var parseHttpRequest = require('./lib/server/parse-http')
var parseUdpRequest = require('./lib/server/parse-udp')
var parseWebSocketRequest = require('./lib/server/parse-websocket')

inherits(Server, EventEmitter)

/**
 * BitTorrent tracker server.
 *
 * HTTP service which responds to GET requests from torrent clients. Requests include
 * metrics from clients that help the tracker keep overall statistics about the torrent.
 * Responses include a peer list that helps the client participate in the torrent.
 *
 * @param {Object}  opts            options object
 * @param {Number}  opts.interval   tell clients to announce on this interval (ms)
 * @param {Number}  opts.trustProxy trust 'x-forwarded-for' header from reverse proxy
 * @param {boolean} opts.http       start an http server? (default: true)
 * @param {boolean} opts.udp        start a udp server? (default: true)
 * @param {boolean} opts.ws         start a websocket server? (default: true)
 * @param {boolean} opts.stats      enable web-based statistics? (default: true)
 * @param {function} opts.filter    black/whitelist fn for disallowing/allowing torrents
 */
function Server (opts) {
  var self = this
  if (!(self instanceof Server)) return new Server(opts)
  EventEmitter.call(self)
  if (!opts) opts = {}

  debug('new server %s', JSON.stringify(opts))

  self.intervalMs = opts.interval
    ? opts.interval
    : 10 * 60 * 1000 // 10 min

  self._trustProxy = !!opts.trustProxy
  if (typeof opts.filter === 'function') self._filter = opts.filter

  self.peersCacheLength = opts.peersCacheLength
  self.peersCacheTtl = opts.peersCacheTtl

  self._listenCalled = false
  self.listening = false
  self.destroyed = false
  self.torrents = {}

  self.http = null
  self.udp4 = null
  self.udp6 = null
  self.ws = null

  // start an http tracker unless the user explictly says no
  if (opts.http !== false) {
    self.http = http.createServer()
    self.http.on('error', function (err) { self._onError(err) })
    self.http.on('listening', onListening)

    // Add default http request handler on next tick to give user the chance to add
    // their own handler first. Handle requests untouched by user's handler.
    process.nextTick(function () {
      self.http.on('request', function (req, res) {
        if (res.headersSent) return
        self.onHttpRequest(req, res)
      })
    })
  }

  // start a udp tracker unless the user explicitly says no
  if (opts.udp !== false) {
    var isNode10 = /^v0.10./.test(process.version)

    self.udp4 = self.udp = dgram.createSocket(
      isNode10 ? 'udp4' : { type: 'udp4', reuseAddr: true }
    )
    self.udp4.on('message', function (msg, rinfo) { self.onUdpRequest(msg, rinfo) })
    self.udp4.on('error', function (err) { self._onError(err) })
    self.udp4.on('listening', onListening)

    self.udp6 = dgram.createSocket(
      isNode10 ? 'udp6' : { type: 'udp6', reuseAddr: true }
    )
    self.udp6.on('message', function (msg, rinfo) { self.onUdpRequest(msg, rinfo) })
    self.udp6.on('error', function (err) { self._onError(err) })
    self.udp6.on('listening', onListening)
  }

  // start a websocket tracker (for WebTorrent) unless the user explicitly says no
  if (opts.ws !== false) {
    if (!self.http) {
      self.http = http.createServer()
      self.http.on('error', function (err) { self._onError(err) })
      self.http.on('listening', onListening)

      // Add default http request handler on next tick to give user the chance to add
      // their own handler first. Handle requests untouched by user's handler.
      process.nextTick(function () {
        self.http.on('request', function (req, res) {
          if (res.headersSent) return
          // For websocket trackers, we only need to handle the UPGRADE http method.
          // Return 404 for all other request types.
          res.statusCode = 404
          res.end('404 Not Found')
        })
      })
    }
    self.ws = new WebSocketServer({
      server: self.http,
      perMessageDeflate: false,
      clientTracking: false
    })
    self.ws.address = function () {
      return self.http.address()
    }
    self.ws.on('error', function (err) { self._onError(err) })
    self.ws.on('connection', function (socket) { self.onWebSocketConnection(socket) })
  }

  if (opts.stats !== false) {
    if (!self.http) {
      self.http = http.createServer()
      self.http.on('error', function (err) { self._onError(err) })
      self.http.on('listening', onListening)
    }

    // Http handler for '/stats' route
    self.http.on('request', function (req, res) {
      if (res.headersSent) return

      var infoHashes = Object.keys(self.torrents)
      var activeTorrents = 0
      var allPeers = {}

      function countPeers (filterFunction) {
        var count = 0
        var key

        for (key in allPeers) {
          if (allPeers.hasOwnProperty(key) && filterFunction(allPeers[key])) {
            count++
          }
        }

        return count
      }

      function groupByClient () {
        var clients = {}
        for (var key in allPeers) {
          if (allPeers.hasOwnProperty(key)) {
            var peer = allPeers[key]

            if (!clients[peer.client.client]) {
              clients[peer.client.client] = {}
            }
            var client = clients[peer.client.client]
            // If the client is not known show 8 chars from peerId as version
            var version = peer.client.version || new Buffer(peer.peerId, 'hex').toString().substring(0, 8)
            if (!client[version]) {
              client[version] = 0
            }
            client[version]++
          }
        }
        return clients
      }

      function printClients (clients) {
        var html = '<ul>\n'
        for (var name in clients) {
          if (clients.hasOwnProperty(name)) {
            var client = clients[name]
            for (var version in client) {
              if (client.hasOwnProperty(version)) {
                html += '<li><strong>' + name + '</strong> ' + version + ' : ' + client[version] + '</li>\n'
              }
            }
          }
        }
        html += '</ul>'
        return html
      }

      if (req.method === 'GET' && (req.url === '/stats' || req.url === '/stats.json')) {
        infoHashes.forEach(function (infoHash) {
          var peers = self.torrents[infoHash].peers
          var keys = peers.keys
          if (keys.length > 0) activeTorrents++

          keys.forEach(function (peerId) {
            // Don't mark the peer as most recently used for stats
            var peer = peers.peek(peerId)
            if (peer == null) return // peers.peek() can evict the peer

            if (!allPeers.hasOwnProperty(peerId)) {
              allPeers[peerId] = {
                ipv4: false,
                ipv6: false,
                seeder: false,
                leecher: false
              }
            }

            if (peer.ip.indexOf(':') >= 0) {
              allPeers[peerId].ipv6 = true
            } else {
              allPeers[peerId].ipv4 = true
            }

            if (peer.complete) {
              allPeers[peerId].seeder = true
            } else {
              allPeers[peerId].leecher = true
            }

            allPeers[peerId].peerId = peer.peerId
            allPeers[peerId].client = peerid(peer.peerId)
          })
        })

        var isSeederOnly = function (peer) { return peer.seeder && peer.leecher === false }
        var isLeecherOnly = function (peer) { return peer.leecher && peer.seeder === false }
        var isSeederAndLeecher = function (peer) { return peer.seeder && peer.leecher }
        var isIPv4 = function (peer) { return peer.ipv4 }
        var isIPv6 = function (peer) { return peer.ipv6 }

        var stats = {
          torrents: infoHashes.length,
          activeTorrents: activeTorrents,
          peersAll: Object.keys(allPeers).length,
          peersSeederOnly: countPeers(isSeederOnly),
          peersLeecherOnly: countPeers(isLeecherOnly),
          peersSeederAndLeecher: countPeers(isSeederAndLeecher),
          peersIPv4: countPeers(isIPv4),
          peersIPv6: countPeers(isIPv6),
          clients: groupByClient()
        }

        if (req.url === '/stats.json' || req.headers['accept'] === 'application/json') {
          res.write(JSON.stringify(stats))
          res.end()
        } else if (req.url === '/stats') {
          res.end('<h1>' + stats.torrents + ' torrents (' + stats.activeTorrents + ' active)</h1>\n' +
            '<h2>Connected Peers: ' + stats.peersAll + '</h2>\n' +
            '<h3>Peers Seeding Only: ' + stats.peersSeederOnly + '</h3>\n' +
            '<h3>Peers Leeching Only: ' + stats.peersLeecherOnly + '</h3>\n' +
            '<h3>Peers Seeding & Leeching: ' + stats.peersSeederAndLeecher + '</h3>\n' +
            '<h3>IPv4 Peers: ' + stats.peersIPv4 + '</h3>\n' +
            '<h3>IPv6 Peers: ' + stats.peersIPv6 + '</h3>\n' +
            '<h3>Clients:</h3>\n' +
            printClients(stats.clients)
          )
        }
      }
    })
  }

  var num = !!self.http + !!self.udp4 + !!self.udp6
  function onListening () {
    num -= 1
    if (num === 0) {
      self.listening = true
      debug('listening')
      self.emit('listening')
    }
  }
}

Server.prototype._onError = function (err) {
  var self = this
  self.emit('error', err)
}

Server.prototype.listen = function (/* port, hostname, onlistening */) {
  var self = this

  if (self._listenCalled || self.listening) throw new Error('server already listening')
  self._listenCalled = true

  var lastArg = arguments[arguments.length - 1]
  if (typeof lastArg === 'function') self.once('listening', lastArg)

  var port = toNumber(arguments[0]) || arguments[0] || 0
  var hostname = typeof arguments[1] !== 'function' ? arguments[1] : undefined

  debug('listen (port: %o hostname: %o)', port, hostname)

  function isObject (obj) {
    return typeof obj === 'object' && obj !== null
  }

  var httpPort = isObject(port) ? (port.http || 0) : port
  var udpPort = isObject(port) ? (port.udp || 0) : port

  // binding to :: only receives IPv4 connections if the bindv6only sysctl is set 0,
  // which is the default on many operating systems
  var httpHostname = isObject(hostname) ? hostname.http : hostname
  var udp4Hostname = isObject(hostname) ? hostname.udp : hostname
  var udp6Hostname = isObject(hostname) ? hostname.udp6 : hostname

  if (self.http) self.http.listen(httpPort, httpHostname)
  if (self.udp4) self.udp4.bind(udpPort, udp4Hostname)
  if (self.udp6) self.udp6.bind(udpPort, udp6Hostname)
}

Server.prototype.close = function (cb) {
  var self = this
  if (!cb) cb = noop
  debug('close')

  self.listening = false
  self.destroyed = true

  if (self.udp4) {
    try {
      self.udp4.close()
    } catch (err) {}
  }

  if (self.udp6) {
    try {
      self.udp6.close()
    } catch (err) {}
  }

  if (self.ws) {
    try {
      self.ws.close()
    } catch (err) {}
  }

  if (self.http) self.http.close(cb)
  else cb(null)
}

Server.prototype.createSwarm = function (infoHash, cb) {
  var self = this
  if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex')

  process.nextTick(function () {
    var swarm = self.torrents[infoHash] = new Swarm(infoHash, self)
    cb(null, swarm)
  })
}

Server.prototype.getSwarm = function (infoHash, cb) {
  var self = this
  if (Buffer.isBuffer(infoHash)) infoHash = infoHash.toString('hex')

  process.nextTick(function () {
    cb(null, self.torrents[infoHash])
  })
}

Server.prototype.onHttpRequest = function (req, res, opts) {
  var self = this
  if (!opts) opts = {}
  opts.trustProxy = opts.trustProxy || self._trustProxy

  var params
  try {
    params = parseHttpRequest(req, opts)
    params.httpReq = req
    params.httpRes = res
  } catch (err) {
    res.end(bencode.encode({
      'failure reason': err.message
    }))

    // even though it's an error for the client, it's just a warning for the server.
    // don't crash the server because a client sent bad data :)
    self.emit('warning', err)
    return
  }

  self._onRequest(params, function (err, response) {
    if (err) {
      self.emit('warning', err)
      response = {
        'failure reason': err.message
      }
    }
    if (self.destroyed) return res.end()

    delete response.action  // only needed for UDP encoding
    res.end(bencode.encode(response))

    if (params.action === common.ACTIONS.ANNOUNCE) {
      self.emit(common.EVENT_NAMES[params.event], params.addr, params)
    }
  })
}

Server.prototype.onUdpRequest = function (msg, rinfo) {
  var self = this

  var params
  try {
    params = parseUdpRequest(msg, rinfo)
  } catch (err) {
    self.emit('warning', err)
    // Do not reply for parsing errors
    return
  }

  self._onRequest(params, function (err, response) {
    if (err) {
      self.emit('warning', err)
      response = {
        action: common.ACTIONS.ERROR,
        'failure reason': err.message
      }
    }
    if (self.destroyed) return

    response.transactionId = params.transactionId
    response.connectionId = params.connectionId

    var buf = makeUdpPacket(response)

    try {
      var udp = (rinfo.family === 'IPv4') ? self.udp4 : self.udp6
      udp.send(buf, 0, buf.length, rinfo.port, rinfo.address)
    } catch (err) {
      self.emit('warning', err)
    }

    if (params.action === common.ACTIONS.ANNOUNCE) {
      self.emit(common.EVENT_NAMES[params.event], params.addr, params)
    }
  })
}

Server.prototype.onWebSocketConnection = function (socket, opts) {
  var self = this
  if (!opts) opts = {}
  opts.trustProxy = opts.trustProxy || self._trustProxy

  socket.peerId = null // as hex
  socket.infoHashes = [] // swarms that this socket is participating in
  socket.onSend = function (err) {
    self._onWebSocketSend(socket, err)
  }

  socket.onMessageBound = function (params) {
    self._onWebSocketRequest(socket, opts, params)
  }
  socket.on('message', socket.onMessageBound)

  socket.onErrorBound = function (err) {
    self._onWebSocketError(socket, err)
  }
  socket.on('error', socket.onErrorBound)

  socket.onCloseBound = function () {
    self._onWebSocketClose(socket)
  }
  socket.on('close', socket.onCloseBound)
}

Server.prototype._onWebSocketRequest = function (socket, opts, params) {
  var self = this

  try {
    params = parseWebSocketRequest(socket, opts, params)
  } catch (err) {
    socket.send(JSON.stringify({
      'failure reason': err.message
    }), socket.onSend)

    // even though it's an error for the client, it's just a warning for the server.
    // don't crash the server because a client sent bad data :)
    self.emit('warning', err)
    return
  }

  if (!socket.peerId) socket.peerId = params.peer_id // as hex

  self._onRequest(params, function (err, response) {
    if (self.destroyed || socket.destroyed) return
    if (err) {
      socket.send(JSON.stringify({
        action: params.action === common.ACTIONS.ANNOUNCE ? 'announce' : 'scrape',
        'failure reason': err.message,
        info_hash: common.hexToBinary(params.info_hash)
      }), socket.onSend)

      self.emit('warning', err)
      return
    }

    response.action = params.action === common.ACTIONS.ANNOUNCE ? 'announce' : 'scrape'

    var peers
    if (response.action === 'announce') {
      peers = response.peers
      delete response.peers

      if (socket.infoHashes.indexOf(params.info_hash) === -1) {
        socket.infoHashes.push(params.info_hash)
      }

      response.info_hash = common.hexToBinary(params.info_hash)

      // WebSocket tracker should have a shorter interval – default: 2 minutes
      response.interval = Math.ceil(self.intervalMs / 1000 / 5)
    }

    // Skip sending update back for 'answer' announce messages – not needed
    if (!params.answer) {
      socket.send(JSON.stringify(response), socket.onSend)
      debug('sent response %s to %s', JSON.stringify(response), params.peer_id)
    }

    if (Array.isArray(params.offers)) {
      debug('got %s offers from %s', params.offers.length, params.peer_id)
      debug('got %s peers from swarm %s', peers.length, params.info_hash)
      peers.forEach(function (peer, i) {
        peer.socket.send(JSON.stringify({
          action: 'announce',
          offer: params.offers[i].offer,
          offer_id: params.offers[i].offer_id,
          peer_id: common.hexToBinary(params.peer_id),
          info_hash: common.hexToBinary(params.info_hash)
        }), peer.socket.onSend)
        debug('sent offer to %s from %s', peer.peerId, params.peer_id)
      })
    }

    if (params.answer) {
      debug('got answer %s from %s', JSON.stringify(params.answer), params.peer_id)

      self.getSwarm(params.info_hash, function (err, swarm) {
        if (self.destroyed) return
        if (err) return self.emit('warning', err)
        if (!swarm) {
          return self.emit('warning', new Error('no swarm with that `info_hash`'))
        }
        // Mark the destination peer as recently used in cache
        var toPeer = swarm.peers.get(params.to_peer_id)
        if (!toPeer) {
          return self.emit('warning', new Error('no peer with that `to_peer_id`'))
        }

        toPeer.socket.send(JSON.stringify({
          action: 'announce',
          answer: params.answer,
          offer_id: params.offer_id,
          peer_id: common.hexToBinary(params.peer_id),
          info_hash: common.hexToBinary(params.info_hash)
        }), toPeer.socket.onSend)
        debug('sent answer to %s from %s', toPeer.peerId, params.peer_id)

        done()
      })
    } else {
      done()
    }

    function done () {
      // emit event once the announce is fully "processed"
      if (params.action === common.ACTIONS.ANNOUNCE) {
        self.emit(common.EVENT_NAMES[params.event], params.peer_id, params)
      }
    }
  })
}

Server.prototype._onWebSocketSend = function (socket, err) {
  var self = this
  if (err) self._onWebSocketError(socket, err)
}

Server.prototype._onWebSocketClose = function (socket) {
  var self = this
  debug('websocket close %s', socket.peerId)
  socket.destroyed = true

  if (socket.peerId) {
    socket.infoHashes.slice(0).forEach(function (infoHash) {
      var swarm = self.torrents[infoHash]
      if (swarm) {
        swarm.announce({
          type: 'ws',
          event: 'stopped',
          numwant: 0,
          peer_id: socket.peerId
        }, noop)
      }
    })
  }

  // ignore all future errors
  socket.onSend = noop
  socket.on('error', noop)

  socket.peerId = null
  socket.infoHashes = null

  if (typeof socket.onMessageBound === 'function') {
    socket.removeListener('message', socket.onMessageBound)
  }
  socket.onMessageBound = null

  if (typeof socket.onErrorBound === 'function') {
    socket.removeListener('error', socket.onErrorBound)
  }
  socket.onErrorBound = null

  if (typeof socket.onCloseBound === 'function') {
    socket.removeListener('close', socket.onCloseBound)
  }
  socket.onCloseBound = null
}

Server.prototype._onWebSocketError = function (socket, err) {
  var self = this
  debug('websocket error %s', err.message || err)
  self.emit('warning', err)
  self._onWebSocketClose(socket)
}

Server.prototype._onRequest = function (params, cb) {
  var self = this
  if (params && params.action === common.ACTIONS.CONNECT) {
    cb(null, { action: common.ACTIONS.CONNECT })
  } else if (params && params.action === common.ACTIONS.ANNOUNCE) {
    self._onAnnounce(params, cb)
  } else if (params && params.action === common.ACTIONS.SCRAPE) {
    self._onScrape(params, cb)
  } else {
    cb(new Error('Invalid action'))
  }
}

Server.prototype._onAnnounce = function (params, cb) {
  var self = this

  self.getSwarm(params.info_hash, function (err, swarm) {
    if (err) return cb(err)
    if (swarm) {
      announce(swarm)
    } else {
      createSwarm()
    }
  })

  function createSwarm () {
    if (self._filter) {
      self._filter(params.info_hash, params, function (err) {
        // Precense of err means that this info_hash is disallowed
        if (err) {
          cb(err)
        } else {
          self.createSwarm(params.info_hash, function (err, swarm) {
            if (err) return cb(err)
            announce(swarm)
          })
        }
      })
    } else {
      self.createSwarm(params.info_hash, function (err, swarm) {
        if (err) return cb(err)
        announce(swarm)
      })
    }
  }

  function announce (swarm) {
    if (!params.event || params.event === 'empty') params.event = 'update'
    swarm.announce(params, function (err, response) {
      if (err) return cb(err)

      if (!response.action) response.action = common.ACTIONS.ANNOUNCE
      if (!response.interval) response.interval = Math.ceil(self.intervalMs / 1000)

      if (params.compact === 1) {
        var peers = response.peers

        // Find IPv4 peers
        response.peers = string2compact(peers.filter(function (peer) {
          return common.IPV4_RE.test(peer.ip)
        }).map(function (peer) {
          return peer.ip + ':' + peer.port
        }))
        // Find IPv6 peers
        response.peers6 = string2compact(peers.filter(function (peer) {
          return common.IPV6_RE.test(peer.ip)
        }).map(function (peer) {
          return '[' + peer.ip + ']:' + peer.port
        }))
      } else if (params.compact === 0) {
        // IPv6 peers are not separate for non-compact responses
        response.peers = response.peers.map(function (peer) {
          return {
            'peer id': common.hexToBinary(peer.peerId),
            ip: peer.ip,
            port: peer.port
          }
        })
      } // else, return full peer objects (used for websocket responses)

      cb(null, response)
    })
  }
}

Server.prototype._onScrape = function (params, cb) {
  var self = this

  if (params.info_hash == null) {
    // if info_hash param is omitted, stats for all torrents are returned
    // TODO: make this configurable!
    params.info_hash = Object.keys(self.torrents)
  }

  series(params.info_hash.map(function (infoHash) {
    return function (cb) {
      self.getSwarm(infoHash, function (err, swarm) {
        if (err) return cb(err)
        if (swarm) {
          swarm.scrape(params, function (err, scrapeInfo) {
            if (err) return cb(err)
            cb(null, {
              infoHash: infoHash,
              complete: (scrapeInfo && scrapeInfo.complete) || 0,
              incomplete: (scrapeInfo && scrapeInfo.incomplete) || 0
            })
          })
        } else {
          cb(null, { infoHash: infoHash, complete: 0, incomplete: 0 })
        }
      })
    }
  }), function (err, results) {
    if (err) return cb(err)

    var response = {
      action: common.ACTIONS.SCRAPE,
      files: {},
      flags: { min_request_interval: Math.ceil(self.intervalMs / 1000) }
    }

    results.forEach(function (result) {
      response.files[common.hexToBinary(result.infoHash)] = {
        complete: result.complete || 0,
        incomplete: result.incomplete || 0,
        downloaded: result.complete || 0 // TODO: this only provides a lower-bound
      }
    })

    cb(null, response)
  })
}

function makeUdpPacket (params) {
  var packet
  switch (params.action) {
    case common.ACTIONS.CONNECT:
      packet = Buffer.concat([
        common.toUInt32(common.ACTIONS.CONNECT),
        common.toUInt32(params.transactionId),
        params.connectionId
      ])
      break
    case common.ACTIONS.ANNOUNCE:
      packet = Buffer.concat([
        common.toUInt32(common.ACTIONS.ANNOUNCE),
        common.toUInt32(params.transactionId),
        common.toUInt32(params.interval),
        common.toUInt32(params.incomplete),
        common.toUInt32(params.complete),
        params.peers
      ])
      break
    case common.ACTIONS.SCRAPE:
      var scrapeResponse = [
        common.toUInt32(common.ACTIONS.SCRAPE),
        common.toUInt32(params.transactionId)
      ]
      for (var infoHash in params.files) {
        var file = params.files[infoHash]
        scrapeResponse.push(
          common.toUInt32(file.complete),
          common.toUInt32(file.downloaded), // TODO: this only provides a lower-bound
          common.toUInt32(file.incomplete)
        )
      }
      packet = Buffer.concat(scrapeResponse)
      break
    case common.ACTIONS.ERROR:
      packet = Buffer.concat([
        common.toUInt32(common.ACTIONS.ERROR),
        common.toUInt32(params.transactionId || 0),
        Buffer.from(String(params['failure reason']))
      ])
      break
    default:
      throw new Error('Action not implemented: ' + params.action)
  }
  return packet
}

function toNumber (x) {
  x = Number(x)
  return x >= 0 ? x : false
}

function noop () {}