bittorrent-tracker/client.js

module.exports = Client

var bencode = require('bencode')
var BN = require('bn.js')
var common = require('./lib/common')
var compact2string = require('compact2string')
var concat = require('concat-stream')
var debug = require('debug')('bittorrent-tracker')
var dgram = require('dgram')
var EventEmitter = require('events').EventEmitter
var extend = require('extend.js')
var hat = require('hat')
var http = require('http')
var inherits = require('inherits')
var once = require('once')
var url = require('url')

inherits(Client, EventEmitter)

/**
 * A Client manages tracker connections for a torrent.
 *
 * @param {string} peerId  this peer's id
 * @param {Number} port    port number that the client is listening on
 * @param {Object} torrent parsed torrent
 * @param {Object} opts    optional options
 * @param {Number} opts.numWant    number of peers to request
 * @param {Number} opts.interval   interval in ms to send announce requests to the tracker
 */
function Client (peerId, port, torrent, opts) {
  var self = this
  if (!(self instanceof Client)) return new Client(peerId, port, torrent, opts)
  EventEmitter.call(self)
  self._opts = opts || {}

  // required
  self._peerId = Buffer.isBuffer(peerId)
    ? peerId
    : new Buffer(peerId, 'utf8')
  self._port = port
  self._infoHash = Buffer.isBuffer(torrent.infoHash)
    ? torrent.infoHash
    : new Buffer(torrent.infoHash, 'hex')
  self.torrentLength = torrent.length

  // optional
  self._numWant = self._opts.numWant || 50
  self._intervalMs = self._opts.interval || (30 * 60 * 1000) // default: 30 minutes

  debug('new client %s', self._infoHash.toString('hex'))

  if (typeof torrent.announce === 'string') torrent.announce = [ torrent.announce ]
  self._trackers = (torrent.announce || [])
    .filter(function (announceUrl) {
      return announceUrl.indexOf('udp://') === 0 || announceUrl.indexOf('http://') === 0
    })
    .map(function (announceUrl) {
      return new Tracker(self, announceUrl, self._opts)
    })
}

/**
 * Simple convenience function to scrape a tracker for an infoHash without
 * needing to create a Client, pass it a parsed torrent, etc.
 * @param  {string}   announceUrl
 * @param  {string}   infoHash
 * @param  {function} cb
 */
Client.scrape = function (announceUrl, infoHash, cb) {
  cb = once(cb)
  var dummy = {
    peerId: new Buffer('01234567890123456789'),
    port: 6881,
    torrent: {
      infoHash: infoHash,
      announce: [ announceUrl ]
    }
  }
  var client = new Client(dummy.peerId, dummy.port, dummy.torrent)
  client.once('error', cb)
  client.once('scrape', function (data) {
    cb(null, data)
  })
  client.scrape()
}

Client.prototype.start = function (opts) {
  var self = this
  self._trackers.forEach(function (tracker) {
    tracker.start(opts)
  })
}

Client.prototype.stop = function (opts) {
  var self = this
  self._trackers.forEach(function (tracker) {
    tracker.stop(opts)
  })
}

Client.prototype.complete = function (opts) {
  var self = this
  self._trackers.forEach(function (tracker) {
    tracker.complete(opts)
  })
}

Client.prototype.update = function (opts) {
  var self = this
  self._trackers.forEach(function (tracker) {
    tracker.update(opts)
  })
}

Client.prototype.scrape = function (opts) {
  var self = this
  self._trackers.forEach(function (tracker) {
    tracker.scrape(opts)
  })
}

Client.prototype.setInterval = function (intervalMs) {
  var self = this
  self._intervalMs = intervalMs

  self._trackers.forEach(function (tracker) {
    tracker.setInterval(intervalMs)
  })
}

inherits(Tracker, EventEmitter)

/**
 * An individual torrent tracker (used by Client)
 *
 * @param {Client} client       parent bittorrent tracker client
 * @param {string} announceUrl  announce url of tracker
 * @param {Object} opts         optional options
 */
