fix: preserve exact bytes of request bodies (if possible)

lib/client/socket.js

@@ -4,7 +4,7 @@ const relay = require('../relay');
 const httpErrors = require('../http-errors');
 const Socket = Primus.createSocket({
   transformer: 'engine.io',
-  parser: 'JSON',
+  parser: 'EJSON',
   plugin: {
     'emitter': require('primus-emitter'),
   }

lib/filters/index.js

@@ -1,6 +1,7 @@
 const pathRegexp = require('path-to-regexp');
 const undefsafe = require('undefsafe');
 const replace = require('../replace-vars');
+const tryJSONParse = require('../try-json-parse');
 
 // reads config that defines
 module.exports = ruleSource => {
@@ -76,9 +77,11 @@ module.exports = ruleSource => {
       }
 
       if (valid && req.body) {
+        const parsedBody = tryJSONParse(req.body);
+
         // validate against the body
         const isValid = valid.some(({ path, value }) => {
-          return undefsafe(req.body, path, value);
+          return undefsafe(parsedBody, path, value);
         });
 
         if (!isValid) {

lib/relay.js

@@ -4,6 +4,7 @@ const parse = require('url').parse;
 const format = require('url').format;
 const Filters = require('./filters');
 const replace = require('./replace-vars');
+const tryJSONParse = require('./try-json-parse');
 
 module.exports = {
   request: requestHandler,
@@ -33,7 +34,9 @@ function requestHandler(filterRules) {
         headers: req.headers,
       }, response => {
         debug('%s %s (%s)', req.method, req.url, response.status);
-        res.status(response.status).send(response.body);
+        res.status(response.status)
+          .set(response.headers)
+          .send(response.body);
       });
     });
   };
@@ -78,12 +81,17 @@ function responseHandler(filterRules, config) {
 
       // if the request is all good - and at this point it is, we'll check
       // whether we want to do variable substitution on the body
-      if (body && body.BROKER_VAR_SUB) {
-        debug('variable swap on ', body.BROKER_VAR_SUB);
-        for (const path of body.BROKER_VAR_SUB) {
-          let source = undefsafe(body, path); // get the value
-          source = replace(source, config); // replace the variables
-          undefsafe(body, path, source); // put it back in
+      if (body) {
+        const parsedBody = tryJSONParse(body);
+
+        if (parsedBody.BROKER_VAR_SUB) {
+          debug('variable swap on ', parsedBody.BROKER_VAR_SUB);
+          for (const path of parsedBody.BROKER_VAR_SUB) {
+            let source = undefsafe(parsedBody, path); // get the value
+            source = replace(source, config); // replace the variables
+            undefsafe(parsedBody, path, source); // put it back in
+          }
+          body = JSON.stringify(parsedBody);
         }
       }
 
@@ -101,7 +109,6 @@ function responseHandler(filterRules, config) {
         headers: headers,
         method,
         body,
-        json: true,
       }, (error, response, body) => {
         debug('%s %s (%s)', method, result, (response || { statusCode: 500 }).statusCode);
 
@@ -116,7 +123,8 @@ function responseHandler(filterRules, config) {
 
         emit({
           status: response.statusCode,
-          body: body
+          body: body,
+          headers: response.headers
         });
       });
     });

lib/server/socket.js

@@ -4,7 +4,7 @@ const debug = require('debug')('broker:server');
 const relay = require('../relay');
 
 module.exports = ({ server, filters, config }) => {
-  const io = new Primus(server, { transformer: 'engine.io', parser: 'JSON' });
+  const io = new Primus(server, { transformer: 'engine.io', parser: 'EJSON' });
   io.plugin('emitter', Emitter);
 
   const connections = new Map();

lib/try-json-parse.js

@@ -0,0 +1,7 @@
+module.exports = (data) => {
+  try {
+    return JSON.parse(Buffer.from(data));
+  } catch (e) {
+    return {};
+  }
+}

lib/webserver.js

@@ -5,13 +5,31 @@ const debug = require('debug')('broker');
 
 module.exports = main;
 
+// bodyparser < 2 initializes req.body to {} for requests with no body. This
+// breaks later serialization, so this pair of middlewares ensures that requests
+// with no body have req.body = undefined. This matches the as-yet-unreleased
+// bodyparser 2.x behaviour.
+const EmptyBody = Symbol('Empty Body');
+const markEmptyRequestBody = (req, res, next) => {
+  req.body = req.body || EmptyBody;
+  next();
+};
+const stripEmptyRequestBody = (req, res, next) => {
+  if (req.body === EmptyBody) {
+    delete req.body;
+  }
+  next();
+}
+
 function main({ key = null, cert = null, port = 7341 } = {}, altPort = null) {
   const http = (!key && !cert); // no https if there's no certs
   const https = http ? require('http') : require('https');
   const app = express();
 
   app.disable('x-powered-by');
-  app.use(bodyParser.json());
+  app.use(markEmptyRequestBody);
+  app.use(bodyParser.raw({ type: '*/*' }));
+  app.use(stripEmptyRequestBody);
   app.use('/healthcheck', require('./healthcheck'));
 
   if (altPort) {

package.json

@@ -45,6 +45,7 @@
     "clite": "^0.3.0",
     "debug": "^2.2.0",
     "dotenv": "^2.0.0",
+    "ejson": "^2.1.2",
     "engine.io": "^1.6.11",
     "engine.io-client": "^1.6.11",
     "express": "^4.14.0",

test/unit/filters.test.js

@@ -1,6 +1,8 @@
 const test = require('tap').test;
 const Filters = require('../../lib/filters');
 
+const jsonBuffer = (body) => Buffer.from(JSON.stringify(body));
+
 test('filter on body', t => {
   const filter = Filters(require(__dirname + '/../fixtures/relay.json'));
 
@@ -10,13 +12,13 @@ test('filter on body', t => {
   filter({
     url: '/',
     method: 'POST',
-    body: {
+    body: jsonBuffer({
       commits: [
         {
           modified: ['package.json', 'file1.txt']
         }
       ]
-    }
+    })
   }, (error, res) => {
     t.equal(error, null, 'no error');
     t.equal(res, '/', 'allows the path request');
@@ -25,7 +27,7 @@ test('filter on body', t => {
   filter({
     url: '/',
     method: 'POST',
-    body: {
+    body: jsonBuffer({
       commits: [
         {
           modified: ['file2.txt']
@@ -34,7 +36,7 @@ test('filter on body', t => {
           modified: ['.snyk', 'file1.txt']
         }
       ]
-    }
+    })
   }, (error, res) => {
     t.equal(error, null, 'no error');
     t.equal(res, '/', 'allows the path request');
@@ -43,7 +45,7 @@ test('filter on body', t => {
   filter({
     url: '/',
     method: 'POST',
-    body: {
+    body: jsonBuffer({
       commits: [
         {
           modified: ['file2.txt']
@@ -52,7 +54,7 @@ test('filter on body', t => {
           modified: ['file3.txt', 'file1.txt']
         }
       ]
-    }
+    })
   }, (error, res) => {
     t.equal(error.message, 'blocked', 'has been blocked');
     t.equal(res, undefined, 'no follow allowed');
@@ -61,9 +63,9 @@ test('filter on body', t => {
   filter({
     url: '/',
     method: 'POST',
-    body: {
+    body: jsonBuffer({
       commits: []
-    }
+    })
   }, error => {
     t.equal(error.message, 'blocked', 'has been blocked');
   });

test/unit/relay-response.test.js

@@ -27,18 +27,20 @@ test('relay swaps values found in BROKER_VAR_SUB', t => {
     url: '/*'
   }], config);
 
+  const body = {
+    BROKER_VAR_SUB: ['url'],
+    url: '${HOST}:${PORT}/webhook'
+  };
+
   route({
     url: '/',
     method: 'POST',
-    body: {
-      BROKER_VAR_SUB: ['url'],
-      url: '${HOST}:${PORT}/webhook'
-    },
+    body: Buffer.from(JSON.stringify(body)),
     headers: {},
   }, () => {
     t.equal(spy.callCount, 1, 'request placed');
     const arg = spy.args[0][0];
-    t.equal(arg.body.url, `${config.HOST}:${config.PORT}/webhook`);
+    t.equal(JSON.parse(arg.body).url, `${config.HOST}:${config.PORT}/webhook`);
     t.end();
   });
 

test/unit/try-json-parse.test.js

@@ -0,0 +1,25 @@
+const test  = require('tap').test;
+
+const tryJSONParse = require('../../lib/try-json-parse');
+
+test('tryJSONParse', t => {
+  const data = {
+    number: '123',
+    animals: ['dog', 'cat'],
+    complex: { nested: 'data' },
+  };
+
+  const dataAsJson = JSON.stringify(data);
+  const dataAsBuffer = Buffer.from(dataAsJson);
+  const dataAsUint8Array = Uint8Array.from(dataAsBuffer); // primus with EJSON returns data as Uint8Arrays
+
+  t.same(tryJSONParse(dataAsJson), data, 'Strings parse');
+  t.same(tryJSONParse(dataAsBuffer), data, 'Buffers parse');
+  t.same(tryJSONParse(dataAsUint8Array), data, 'Uint8Arrays parse');
+  t.same(tryJSONParse(undefined), {}, 'undefined parses as empty');
+  t.same(tryJSONParse(null), {}, 'null parses as empty');
+  t.same(tryJSONParse(data), {}, 'objects parse as empty');
+  t.same(tryJSONParse("nonsense"), {}, 'malformed strings parse as empty');
+
+  t.end();
+});