Tests and guards for failure cases

package.json

@@ -39,7 +39,7 @@
     "varint": "^4.0.1"
   },
   "devDependencies": {
-    "aegir": "^6.0.0",
+    "aegir": "^9.1.2",
     "chai": "^3.5.0",
     "pull-block": "^1.0.1",
     "pull-stream": "^3.4.3"

src/decode.js

@@ -20,7 +20,7 @@ function decode (opts) {
   return (read) => {
     reader(read)
     function next () {
-      decodeFromReader(reader, opts, (err, msg) => {
+      _decodeFromReader(reader, opts, (err, msg) => {
         if (err) return p.end(err)
 
         p.push(msg)
@@ -33,12 +33,23 @@ function decode (opts) {
   }
 }
 
+// wrapper to detect sudden pull-stream disconnects
 function decodeFromReader (reader, opts, cb) {
   if (typeof opts === 'function') {
     cb = opts
     opts = {}
   }
 
+  _decodeFromReader(reader, opts, function onComplete (err, msg) {
+    if (err) {
+      if (err === true) return cb(new Error('Unexpected end of input from reader.'))
+      return cb(err)
+    }
+    cb(null, msg)
+  })
+}
+
+function _decodeFromReader (reader, opts, cb) {
   opts = Object.assign({
     fixed: false,
     bytes: 4
@@ -88,6 +99,9 @@ function readVarintMessage (reader, cb) {
 
         rawMsgSize = []
 
+        if (msg.length < msgSize) {
+          return cb(new Error('Message length does not match prefix specified length.'))
+        }
         cb(null, msg)
       })
     })

test/fromReader.spec.js

@@ -0,0 +1,57 @@
+/* eslint-env mocha */
+'use strict'
+
+const pull = require('pull-stream')
+const Reader = require('pull-reader')
+const expect = require('chai').expect
+
+const lp = require('../src')
+
+describe('pull-length-prefixed decodeFromReader', () => {
+  it('basic', (done) => {
+    const input = [
+      new Buffer('haay wuurl!')
+    ]
+
+    const reader = Reader(1e3)
+
+    // length-prefix encode input
+    pull(
+      pull.values(input),
+      lp.encode(),
+      reader
+    )
+
+    // decode from reader
+    lp.decodeFromReader(reader, function (err, output) {
+      if (err) throw err
+      expect(
+        output
+      ).to.be.eql(
+        input[0]
+      )
+      done()
+    })
+  })
+
+  it('empty input', (done) => {
+    const input = []
+
+    const reader = Reader(1e3)
+
+    // length-prefix encode input
+    pull(
+      pull.values(input),
+      lp.encode(),
+      reader
+    )
+
+    // decode from reader
+    lp.decodeFromReader(reader, function (err, output) {
+      expect(err).to.exist
+      expect(err).to.be.instanceof(Error)
+      expect(output).to.not.exist
+      done()
+    })
+  })
+})

test/index.spec.js

@@ -89,6 +89,28 @@ describe('pull-length-prefixed', () => {
     )
   })
 
+  it.skip('invalid prefix', (done) => {
+    const input = [
+      new Buffer('br34k mai h34rt')
+    ]
+
+    pull(
+      // encode valid input
+      pull.values(input),
+      lp.encode(),
+      // corrupt data
+      pull.map(data => data.slice(0, -6)),
+      // attempt decode
+      lp.decode(),
+      pull.collect((err, output) => {
+        expect(err).to.exist
+        expect(err).to.be.instanceof(Error)
+        expect(output).to.not.exist
+        done()
+      })
+    )
+  })
+
   const sizes = [1, 2, 4, 6, 10, 100, 1000]
 
   sizes.forEach((size) => {