diff --git a/lib/check-signature.js b/lib/check-signature.js new file mode 100644 index 000000000..8f3d5edad --- /dev/null +++ b/lib/check-signature.js @@ -0,0 +1,45 @@ +const Nacl = require('tweetnacl/nacl-fast'); + +// XXX npm "os" and "child_process" + +// TODO if this process is using too much CPU, we can use "cluster" to add load balancing to this code + +console.log('New child process', process.pid); + +process.on('message', function (data) { + console.log('In process', process.pid) + console.log(+new Date(), "Message received by subprocess"); + if (!data || !data.key || !data.msg || !data.txid) { + process.send({ + error:'E_INVAL' + }); + return; + } + const txid = data.txid; + + const signedMsg = Nacl.util.decodeBase64(data.msg); + var validateKey; + try { + validateKey = Nacl.util.decodeBase64(data.key); + } catch (e) { + process.send({ + txid: txid, + error:'E_BADKEY' + }); + return; + } + // validate the message + const validated = Nacl.sign.open(signedMsg, validateKey); + if (!validated) { + process.send({ + txid: txid, + error:'FAILED' + }); + return; + } + console.log(+new Date(), "Verification done in the subprocess"); + process.send({ + txid: txid, + success: true + }); +}); diff --git a/lib/commands/core.js b/lib/commands/core.js index d7add69b4..bb5156749 100644 --- a/lib/commands/core.js +++ b/lib/commands/core.js @@ -3,6 +3,7 @@ const Core = module.exports; const Util = require("../common-util"); const escapeKeyCharacters = Util.escapeKeyCharacters; +const { fork } = require('child_process'); /* Use Nacl for checking signatures of messages */ const Nacl = require("tweetnacl/nacl-fast"); diff --git a/lib/hk-util.js b/lib/hk-util.js index cb9e9b8ef..a9b9f0eb1 100644 --- a/lib/hk-util.js +++ b/lib/hk-util.js @@ -6,6 +6,8 @@ const nThen = require('nthen'); const Util = require("./common-util"); const MetaRPC = require("./commands/metadata"); const Nacl = require('tweetnacl/nacl-fast'); +const { fork } = require('child_process'); +const numCPUs = require('os').cpus().length; const now = function () { return (new Date()).getTime(); }; const ONE_DAY = 1000 * 60 * 60 * 24; // one day in milliseconds @@ -921,7 +923,36 @@ HK.onDirectMessage = function (Env, Server, seq, userId, json) { * adds timestamps to incoming messages * writes messages to the store */ + + +const onChecked = Util.mkEvent(); +// Create our workers +const workers = []; +for (let i = 0; i < numCPUs; i++) { + workers.push(fork('lib/check-signature.js')); +} +var nextWorker = 0; +var initWorker = function (worker) { + worker.on('message', function (res) { + onChecked.fire(res); + }); + // Spawn a new process in one ends + worker.on('exit', function () { + // XXX make sure it's dead? + var idx = workers.indexOf(worker); + if (idx !== -1) { + workers.splice(idx, 1); + } + // Spawn a new one + var w = fork('lib/check-signature.js'); + workers.push(w); + initWorker(w); + }); +}; +workers.forEach(initWorker); + HK.onChannelMessage = function (Env, Server, channel, msgStruct) { + console.log(+new Date(), "onChannelMessage"); const Log = Env.Log; // TODO our usage of 'channel' here looks prone to errors @@ -962,20 +993,38 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) { let signedMsg = (isCp) ? msgStruct[4].replace(CHECKPOINT_PATTERN, '') : msgStruct[4]; // convert the message from a base64 string into a Uint8Array - // FIXME this can fail and the client won't notice - signedMsg = Nacl.util.decodeBase64(signedMsg); - - // FIXME this can blow up - // TODO check that that won't cause any problems other than not being able to append... - const validateKey = Nacl.util.decodeBase64(metadata.validateKey); - // validate the message - const validated = Nacl.sign.open(signedMsg, validateKey); - if (!validated) { - // don't go any further if the message fails validation - w.abort(); - Log.info("HK_SIGNED_MESSAGE_REJECTED", 'Channel '+channel.id); - return; - } + const txid = Util.uid(); + const next = w(); + + // Listen for messages + const onCheck = function (res) { + if (!res) { return; } + if (res.txid !== txid) { return; } + // Wev'e received an answer, remove this handler + console.log(+new Date(), "Received verification response"); + onChecked.unreg(onCheck); + if (res.error) { + // don't go any further if the message fails validation + if (res.error === 'FAILED') { + // Signature doesn't match the public key + Log.info("HK_SIGNED_MESSAGE_REJECTED", 'Channel '+channel.id); + } + w.abort(); + return; + } + // Success, continue to next block + next(); + }; + onChecked.reg(onCheck); + + console.log(+new Date(), "Send verification request"); + // Send the request + workers[nextWorker].send({ + txid: txid, + msg: signedMsg, + key: metadata.validateKey + }); + if (++nextWorker > numCPUs) { nextWorker = 0; } }).nThen(function () { // do checkpoint stuff... @@ -1002,7 +1051,9 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) { msgStruct.push(now()); // storeMessage + console.log(+new Date(), "Storing message"); storeMessage(Env, channel, JSON.stringify(msgStruct), isCp, getHash(msgStruct[4], Log)); + console.log(+new Date(), "Message stored"); }); };