feat: support private & public rulesets

BREAKING CHANGE:

The accept.json file format has now changes from being rooted as an array to being an object with `private` and `public` accept rules.

README.md

@@ -99,12 +99,13 @@ The second, `${PARAM}` is populated with the matching value in your configuratio
 
 The final result is that the broker will accept and forward `GET` requests to my local server that will respond to `https://[email protected]/snyk/broker/master/package.json`.
 
-# TODO / Aims
+### private
 
-- [x] Proxy e2e socket (server -> client -> internal -> client -> server)
-- [x] Can serve as both client and server
-- [x] Client can forward requests from internal to server
-- [ ] Filter relays (i.e. whether the local webserver should accept the inbound request)
+Private filters are for requests that come from the broker server into your client and ask for resources inside your private infrastructure (such as a github enterprise instance).
+
+### public
+
+Public filters are for requests that a recieved on your broker client and are intended to be forwarded to the broker server (such as a github webhook).
 
 # Notes
 

lib/client/index.js

@@ -2,30 +2,29 @@ const debug = require('debug')('broker:client');
 const socket = require('./socket');
 const relay = require('../relay');
 
-module.exports = ({ port = null }) => {
-  debug('running client');
-  // we import config here to allow the tests to invalidate the require cache
-  const config = require('../config');
+module.exports = ({ port = null, config = {}, filters = {} }) => {
+  debug('running');
 
   // start the local webserver to listen for relay requests
   const { app, server } = require('../webserver')(config, port);
 
   const io = socket({
     id: config.brokerId,
     url: config.brokerUrl,
+    filters: filters.private,
   });
 
   app.all('/*', (req, res, next) => {
     res.locals.io = io;
     next();
-  }, relay.request);
+  }, relay.request(filters.public));
 
   return {
     io,
-    close: () => {
+    close: done => {
       debug('closing');
       server.close();
-      io.end();
+      io.destroy(done || (() => debug('closed')));
     },
   };
 };

lib/client/socket.js

@@ -1,6 +1,6 @@
 const debug = require('debug')('broker:client');
-const relay = require('../relay');
 const Primus = require('primus');