function Tracker (client, announceUrl, opts) {
  var self = this
  EventEmitter.call(self)
  self._opts = opts || {}

  self.client = client

  debug('new tracker %s', announceUrl)

  self._announceUrl = announceUrl
  self._intervalMs = self.client._intervalMs // use client interval initially
  self._interval = null

  if (self._announceUrl.indexOf('udp://') === 0) {
    self._requestImpl = self._requestUdp
  } else if (self._announceUrl.indexOf('http://') === 0) {
    self._requestImpl = self._requestHttp
  }
}

Tracker.prototype.start = function (opts) {
  var self = this
  opts = opts || {}
  opts.event = 'started'

  debug('sent `start` %s', self._announceUrl)
  self._announce(opts)
  self.setInterval(self._intervalMs) // start announcing on intervals
}

Tracker.prototype.stop = function (opts) {
  var self = this
  opts = opts || {}
  opts.event = 'stopped'

  debug('sent `stop` %s', self._announceUrl)
  self._announce(opts)
  self.setInterval(0) // stop announcing on intervals
}

Tracker.prototype.complete = function (opts) {
  var self = this
  opts = opts || {}
  opts.event = 'completed'
  opts.downloaded = opts.downloaded || self.torrentLength || 0

  debug('sent `complete` %s', self._announceUrl)
  self._announce(opts)
}

Tracker.prototype.update = function (opts) {
  var self = this
  opts = opts || {}

  debug('sent `update` %s', self._announceUrl)
  self._announce(opts)
}

/**
 * Send an announce request to the tracker.
 * @param {Object} opts
 * @param {number=} opts.uploaded
 * @param {number=} opts.downloaded
 * @param {number=} opts.left (if not set, calculated automatically)
 */
Tracker.prototype._announce = function (opts) {
  var self = this
  opts = extend({
    uploaded: 0, // default, user should provide real value
    downloaded: 0 // default, user should provide real value
  }, opts)

  if (self.client.torrentLength != null && opts.left == null) {
    opts.left = self.client.torrentLength - (opts.downloaded || 0)
  }

  self._requestImpl(self._announceUrl, opts)
}

/**
 * Send a scrape request to the tracker.
 */
Tracker.prototype.scrape = function () {
  var self = this

  self._scrapeUrl = self._scrapeUrl || getScrapeUrl(self._announceUrl)

  if (!self._scrapeUrl) {
    debug('scrape not supported %s', self._announceUrl)
    self.client.emit('error', new Error('scrape not supported for announceUrl ' + self._announceUrl))
    return
  }

  debug('sent `scrape` %s', self._announceUrl)
  self._requestImpl(self._scrapeUrl, { _scrape: true })
}

Tracker.prototype.setInterval = function (intervalMs) {
  var self = this
  clearInterval(self._interval)

  self._intervalMs = intervalMs
  if (intervalMs) {
    self._interval = setInterval(self.update.bind(self), self._intervalMs)
  }
}

Tracker.prototype._requestHttp = function (requestUrl, opts) {
  var self = this

  if (opts._scrape) {
    opts = extend({
      info_hash: self.client._infoHash.toString('binary')
    }, opts)
  } else {
    opts = extend({
      info_hash: self.client._infoHash.toString('binary'),
      peer_id: self.client._peerId.toString('binary'),
      port: self.client._port,
      compact: 1,
      numwant: self.client._numWant
    }, opts)

    if (self._trackerId) {
      opts.trackerid = self._trackerId
    }
  }

  var fullUrl = requestUrl + '?' + common.querystringStringify(opts)

  var req = http.get(fullUrl, function (res) {
    if (res.statusCode !== 200) {
      res.resume() // consume the whole stream
      self.client.emit('warning', new Error('Invalid response code ' + res.statusCode + ' from tracker ' + requestUrl))
      return
    }
    res.pipe(concat(function (data) {
      if (data && data.length) self._handleResponse(requestUrl, data)
    }))
  })

  req.on('error', function (err) {
    self.client.emit('warning', err)
  })
}

