From 9f85ead558c8375e2a8eedc9039da5c0ec1a8234 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Sun, 4 Dec 2016 19:04:22 -0800 Subject: [PATCH] Turn errors from scripts and extensions into faults --- lib/config.coffee | 1 + lib/cwmp.coffee | 36 +++++++++++-- lib/extension-wrapper.coffee | 26 ---------- lib/extension-wrapper.js | 62 +++++++++++++++++++++++ lib/extensions.coffee | 55 -------------------- lib/extensions.js | 98 ++++++++++++++++++++++++++++++++++++ lib/sandbox.js | 28 ++++++++--- lib/session.coffee | 53 +++++++++++++------ 8 files changed, 254 insertions(+), 105 deletions(-) delete mode 100644 lib/extension-wrapper.coffee create mode 100644 lib/extension-wrapper.js delete mode 100644 lib/extensions.coffee create mode 100644 lib/extensions.js diff --git a/lib/config.coffee b/lib/config.coffee index 2137822..40fab86 100644 --- a/lib/config.coffee +++ b/lib/config.coffee @@ -42,6 +42,7 @@ options = { FS_IP : {type : 'string', default : '192.168.0.1'}, DOWNLOAD_TIMEOUT: {type : 'int', default : 3600}, + EXT_TIMEOUT: {type: 'int', default: 3000}, PRESETS_CACHE_DURATION : {type : 'int', default : 86400}, LOG_INFORMS : {type : 'bool', default : true}, DEBUG : {type : 'bool', default : false}, diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 21d436e..296233d 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -207,9 +207,12 @@ applyPresets = (currentRequest) -> for k, v of parameters declarations.push([v, 1, {value: 1}]) - session.rpcRequest(currentRequest.sessionData, declarations, (err, id, rpcRequest) -> + session.rpcRequest(currentRequest.sessionData, declarations, (err, faults, id, rpcRequest) -> throw err if err + if faults? + throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(faults)) + if rpcRequest? return sendRpcRequest(currentRequest, id, rpcRequest) @@ -227,9 +230,18 @@ applyPresets = (currentRequest) -> if query.testFilter(parameters, p.precondition) session.addProvisions(currentRequest.sessionData, p.channel, p.provisions) - session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest, doneProvisions) -> + session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) -> throw err if err + if faults? + for channel, fault of faults + if not fault.retries? and Object.keys(faults).length == 1 + fault.retries = 0 + currentRequest.sessionData.faults[channel] = fault + util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})") + + return nextRpc(currentRequest) + for p, i in doneProvisions or [] delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]] @@ -240,8 +252,26 @@ applyPresets = (currentRequest) -> nextRpc = (currentRequest) -> - session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest, doneProvisions) -> + session.rpcRequest(currentRequest.sessionData, null, (err, faults, id, rpcRequest, doneProvisions) -> throw err if err + + if faults? + for channel, fault of faults + if channel.startsWith('_task_') + taskId = channel.slice(6) + for t, j in currentRequest.sessionData.tasks + if t._id == taskId + currentRequest.sessionData.tasks.splice(j, 1) + fault.expiry = t.expiry if t.expiry? + break + + if not fault.retries? and Object.keys(faults).length == 1 + fault.retries = 0 + currentRequest.sessionData.faults[channel] = fault + util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})") + + return nextRpc(currentRequest) + if rpcRequest? return sendRpcRequest(currentRequest, id, rpcRequest) diff --git a/lib/extension-wrapper.coffee b/lib/extension-wrapper.coffee deleted file mode 100644 index 5a5afa1..0000000 --- a/lib/extension-wrapper.coffee +++ /dev/null @@ -1,26 +0,0 @@ -### -# Copyright 2013-2016 Zaid Abdulla -# -# GenieACS is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# GenieACS is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with GenieACS. If not, see . -### - -process.chdir(process.argv[2]) -script = require("#{process.argv[2]}/#{process.argv[3]}") - - -process.on('message', (message) -> - script[message[1][0]](message[1].slice(1), (err, res) -> - process.send([message[0], err, res]) - ) -) diff --git a/lib/extension-wrapper.js b/lib/extension-wrapper.js new file mode 100644 index 0000000..b08ff16 --- /dev/null +++ b/lib/extension-wrapper.js @@ -0,0 +1,62 @@ +/** + * Copyright 2013-2016 Zaid Abdulla + * + * GenieACS is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * GenieACS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with GenieACS. If not, see . + */ +"use strict"; + +const jobs = new Set(); +const fileName = process.argv[2]; +let script; + +function errorToFault(err) { + if (!err) + return null; + + return { + code: `ext.${err.name}`, + message: err.message, + detail: { + name: err.name, + message: err.message, + stack: err.stack + } + } +} + +process.on("uncaughtException", function(err) { + let fault = errorToFault(err); + jobs.forEach(function(jobId) { + process.send([jobId, fault]); + }); + jobs.clear(); + process.disconnect(); +}); + +process.on("message", function(message) { + jobs.add(message[0]); + + if (!script) { + let cwd = process.env["GENIEACS_CONFIG_DIR"] + "/ext"; + process.chdir(cwd); + script = require(`${cwd}/${fileName}`); + } + + script[message[1][0]](message[1].slice(1), function(err, res) { + if (!jobs.delete(message[0])) + return; + + process.send([message[0], errorToFault(err), res]); + }); +}); diff --git a/lib/extensions.coffee b/lib/extensions.coffee deleted file mode 100644 index 484c1b8..0000000 --- a/lib/extensions.coffee +++ /dev/null @@ -1,55 +0,0 @@ -### -# Copyright 2013-2016 Zaid Abdulla -# -# GenieACS is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# GenieACS is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with GenieACS. If not, see . -### - -childProcess = require 'child_process' -crypto = require 'crypto' -path = require 'path' - -config = require './config' - - -TIMEOUT = 5000 - -processes = {} -queue = {} - -messageHandler = (message) -> - func = queue[message[0]] - delete queue[message[0]] - func(message[1], message[2]) - - -run = (args, callback) -> - scriptName = args[0] - if not processes[scriptName]? - cwd = path.resolve(config.get('CONFIG_DIR'), 'ext') - p = childProcess.fork('lib/extension-wrapper', [cwd, scriptName]) - p.on('message', messageHandler) - processes[scriptName] = p - - id = crypto.randomBytes(8).toString('hex') - queue[id] = callback - setTimeout(() -> - if id of queue - delete queue[id] - return callback(new Error('Extension timeout')) - , TIMEOUT) - - processes[scriptName].send([id, args.slice(1)]) - - -exports.run = run diff --git a/lib/extensions.js b/lib/extensions.js new file mode 100644 index 0000000..e606ec5 --- /dev/null +++ b/lib/extensions.js @@ -0,0 +1,98 @@ +/** + * Copyright 2013-2016 Zaid Abdulla + * + * GenieACS is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * GenieACS is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with GenieACS. If not, see . + */ +"use strict"; + +const childProcess = require("child_process"); +const crypto = require ("crypto"); + +const config = require("./config"); + +const TIMEOUT = config.get('EXT_TIMEOUT'); + +const processes = {}; +const jobs = {}; + +function messageHandler(message) { + let func = jobs[message[0]]; + if (func) { + delete jobs[message[0]]; + func(null, message[1], message[2]); + } +} + + +function run(args, callback) { + let scriptName = args[0]; + + if (!processes[scriptName]) { + let p = childProcess.fork('lib/extension-wrapper', [scriptName]); + + p.on('error', function() { + kill(processes[scriptName]); + delete processes[scriptName]; + }); + + p.on('disconnect', function() { + kill(processes[scriptName]); + delete processes[scriptName]; + }); + + p.on('message', messageHandler); + processes[scriptName] = p; + } + + let id = crypto.randomBytes(8).toString('hex'); + jobs[id] = callback; + setTimeout(function() { + if (id in jobs) { + delete jobs[id]; + return callback(null, {code: "timeout", message: "Extension timed out"}); + } + }, TIMEOUT); + + return processes[scriptName].send([id, args.slice(1)]); +}; + + +function kill(process) { + if (process.signalCode !== null || process.exitCode !== null) + return; + + let timeToKill = Date.now() + 5000; + + process.kill(); + + let t = setInterval(function() { + if (process.signalCode !== null || process.exitCode !== null) { + clearInterval(t); + } + else if (Date.now() > timeToKill) { + process.kill('SIGKILL'); + clearInterval(t); + } + }, 100); +} + + +function killAll() { + for (let n in processes) + kill(processes[n]); +} + + +exports.run = run; +exports.killAll = killAll; diff --git a/lib/sandbox.js b/lib/sandbox.js index 4c3042d..087f0d9 100644 --- a/lib/sandbox.js +++ b/lib/sandbox.js @@ -204,6 +204,22 @@ Object.defineProperty(sandbox.context, 'commit', {value: commit}); Object.defineProperty(sandbox.context, 'ext', {value: ext}); +function errorToFault(err) { + if (!err) + return null; + + return { + code: `script.${err.name}`, + message: err.message, + detail: { + name: err.name, + message: err.message, + stack: err.stack + } + } +} + + function run(script, globals, timestamp, deviceData, extensionsCache, startRevision, maxRevision, callback) { sandbox.timestamp = timestamp; sandbox.deviceData = deviceData; @@ -233,7 +249,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis else if (err === EXT) status = 2; else - return callback(err); + return callback(null, errorToFault(err)); } let ext = sandbox.extensions; @@ -245,17 +261,17 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis let counter = 1; for (let key in sandbox.extensions) { ++ counter; - extensions.run(sandbox.extensions[key], function(err, res) { - if (err && counter > 0) { + extensions.run(sandbox.extensions[key], function(err, fault, res) { + if ((err || fault) && counter > 0) { counter = 0; - return callback(err); + return callback(err, fault); } extCache[key] = res; if (-- counter === 0) if (status === 2) return run.apply(null, args); else - return callback(null, clr, decs, status === 0, ret); + return callback(null, null, clr, decs, status === 0, ret); }); } @@ -263,7 +279,7 @@ function run(script, globals, timestamp, deviceData, extensionsCache, startRevis if (status === 2) return run.apply(null, args); else - return callback(null, clr, decs, status === 0, ret); + return callback(null, null, clr, decs, status === 0, ret); } diff --git a/lib/session.coffee b/lib/session.coffee index 7dc2165..22ab9a4 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -21,7 +21,6 @@ db = require './db' device = require './device' sandbox = require './sandbox' cache = require './cache' -extensions = require './extensions' PathSet = require './path-set' VersionedMap = require './versioned-map' InstanceSet = require './instance-set' @@ -379,12 +378,12 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback) sandbox.run(sessionData.cache.provisions[provision[0]].script, {args: provision[1]}, sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, startRevision, endRevision, - (err, _clear, _declarations, _done) -> + (err, _fault, _clear, _declarations, _done) -> -- counter - if err + if err or _fault if counter >= 0 counter = 0 - return callback(err) + return callback(err, _fault) return done &&= _done @@ -396,11 +395,11 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback) allClear = allClear.concat(_clear) if counter == 0 - return callback(null, done, allDeclarations, allClear) + return callback(null, null, done, allDeclarations, allClear) ) if -- counter == 0 - return callback(null, done, allDeclarations, allClear) + return callback(null, null, done, allDeclarations, allClear) runVirtualParameters = (sessionData, provisions, startRevision, endRevision, callback) -> @@ -415,12 +414,12 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, globals, sessionData.timestamp, sessionData.deviceData, sessionData.extensionsCache, startRevision, endRevision, - (err, _clear, _declarations, _done, _returnValue) -> + (err, _fault, _clear, _declarations, _done, _returnValue) -> -- counter - if err + if err or _fault if counter >= 0 counter = 0 - return callback(err) + return callback(err, _fault) return done &&= _done @@ -441,7 +440,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear) clear(sessionData, toClear, (err) -> - return callback(err, done, allDeclarations, allClear) + return callback(err, null, done, allDeclarations, allClear) ) ) @@ -452,7 +451,7 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear) clear(sessionData, toClear, (err) -> - return callback(err, done, allDeclarations, allClear) + return callback(err, null, done, allDeclarations, allClear) ) @@ -586,7 +585,7 @@ runDeclarations = (sessionData, declarations) -> rpcRequest = (sessionData, _declarations, callback) -> if sessionData.rpcRequest? - return callback(null, generateRpcId(sessionData), sessionData.rpcRequest) + return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest) if sessionData.declarations.length < sessionData.provisions.length inception = sessionData.declarations.length @@ -595,9 +594,33 @@ rpcRequest = (sessionData, _declarations, callback) -> sessionData.deviceData.attributes.revision = revision run = if inception == 0 then runProvisions else runVirtualParameters - return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, done, decs, toClear) -> + return run(sessionData, sessionData.provisions[inception], sessionData.revisions[inception - 1] ? 0, sessionData.revisions[inception], (err, fault, done, decs, toClear) -> return callback(err) if err + if fault + faults = {} + for p, i in sessionData.provisions[0] + channel = sessionData.channels[i] + f = faults[channel] + if not f? + f = { + provisions: [] + timestamp: sessionData.timestamp + code: fault.code + message: fault.message + detail: fault.detail + } + + if sessionData.faults[channel]? + f.retries = (sessionData.faults[channel].retries or 0) + 1 + + faults[channel] = f + + f.provisions.push(p) + + clearProvisions(sessionData) + return callback(null, faults) + sessionData.declarations.push(decs) sessionData.doneProvisions |= 1 << inception if done or not decs.length @@ -777,7 +800,7 @@ rpcRequest = (sessionData, _declarations, callback) -> return rpcRequest(sessionData, null, callback) if sessionData.rpcRequest - return callback(null, generateRpcId(sessionData), sessionData.rpcRequest) + return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest) ++ sessionData.revisions[inception] sessionData.declarations.pop() @@ -791,7 +814,7 @@ rpcRequest = (sessionData, _declarations, callback) -> sessionData.deviceData.timestamps.collapse(1) sessionData.deviceData.attributes.collapse(1) sessionData.extensionsCache = {} - return callback(null, null, null, doneProvisions) + return callback(null, null, null, null, doneProvisions) rev = sessionData.revisions[sessionData.revisions.length - 1] sessionData.deviceData.timestamps.collapse(rev + 1)