check authenticated rpc signatures in separate threads

lib/check-signature.js

@@ -1,47 +0,0 @@
-/* jshint esversion: 6 */
-/* global process */
-const Nacl = require('tweetnacl/nacl-fast');
-
-// 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) {
-        return void process.send({
-            error:'E_INVAL'
-        });
-    }
-    const txid = data.txid;
-
-    var signedMsg;
-    try {
-        signedMsg = Nacl.util.decodeBase64(data.msg);
-    } catch (e) {
-        return void process.send({
-            txid: txid,
-            error: 'E_BAD_MESSAGE',
-        });
-    }
-
-    var validateKey;
-    try {
-        validateKey = Nacl.util.decodeBase64(data.key);
-    } catch (e) {
-        return void process.send({
-            txid: txid,
-            error:'E_BADKEY'
-        });
-    }
-    // validate the message
-    const validated = Nacl.sign.open(signedMsg, validateKey);
-    if (!validated) {
-        return void process.send({
-            txid: txid,
-            error:'FAILED'
-        });
-    }
-    process.send({
-        txid: txid,
-    });
-});

lib/commands/core.js

@@ -129,44 +129,6 @@ Core.isValidCookie = function (Sessions, publicKey, cookie) {
     return true;
 };
 
-Core.checkSignature = function (Env, signedMsg, signature, publicKey) {
-    if (!(signedMsg && publicKey)) { return false; }
-
-    var signedBuffer;
-    var pubBuffer;
-    var signatureBuffer;
-
-    try {
-        signedBuffer = Nacl.util.decodeUTF8(signedMsg);
-    } catch (e) {
-        Env.Log.error('INVALID_SIGNED_BUFFER', signedMsg);
-        return null;
-    }
-
-    try {
-        pubBuffer = Nacl.util.decodeBase64(publicKey);
-    } catch (e) {
-        return false;
-    }
-
-    try {
-        signatureBuffer = Nacl.util.decodeBase64(signature);
-    } catch (e) {
-        return false;
-    }
-
-    if (pubBuffer.length !== 32) {
-        Env.Log.error('PUBLIC_KEY_LENGTH', publicKey);
-        return false;
-    }
-
-    if (signatureBuffer.length !== 64) {
-        return false;
-    }
-
-    return Nacl.sign.detached.verify(signedBuffer, signatureBuffer, pubBuffer);
-};
-
 // E_NO_OWNERS
 Core.hasOwners = function (metadata) {
     return Boolean(metadata && Array.isArray(metadata.owners));

lib/hk-util.js

@@ -849,7 +849,7 @@ HK.initializeIndexWorkers = function (Env, config, _cb) {
             if (idx !== -1) {
                 workers.splice(idx, 1);
             }
-            var w = fork('lib/compute-index');
+            var w = fork('lib/workers/compute-index');
             initWorker(w, function (err) {
                 if (err) {
                     throw new Error(err);
@@ -878,7 +878,7 @@ HK.initializeIndexWorkers = function (Env, config, _cb) {
 
     nThen(function (w) {
         OS.cpus().forEach(function () {
-            initWorker(fork('lib/compute-index'), w(function (err) {
+            initWorker(fork('lib/workers/compute-index'), w(function (err) {
                 if (!err) { return; }
                 w.abort();
                 return void cb(err);
@@ -898,7 +898,7 @@ HK.initializeValidationWorkers = function (Env) {
     // Create our workers
     const workers = [];
     for (let i = 0; i < numCPUs; i++) {
-        workers.push(fork('lib/check-signature.js'));
+        workers.push(fork('lib/workers/check-signature.js'));
     }
 
     const response = Util.response();
@@ -911,13 +911,12 @@ HK.initializeValidationWorkers = function (Env) {
         });
         // 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');
+            var w = fork('lib/workers/check-signature.js');
             workers.push(w);
             initWorker(w);
         });
@@ -925,27 +924,39 @@ HK.initializeValidationWorkers = function (Env) {
     workers.forEach(initWorker);
 
     var nextWorker = 0;
-    Env.validateMessage = function (signedMsg, key, _cb) {
+    const send = function (msg, _cb) {
         // let's be paranoid about asynchrony and only calling back once..
-        var cb = Util.once(Util.mkAsync(_cb));
-
-        var txid = Util.uid();
-
-        // expect a response within 15s
-        response.expect(txid, cb, 15000);
-
         nextWorker = (nextWorker + 1) % workers.length;
         if (workers.length === 0 || typeof(workers[nextWorker].send) !== 'function') {
             console.error(workers);
             throw new Error("INVALID_WORKERS");
         }
 
+        var cb = Util.once(Util.mkAsync(_cb));
+        var txid = msg.txid = Util.uid();
+
+        // expect a response within 15s
+        response.expect(txid, cb, 15000);
+
         // Send the request
-        workers[nextWorker].send({
+        workers[nextWorker].send(msg);
-            txid: txid,
+    };
+
+    Env.validateMessage = function (signedMsg, key, cb) {
+        send({
             msg: signedMsg,
             key: key,
-        });
+            command: 'INLINE',
+        }, cb);
+    };
+
+    Env.checkSignature = function (signedMsg, signature, publicKey, cb) {
+        send({
+            command: 'DETACHED',
+            sig: signature,
+            msg: signedMsg,
+            key: publicKey,
+        }, cb);
     };
 };
 

lib/rpc.js

@@ -175,11 +175,13 @@ var rpc = function (Env, Server, userId, data, respond) {
     if (isAuthenticatedCall(command)) {
         // check the signature on the message
         // refuse the command if it doesn't validate
-        if (Core.checkSignature(Env, serialized, signature, publicKey) === true) {
+        return void Env.checkSignature(serialized, signature, publicKey, function (err) {
+            if (err) {
+                return void respond("INVALID_SIGNATURE_OR_PUBLIC_KEY");
+            }
             HK.authenticateNetfluxSession(Env, userId, publicKey);
             return void handleAuthenticatedMessage(Env, publicKey, msg, respond, Server);
-        }
+        });
-        return void respond("INVALID_SIGNATURE_OR_PUBLIC_KEY");
     }
     Env.Log.warn('INVALID_RPC_CALL', command);
     return void respond("INVALID_RPC_CALL");

lib/workers/check-signature.js

@@ -0,0 +1,96 @@
+/* jshint esversion: 6 */
+/* global process */
+const Nacl = require('tweetnacl/nacl-fast');
+
+const COMMANDS = {};
+
+COMMANDS.INLINE = function (data, cb) {
+    var signedMsg;
+    try {
+        signedMsg = Nacl.util.decodeBase64(data.msg);
+    } catch (e) {
+        return void cb('E_BAD_MESSAGE');
+    }
+
+    var validateKey;
+    try {
+        validateKey = Nacl.util.decodeBase64(data.key);
+    } catch (e) {
+        return void cb("E_BADKEY");
+    }
+    // validate the message
+    const validated = Nacl.sign.open(signedMsg, validateKey);
+    if (!validated) {
+        return void cb("FAILED");
+    }
+    cb();
+};
+
+const checkDetachedSignature = function (signedMsg, signature, publicKey) {
+    if (!(signedMsg && publicKey)) { return false; }
+
+    var signedBuffer;
+    var pubBuffer;
+    var signatureBuffer;
+
+    try {
+        signedBuffer = Nacl.util.decodeUTF8(signedMsg);
+    } catch (e) {
+        throw new Error("INVALID_SIGNED_BUFFER");
+    }
+
+    try {
+        pubBuffer = Nacl.util.decodeBase64(publicKey);
+    } catch (e) {
+        throw new Error("INVALID_PUBLIC_KEY");
+    }
+
+    try {
+        signatureBuffer = Nacl.util.decodeBase64(signature);
+    } catch (e) {
+        throw new Error("INVALID_SIGNATURE");
+    }
+
+    if (pubBuffer.length !== 32) {
+        throw new Error("INVALID_PUBLIC_KEY_LENGTH");
+    }
+
+    if (signatureBuffer.length !== 64) {
+        throw new Error("INVALID_SIGNATURE_LENGTH");
+    }
+
+    if (Nacl.sign.detached.verify(signedBuffer, signatureBuffer, pubBuffer) !== true) {
+        throw new Error("FAILED");
+    }
+};
+
+COMMANDS.DETACHED = function (data, cb) {
+    try {
+        checkDetachedSignature(data.msg, data.sig, data.key)
+    } catch (err) {
+        return void cb(err && err.message);
+    }
+    cb();
+};
+
+process.on('message', function (data) {
+    if (!data || !data.key || !data.msg || !data.txid) {
+        return void process.send({
+            error:'E_INVAL'
+        });
+    }
+
+    const cb = function (err) {
+        process.send({
+            txid: data.txid,
+            error: err,
+        });
+    };
+
+    const command = COMMANDS[data.command];
+    if (typeof(command) !== 'function') {
+        return void cb("E_BAD_COMMAND");
+    }
+
+    command(data, cb);
+});

lib/workers/compute-index.js

@@ -1,9 +1,9 @@
 /* jshint esversion: 6 */
 /* global process */
 
-const HK = require("./hk-util");
+const HK = require("../hk-util");
-const Store = require("./storage/file");
+const Store = require("../storage/file");
-const Util = require("./common-util");
+const Util = require("../common-util");
 const nThen = require("nthen");
 
 const Env = {};