From 27e9882071e2c5238f0af4c87574e9e4e19a75c7 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Tue, 19 Apr 2016 03:18:48 +0300 Subject: [PATCH] Initial implementation of user extensions --- lib/extension-wrapper.coffee | 27 +++++++++++++++++ lib/extensions.coffee | 55 ++++++++++++++++++++++++++++++++++ lib/sandbox.coffee | 28 ++++++++++++++---- lib/session.coffee | 57 +++++++++++++++++++++++++++++++++--- 4 files changed, 158 insertions(+), 9 deletions(-) create mode 100644 lib/extension-wrapper.coffee create mode 100644 lib/extensions.coffee diff --git a/lib/extension-wrapper.coffee b/lib/extension-wrapper.coffee new file mode 100644 index 0000000..ac22510 --- /dev/null +++ b/lib/extension-wrapper.coffee @@ -0,0 +1,27 @@ +### +# 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]](message.slice(2), (err, res) -> + throw err if err + process.send([message[0], res]) + ) +) diff --git a/lib/extensions.coffee b/lib/extensions.coffee new file mode 100644 index 0000000..8520f94 --- /dev/null +++ b/lib/extensions.coffee @@ -0,0 +1,55 @@ +### +# 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(null, message[1]) + + +run = (args, callback) -> + scriptName = args[0] + if not processes[scriptName]? + cwd = path.resolve(config.get('CONFIG_DIR'), 'extensions') + 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/sandbox.coffee b/lib/sandbox.coffee index b736949..a7122be 100644 --- a/lib/sandbox.coffee +++ b/lib/sandbox.coffee @@ -22,7 +22,7 @@ device = require './device' # Used for throwing to exit user script -class Declare; +class Exit; sandbox = { @@ -31,9 +31,25 @@ sandbox = { revision: null maxRevision: null declarations: null + extensionsCache: null + extensions: null context: vm.createContext() } + +sandbox.context.ext = () -> + if sandbox.extensions?.length + throw new Error('Ext function should not be called from within a try/catch block') + + extCall = (String(a) for a in arguments) + key = JSON.stringify(extCall) + if not sandbox.extensionsCache[sandbox.revision]?[key]? + sandbox.extensions = [extCall] + throw new Exit() + + return sandbox.extensionsCache[sandbox.revision][key] + + sandbox.context.declare = (decs) -> if ++ sandbox.revision > sandbox.maxRevision + 1 throw new Error('Declare function should not be called from within a try/catch block') @@ -42,7 +58,7 @@ sandbox.context.declare = (decs) -> for k, v of decs sandbox.declarations.push([common.parsePath(k)].concat(toIndexDeclaration(v))) - throw new Declare() + throw new Exit() res = {} for k, v of decs @@ -99,20 +115,22 @@ toIndexDeclaration = (keyDeclaraiton) -> return indexDeclaraiton -run = (script, args, deviceData, startRevision, maxRevision) -> +run = (script, args, deviceData, extensionsCache, startRevision, maxRevision) -> sandbox.deviceData = deviceData + sandbox.extensionsCache = extensionsCache sandbox.args = args sandbox.revision = startRevision sandbox.maxRevision = maxRevision sandbox.declarations = [] + sandbox.extensions = null try ret = script.runInNewContext(sandbox.context) return {done: true, declarations: sandbox.declarations, returnValue: ret} catch err - throw err if err not instanceof Declare + throw err if err not instanceof Exit - return {done: false, declarations: sandbox.declarations} + return {done: false, declarations: sandbox.declarations, extensions: sandbox.extensions} exports.run = run diff --git a/lib/session.coffee b/lib/session.coffee index ef74092..40ebb98 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -23,6 +23,7 @@ db = require './db' device = require './device' sandbox = require './sandbox' cache = require './cache' +extensions = require './extensions' init = (deviceId, cwmpVersion, timeout) -> @@ -40,6 +41,7 @@ init = (deviceId, cwmpVersion, timeout) -> revisions: [0] rpcCount: 0 cycle: 0 + extensionsCache: [] } return sessionData @@ -100,6 +102,7 @@ addProvisions = (sessionData, provisions) -> sessionData.rpcCount = 0 sessionData.revisions = [0] device.collapse(sessionData.deviceData, 1) + sessionData.extensionsCache.length = 0 clearProvisions = (sessionData) -> @@ -109,6 +112,7 @@ clearProvisions = (sessionData) -> sessionData.rpcCount = 0 sessionData.revisions = [0] device.collapse(sessionData.deviceData, 1) + sessionData.extensionsCache.length = 0 generateRpcRequest = (sessionData) -> @@ -334,6 +338,28 @@ commitVirtualParameter = (sessionData, parameterDeclaration, revision, update) - device.set(sessionData.deviceData, parameterDeclaration[0], revision, v) +runExtensions = (sessionData, revision, _extensions, callback) -> + sessionData.extensionsCache[revision] ?= {} + obj = {} + for e in _extensions + obj[JSON.stringify(e)] = e + + counter = 1 + for k, v of obj + ++ counter + do (k, v) -> + extensions.run(v, (err, res) -> + sessionData.extensionsCache[revision][k] = res + + if -- counter == 0 or err + counter = 0 + return callback(err) + ) + + if -- counter == 0 + return callback() + + rpcRequest = (sessionData, declarations, callback) -> allDeclarations = declarations?.slice() ? [] @@ -341,6 +367,7 @@ rpcRequest = (sessionData, declarations, callback) -> return callback(err) if err or presetsHash != sessionData.presetsHash done = true + _extensions = [] for provision in sessionData.provisions if not provisions[provision[0]]? switch provision[0] @@ -355,13 +382,22 @@ rpcRequest = (sessionData, declarations, callback) -> allDeclarations.push([[['Tags', provision[1]], null, null, null, null, null, null, null, [provision[2], 'xsd:boolean']]]) continue - ret = sandbox.run(provisions[provision[0]].script, provision.slice(1), sessionData.deviceData, 0, sessionData.revisions[0] >> 1) + ret = sandbox.run(provisions[provision[0]].script, provision.slice(1), sessionData.deviceData, sessionData.extensionsCache, 0, sessionData.revisions[0] >> 1) + if ret.extensions?.length + _extensions = _extensions.concat(ret.extensions) done &&= ret.done allDeclarations = allDeclarations.concat(ret.declarations) if done and (not allDeclarations?.length or sessionData.revisions[0] > 1) return callback() + if _extensions.length + runExtensions(sessionData, (sessionData.revisions[0] >> 1), _extensions, (err) -> + return callback(err) if err + return rpcRequest(sessionData, declarations, callback) + ) + return + doVirtualParameters = (iter, virtualParameterDeclarations, cb) -> if not virtualParameterDeclarations? virtualParameterDeclarations = extractVirtualParameterDeclarations(sessionData) @@ -371,16 +407,27 @@ rpcRequest = (sessionData, declarations, callback) -> sessionData.revisions[iter] ?= 0 decs = [] virtualParameterUpdates = [] + _extensions = [] + lastRevision = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) + firstRevision = lastRevision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1) for vpd, i in virtualParameterDeclarations vpName = vpd[0][1] continue if not virtualParameters[vpName]? - lastRevision = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) - firstRevision = lastRevision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1) - ret = sandbox.run(virtualParameters[vpName].script, [vpd.slice(1)], sessionData.deviceData, firstRevision, lastRevision) + ret = sandbox.run(virtualParameters[vpName].script, [vpd.slice(1)], sessionData.deviceData, sessionData.extensionsCache, firstRevision, lastRevision) + if ret.extensions?.length + _extensions = _extensions.concat(ret.extensions) + decs = decs.concat(ret.declarations) if ret.done virtualParameterUpdates.push(ret.returnValue) + if _extensions.length + runExtensions(sessionData, lastRevision, _extensions, (err) -> + return callback(err) if err + return doVirtualParameters(iter, virtualParameterDeclarations, cb) + ) + return + if virtualParameterUpdates.length == virtualParameterDeclarations.length rev = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) for vpd, i in virtualParameterDeclarations @@ -392,6 +439,8 @@ rpcRequest = (sessionData, declarations, callback) -> sessionData.revisions.length = iter rev = sessionData.revisions.reduce((a, b) -> (a >> 1) + (b >> 1)) device.collapse(sessionData.deviceData, rev + 1) + if sessionData.extensionsCache.length > rev + 1 + sessionData.extensionsCache.length = rev + 1 return cb(true) applyDeclarations(iter, decs, () ->