Merge branch 'fix/body-preservation' into develop

lib/client/index.js

@@ -9,6 +9,7 @@ module.exports = ({ port = null, config = {}, filters = {} }) => {
     id: config.brokerId,
     url: config.brokerUrl,
     filters: filters.private,
+    config,
   });
 
   // start the local webserver to listen for relay requests

lib/client/socket.js

@@ -4,13 +4,13 @@ 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'),
   }
 });
 
-module.exports = ({ url, id, filters }) => {
+module.exports = ({ url, id, filters, config }) => {
   if (!id) { // null, undefined, empty, etc.
     debug('missing client id');
     const error = new ReferenceError('BROKER_ID is required to successfully identify itself to the server');
@@ -29,7 +29,7 @@ module.exports = ({ url, id, filters }) => {
 
   debug('connecting to %s', url);
 
-  const response = relay.response(filters);
+  const response = relay.response(filters, config);
 
   // RS note: this bind doesn't feel right, it feels like a sloppy way of
   // getting the filters into the request function.

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,7 +4,7 @@ const parse = require('url').parse;
 const format = require('url').format;
 const Filters = require('./filters');
 const replace = require('./replace-vars');
-const config = require('./config');
+const tryJSONParse = require('./try-json-parse');
 
 module.exports = {
   request: requestHandler,
@@ -34,13 +34,15 @@ 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);
       });
     });
   };
 }
 
-function responseHandler(filterRules) {
+function responseHandler(filterRules, config) {
   const filters = Filters(filterRules);
   const debug = require('debug')('broker:' + (process.env.BROKER_TYPE || 'relay'));
 
@@ -79,12 +81,17 @@ function responseHandler(filterRules) {
 
       // 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);
         }
       }
 
@@ -102,7 +109,6 @@ function responseHandler(filterRules) {
         headers: headers,
         method,
         body,
-        json: true,
       }, (error, response, body) => {
         debug('%s %s (%s)', method, result, (response || { statusCode: 500 }).statusCode);
 
@@ -117,7 +123,8 @@ function responseHandler(filterRules) {
 
         emit({
           status: response.statusCode,
-          body: body
+          body: body,
+          headers: response.headers
         });
       });
     });

lib/server/index.js

@@ -12,6 +12,7 @@ module.exports = ({ config = {}, port = null, filters = {} }) => {
   const { io, connections } = socket({
     server,
     filters: filters.private,
+    config,
   });
 
   app.all('/broker/:id/*', (req, res, next) => {

lib/server/socket.js

@@ -3,12 +3,12 @@ const Emitter = require('primus-emitter');
 const debug = require('debug')('broker:server');
 const relay = require('../relay');
 
-module.exports = ({ server, filters }) => {
-  const io = new Primus(server, { transformer: 'engine.io', parser: 'JSON' });
+module.exports = ({ server, filters, config }) => {
+  const io = new Primus(server, { transformer: 'engine.io', parser: 'EJSON' });
   io.plugin('emitter', Emitter);
 
   const connections = new Map();
-  const response = relay.response(filters);
+  const response = relay.response(filters, config);
 
   io.on('error', error => console.error(error.stack));
   io.on('offline', () => console.error('Internet access has gone offline'));

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/fixtures/client/filters.json

@@ -1,21 +1,51 @@
 {
   "private": [
     {
-      "path": "/magic-path/${secret}/package.json",
+      "path": "/echo-param/${param}",
+      "method": "GET",
+      "origin": "http://localhost:${originPort}"
+    },
+
+    {
+      "path": "/echo-body",
+      "method": "POST",
+      "origin": "http://localhost:${originPort}"
+    },
+
+    {
+      "path": "/echo-body/filtered",
       "method": "POST",
-      "origin": "http://localhost:${port}"
+      "origin": "http://localhost:${originPort}",
+      "valid": [
+        {
+          "//": "accept requests with 'proxy.*: please' in their body",
+          "path": "proxy.*",
+          "value": "please"
+        }
+      ]
     }
   ],
   "public": [
     {
-      "path": "/magic-path/${secret}/package.json",
-      "method": "POST",
-      "origin": "http://localhost:${port}"
+      "path": "/echo-param/${param}",
+      "method": "GET"
     },
-      {
-        "path": "/service/:package",
-        "method": "GET",
-        "origin": "http://localhost:${port}"
-      }
+
+    {
+      "path": "/echo-body",
+      "method": "POST"
+    },
+
+    {
+      "path": "/echo-body/filtered",
+      "method": "POST",
+      "valid": [
+        {
+          "//": "accept requests with 'proxy.*: please' in their body",
+          "path": "proxy.*",
+          "value": "please"
+        }
+      ]
+    }
   ]
 }

test/fixtures/server/filters.json

@@ -1,24 +1,27 @@
 {
   "private": [
     {
-      "path": "/service/:package",
+      "path": "/echo-param/${param}",
       "method": "GET",
-      "origin": "http://localhost:${port}"
+      "origin": "http://localhost:${originPort}"
+    },
+
+    {
+      "path": "/echo-body/:param?",
+      "method": "POST",
+      "origin": "http://localhost:${originPort}"
     }
   ],
   "public": [
     {
-      "path": "/service/:package",
-      "method": "GET",
-      "origin": "http://localhost:${port}"
+      "path": "/*",
+      "method": "GET"
     },
 
     {
-      "path": "/magic-path/${secret}/package.json",
-      "method": "POST",
-      "origin": "http://localhost:${port}"
+      "path": "/*",
+      "method": "POST"
     }
 
   ]
-
 }