Decode messages in stream mode

lib/objects.js

@@ -437,8 +437,9 @@ var pubkey = exports.pubkey = {
 function tryDecryptMsg(identities, buf) {
   function inner(i) {
     if (i > last) {
-      var err = new Error("Failed to decrypt msg with given identities");
+      return PPromise.reject(
-      return PPromise.reject(err);
+        new Error("Failed to decrypt msg with given identities")
+      );
     }
     return bmcrypto
       .decrypt(identities[i].encPrivateKey, buf)
@@ -748,8 +749,9 @@ var DEFAULT_ENCODING = msg.TRIVIAL;
 function tryDecryptBroadcastV4(subscriptions, buf) {
   function inner(i) {
     if (i > last) {
-      var err = new Error("Failed to decrypt broadcast with given identities");
+      return PPromise.reject(
-      return PPromise.reject(err);
+        new Error("Failed to decrypt broadcast with given identities")
+      );
     }
     return bmcrypto
       .decrypt(subscriptions[i].getBroadcastPrivateKey(), buf)

lib/structs.js

@@ -28,6 +28,48 @@ function getmsgchecksum(data) {
   return bmcrypto.sha512(data).slice(0, 4);
 }
 
+// \ :3 /
+function findMagic(buf) {
+  var i;
+  var len = buf.length;
+  var firstb = false;
+  var secondb = false;
+  var thirdb = false;
+  for (i = 0; i < len; ++i) {
+    switch (buf[i]) {
+      case 0xE9:
+        firstb = true;
+        break;
+      case 0xBE:
+        if (firstb) { secondb = true; }
+        break;
+      case 0xB4:
+        if (firstb && secondb) { thirdb = true; }
+        break;
+      case 0xD9:
+        if (firstb && secondb && thirdb) { return i - 3; }
+        break;
+      default:
+        firstb = false;
+        secondb = false;
+        thirdb = false;
+    }
+  }
+  // If we reached the end of the buffer but part of the magic matches
+  // we'll still return index of the magic's start position.
+  if (firstb) {
+    if (secondb) {
+      --i;
+    }
+    if (thirdb) {
+      --i;
+    }
+    return i - 1;  // Compensate for last i's increment
+  } else {
+    return -1;
+  }
+}
+
 /**
  * Message structure.
  * @see {@link https://bitmessage.org/wiki/Protocol_specification#Message_structure}
@@ -41,6 +83,92 @@ var message = exports.message = {
    */
   MAGIC: 0xE9BEB4D9,
 
+  /**
+   * Decode message in "stream" mode.  
+   * NOTE: message payload and `rest` are copied (so the runtime can GC
+   * processed buffer data).
+   * @param {Buffer} buf - Data buffer
+   * @return {?{?message: Object, ?error: Error, rest: Buffer}} Decoded
+   * result.
+   */
+  tryDecode: function(buf) {
+    if (buf.length < 24) {
+      // Message is not yet fully received, just skip to next process
+      // cycle.
+      return;
+    }
+    var res = {};
+
+    // Magic.
+    var mindex = findMagic(buf);
+    if (mindex !== 0) {
+      if (mindex === -1) {
+        res.error = new Error("Magic not found, skipping buffer data");
+        res.rest = new Buffer(0);
+      } else {
+        res.error = new Error(
+          "Magic in the middle of buffer, skipping some data at start"
+        );
+        res.rest = new Buffer(buf.length - mindex);
+        buf.copy(res.rest, 0, mindex);
+      }
+      return res;
+    }
+
+    // Payload length.
+    var payloadLength = buf.readUInt32BE(16, true);
+    var msgLength = 24 + payloadLength;
+    // See also: <https://github.com/Bitmessage/PyBitmessage/issues/767>.
+    if (payloadLength > 1600003) {
+      res.error = new Error("Message is too large, skipping it");
+      if (buf.length > msgLength) {
+        res.rest = new Buffer(buf.length - msgLength);
+        buf.copy(res.rest, 0, msgLength);
+      } else {
+        res.rest = new Buffer(0);
+      }
+      return res;
+    }
+    if (buf.length < msgLength) {
+      // Message is not yet fully received, just skip to next process
+      // cycle.
+      return;
+    }
+
+    // Now we can set `rest` value.
+    res.rest = new Buffer(buf.length - msgLength);
+    buf.copy(res.rest, 0, msgLength);
+
+    // Command.
+    var command = buf.slice(4, 16);
+    var firstNonNull = 0;
+    var i;
+    for (i = 11; i >=0; i--) {
+      if (command[i] > 127) {
+        res.error = new Error(
+          "Non-ASCII characters in command, skipping message"
+        );
+        return res;
+      }
+      if (!firstNonNull && command[i] !== 0) {
+        firstNonNull = i + 1;
+      }
+    }
+    command = command.slice(0, firstNonNull).toString("ascii");
+
+    // Payload.
+    var payload = new Buffer(payloadLength);
+    buf.copy(payload, 0, 24, msgLength);
+    var checksum = buf.slice(20, 24);
+    if (!bufferEqual(checksum, getmsgchecksum(payload))) {
+      res.error = new Error("Bad checksum, skipping message");
+      return res;
+    }
+
+    res.message = {command: command, payload: payload, length: msgLength};
+    return res;
+  },
+
   /**
    * Decode message structure.  
    * NOTE: `payload` is copied, `rest` references input buffer.

tests/unit.js

@@ -142,6 +142,37 @@ describe("Common structures", function() {
       expect(res.command).to.equal("ping");
       expect(res.payload.toString("hex")).to.equal("");
     });
+
+    it("should decode messages in stream mode", function() {
+      var res = message.tryDecode(Buffer(""));
+      expect(res).to.not.exist;
+
+      res = message.tryDecode(Buffer(25));
+      expect(res.error).to.match(/magic not found/i);
+      expect(res.rest.toString("hex")).to.equal("");
+      expect(res).to.not.have.property("message");
+
+      res = message.tryDecode(message.encode("test", Buffer([1,2,3])));
+      expect(res).to.not.have.property("error");
+      expect(res.message.command).to.equal("test");
+      expect(res.message.payload.toString("hex")).to.equal("010203");
+      expect(res.rest.toString("hex")).to.equal("");
+
+      var encoded = message.encode("cmd", Buffer("buf"));
+      encoded[20] ^= 1;  // Corrupt checksum
+      encoded = Buffer.concat([encoded, Buffer("rest")]);
+      res = message.tryDecode(encoded);
+      expect(res.error).to.match(/bad checksum/i);
+      expect(res.rest.toString()).to.equal("rest");
+      expect(res).to.not.have.property("message");
+
+      encoded = Buffer.concat([Buffer(10), encoded]);
+      res = message.tryDecode(encoded);
+      expect(res.error).to.match(/magic in the middle/i);
+      expect(res.rest).to.have.length(31);
+      expect(res.rest.readUInt32BE(0)).to.equal(message.MAGIC);
+      expect(res).to.not.have.property("message");
+    });
   });
 
   describe("object", function() {