Connection.js

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

var config             = imports.config || require('./config');
var log                = imports.log || require('./util/log');
var network            = imports.network || require('./networks')[config.network];

var MAX_RECEIVE_BUFFER = 10000000;
var PROTOCOL_VERSION   = 70000;

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

var Block           = require('./Block');
var Transaction     = require('./Transaction');
var util            = imports.util || require('./util/util');
var Parser          = imports.Parser || require('./util/BinaryParser');
var buffertools     = imports.buffertools || require('buffertools');
var doubleSha256    = imports.doubleSha256 || util.twoSha256;
var nonce           = util.generateNonce();

var BIP0031_VERSION = 60000;

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

  // 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 = !!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.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) {
  var put = new Put();
  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 = 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 = 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);