Connection.js

var imports = require('soop').imports();

var log = imports.log || require('../util/log');

var MAX_RECEIVE_BUFFER = 10000000;
var PROTOCOL_VERSION = 70000;

var Put = imports.Put || require('bufferput');
var Buffers = imports.Buffers || require('buffers');
require('../patches/Buffers.monkey').patch(Buffers);

var bitcoreDefaults = imports.config || require('../config');
var networks = imports.networks || require('../networks');
var Block = imports.Block || require('./Block');
var Transaction = imports.Transaction || require('./Transaction');
var util = imports.util || require('../util');
var Parser = imports.Parser || require('../util/BinaryParser');
var buffertools = imports.buffertools || require('buffertools');
var doubleSha256 = imports.doubleSha256 || util.twoSha256;
var SecureRandom = imports.SecureRandom || require('./SecureRandom');
var nonce = SecureRandom.getPseudoRandomBuffer(8);

var BIP0031_VERSION = 60000;

function Connection(socket, peer, opts) {
  Connection.super(this, arguments);

  this.config = opts || bitcoreDefaults;

  this.network = networks[this.config.network] || networks.livenet;
  this.socket = socket;
  this.peer = peer;

  // check for socks5 proxy options and construct a proxied socket
  if (this.config.proxy) {
    var Socks5Client = imports.Socks5Client || require('socks5-client');
    this.socket = new Socks5Client(this.config.proxy.host, this.config.proxy.port);
  }

  // A connection is considered "active" once we have received verack
  this.active = false;
  // The version incoming packages are interpreted as
  this.recvVer = 0;
  // The version outgoing packages are sent as
  this.sendVer = 0;
  // The (claimed) height of the remote peer's block chain
  this.bestHeight = 0;
  // Is this an inbound connection?
  this.inbound = !!this.socket.server;
  // Have we sent a getaddr on this connection?
  this.getaddr = false;

  // Receive buffer
  this.buffers = new Buffers();

  // Starting 20 Feb 2012, Version 0.2 is obsolete
  // This is the same behavior as the official client
  if (new Date().getTime() > 1329696000000) {
    this.recvVer = 209;
    this.sendVer = 209;
  }

  this.setupHandlers();
}
Connection.parent = imports.parent || require('events').EventEmitter;

Connection.prototype.open = function(callback) {
  if (typeof callback === 'function') this.once('connect', callback);
  this.socket.connect(this.peer.port, this.peer.host);
  return this;
};

Connection.prototype.setupHandlers = function() {
  this.socket.addListener('connect', this.handleConnect.bind(this));
  this.socket.addListener('error', this.handleError.bind(this));
  this.socket.addListener('end', this.handleDisconnect.bind(this));
  this.socket.addListener('data', (function(data) {
    var dumpLen = 35;
    log.debug('[' + this.peer + '] ' +
      'Recieved ' + data.length + ' bytes of data:');
    log.debug('... ' + buffertools.toHex(data.slice(0, dumpLen > data.length ?
        data.length : dumpLen)) +
      (data.length > dumpLen ? '...' : ''));
  }).bind(this));
  this.socket.addListener('data', this.handleData.bind(this));
};

Connection.prototype.handleConnect = function() {
  if (!this.inbound) {
    this.sendVersion();
  }
  this.emit('connect', {
    conn: this,
    socket: this.socket,
    peer: this.peer
  });
};

Connection.prototype.handleError = function(err) {
  if (err.errno == 110 || err.errno == 'ETIMEDOUT') {
    log.info('connection timed out for ' + this.peer);
  } else if (err.errno == 111 || err.errno == 'ECONNREFUSED') {
    log.info('connection refused for ' + this.peer);
  } else {
    log.warn('connection with ' + this.peer + ' ' + err.toString());
  }
  this.emit('error', {
    conn: this,
    socket: this.socket,
    peer: this.peer,
    err: err
  });
};

