Initial implementation of user extensions

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 <http://www.gnu.org/licenses/>.
+###
+
+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])
+  )
+)

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 <http://www.gnu.org/licenses/>.
+###
+
+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

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

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))
+        ret = sandbox.run(virtualParameters[vpName].script, [vpd.slice(1)], sessionData.deviceData, sessionData.extensionsCache, firstRevision, lastRevision)
-        firstRevision = lastRevision - (sessionData.revisions[sessionData.revisions.length - 1] >> 1)
+        if ret.extensions?.length
-        ret = sandbox.run(virtualParameters[vpName].script, [vpd.slice(1)], sessionData.deviceData, firstRevision, lastRevision)
+          _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, () ->