+const relay = require('../relay');
 const httpErrors = require('../http-errors');
 const Socket = Primus.createSocket({
   transformer: 'engine.io',
@@ -10,7 +10,7 @@ const Socket = Primus.createSocket({
   }
 });
 
-module.exports = ({ url, id }) => {
+module.exports = ({ url, id, filters }) => {
   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,12 +29,12 @@ module.exports = ({ url, id }) => {
 
   debug('connecting to %s', url);
 
-  const response = relay.response();
+  const response = relay.response(filters);
 
   // RS note: this bind doesn't feel right, it feels like a sloppy way of
   // getting the filters into the request function.
   io.on('request', response);
-  io.on('error', ({ message, type, description }) => {
+  io.on('error', ({ type, description }) => {
     if (type === 'TransportError') {
       console.error(`Failed to connect to broker server: ${httpErrors[description]}`);
     }

lib/filters/index.js

@@ -1,17 +1,22 @@
 const pathRegexp = require('path-to-regexp');
-const debug = require('debug')('broker');
 const undefsafe = require('undefsafe');
 
 // reads config that defines
-module.exports = function (filename) {
+module.exports = ruleSource => {
+  const debug = require('debug')('broker:' + (process.env.BROKER_TYPE || 'filter'));
+  debug('loading new rules', ruleSource);
+
   let rules = [];
   const config = require('../config');
 
-  if (filename) {
+  // polymorphic support
+  if (Array.isArray(ruleSource)) {
+    rules = ruleSource;
+  } else if (ruleSource) {
     try {
-      rules = require(filename);
+      rules = require(ruleSource);
     } catch (e) {
-      console.warn(`Unable to parse ${filename}, ignoring for now: ${e.message}`);
+      console.warn(`Unable to parse ${ruleSource}, ignoring for now: ${e.message}`);
     }
   }
 
@@ -55,7 +60,7 @@ module.exports = function (filename) {
       let url = req.url.split('?').shift(); // strip the querystring
       const res = regexp.exec(url);
       if (!res) {
-        // console.log('false', path, req.url, regexp.source);
+        // debug('bad regexp match', path, req.url, regexp.source);
         // no url match
         return false;
       }
@@ -87,6 +92,7 @@ module.exports = function (filename) {
 
   return (url, callback) => {
     let res = false;
+    debug(`testing ${tests.length} rules`);
     for (const test of tests) {
       res = test(url);
       if (res) {

lib/index.js

@@ -1,21 +1,28 @@
 require('clarify'); // clean the stacktraces
+const path = require('path');
 
 const app = module.exports = {
   client: require('./client'),
   server: require('./server'),
+  main: main,
 };
 
 function main({ port } = {}) {
   // note: the config is loaded in the main function to allow us to mock in tests
   const config = require('./config');
+  const client = !!config.brokerUrl;
+  const method = client ? 'client' : 'server';
+  process.env.BROKER_TYPE = method;
 
-  // if the config has the broker server, then we must assume it's in client
-  // mode.
-  if (config.brokerUrl) {
-    return app.client(config, port);
+  let filters = {};
+  if (config.accept) {
+    require('debug')(`broker:${method}`)('loading rules from %s', config.accept);
+    filters = require(path.resolve(process.cwd(), config.accept));
   }
 
-  return app.server(config, port);
+
+  // if the config has the broker server, then we must assume it's a client
+  return app[method]({ config, port, filters });
 }
 
 if (!module.parent) {

lib/relay.js

@@ -1,44 +1,51 @@
-const debug = require('debug')('broker');
 const request = require('request');
-const path = require('path');
+const Filters = require('./filters');
 const parse = require('url').parse;
 
 module.exports = {
   request: requestHandler,
   response: responseHandler,
 };
 
-function requestHandler(req, res) {
-  debug('send socket request for', req.url);
+function requestHandler(filterRules) {
+  const debug = require('debug')('broker:' + (process.env.BROKER_TYPE || 'relay'));
+  const filters = Filters(filterRules);
 
-  // send the socket request containing the http request we're after
-  res.locals.io.send('request', {
-    url: req.url, // strip the leading
-    method: req.method,
-    body: req.body,
-    headers: {
-      'user-agent': req.headers['user-agent'],
-      authorization: req.headers.authorization,
-    },
-  }, response => {
-    console.log('%s %s (%s)', req.method, req.url, response.status);
-    res.status(response.status).send(response.body);
-  });
-}
+  return (req, res) => {
+    filters(req, (error, result) => {
+      if (error) {
+        return res.status(400).send(error);
+      }
 
-function responseHandler(filterPath) {
-  const config = require('./config');
+      const url = parse(result);
+      req.url = url.pathname;
+      debug('send socket request for', req.url);
 
-  if (!filterPath && config.accept) {
-    filterPath = path.resolve(process.cwd(), config.accept);
-  }
+      // send the socket request containing the http request we're after
+      res.locals.io.send('request', {
+        url: req.url, // strip the leading
+        method: req.method,
+        body: req.body,
+        headers: {
+          'user-agent': req.headers['user-agent'],
+          authorization: req.headers.authorization,
+        },
+      }, response => {
+        debug('%s %s (%s)', req.method, req.url, response.status);
+        res.status(response.status).send(response.body);
+      });
+    });
+  };
+}
 
-  const filters = require('./filters')(filterPath);
+function responseHandler(filterRules) {
+  const filters = Filters(filterRules);
+  const debug = require('debug')('broker:' + (process.env.BROKER_TYPE || 'relay'));
 
   return ({ url, headers, method, body = null } = {}, emit) => {
     // run the request through the filter
     debug('request captured', url, method);
-    filters({ url, method }, (error, result) => {
+    filters({ url, method, body }, (error, result) => {
       if (error) {
         debug('blocked %s %s', method, url);
         return emit({

lib/server/index.js

@@ -2,15 +2,17 @@ const debug = require('debug')('broker:server');
 const socket = require('./socket');
 const relay = require('../relay');
 
-module.exports = function ({ port = null }) {
-  debug('running server');
-  // we import config here to allow the tests to invalidate the require cache
-  const config = require('../config');
+module.exports = ({ config = {}, port = null, filters = {} }) => {
+  debug('running');
 
+  // start the local webserver to listen for relay requests
   const { app, server } = require('../webserver')(config, port);
 
   // bind the socket server to the web server
-  const { io, connections } = socket(server);
+  const { io, connections } = socket({
+    server,
+    filters: filters.private,
+  });
 
   app.all('/broker/:id/*', (req, res, next) => {
     const id = req.params.id;
@@ -21,20 +23,22 @@ module.exports = function ({ port = null }) {
       return res.status(404).send(null);
     }
 
+
     res.locals.io = connections.get(id);
 
      // strip the leading url
     req.url = req.url.slice(`/broker/${id}`.length);
+    debug('request for %s', req.url);
 
     next();
-  }, relay.request);
+  }, relay.request(filters.public));
 
   return {
     io,
-    close: () => {
+    close: done => {
       debug('closing');
       server.close();
-      io.end();
+      io.destroy(done || (() => debug('closed')));
     },
   };
 };

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) => {
+module.exports = ({ server, filters }) => {
   const io = new Primus(server, { transformer: 'engine.io', parser: 'JSON' });
   io.plugin('emitter', Emitter);
 
   const connections = new Map();
-  const response = relay.response();
+  const response = relay.response(filters);
 
   io.on('error', error => console.error(error.stack));
   io.on('offline', () => console.error('Internet access has gone offline'));

test/fixtures/client/filters.json

@@ -1,7 +1,21 @@
-[
-  {
-    "path": "/magic-path/${secret}/package.json",
-    "method": "POST",
-    "origin": "http://localhost:${port}"
-  }
-]
+{
+  "private": [
+    {
+      "path": "/magic-path/${secret}/package.json",
+      "method": "POST",
+      "origin": "http://localhost:${port}"
+    }
+  ],
+  "public": [
+    {
+      "path": "/magic-path/${secret}/package.json",
+      "method": "POST",
+      "origin": "http://localhost:${port}"
+    },
+      {
+        "path": "/service/:package",
+        "method": "GET",
+        "origin": "http://localhost:${port}"
+      }
+  ]
+}

test/fixtures/relay.json

@@ -4,10 +4,18 @@
     "method": "POST",
     "path": "/",
     "valid": [
+      {
+        "path": "commits.*.added.*",
+        "value": "package.json"
+      },
       {
         "path": "commits.*.modified.*",
         "value": "package.json"
       },
+      {
+        "path": "commits.*.added.*",
+        "value": ".snyk"
+      },
       {
         "path": "commits.*.modified.*",
         "value": ".snyk"

test/fixtures/server/filters.json

@@ -1,7 +1,24 @@
-[
-  {
-    "path": "/service/:package",
-    "method": "GET",
-    "origin": "http://localhost:${port}"
-  }
-]
+{
+  "private": [
+    {
+      "path": "/service/:package",
+      "method": "GET",
+      "origin": "http://localhost:${port}"
+    }
+  ],
+  "public": [
+    {
+      "path": "/service/:package",
+      "method": "GET",
+      "origin": "http://localhost:${port}"
+    },
+
+    {
+      "path": "/magic-path/${secret}/package.json",
+      "method": "POST",
+      "origin": "http://localhost:${port}"
+    }
+
+  ]
+
+}