Connection.prototype.handleDisconnect = function() {
  this.emit('disconnect', {
    conn: this,
    socket: this.socket,
    peer: this.peer
  });
};

Connection.prototype.handleMessage = function(message) {
  if (!message) {
    // Parser was unable to make sense of the message, drop it
    return;
  }

  try {
    switch (message.command) {
      case 'version':
        // Did we connect to ourself?
        if (buffertools.compare(nonce, message.nonce) === 0) {
          this.socket.end();
          return;
        }

        if (this.inbound) {
          this.sendVersion();
        }

        if (message.version >= 209) {
          this.sendMessage('verack', new Buffer([]));
        }
        this.sendVer = Math.min(message.version, PROTOCOL_VERSION);
        if (message.version < 209) {
          this.recvVer = Math.min(message.version, PROTOCOL_VERSION);
        } else {
          // We won't start expecting a checksum until after we've received
          // the 'verack' message.
          this.once('verack', (function() {
            this.recvVer = message.version;
          }).bind(this));
        }
        this.bestHeight = message.start_height;
        break;

      case 'verack':
        this.recvVer = Math.min(message.version, PROTOCOL_VERSION);
        this.active = true;
        break;

      case 'ping':
        if ('object' === typeof message.nonce) {
          this.sendPong(message.nonce);
        }
        break;
    }
  } catch (e) {
    log.err('Error while handling "' + message.command + '" message from ' +
      this.peer + ':\n' +
      (e.stack ? e.stack : e.toString()));
    return;
  }
  this.emit(message.command, {
    conn: this,
    socket: this.socket,
    peer: this.peer,
    message: message
  });
};

Connection.prototype.sendPong = function(nonce) {
  this.sendMessage('pong', nonce);
};

Connection.prototype.sendVersion = function() {
  var subversion = '/BitcoinX:0.1/';

  var put = new Put();
  put.word32le(PROTOCOL_VERSION); // version
  put.word64le(1); // services
  put.word64le(Math.round(new Date().getTime() / 1000)); // timestamp
  put.pad(26); // addr_me
  put.pad(26); // addr_you
  put.put(nonce);
  put.varint(subversion.length);
  put.put(new Buffer(subversion, 'ascii'));
  put.word32le(0);

  this.sendMessage('version', put.buffer());
};

Connection.prototype.sendGetBlocks = function(starts, stop, wantHeaders) {
  // Default value for stop is 0 to get as many blocks as possible (500)
  stop = stop || util.NULL_HASH;

  var put = new Put();

  // https://en.bitcoin.it/wiki/Protocol_specification#getblocks
  put.word32le(this.sendVer);
  put.varint(starts.length);

  for (var i = 0; i < starts.length; i++) {
    if (starts[i].length != 32) {
      throw new Error('Invalid hash length');
    }

    put.put(starts[i]);
  }

  var stopBuffer = new Buffer(stop, 'binary');
  if (stopBuffer.length != 32) {
    throw new Error('Invalid hash length');
  }

  put.put(stopBuffer);

  var command = 'getblocks';
  if (wantHeaders)
    command = 'getheaders';
  this.sendMessage(command, put.buffer());
};

Connection.prototype.sendGetHeaders = function(starts, stop) {
  this.sendGetBlocks(starts, stop, true);
};

Connection.prototype.sendGetData = function(invs) {
  var put = new Put();
  put.varint(invs.length);
  for (var i = 0; i < invs.length; i++) {
    put.word32le(invs[i].type);
    put.put(invs[i].hash);
  }
  this.sendMessage('getdata', put.buffer());
};

Connection.prototype.sendGetAddr = function(invs) {
  var put = new Put();
  this.sendMessage('getaddr', put.buffer());
};

Connection.prototype.sendInv = function(data) {
  if (!Array.isArray(data)) data = [data];
  var put = new Put();
  put.varint(data.length);
  data.forEach(function(value) {
    if (value instanceof Block) {
      // Block
      put.word32le(2); // MSG_BLOCK
    } else {
      // Transaction
      put.word32le(1); // MSG_TX
    }
    put.put(value.getHash());
  });
  this.sendMessage('inv', put.buffer());
};