Tracker.prototype._requestUdp = function (requestUrl, opts) {
  var self = this
  opts = opts || {}
  var parsedUrl = url.parse(requestUrl)
  var socket = dgram.createSocket('udp4')
  var transactionId = new Buffer(hat(32), 'hex')

  var stopped = opts.event === 'stopped'
  // if we're sending a stopped message, we don't really care if it arrives, so set
  // a short timer and don't call error
  var timeout = setTimeout(function () {
    timeout = null
    cleanup()
    if (!stopped) {
      error('tracker request timed out')
    }
  }, stopped ? 1500 : 15000)

  if (timeout && timeout.unref) {
    timeout.unref()
  }

  send(Buffer.concat([
    common.CONNECTION_ID,
    common.toUInt32(common.ACTIONS.CONNECT),
    transactionId
  ]))

  socket.on('error', error)

  socket.on('message', function (msg) {
    if (msg.length < 8 || msg.readUInt32BE(4) !== transactionId.readUInt32BE(0)) {
      return error('tracker sent back invalid transaction id')
    }

    var action = msg.readUInt32BE(0)
    switch (action) {
      case 0: // handshake
        if (msg.length < 16) {
          return error('invalid udp handshake')
        }

        if (opts._scrape) {
          scrape(msg.slice(8, 16))
        } else {
          announce(msg.slice(8, 16), opts)
        }

        return

      case 1: // announce
        cleanup()
        if (msg.length < 20) {
          return error('invalid announce message')
        }

        var interval = msg.readUInt32BE(8)
        if (interval && !self._opts.interval && self._intervalMs !== 0) {
          // use the interval the tracker recommends, UNLESS the user manually specifies an
          // interval they want to use
          self.setInterval(interval * 1000)
        }

        self.client.emit('update', {
          announce: self._announceUrl,
          complete: msg.readUInt32BE(16),
          incomplete: msg.readUInt32BE(12)
        })

        var addrs
        try {
          addrs = compact2string.multi(msg.slice(20))
        } catch (err) {
          return self.client.emit('warning', err)
        }
        addrs.forEach(function (addr) {
          self.client.emit('peer', addr)
        })
        break

      case 2: // scrape
        cleanup()
        if (msg.length < 20) {
          return error('invalid scrape message')
        }
        self.client.emit('scrape', {
          announce: self._announceUrl,
          complete: msg.readUInt32BE(8),
          downloaded: msg.readUInt32BE(12),
          incomplete: msg.readUInt32BE(16)
        })
        break

      case 3: // error
        cleanup()
        if (msg.length < 8) {
          return error('invalid error message')
        }
        self.client.emit('error', new Error(msg.slice(8).toString()))
        break
    }
  })

  function send (message) {
    if (!parsedUrl.port) {
      parsedUrl.port = 80
    }
    socket.send(message, 0, message.length, parsedUrl.port, parsedUrl.hostname)
  }

  function error (message) {
    // errors will often happen if a tracker is offline, so don't treat it as fatal
    self.client.emit('warning', new Error(message + ' (' + requestUrl + ')'))
    cleanup()
  }

  function cleanup () {
    if (timeout) {
      clearTimeout(timeout)
      timeout = null
    }
    try { socket.close() } catch (err) {}
  }

  function genTransactionId () {
    transactionId = new Buffer(hat(32), 'hex')
  }

  function announce (connectionId, opts) {
    opts = opts || {}
    genTransactionId()

    send(Buffer.concat([
      connectionId,
      common.toUInt32(common.ACTIONS.ANNOUNCE),
      transactionId,
      self.client._infoHash,
      self.client._peerId,
      toUInt64(opts.downloaded || 0),
      opts.left ? toUInt64(opts.left) : new Buffer('FFFFFFFFFFFFFFFF', 'hex'),
      toUInt64(opts.uploaded || 0),
      common.toUInt32(common.EVENTS[opts.event] || 0),
      common.toUInt32(0), // ip address (optional)
      common.toUInt32(0), // key (optional)
      common.toUInt32(self.client._numWant),
      toUInt16(self.client._port || 0)
    ]))
  }

  function scrape (connectionId) {
    genTransactionId()

    send(Buffer.concat([
      connectionId,
      common.toUInt32(common.ACTIONS.SCRAPE),
      transactionId,
      self.client._infoHash
    ]))
  }
}

