refactor(proxy): proxy blockchain accounts so that they are available in the Dapp

Author: michaelsbradleyjr

package.json

@@ -131,6 +131,7 @@
     "multihashes": "0.4.14",
     "neo-blessed": "0.2.0",
     "netcat": "1.3.5",
+    "node-http-proxy-json": "0.1.6",
     "node-ipc": "9.1.1",
     "node-sass": "4.9.3",
     "npmlog": "4.1.2",

src/lib/modules/blockchain_process/blockchain.js

@@ -147,14 +147,17 @@ Blockchain.prototype.initProxy = function () {
 };
 
 Blockchain.prototype.setupProxy = async function () {
+  const AccountParser = require('../../utils/accountParser');
   if (!this.proxyIpc) this.proxyIpc = new Ipc({ipcRole: 'client'});
 
+  const addresses = AccountParser.parseAccountsConfig(this.userConfig.accounts, false, this.logger);
+
   let wsProxy;
   if (this.config.wsRPC) {
-    wsProxy = proxy.serve(this.proxyIpc, this.config.wsHost, this.config.wsPort, true, this.config.wsOrigins, this.certOptions);
+    wsProxy = proxy.serve(this.proxyIpc, this.config.wsHost, this.config.wsPort, true, this.config.wsOrigins, addresses, this.certOptions);
   }
 
-  [this.rpcProxy, this.wsProxy] = await Promise.all([proxy.serve(this.proxyIpc, this.config.rpcHost, this.config.rpcPort, false, undefined, this.certOptions), wsProxy]);
+  [this.rpcProxy, this.wsProxy] = await Promise.all([proxy.serve(this.proxyIpc, this.config.rpcHost, this.config.rpcPort, false, null, addresses, this.certOptions), wsProxy]);
 };
 
 Blockchain.prototype.shutdownProxy = function () {

src/lib/modules/blockchain_process/httpProxyOverride.js

@@ -0,0 +1,123 @@
+/* global require */
+
+const http = require('http');
+const https = require('https');
+const httpProxyWsIncoming = require('http-proxy/lib/http-proxy/passes/ws-incoming');
+const common = require('http-proxy/lib/http-proxy/common');
+
+const CRLF = '\r\n';
+
+httpProxyWsIncoming.stream = (req, socket, options, head, server, cb) => {
+  const createHttpHeader = function(line, headers) {
+    return Object.keys(headers).reduce(function (head, key) {
+      const value = headers[key];
+      if (!Array.isArray(value)) {
+        head.push(`${key}: ${value}`);
+        return head;
+      }
+      for (let i = 0; i < value.length; i++) {
+        head.push(`${key}: ${value[i]}`);
+      }
+      return head;
+    }, [line])
+      .join(CRLF) + `${CRLF}${CRLF}`;
+  };
+
+  common.setupSocket(socket);
+
+  if (head && head.length) socket.unshift(head);
+
+  const protocol = common.isSSL.test(options.target.protocol) ? https : http;
+
+  const proxyReq = protocol.request(
+    common.setupOutgoing(options.ssl || {}, options, req)
+  );
+
+  // Enable developers to modify the proxyReq before headers are sent
+  if (server) {
+    server.emit('proxyReqWs', proxyReq, req, socket, options, head);
+  }
+
+  // Error Handler
+  proxyReq.on('error', onOutgoingError);
+  proxyReq.on('response', function (res) {
+    // if upgrade event isn't going to happen, close the socket
+    if (!res.upgrade) {
+      const {httpVersion, statusCode, statusMessage, headers} = res;
+      socket.write(createHttpHeader(
+        `HTTP/${httpVersion} ${statusCode} ${statusMessage}`,
+        headers
+      ));
+      res.pipe(socket);
+    }
+  });
+
+  proxyReq.on('upgrade', function(proxyRes, proxySocket, proxyHead) {
+    proxySocket.on('error', onOutgoingError);
+
+    // Allow us to listen when the websocket has completed
+    proxySocket.on('end', function () {
+      server.emit('close', proxyRes, proxySocket, proxyHead);
+    });
+
+    // The pipe below will end proxySocket if socket closes cleanly, but not
+    // if it errors (eg, vanishes from the net and starts returning
+    // EHOSTUNREACH). We need to do that explicitly.
+    socket.on('error', function () {
+      proxySocket.end();
+    });
+
+    common.setupSocket(proxySocket);
+
+    if (proxyHead && proxyHead.length) proxySocket.unshift(proxyHead);
+
+    // Remark: Handle writing the headers to the socket when switching protocols
+    // Also handles when a header is an array
+    socket.write(createHttpHeader(
+      'HTTP/1.1 101 Switching Protocols',
+      proxyRes.headers
+    ));
+
+    let proxyStream = proxySocket;
+
+    if (options.createWsServerTransformStream) {
+      const wsServerTransformStream = options.createWsServerTransformStream(
+        req,
+        proxyReq,
+        proxyRes,
+      );
+
+      wsServerTransformStream.on('error', onOutgoingError);
+      proxyStream = proxyStream.pipe(wsServerTransformStream);
+    }
+
+    proxyStream = proxyStream.pipe(socket);
+
+    if (options.createWsClientTransformStream) {
+      const wsClientTransformStream = options.createWsClientTransformStream(
+        req,
+        proxyReq,
+        proxyRes,
+      );
+
+      wsClientTransformStream.on('error', onOutgoingError);
+      proxyStream = proxyStream.pipe(wsClientTransformStream);
+    }
+
+    proxyStream.pipe(proxySocket);
+
+    server.emit('open', proxySocket);
+    server.emit('proxySocket', proxySocket);  //DEPRECATED.
+  });
+
+  return proxyReq.end(); // XXX: CHECK IF THIS IS THIS CORRECT
+
+  function onOutgoingError(err) {
+    if (cb) {
+      cb(err, req, socket);
+    } else {
+      server.emit('error', err, req, socket);
+    }
+    socket.end();
+  }
+};

src/lib/modules/blockchain_process/proxy.js

@@ -1,14 +1,30 @@
 /* global Buffer __ exports require */
 
+require('./httpProxyOverride');
 const Asm = require('stream-json/Assembler');
 const {canonicalHost, defaultHost} = require('../../utils/host');
 const constants = require('../../constants.json');
+const {Duplex} = require('stream');
 const http = require('http');
 const httpProxy = require('http-proxy');
 const {parser: jsonParser} = require('stream-json');
 const pump = require('pump');
 const utils = require('../../utils/utils');
 const WsParser = require('simples/lib/parsers/ws');
+const WsWrapper = require('simples/lib/ws/wrapper');
+const modifyResponse = require('node-http-proxy-json');
+
+const METHODS_TO_MODIFY = {accounts: 'eth_accounts'};
+
+const modifyPayload = (toModifyPayloads, body, accounts) => {
+  switch (toModifyPayloads[body.id]) {
+    case METHODS_TO_MODIFY.accounts:
+      body.result = body.result.concat(accounts);
+      break;
+    default:
+  }
+  return body;
+};
 
 const hex = (n) => {
   let _n = n.toString(16);
@@ -32,14 +48,18 @@ const parseJsonMaybe = (string) => {
   return object;
 };
 
-exports.serve = async (ipc, host, port, ws, origin, certOptions={}) => {
+exports.serve = async (ipc, host, port, ws, origin, accounts, certOptions={}) => {
   const commList = {};
   const receipts = {};
   const transactions = {};
+  const toModifyPayloads = {};
 
   const trackRequest = (req) => {
     if (!req) return;
     try {
+      if (Object.values(METHODS_TO_MODIFY).includes(req.method)) {
+        toModifyPayloads[req.id] = req.method;
+      }
       if (req.method === 'eth_sendTransaction') {
         commList[req.id] = {
           type: 'contract-log',
@@ -121,7 +141,32 @@ exports.serve = async (ipc, host, port, ws, origin, certOptions={}) => {
       host: canonicalHost(host),
       port: port
     },
-    ws: ws
+    ws: ws,
+    createWsServerTransformStream: (_req, _proxyReq, _proxyRes) => {
+      const parser = new WsParser(0, true);
+      parser.on('frame', ({data: buffer}) => {
+        let object = parseJsonMaybe(buffer.toString());
+        if (object) {
+          object = modifyPayload(toModifyPayloads, object, accounts);
+          // track the modified response
+          trackResponse(object);
+          // send the modified response
+          WsWrapper.wrap(
+            {connection: dupl, masked: 0},
+            Buffer.from(JSON.stringify(object)),
+            () => {}
+          );
+        }
+      });
+      const dupl = new Duplex({
+        read(_size) {},
+        write(chunk, encoding, callback) {
+          parser.write(chunk);
+          callback();
+        }
+      });
+      return dupl;
+    }
   });
 
   proxy.on('error', (err) => {
@@ -131,15 +176,14 @@ exports.serve = async (ipc, host, port, ws, origin, certOptions={}) => {
     );
   });
 
-  proxy.on('proxyRes', (proxyRes, req, _res) => {
-    if (req.method === 'POST') {
-      // messages FROM the target
-      Asm.connectTo(
-        pump(proxyRes, jsonParser())
-      ).on('done', ({current: object}) => {
-        trackResponse(object);
-      });
-    }
+  proxy.on('proxyRes', (proxyRes, req, res) => {
+    modifyResponse(res, proxyRes, (body) => {
+      if (body) {
+        body = modifyPayload(toModifyPayloads, body, accounts);
+        trackResponse(body);
+      }
+      return body;
+    });
   });
 
   const server = http.createServer((req, res) => {
@@ -158,17 +202,11 @@ exports.serve = async (ipc, host, port, ws, origin, certOptions={}) => {
   });
 
   if (ws) {
-    server.on('upgrade', function (msg, socket, head) {
+    server.on('upgrade', (msg, socket, head) => {
       proxy.ws(msg, socket, head);
     });
 
-    proxy.on('open', (proxySocket) => {
-      // messages FROM the target
-      pump(proxySocket, new WsParser(0, true)).on('frame', ({data: buffer}) => {
-        const object = parseJsonMaybe(buffer.toString());
-        trackResponse(object);
-      });
-    });
+    proxy.on('open', (_proxySocket) => { /* messages FROM the target */ });
 
     proxy.on('proxyReqWs', (_proxyReq, _req, socket) => {
       // messages TO the target

src/lib/modules/blockchain_process/simulator.js

@@ -67,7 +67,7 @@ class Simulator {
   }
 
   runCommand(cmds, useProxy, host, port) {
-    const ganache_main = require.resolve('ganache-cli', {paths: fs.embarkPath('node_modules')});
+    const ganache_main = require.resolve('ganache-cli', {paths: [fs.embarkPath('node_modules')]});
     const ganache_json = pkgUp.sync(path.dirname(ganache_main));
     const ganache_root = path.dirname(ganache_json);
     const ganache_bin = require(ganache_json).bin;

src/lib/modules/code_generator/index.js

@@ -290,7 +290,7 @@ class CodeGenerator {
       function getWeb3Location(next) {
         self.events.request("version:get:web3", function(web3Version) {
           if (web3Version === "1.0.0-beta") {
-            return next(null, require.resolve("web3", {paths: fs.embarkPath("node_modules")}));
+            return next(null, require.resolve("web3", {paths: [fs.embarkPath("node_modules")]}));
           }
           self.events.request("version:getPackageLocation", "web3", web3Version, function(err, location) {
             return next(null, fs.dappPath(location));
@@ -356,7 +356,7 @@ class CodeGenerator {
       function getWeb3Location(next) {
         self.events.request("version:get:web3", function(web3Version) {
           if (web3Version === "1.0.0-beta") {
-            return next(null, require.resolve("web3", {paths: fs.embarkPath("node_modules")}));
+            return next(null, require.resolve("web3", {paths: [fs.embarkPath("node_modules")]}));
           }
           self.events.request("version:getPackageLocation", "web3", web3Version, function(err, location) {
             return next(null, fs.dappPath(location));