Connection.prototype.sendHeaders = function(headers) {
  var put = new Put();
  put.varint(headers.length);
  headers.forEach(function(header) {
    put.put(header);

    // Indicate 0 transactions
    put.word8(0);
  });
  this.sendMessage('headers', put.buffer());
};

Connection.prototype.sendTx = function(tx) {
  this.sendMessage('tx', tx.serialize());
};

Connection.prototype.sendBlock = function(block, txs) {
  var put = new Put();

  // Block header
  put.put(block.getHeader());

  // List of transactions
  put.varint(txs.length);
  txs.forEach(function(tx) {
    put.put(tx.serialize());
  });

  this.sendMessage('block', put.buffer());
};

Connection.prototype.sendMessage = function(command, payload) {
  try {
    var magic = this.network.magic;
    var commandBuf = new Buffer(command, 'ascii');
    if (commandBuf.length > 12) throw 'Command name too long';

    var checksum;
    if (this.sendVer >= 209) {
      checksum = doubleSha256(payload).slice(0, 4);
    } else {
      checksum = new Buffer([]);
    }

    var message = new Put(); // -- HEADER --
    message.put(magic); // magic bytes
    message.put(commandBuf); // command name
    message.pad(12 - commandBuf.length); // zero-padded
    message.word32le(payload.length); // payload length
    message.put(checksum); // checksum
    // -- BODY --
    message.put(payload); // payload data

    var buffer = message.buffer();

    log.debug('[' + this.peer + '] ' +
      'Sending message ' + command + ' (' + payload.length + ' bytes)');

    this.socket.write(buffer);
  } catch (err) {
    // TODO: We should catch this error one level higher in order to better
    //       determine how to react to it. For now though, ignoring it will do.
    log.err('Error while sending message to peer ' + this.peer + ': ' +
      (err.stack ? err.stack : err.toString()));
  }
};

Connection.prototype.handleData = function(data) {
  this.buffers.push(data);

  if (this.buffers.length > MAX_RECEIVE_BUFFER) {
    log.err('Peer ' + this.peer + ' exceeded maxreceivebuffer, disconnecting.' +
      (err.stack ? err.stack : err.toString()));
    this.socket.destroy();
    return;
  }

  this.processData();
};

Connection.prototype.processData = function() {
  // If there are less than 20 bytes there can't be a message yet.
  if (this.buffers.length < 20) return;

  var magic = this.network.magic;
  var i = 0;
  for (;;) {
    if (this.buffers.get(i) === magic[0] &&
      this.buffers.get(i + 1) === magic[1] &&
      this.buffers.get(i + 2) === magic[2] &&
      this.buffers.get(i + 3) === magic[3]) {
      if (i !== 0) {
        log.debug('[' + this.peer + '] ' +
          'Received ' + i +
          ' bytes of inter-message garbage: ');
        log.debug('... ' + this.buffers.slice(0, i));

        this.buffers.skip(i);
      }
      break;
    }

    if (i > (this.buffers.length - 4)) {
      this.buffers.skip(i);
      return;
    }
    i++;
  }

  var payloadLen = (this.buffers.get(16)) +
    (this.buffers.get(17) << 8) +
    (this.buffers.get(18) << 16) +
    (this.buffers.get(19) << 24);

  var startPos = (this.recvVer >= 209) ? 24 : 20;
  var endPos = startPos + payloadLen;

  if (this.buffers.length < endPos) return;

  var command = this.buffers.slice(4, 16).toString('ascii').replace(/\0+$/, '');
  var payload = this.buffers.slice(startPos, endPos);
  var checksum = (this.recvVer >= 209) ? this.buffers.slice(20, 24) : null;

  log.debug('[' + this.peer + '] ' +
    'Received message ' + command +
    ' (' + payloadLen + ' bytes)');

  if (checksum !== null) {
    var checksumConfirm = doubleSha256(payload).slice(0, 4);
    if (buffertools.compare(checksumConfirm, checksum) !== 0) {
      log.err('[' + this.peer + '] ' +
        'Checksum failed', {
          cmd: command,
          expected: checksumConfirm.toString('hex'),
          actual: checksum.toString('hex')
        });
      return;
    }
  }

  var message;
  try {
    message = this.parseMessage(command, payload);
  } catch (e) {
    log.err('Error while parsing message ' + command + ' from ' +
      this.peer + ':\n' +
      (e.stack ? e.stack : e.toString()));
  }

  if (message) {
    this.handleMessage(message);
  }

  this.buffers.skip(endPos);
  this.processData();
};