Tracker.prototype._handleResponse = function (requestUrl, data) {
  var self = this

  try {
    data = bencode.decode(data)
  } catch (err) {
    return self.client.emit('warning', new Error('Error decoding tracker response: ' + err.message))
  }
  var failure = data['failure reason']
  if (failure) {
    return self.client.emit('warning', new Error(failure))
  }

  var warning = data['warning message']
  if (warning) {
    self.client.emit('warning', new Error(warning))
  }

  if (requestUrl === self._announceUrl) {
    var interval = data.interval || data['min interval']
    if (interval && !self._opts.interval && self._intervalMs !== 0) {
      // use the interval the tracker recommends, UNLESS the user manually specifies an
      // interval they want to use
      self.setInterval(interval * 1000)
    }

    var trackerId = data['tracker id']
    if (trackerId) {
      // If absent, do not discard previous trackerId value
      self._trackerId = trackerId
    }

    self.client.emit('update', {
      announce: self._announceUrl,
      complete: data.complete,
      incomplete: data.incomplete
    })

    if (Buffer.isBuffer(data.peers)) {
      // tracker returned compact response
      var addrs
      try {
        addrs = compact2string.multi(data.peers)
      } catch (err) {
        return self.client.emit('warning', err)
      }
      addrs.forEach(function (addr) {
        self.client.emit('peer', addr)
      })
    } else if (Array.isArray(data.peers)) {
      // tracker returned normal response
      data.peers.forEach(function (peer) {
        self.client.emit('peer', peer.ip + ':' + peer.port)
      })
    }

    if (Buffer.isBuffer(data.peers6)) {
      // tracker returned compact response
      var addrs
      try {
        addrs = compact2string.multi6(data.peers6)
      } catch (err) {
        return self.client.emit('warning', err)
      }
      addrs.forEach(function (addr) {
        self.client.emit('peer', addr)
      })
    } else if (Array.isArray(data.peers6)) {
      // tracker returned normal response
      data.peers.forEach(function (peer) {
        var ip = /:/.test(peer.ip) ?
          '[' + peer.ip + ']' :
          peer.ip
        self.client.emit('peer', ip + ':' + peer.port)
      })
    }
  } else if (requestUrl === self._scrapeUrl) {
    // NOTE: the unofficial spec says to use the 'files' key but i've seen 'host' in practice
    data = data.files || data.host || {}
    data = data[self.client._infoHash.toString('binary')]

    if (!data) {
      self.client.emit('warning', new Error('invalid scrape response'))
    } else {
      // TODO: optionally handle data.flags.min_request_interval (separate from announce interval)
      self.client.emit('scrape', {
        announce: self._announceUrl,
        complete: data.complete,
        incomplete: data.incomplete,
        downloaded: data.downloaded
      })
    }
  }
}

function toUInt16 (n) {
  var buf = new Buffer(2)
  buf.writeUInt16BE(n, 0)
  return buf
}

var MAX_UINT = 4294967295

function toUInt64 (n) {
  if (n > MAX_UINT || typeof n === 'string') {
    var bytes = new BN(n).toArray()
    while (bytes.length < 8) {
      bytes.unshift(0)
    }
    return new Buffer(bytes)
  }
  return Buffer.concat([common.toUInt32(0), common.toUInt32(n)])
}

var UDP_TRACKER = /^udp:\/\//
var HTTP_SCRAPE_SUPPORT = /\/(announce)[^\/]*$/

function getScrapeUrl (announceUrl) {
  if (announceUrl.match(UDP_TRACKER)) return announceUrl
  var match = announceUrl.match(HTTP_SCRAPE_SUPPORT)
  if (match) {
    var i = match.index
    return announceUrl.slice(0, i) + '/scrape' + announceUrl.slice(i + 9)
  }
  return null
}