diff --git a/acs.coffee b/acs.coffee index dd4e2ca..2b82c0d 100644 --- a/acs.coffee +++ b/acs.coffee @@ -11,6 +11,8 @@ presets = require './presets' mongodb = require 'mongodb' fs = require 'fs' apiFunctions = require './api-functions' +customCommands = require './custom-commands' +crypto = require 'crypto' OBJECT_REGEX = /\.$/ INSTANCE_REGEX = /\.[\d]+\.$/ @@ -45,10 +47,6 @@ updateDevice = (currentRequest, actions, callback) -> now = new Date(Date.now()) updates = {} deletes = {} - if actions.inform? - updates['_lastInform'] = now - updates['_lastBoot'] = now if '1 BOOT' in actions.inform - updates['_lastBootstrap'] = now if '0 BOOTSTRAP' in actions.inform if actions.parameterValues? for p in actions.parameterValues @@ -94,35 +92,118 @@ updateDevice = (currentRequest, actions, callback) -> updates["_customCommands.#{commandName}._value"] = p[1] updates["_customCommands.#{commandName}._timestamp"] = now + if actions.set? + common.extend(updates, actions.set) + if Object.keys(updates).length > 0 or Object.keys(deletes).length > 0 db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates, '$unset' : deletes}, {}, (err, count) -> - if (err) - callback(err) if callback? - return - - if count == 0 - util.log("#{currentRequest.deviceId}: New device detected") - db.devicesCollection.update({'_id' : currentRequest.deviceId}, {'$set' : updates}, {upsert: true}, (err) -> - if err? - callback(err) if callback? - return - - task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} - apiFunctions.insertTasks(task, {}, (err, t) -> - callback(err) if callback? - ) - ) - else if updates['_lastBootstrap']? - # reinitialize on bootstrap event (e.g. firmware upgrade) - task = {device : currentRequest.deviceId, name : 'init', timestamp : config.INIT_TIMESTAMP} - apiFunctions.insertTasks(task, {}, (err) -> - callback(err) if callback? - ) - else - callback() if callback? + callback?(err) ) else - callback() if callback? + callback?() + + +inform = (currentRequest, cwmpRequest) -> + # If overloaded, ask CPE to retry in 5 mins + if Date.now() < holdUntil + res = {code : 503, headers : {'Content-Length' : 0, 'Retry-After' : 300}, data : ''} + return writeResponse(currentRequest, res) + + now = new Date(Date.now()) + + deviceId = common.getDeviceId(cwmpRequest.methodRequest.deviceId) + currentRequest.deviceId = deviceId + util.log("#{deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") if config.LOG_INFORMS + + parameterNames = (p[0] for p in cwmpRequest.methodRequest.parameterList) + + informHash = crypto.createHash('md5').update(JSON.stringify(parameterNames)).update(JSON.stringify(config.CUSTOM_COMMANDS)).digest('hex') + + actions = {parameterValues : cwmpRequest.methodRequest.parameterList, set : {}} + actions.set._lastInform = now + actions.set._lastBoot = now if '1 BOOT' in cwmpRequest.methodRequest.event + actions.set._lastBootstrap = lastBootstrap = now if '0 BOOTSTRAP' in cwmpRequest.methodRequest.event + + db.redisClient.get("#{deviceId}_inform_hash", (err, res) -> + throw err if err + + if not res? + db.redisClient.setex("#{deviceId}_inform_hash", config.PRESETS_CACHE_DURATION, informHash, (err, res) -> + throw err if err + ) + + updateAndRespond = () -> + updateDevice(currentRequest, actions, (err) -> + throw err if err + res = tr069.response(cwmpRequest.id, {methodResponse : {type : 'InformResponse'}}, {deviceId : deviceId}) + writeResponse(currentRequest, res) + ) + + if res == informHash + return updateAndRespond() + + # populate projection and parameterStructure + parameterStructure = {} + projection = {} + projection._customCommands = 1 + projection._timestamp = 1 + projection._lastBootstrap = 1 + for param in parameterNames + ref = parameterStructure + path = '' + for p in param.split('.') + ref[p] ?= {_path : path} + ref = ref[p] + path += p + '.' + projection[path + '_timestamp'] = 1 + + db.devicesCollection.findOne({'_id' : deviceId}, projection, (err, device) -> + lastBootstrap ?= device?._lastBootstrap + _tasks = [] + + # For any parameters that aren't in the DB model, issue a refresh task + traverse = (reference, actual) -> + for k of reference + continue if k[0] == '_' + if not actual[k]?._timestamp? or actual[k]._timestamp > lastBootstrap + _tasks.push({device: deviceId, name : 'refreshObject', objectName : reference[k]._path}) + else + traverse(reference[k], actual[k]) + + if device? + traverse(parameterStructure, device) + + # Update custom commands if needed + deviceCustomCommands = {} + deviceCustomCommands[k] = v._timestamp for k,v of device._customCommands + + for cmd in customCommands.getDeviceCustomCommandNames(deviceId) + if not (deviceCustomCommands[cmd]?._timestamp < lastBootstrap) + _tasks.push({device: deviceId, name: 'customCommand', command: "#{cmd} init"}) + delete deviceCustomCommands[cmd] + + for cmd of deviceCustomCommands + actions.deletedObjects ?= [] + actions.deletedObjects.push("_customCommands.#{cmd}") + + apiFunctions.insertTasks(_tasks, {}, () -> + updateAndRespond() + ) + else + db.devicesCollection.insert({_id : deviceId, _registered : now}, (err) -> + throw err if err + util.log("#{deviceId}: New device registered") + _tasks.push({device: deviceId, name : 'refreshObject', objectName : ''}) + + for cmd in customCommands.getDeviceCustomCommandNames(deviceId) + _tasks.push({device: deviceId, name: 'customCommand', command: "#{cmd} init"}) + + apiFunctions.insertTasks(_tasks, {}, () -> + updateAndRespond() + ) + ) + ) + ) runTask = (currentRequest, task, methodResponse) -> @@ -290,18 +371,14 @@ listener = (httpRequest, httpResponse) -> return body.toString(encoding || 'utf8', 0, body.byteLength) httpRequest.addListener 'end', () -> - currentRequest = {} - currentRequest.httpRequest = httpRequest - currentRequest.httpResponse = httpResponse - cwmpResponse = {} cwmpRequest = tr069.request(httpRequest) - # get deviceId either from inform xml or cookie - if cwmpRequest.methodRequest? and cwmpRequest.methodRequest.type is 'Inform' - cookies.deviceId = currentRequest.deviceId = common.getDeviceId(cwmpRequest.methodRequest.deviceId) - else - currentRequest.deviceId = cwmpRequest.cookies.deviceId + currentRequest = { + httpRequest : httpRequest + httpResponse : httpResponse + deviceId : cwmpRequest.cookies?.deviceId + } if config.DEBUG_DEVICES[currentRequest.deviceId] dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n" @@ -311,22 +388,7 @@ listener = (httpRequest, httpResponse) -> if cwmpRequest.methodRequest? if cwmpRequest.methodRequest.type is 'Inform' - if Date.now() < holdUntil - # Ask CPE to retry in 5 mins - res = {code : 503, headers : {'Content-Length' : 0, 'Retry-After' : 300}, data : ''} - writeResponse(currentRequest, res) - return - - cwmpResponse.methodResponse = {type : 'InformResponse'} - - if config.LOG_INFORMS - util.log("#{currentRequest.deviceId}: Inform (#{cwmpRequest.methodRequest.event}); retry count #{cwmpRequest.methodRequest.retryCount}") - - updateDevice(currentRequest, {'inform' : cwmpRequest.methodRequest.event, 'parameterValues' : cwmpRequest.methodRequest.parameterList}, (err) -> - throw new Error(err) if err? - res = tr069.response(cwmpRequest.id, cwmpResponse, cookies) - writeResponse(currentRequest, res) - ) + inform(currentRequest, cwmpRequest) else if cwmpRequest.methodRequest.type is 'TransferComplete' # do nothing util.log("#{currentRequest.deviceId}: Transfer complete") diff --git a/api-functions.coffee b/api-functions.coffee index 1fe2a47..841a6f8 100644 --- a/api-functions.coffee +++ b/api-functions.coffee @@ -141,8 +141,10 @@ addAliases = (device, aliases) -> insertTasks = (tasks, aliases, callback) -> - if common.typeOf(tasks) isnt common.ARRAY_TYPE + if tasks? and common.typeOf(tasks) isnt common.ARRAY_TYPE tasks = [tasks] + else if not tasks? or tasks.length == 0 + return callback(tasks) counter = tasks.length diff --git a/config.coffee b/config.coffee index e3aef0d..1fa24b9 100644 --- a/config.coffee +++ b/config.coffee @@ -1,4 +1,3 @@ -exports.INIT_TIMESTAMP = new Date(1983, 9, 16) # arbitrary date exports.DATABASE_NAME = 'genieacs' exports.ACS_PORT = 7547 # CWMP port as assigned by IANA exports.ACS_HTTPS_PORT = 7548 diff --git a/custom-commands.coffee b/custom-commands.coffee index 314fa03..6dff4f3 100644 --- a/custom-commands.coffee +++ b/custom-commands.coffee @@ -25,3 +25,11 @@ exports.getDeviceCustomCommands = (deviceId) -> if eval(v).test(deviceId) commands[k] = exports.getFileCommands(k) return commands + + +exports.getDeviceCustomCommandNames = (deviceId) -> + commands = [] + for k,v of config.CUSTOM_COMMANDS + if eval(v).test(deviceId) + commands.push(k) + return commands diff --git a/tasks.coffee b/tasks.coffee index d7e4509..e772ff0 100644 --- a/tasks.coffee +++ b/tasks.coffee @@ -11,70 +11,13 @@ exports.STATUS_COMPLETED = STATUS_COMPLETED = 4 exports.STATUS_SAVE = STATUS_SAVE = 128 # not mutually exclusive with the rest -initCustomCommands = (deviceId, callback) -> - updates = {customCommands : []} - counter = 0 - for k,v of customCommands.getDeviceCustomCommands(deviceId) - if 'init' in v - ++counter - customCommands.execute(deviceId, "#{k} init", (err, value) -> - updates.customCommands.push([k, value]) - callback(updates) if --counter <= 0 - ) - callback(updates) if counter <= 0 - - -this.init = (task, methodResponse, callback) -> - task.session = {} if not task.session? - - if not task.session.subtask? - task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : '', nextLevel : false} - - if task.session.subtask.name is 'getParameterNames' - if methodResponse.faultcode? - task.fault = methodResponse - callback(null, STATUS_FAULT) - return - - this.getParameterNames(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) => - if status & STATUS_COMPLETED - parameterNames = [] - for p in deviceUpdates.parameterNames - if not common.endsWith(p[0], '.') - parameterNames.push(p[0]) - task.session.subtask = {name : 'getParameterValues', parameterNames : parameterNames} - this.getParameterValues(task.session.subtask, {}, (err, status, cwmpResponse) -> - # ignore deviceUpdates returned by firt call to getParameterValues - callback(err, status, cwmpResponse, deviceUpdates) - ) - else if status & STATUS_OK - callback(err, STATUS_OK, cwmpResponse, deviceUpdates) - else - throw Error('Unexpected subtask status') - ) - else if task.session.subtask.name is 'getParameterValues' - if methodResponse.faultcode? - # Ignore GetParameterValues errors. A workaround for the crappy Seewon devices. - methodResponse = {parameterList : {}} - - this.getParameterValues(task.session.subtask, methodResponse, (err, status, cwmpResponse, deviceUpdates) -> - if status & STATUS_COMPLETED - initCustomCommands(task.device, (updates) -> - common.extend(deviceUpdates, updates) - callback(err, status, cwmpResponse, deviceUpdates) - ) - else - callback(err, status, cwmpResponse, deviceUpdates) - ) - else - throw Error('Unexpected subtask name') - - this.refreshObject = (task, methodResponse, callback) -> task.session = {} if not task.session? if not task.session.subtask? - task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : "#{task.objectName}.", nextLevel : false} + path = task.objectName + path += '.' if path != '' and not common.endsWith(path, '.') + task.session.subtask = {device : task.device, name : 'getParameterNames', parameterPath : path, nextLevel : false} if methodResponse.faultcode? task.fault = methodResponse @@ -142,6 +85,20 @@ this.getParameterNames = (task, methodResponse, callback) -> return if methodResponse.type is 'GetParameterNamesResponse' + # Make sure that for each parameters, all its parents are included + found = {} + for p in methodResponse.parameterList + param = p[0] + i = param.length + while (i = param.lastIndexOf('.', i-1)) > task.parameterPath.length + pp = param.slice(0, i + 1) + break if found[pp]? + found[pp] = 0 + found[p[0]] = 1 + + for k, v of found + methodResponse.parameterList.push([k]) if v == 0 + deviceUpdates = {parameterNames : methodResponse.parameterList} path = if common.endsWith(task.parameterPath, '.') then task.parameterPath.slice(0, -1) else task.parameterPath projection = {} @@ -150,17 +107,14 @@ this.getParameterNames = (task, methodResponse, callback) -> # delete nonexisting params db.devicesCollection.findOne({_id : task.device}, projection, (err, device) -> if device + root = device + rootPath = '' if !!task.parameterPath - root = device - rootPath = '' ps = path.split('.') for p in ps rootPath += "#{p}." root = root[p] break if not root? - else - root = device['InternetGatewayDevice'] - rootPath = 'InternetGatewayDevice.' if root deviceUpdates.deletedObjects = findMissingParameters(root, methodResponse.parameterList, rootPath)