Connection.prototype.parseMessage = function(command, payload) {
  var parser = new Parser(payload);

  var data = {
    command: command
  };

  var i;

  switch (command) {
    case 'version': // https://en.bitcoin.it/wiki/Protocol_specification#version
      data.version = parser.word32le();
      data.services = parser.word64le();
      data.timestamp = parser.word64le();
      data.addr_me = parser.buffer(26);
      data.addr_you = parser.buffer(26);
      data.nonce = parser.buffer(8);
      data.subversion = parser.varStr();
      data.start_height = parser.word32le();
      break;

    case 'inv':
    case 'getdata':
      data.count = parser.varInt();

      data.invs = [];
      for (i = 0; i < data.count; i++) {
        data.invs.push({
          type: parser.word32le(),
          hash: parser.buffer(32)
        });
      }
      break;

    case 'headers':
      data.count = parser.varInt();

      data.headers = [];
      for (i = 0; i < data.count; i++) {
        var header = new Block();
        header.parse(parser);
        data.headers.push(header);
      }
      break;

    case 'block':
      var block = new Block();
      block.parse(parser);

      data.block = block;
      data.version = block.version;
      data.prev_hash = block.prev_hash;
      data.merkle_root = block.merkle_root;
      data.timestamp = block.timestamp;
      data.bits = block.bits;
      data.nonce = block.nonce;

      data.txs = block.txs;

      data.size = payload.length;
      break;

    case 'tx':
      var tx = new Transaction();
      tx.parse(parser);
      return {
        command: command,
        version: tx.version,
        lock_time: tx.lock_time,
        ins: tx.ins,
        outs: tx.outs,
        tx: tx,
      };

    case 'getblocks':
    case 'getheaders':
      // parse out the version
      data.version = parser.word32le();

      // TODO: Limit block locator size?
      // reference implementation limits to 500 results
      var startCount = parser.varInt();

      data.starts = [];
      for (i = 0; i < startCount; i++) {
        data.starts.push(parser.buffer(32));
      }
      data.stop = parser.buffer(32);
      break;

    case 'addr':
      var addrCount = parser.varInt();

      // Enforce a maximum number of addresses per message
      if (addrCount > 1000) {
        addrCount = 1000;
      }

      data.addrs = [];
      for (i = 0; i < addrCount; i++) {
        // TODO: Time actually depends on the version of the other peer (>=31402)
        data.addrs.push({
          time: parser.word32le(),
          services: parser.word64le(),
          ip: parser.buffer(16),
          port: parser.word16be()
        });
      }
      break;

    case 'alert':
      data.payload = parser.varStr();
      data.signature = parser.varStr();
      break;

    case 'ping':
      if (this.recvVer > BIP0031_VERSION) {
        data.nonce = parser.buffer(8);
      }
      break;

    case 'getaddr':
    case 'verack':
    case 'reject':
      // Empty message, nothing to parse
      break;

    default:
      log.err('Connection.parseMessage(): Command not implemented', {
        cmd: command
      });

      // This tells the calling function not to issue an event
      return null;
  }

  return data;
};

module.exports = require('soop')(Connection);