diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 4509ac9..9391eb3 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -63,16 +63,16 @@ throwError = (err, httpResponse) -> throw err -writeResponse = (currentRequest, res) -> - if config.get('DEBUG', currentRequest.sessionData.deviceId) +writeResponse = (sessionContext, res) -> + if config.get('DEBUG', sessionContext.deviceId) dump = "# RESPONSE #{new Date(Date.now())}\n" + JSON.stringify(res.headers) + "\n#{res.data}\n\n" - require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> + require('fs').appendFile("./debug/#{sessionContext.deviceId}.dump", dump, (err) -> throwError(err) if err ) # respond using the same content-encoding as the request - if currentRequest.httpRequest.headers['content-encoding']? and res.data.length > 0 - switch currentRequest.httpRequest.headers['content-encoding'] + if sessionContext.httpRequest.headers['content-encoding']? and res.data.length > 0 + switch sessionContext.httpRequest.headers['content-encoding'] when 'gzip' res.headers['Content-Encoding'] = 'gzip' compress = zlib.gzip @@ -82,44 +82,50 @@ writeResponse = (currentRequest, res) -> if compress? compress(res.data, (err, data) -> - return throwError(err, currentRequest.httpResponse) if err + return throwError(err, sessionContext.httpResponse) if err res.headers['Content-Length'] = data.length - currentRequest.httpResponse.writeHead(res.code, res.headers) - currentRequest.httpResponse.end(data) + sessionContext.httpResponse.writeHead(res.code, res.headers) + sessionContext.httpResponse.end(data) ) else res.headers['Content-Length'] = res.data.length - currentRequest.httpResponse.writeHead(res.code, res.headers) - currentRequest.httpResponse.end(res.data) + sessionContext.httpResponse.writeHead(res.code, res.headers) + sessionContext.httpResponse.end(res.data) + + delete sessionContext.httpRequest + delete sessionContext.httpResponse -recordFault = (currentRequest, fault, provisions, channels) -> - sessionData = currentRequest.sessionData - faults = sessionData.faults +recordFault = (sessionContext, fault, provisions, channels) -> + if not provisions + provisions = sessionContext.provisions + channels = sessionContext.channels + + faults = sessionContext.faults for channel of channels - provs = sessionData.faults[channel]?.provisions or [] + provs = sessionContext.faults[channel]?.provisions or [] faults[channel] = Object.assign({provisions: provs}, fault) - if sessionData.retries[channel]? - ++ sessionData.retries[channel] + if sessionContext.retries[channel]? + ++ sessionContext.retries[channel] else - sessionData.retries[channel] = 0 + sessionContext.retries[channel] = 0 if Object.keys(channels).length != 1 faults[channel].retryNow = true if channels[channel] == 0 faults[channel].precondition = true - sessionData.faultsTouched ?= {} - sessionData.faultsTouched[channel] = true + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[channel] = true logger.accessError({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'Channel has faulted' fault: fault channel: channel - retries: sessionData.retries[channel] + retries: sessionContext.retries[channel] }) for provision, i in provisions @@ -132,51 +138,51 @@ recordFault = (currentRequest, fault, provisions, channels) -> faults[channel].provisions = [] appendProvisions(faults[channel].provisions, provs) - session.clearProvisions(sessionData) + session.clearProvisions(sessionContext) return -inform = (currentRequest, cwmpRequest) -> - session.inform(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse) -> - return throwError(err, currentRequest.httpResponse) if err +inform = (sessionContext, rpc) -> + session.inform(sessionContext, rpc.cpeRequest, (err, acsResponse) -> + return throwError(err, sessionContext.httpResponse) if err res = soap.response({ - id : cwmpRequest.id, - methodResponse : rpcResponse, - cwmpVersion : currentRequest.sessionData.cwmpVersion + id: rpc.id, + acsResponse: acsResponse, + cwmpVersion : sessionContext.cwmpVersion }) - if !!cookiesPath = config.get('COOKIES_PATH', currentRequest.sessionData.deviceId) - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionData.sessionId}; Path=#{cookiesPath}" + if !!cookiesPath = config.get('COOKIES_PATH', sessionContext.deviceId) + res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}; Path=#{cookiesPath}" else - res.headers['Set-Cookie'] = "session=#{currentRequest.sessionData.sessionId}" + res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}" - writeResponse(currentRequest, res) + writeResponse(sessionContext, res) ) -transferComplete = (currentRequest, cwmpRequest) -> - session.transferComplete(currentRequest.sessionData, cwmpRequest.methodRequest, (err, rpcResponse, operation, fault) -> - return throwError(err, currentRequest.httpResponse) if err +transferComplete = (sessionContext, rpc) -> + session.transferComplete(sessionContext, rpc.cpeRequest, (err, acsResponse, operation, fault) -> + return throwError(err, sessionContext.httpResponse) if err if not operation? logger.accessWarn({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'Unrecognized command key' - cpeRequest: cwmpRequest.methodRequest + rpc: rpc }) if fault for k, v of operation.retries - currentRequest.sessionData.retries[k] = v + sessionContext.retries[k] = v - recordFault(currentRequest, fault, operation.provisions, operation.channels) + recordFault(sessionContext, fault) res = soap.response({ - id : cwmpRequest.id, - methodResponse : rpcResponse, - cwmpVersion : currentRequest.sessionData.cwmpVersion + id : rpc.id, + acsResponse : acsResponse, + cwmpVersion : sessionContext.cwmpVersion }) - writeResponse(currentRequest, res) + writeResponse(sessionContext, res) ) @@ -227,38 +233,34 @@ appendProvisions = (original, toAppend) -> return modified -applyPresets = (currentRequest) -> - sessionData = currentRequest.sessionData - +applyPresets = (sessionContext) -> cache.getPresets((err, presetsHash, presets) -> - return throwError(err, currentRequest.httpResponse) if err + return throwError(err, sessionContext.httpResponse) if err # Filter presets based on existing faults blackList = {} whiteList = null whiteListProvisions = null - RETRY_DELAY = config.get('RETRY_DELAY', sessionData.deviceId) + RETRY_DELAY = config.get('RETRY_DELAY', sessionContext.deviceId) - if sessionData.faults? - for channel, fault of sessionData.faults + if sessionContext.faults? + for channel, fault of sessionContext.faults if fault.retryNow retryTimestamp = 0 else - retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, sessionData.retries[channel])) * 1000 + retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, sessionContext.retries[channel])) * 1000 - if retryTimestamp <= sessionData.timestamp + if retryTimestamp <= sessionContext.timestamp whiteList = channel whiteListProvisions = fault.provisions break blackList[channel] = if fault.precondition then 1 else 2 - sessionData.presetsHash = presetsHash - deviceEvents = {} - for p in sessionData.deviceData.paths.find(['Events', '*'], false, true) - if sessionData.timestamp <= sessionData.deviceData.attributes.get(p)?.value?[1][0] + for p in sessionContext.deviceData.paths.find(['Events', '*'], false, true) + if sessionContext.timestamp <= sessionContext.deviceData.attributes.get(p)?.value?[1][0] deviceEvents[p[1]] = true parameters = {} @@ -277,11 +279,11 @@ applyPresets = (currentRequest) -> break continue if (not eventsMatch) or (preset.schedule and not - (preset.schedule.schedule and testSchedule(preset.schedule, sessionData.timestamp)[0])) + (preset.schedule.schedule and testSchedule(preset.schedule, sessionContext.timestamp)[0])) filteredPresets.push(preset) for k of preset.precondition - sessionData.channels[preset.channel] = 0 + sessionContext.channels[preset.channel] = 0 p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0] parameters[p] = common.parsePath(p) @@ -289,28 +291,28 @@ applyPresets = (currentRequest) -> for k, v of parameters declarations.push([v, 1, {value: 1}]) - session.rpcRequest(sessionData, declarations, (err, fault, id, rpcRequest) -> - return throwError(err, currentRequest.httpResponse) if err + session.rpcRequest(sessionContext, declarations, (err, fault, id, acsRequest) -> + return throwError(err, sessionContext.httpResponse) if err if fault - recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels) - session.clearProvisions(sessionData) - return applyPresets(currentRequest) + recordFault(sessionContext, fault) + session.clearProvisions(sessionContext) + return applyPresets(sessionContext) - if rpcRequest? - return sendRpcRequest(currentRequest, id, rpcRequest) + if acsRequest + return sendAcsRequest(sessionContext, id, acsRequest) - session.clearProvisions(sessionData) + session.clearProvisions(sessionContext) for k, v of parameters - unpacked = device.unpack(sessionData.deviceData, v) - if unpacked[0] and (vv = sessionData.deviceData.attributes.get(unpacked[0]).value?[1])? + unpacked = device.unpack(sessionContext.deviceData, v) + if unpacked[0] and (vv = sessionContext.deviceData.attributes.get(unpacked[0]).value?[1])? parameters[k] = vv[0] else delete parameters[k] if whiteList? - session.addProvisions(sessionData, whiteList, whiteListProvisions) + session.addProvisions(sessionContext, whiteList, whiteListProvisions) appendProvisionsToFaults = {} for p in filteredPresets @@ -319,221 +321,222 @@ applyPresets = (currentRequest) -> appendProvisionsToFaults[p.channel] = (appendProvisionsToFaults[p.channel] or []).concat(p.provisions) else - session.addProvisions(sessionData, p.channel, p.provisions) + session.addProvisions(sessionContext, p.channel, p.provisions) for channel, provisions of appendProvisionsToFaults - if appendProvisions(sessionData.faults[channel].provisions, provisions) - sessionData.faultsTouched ?= {} - sessionData.faultsTouched[channel] = true + if appendProvisions(sessionContext.faults[channel].provisions, provisions) + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[channel] = true - sessionData.presetCycles = (sessionData.presetCycles or 0) + 1 + sessionContext.presetCycles = (sessionContext.presetCycles or 0) + 1 - if sessionData.presetCycles > MAX_CYCLES + if sessionContext.presetCycles > MAX_CYCLES fault = { code: 'endless_cycle' message: 'The provision seems to be repeating indefinitely' - timestamp: sessionData.timestamp + timestamp: sessionContext.timestamp } - recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels) - session.clearProvisions(sessionData) - return sendRpcRequest(currentRequest) + recordFault(sessionContext, fault) + session.clearProvisions(sessionContext) + return sendAcsRequest(sessionContext) - session.rpcRequest(sessionData, null, (err, fault, id, rpcRequest) -> - return throwError(err, currentRequest.httpResponse) if err + session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) -> + return throwError(err, sessionContext.httpResponse) if err if fault - recordFault(currentRequest, fault, sessionData.provisions, sessionData.channels) - session.clearProvisions(sessionData) - return applyPresets(currentRequest) + recordFault(sessionContext, fault) + session.clearProvisions(sessionContext) + return applyPresets(sessionContext) - if not rpcRequest? - for channel, flags of sessionData.channels - if channel of sessionData.faults - delete sessionData.faults[channel] - sessionData.faultsTouched ?= {} - sessionData.faultsTouched[channel] = true + if not acsRequest + for channel, flags of sessionContext.channels + if channel of sessionContext.faults + delete sessionContext.faults[channel] + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[channel] = true if whiteList? - return applyPresets(currentRequest) + return applyPresets(sessionContext) - sendRpcRequest(currentRequest, id, rpcRequest) + sendAcsRequest(sessionContext, id, acsRequest) ) ) ) -nextRpc = (currentRequest) -> - session.rpcRequest(currentRequest.sessionData, null, (err, fault, id, rpcRequest) -> - return throwError(err, currentRequest.httpResponse) if err +nextRpc = (sessionContext) -> + session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) -> + return throwError(err, sessionContext.httpResponse) if err if fault - recordFault(currentRequest, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) - session.clearProvisions(currentRequest.sessionData) - return nextRpc(currentRequest) + recordFault(sessionContext, fault) + session.clearProvisions(sessionContext) + return nextRpc(sessionContext) - if rpcRequest? - return sendRpcRequest(currentRequest, id, rpcRequest) + if acsRequest + return sendAcsRequest(sessionContext, id, acsRequest) - for channel, flags of currentRequest.sessionData.channels - if flags and channel of currentRequest.sessionData.faults - delete currentRequest.sessionData.faults[channel] - currentRequest.sessionData.faultsTouched ?= {} - currentRequest.sessionData.faultsTouched[channel] = true + for channel, flags of sessionContext.channels + if flags and channel of sessionContext.faults + delete sessionContext.faults[channel] + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[channel] = true if channel.startsWith('task_') taskId = channel.slice(5) - currentRequest.sessionData.doneTasks ?= [] - currentRequest.sessionData.doneTasks.push(taskId) - for t, j in currentRequest.sessionData.tasks + sessionContext.doneTasks ?= [] + sessionContext.doneTasks.push(taskId) + for t, j in sessionContext.tasks if t._id == taskId - currentRequest.sessionData.tasks.splice(j, 1) + sessionContext.tasks.splice(j, 1) break - session.clearProvisions(currentRequest.sessionData) + session.clearProvisions(sessionContext) - for task in currentRequest.sessionData.tasks + for task in sessionContext.tasks channel = "task_#{task._id}" # Delete if expired - if task.expiry <= currentRequest.sessionData.timestamp + if task.expiry <= sessionContext.timestamp logger.accessInfo({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'Task expired' task: task }) - currentRequest.sessionData.doneTasks ?= [] - currentRequest.sessionData.doneTasks.push(String(task._id)) - if channel of currentRequest.sessionData.faults - delete currentRequest.sessionData.faults[channel] - currentRequest.sessionData.faultsTouched ?= {} - currentRequest.sessionData.faultsTouched[channel] = true + sessionContext.doneTasks ?= [] + sessionContext.doneTasks.push(String(task._id)) + if channel of sessionContext.faults + delete sessionContext.faults[channel] + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[channel] = true continue - if currentRequest.sessionData.faults[channel] + if sessionContext.faults[channel] continue switch task.name when 'getParameterValues' # Set channel in case params array is empty - currentRequest.sessionData.channels["task_#{task._id}"] = 0 + sessionContext.channels["task_#{task._id}"] = 0 for p in task.parameterNames - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['refresh', p]]) when 'setParameterValues' # Set channel in case params array is empty - currentRequest.sessionData.channels["task_#{task._id}"] = 0 + sessionContext.channels["task_#{task._id}"] = 0 for p in task.parameterValues - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['value', p[0], p[1]]]) when 'refreshObject' - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['refresh', task.objectName]]) when 'reboot' - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['reboot']]) when 'factoryReset' - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['reset']]) when 'download' - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['download', task.fileType, task.fileName, task.targetFileName]]) when 'addObject' alias = ("#{p[0]}:#{JSON.stringify(p[1])}" for p in task.parameterValues or []).join(',') - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['instances', "#{task.objectName}.[#{alias}]", "+1"]]); when 'deleteObject' - session.addProvisions(currentRequest.sessionData, "task_#{task._id}", + session.addProvisions(sessionContext, "task_#{task._id}", [['instances', task.objectName, 0]]); else - return throwError(new Error('Task name not recognized'), currentRequest.httpResponse) if err + return throwError(new Error('Task name not recognized'), sessionContext.httpResponse) if err - return nextRpc(currentRequest) + return nextRpc(sessionContext) - return applyPresets(currentRequest) + return applyPresets(sessionContext) ) -sendRpcRequest = (currentRequest, id, rpcRequest) -> - if not rpcRequest? - session.end(currentRequest.sessionData, (err, isNew) -> - return throwError(err, currentRequest.httpResponse) if err +sendAcsRequest = (sessionContext, id, acsRequest) -> + if not acsRequest? + session.end(sessionContext, (err, isNew) -> + return throwError(err, sessionContext.httpResponse) if err - delete currentSessions.get(currentRequest.httpRequest.connection)[currentRequest.sessionData.sessionId] + delete currentSessions.get(sessionContext.httpRequest.connection)[sessionContext.sessionId] if isNew logger.accessInfo({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'New device registered' }) - db.clearTasks(currentRequest.sessionData.deviceId, currentRequest.sessionData.doneTasks, (err) -> - return throwError(err, currentRequest.httpResponse) if err + db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) -> + return throwError(err, sessionContext.httpResponse) if err counter = 3 - for k of currentRequest.sessionData.faultsTouched + for k of sessionContext.faultsTouched counter += 2 - if currentRequest.sessionData.faults[k] - currentRequest.sessionData.faults[k].retries = currentRequest.sessionData.retries[k] - db.saveFault(currentRequest.sessionData.deviceId, k, currentRequest.sessionData.faults[k], (err) -> + if sessionContext.faults[k] + sessionContext.faults[k].retries = sessionContext.retries[k] + db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) -> if err - throwError(err, currentRequest.httpResponse) if counter & 1 + throwError(err, sessionContext.httpResponse) if counter & 1 return counter = 0 if (counter -= 2) == 1 - return writeResponse(currentRequest, soap.response(null)) + return writeResponse(sessionContext, soap.response(null)) ) else - db.deleteFault(currentRequest.sessionData.deviceId, k, (err) -> + db.deleteFault(sessionContext.deviceId, k, (err) -> if err - throwError(err, currentRequest.httpResponse) if counter & 1 + throwError(err, sessionContext.httpResponse) if counter & 1 return counter = 0 if (counter -= 2) == 1 - return writeResponse(currentRequest, soap.response(null)) + return writeResponse(sessionContext, soap.response(null)) ) if (counter -= 2) == 1 - return writeResponse(currentRequest, soap.response(null)) + return writeResponse(sessionContext, soap.response(null)) ) ) return - if rpcRequest.type is 'Download' - if not rpcRequest.url? + if acsRequest.name is 'Download' + if not acsRequest.url? FS_PORT = config.get('FS_PORT') FS_IP = config.get('FS_IP') FS_SSL = config.get('FS_SSL') - rpcRequest.url = if FS_SSL then 'https://' else 'http://' - rpcRequest.url += FS_IP - rpcRequest.url += ":#{FS_PORT}" if FS_PORT != 80 - rpcRequest.url += "/#{encodeURIComponent(rpcRequest.fileName)}" + acsRequest.url = if FS_SSL then 'https://' else 'http://' + acsRequest.url += FS_IP + acsRequest.url += ":#{FS_PORT}" if FS_PORT != 80 + acsRequest.url += "/#{encodeURIComponent(acsRequest.fileName)}" - if not rpcRequest.fileSize? + if not acsRequest.fileSize? return cache.getFiles((err, hash, files) -> - return throwError(err, currentRequest.httpResponse) if err + return throwError(err, sessionContext.httpResponse) if err - if rpcRequest.fileName of files - rpcRequest.fileSize = files[rpcRequest.fileName].length + if acsRequest.fileName of files + acsRequest.fileSize = files[acsRequest.fileName].length else - rpcRequest.fileSize = 0 + acsRequest.fileSize = 0 - return sendRpcRequest(currentRequest, id, rpcRequest) + return sendAcsRequest(sessionContext, id, acsRequest) ) - logger.accessInfo({ - currentRequest: currentRequest - message: 'ACS request' - acsRequest: rpcRequest - acsRequestId: id - }) - - res = soap.response({ + rpc = { id : id, - methodRequest : rpcRequest, - cwmpVersion : currentRequest.sessionData.cwmpVersion + acsRequest : acsRequest, + cwmpVersion : sessionContext.cwmpVersion + } + + logger.accessInfo({ + sessionContext: sessionContext + message: 'ACS request' + rpc: rpc }) - writeResponse(currentRequest, res) + res = soap.response(rpc) + + writeResponse(sessionContext, res) getSession = (httpRequest, callback) -> @@ -544,18 +547,18 @@ getSession = (httpRequest, callback) -> return callback() if not sessionId? - sessionData = currentSessions.get(httpRequest.connection)?[sessionId] + sessionContext = currentSessions.get(httpRequest.connection)?[sessionId] - if sessionData? - return callback(null, sessionData) + if sessionContext? + return callback(null, sessionContext) setTimeout(() -> - db.redisClient.eval('local v=redis.call("get",KEYS[1]);redis.call("del",KEYS[1]);return v;', 1, "session_#{sessionId}", (err, sessionDataString) -> - return callback(err) if err or not sessionDataString - session.deserialize(sessionDataString, (err, sessionData) -> + db.redisClient.eval('local v=redis.call("get",KEYS[1]);redis.call("del",KEYS[1]);return v;', 1, "session_#{sessionId}", (err, sessionContextString) -> + return callback(err) if err or not sessionContextString + session.deserialize(sessionContextString, (err, sessionContext) -> return callback(err) if err - currentSessions.get(httpRequest.connection)[sessionId] = sessionData - callback(null, sessionData) + currentSessions.get(httpRequest.connection)[sessionId] = sessionContext + callback(null, sessionContext) ) ) , 100 @@ -569,11 +572,13 @@ onConnection = (socket) -> currentSessions.set(socket, {}) socket.on('close', () -> sessions = currentSessions.get(socket) - for sessionId, sessionData of sessions - session.serialize(sessionData, (err, sessionDataString) -> + for sessionId, sessionContext of sessions + delete sessionContext.httpRequest + delete sessionContext.httpResponse + session.serialize(sessionContext, (err, sessionContextString) -> return throwError(err) if err # TODO don't set if process is shutting down - db.redisClient.setex("session_#{sessionId}", sessionData.timeout, sessionDataString, (err) -> + db.redisClient.setex("session_#{sessionId}", sessionContext.timeout, sessionContextString, (err) -> return throwError(err) if err ) ) @@ -618,121 +623,114 @@ listener = (httpRequest, httpResponse) -> return body stream.on('end', () -> - cwmpRequest = null - getSession(httpRequest, f = (err, sessionData) -> + rpc = null + getSession(httpRequest, f = (err, sessionContext) -> return throwError(err, httpResponse) if err - cwmpRequest ?= soap.request(httpRequest, sessionData?.cwmpVersion) - if not sessionData? - if cwmpRequest.methodRequest?.type isnt 'Inform' + rpc ?= soap.request(httpRequest, sessionContext?.cwmpVersion) + if not sessionContext? + if rpc.cpeRequest?.name isnt 'Inform' httpResponse.writeHead(400) httpResponse.end('Session is expired') return - deviceId = common.generateDeviceId(cwmpRequest.methodRequest.deviceId) - return session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionData) -> + deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId) + return session.init(deviceId, rpc.cwmpVersion, rpc.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionContext) -> return throwError(err, httpResponse) if err - sessionData.sessionId = crypto.randomBytes(8).toString('hex') + sessionContext.sessionId = crypto.randomBytes(8).toString('hex') - currentSessions.get(httpRequest.connection)[sessionData.sessionId] = sessionData + currentSessions.get(httpRequest.connection)[sessionContext.sessionId] = sessionContext - httpRequest.connection.setTimeout(sessionData.timeout * 1000) + httpRequest.connection.setTimeout(sessionContext.timeout * 1000) - db.getDueTasksAndFaultsAndOperations(deviceId, sessionData.timestamp, (err, dueTasks, faults, operations) -> + db.getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations) -> return throwError(err, httpResponse) if err - sessionData.tasks = dueTasks - sessionData.faults = faults - sessionData.retries = {} + sessionContext.tasks = dueTasks + sessionContext.faults = faults + sessionContext.retries = {} for k, v of faults - sessionData.retries[k] = v.retries - sessionData.operations = operations + sessionContext.retries[k] = v.retries + sessionContext.operations = operations # Delete expired faults - for k, v of sessionData.faults - if v.expiry >= sessionData.timestamp - delete sessionData.faults[k] - sessionData.faultsTouched ?= {} - sessionData.faultsTouched[k] = true + for k, v of sessionContext.faults + if v.expiry >= sessionContext.timestamp + delete sessionContext.faults[k] + sessionContext.faultsTouched ?= {} + sessionContext.faultsTouched[k] = true - f(null, sessionData) + f(null, sessionContext) ) ) - currentRequest = { - httpRequest : httpRequest, - httpResponse : httpResponse, - sessionData : sessionData - } + sessionContext.httpRequest = httpRequest + sessionContext.httpResponse = httpResponse - if config.get('DEBUG', currentRequest.sessionData.deviceId) + if config.get('DEBUG', sessionContext.deviceId) dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{httpRequest.getBody()}\n\n" - require('fs').appendFile("./debug/#{currentRequest.sessionData.deviceId}.dump", dump, (err) -> + require('fs').appendFile("./debug/#{sessionContext.deviceId}.dump", dump, (err) -> return throwError(err) if err ) - if cwmpRequest.methodRequest? - if cwmpRequest.methodRequest.type is 'Inform' + if rpc.cpeRequest? + if rpc.cpeRequest.name is 'Inform' logger.accessInfo({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'Inform' - cpeRequest: cwmpRequest.methodRequest - cpeRequestId: cwmpRequest.id - }); - inform(currentRequest, cwmpRequest) - else if cwmpRequest.methodRequest.type is 'TransferComplete' - logger.accessInfo({ - currentRequest: currentRequest - message: 'CPE request' - cpeRequest: cwmpRequest.methodRequest - cpeRequestId: cwmpRequest.id - }); - transferComplete(currentRequest, cwmpRequest) - else if cwmpRequest.methodRequest.type is 'GetRPCMethods' - logger.accessInfo({ - currentRequest: currentRequest - message: 'CPE request' - cpeRequest: cwmpRequest.methodRequest - cpeRequestId: cwmpRequest.id - }); - res = soap.response({ - id : cwmpRequest.id, - methodResponse : {type : 'GetRPCMethodsResponse', methodList : ['Inform', 'GetRPCMethods', 'TransferComplete', 'RequestDownload']}, - cwmpVersion : currentRequest.sessionData.cwmpVersion + rpc: rpc }) - writeResponse(currentRequest, res) + inform(sessionContext, rpc) + else if rpc.cpeRequest.name is 'TransferComplete' + logger.accessInfo({ + sessionContext: sessionContext + message: 'CPE request' + rpc: rpc + }) + transferComplete(sessionContext, rpc) + else if rpc.cpeRequest.name is 'GetRPCMethods' + logger.accessInfo({ + sessionContext: sessionContext + message: 'CPE request' + rpc: rpc + }) + res = soap.response({ + id : rpc.id + acsResponse : {name: 'GetRPCMethodsResponse', methodList: ['Inform', 'GetRPCMethods', 'TransferComplete', 'RequestDownload']} + cwmpVersion : sessionContext.cwmpVersion + }) + writeResponse(sessionContext, res) else - return throwError(new Error('ACS method not supported'), currentRequest.httpResponse) - else if cwmpRequest.methodResponse? - session.rpcResponse(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.methodResponse, (err) -> - return throwError(err, currentRequest.httpResponse) if err - nextRpc(currentRequest) + return throwError(new Error('ACS method not supported'), sessionContext.httpResponse) + else if rpc.cpeResponse + session.rpcResponse(sessionContext, rpc.id, rpc.cpeResponse, (err) -> + return throwError(err, sessionContext.httpResponse) if err + nextRpc(sessionContext) ) - else if cwmpRequest.fault? + else if rpc.cpeFault logger.accessWarn({ - currentRequest: currentRequest + sessionContext: sessionContext message: 'CPE fault' - cpeFault: cwmpRequest.fault.detail - cpeRequestId: cwmpRequest.id - }); - session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, fault) -> - return throwError(err, currentRequest.httpResponse) if err + rpc:rpc + }) + session.rpcFault(sessionContext, rpc.id, rpc.cpeFault, (err, fault) -> + return throwError(err, sessionContext.httpResponse) if err if fault - recordFault(currentRequest, fault, currentRequest.sessionData.provisions, currentRequest.sessionData.channels) - session.clearProvisions(currentRequest.sessionData) - nextRpc(currentRequest) + recordFault(sessionContext, fault) + session.clearProvisions(sessionContext) + nextRpc(sessionContext) ) else # CPE sent empty response - session.timeoutOperations(currentRequest.sessionData, (err, faults, operations) -> - return throwError(err, currentRequest.httpResponse) if err + session.timeoutOperations(sessionContext, (err, faults, operations) -> + return throwError(err, sessionContext.httpResponse) if err for fault, i in faults for k, v of operations[i].retries - sessionData.retries[k] = v + sessionContext.retries[k] = v - recordFault(currentRequest, fault, operations[i].provisions, operations[i].channels) + recordFault(sessionContext, fault, operations[i].provisions, operations[i].channels) - nextRpc(currentRequest) + nextRpc(sessionContext) ) ) ) diff --git a/lib/default-provisions.js b/lib/default-provisions.js index aa1e7a0..f4ee714 100644 --- a/lib/default-provisions.js +++ b/lib/default-provisions.js @@ -23,14 +23,14 @@ const device = require("./device"); const MAX_DEPTH = config.get("MAX_DEPTH"); -function refresh(sessionData, provision, declarations, startRevision, endRevision) { +function refresh(sessionContext, provision, declarations, startRevision, endRevision) { let path = common.parsePath(provision[1]).slice(); let l = path.length; path.length = MAX_DEPTH; path.fill("*", l); let every = 1000 * (provision[2] || 1); - let offset = Math.abs(common.djb2(sessionData.deviceId)) % 86400000; - let t = Math.trunc((sessionData.timestamp + offset) / every) * every - offset; + let offset = Math.abs(common.djb2(sessionContext.deviceId)) % 86400000; + let t = Math.trunc((sessionContext.timestamp + offset) / every) * every - offset; for (let i = l; i < path.length; ++ i) declarations.push([path.slice(0, i), t, {object: 1, writable: 1, value: t}]); @@ -39,7 +39,7 @@ function refresh(sessionData, provision, declarations, startRevision, endRevisio } -function value(sessionData, provision, declarations, startRevision, endRevision) { +function value(sessionContext, provision, declarations, startRevision, endRevision) { let path = common.parsePath(provision[1]); declarations.push([path, 1, {value: 1}, null, {value: [provision[2]]}]); @@ -47,7 +47,7 @@ function value(sessionData, provision, declarations, startRevision, endRevision) } -function tag(sessionData, provision, declarations, startRevision, endRevision) { +function tag(sessionContext, provision, declarations, startRevision, endRevision) { let path = ["Tags", provision[1]]; declarations.push([path, 1, {value: 1}, null, {value: [provision[2]]}]); @@ -55,33 +55,33 @@ function tag(sessionData, provision, declarations, startRevision, endRevision) { } -function reboot(sessionData, provision, declarations, startRevision, endRevision) { - declarations.push([["Reboot"], 1, {value: 1}, null, {value: [sessionData.timestamp]}]); +function reboot(sessionContext, provision, declarations, startRevision, endRevision) { + declarations.push([["Reboot"], 1, {value: 1}, null, {value: [sessionContext.timestamp]}]); return true; } -function reset(sessionData, provision, declarations, startRevision, endRevision) { - declarations.push([["FactoryReset"], 1, {value: 1}, null, {value: [sessionData.timestamp]}]); +function reset(sessionContext, provision, declarations, startRevision, endRevision) { + declarations.push([["FactoryReset"], 1, {value: 1}, null, {value: [sessionContext.timestamp]}]); return true; } -function download(sessionData, provision, declarations, startRevision, endRevision) { +function download(sessionContext, provision, declarations, startRevision, endRevision) { let alias = ["FileType", provision[1] || "", "FileName", provision[2] || "", "TargetFileName", provision[3] || ""]; declarations.push([["Downloads", alias], 1, {}, 1]); declarations.push([["Downloads", alias, "Download"], 1, {value: 1}, null, - {value: [sessionData.timestamp]}]); + {value: [sessionContext.timestamp]}]); return true; } -function instances(sessionData, provision, declarations, startRevision, endRevision) { +function instances(sessionContext, provision, declarations, startRevision, endRevision) { let count = Number(provision[2]); if (Number.isNaN(count)) @@ -95,7 +95,7 @@ function instances(sessionData, provision, declarations, startRevision, endRevis if (endRevision === startRevision) return false; - let unpacked = device.unpack(sessionData.deviceData, path, startRevision + 1); + let unpacked = device.unpack(sessionContext.deviceData, path, startRevision + 1); count = Math.max(0, unpacked.length + count); } diff --git a/lib/logger.js b/lib/logger.js index a2d2ceb..ce28952 100644 --- a/lib/logger.js +++ b/lib/logger.js @@ -131,10 +131,10 @@ function close() { function flatten(details) { - if (details.currentRequest) { - details.deviceId = details.currentRequest.sessionData.deviceId; - details.remoteAddress = details.currentRequest.httpRequest.connection.remoteAddress; - delete details.currentRequest; + if (details.sessionContext) { + details.deviceId = details.sessionContext.deviceId; + details.remoteAddress = details.sessionContext.httpRequest.connection.remoteAddress; + delete details.sessionContext; } if (details.exception) { @@ -149,30 +149,31 @@ function flatten(details) { delete details.task; } - if (details.acsRequest) { - details.acsRequestName = details.acsRequest.type; - if (details.acsRequest.commandKey) - details.acsRequestCommandKey = details.acsRequest.commandKey; - delete details.acsRequest; - } - - if (details.cpeRequest) { - if (details.cpeRequest.type === "Inform") { - details.informEvent = details.cpeRequest.event.join(","); - details.informRetryCount = details.cpeRequest.retryCount; + if (details.rpc) { + if (details.rpc.acsRequest) { + details.acsRequestId = details.rpc.id; + details.acsRequestName = details.rpc.acsRequest.name; + if (details.rpc.acsRequest.commandKey) + details.acsRequestCommandKey = details.rpc.acsRequest.commandKey; } - else { - details.cpeRequestName = details.cpeRequest.type; - if (details.cpeRequest.commandKey) - details.cpeRequestCommandKey = details.cpeRequest.commandKey; + else if (details.rpc.cpeRequest) { + details.cpeRequestId = details.rpc.id; + if (details.rpc.cpeRequest.name === "Inform") { + details.informEvent = details.rpc.cpeRequest.event.join(","); + details.informRetryCount = details.rpc.cpeRequest.retryCount; + } + else { + details.cpeRequestName = details.rpc.cpeRequest.type; + if (details.rpc.cpeRequest.commandKey) + details.cpeRequestCommandKey = details.rpc.cpeRequest.commandKey; + } } - delete details.cpeRequest; - } - - if (details.cpeFault) { - details.cpeFaultCode = details.cpeFault.faultCode; - details.cpeFaultString = details.cpeFault.faultString; - delete details.cpeFault; + else if (details.rpc.cpeFault) { + details.acsRequestId = details.rpc.id; + details.cpeFaultCode = details.rpc.cpeFault.faultCode; + details.cpeFaultString = details.rpc.cpeFault.faultString; + } + delete details.rpc; } if (details.fault) { @@ -180,6 +181,7 @@ function flatten(details) { details.faultMessage = details.fault.message; delete details.fault; } + return details; } diff --git a/lib/session.coffee b/lib/session.coffee index 6317502..9f56ecf 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -28,6 +28,9 @@ defaultProvisions = require './default-provisions' MAX_ITERATIONS = 64 +provisionsCache = new WeakMap() +virtualParametersCache = new WeakMap() + initDeviceData = () -> return { @@ -43,7 +46,7 @@ initDeviceData = () -> init = (deviceId, cwmpVersion, timeout, callback) -> timestamp = Date.now() - sessionData = { + sessionContext = { timestamp : timestamp deviceId : deviceId deviceData : initDeviceData() @@ -63,58 +66,56 @@ init = (deviceId, cwmpVersion, timeout, callback) -> cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> return callback(err) if err - sessionData.cache = { - hash: hash - provisions: provisions - virtualParameters: virtualParameters - } + sessionContext.presetsHash = hash + provisionsCache.set(sessionContext, provisions) + virtualParametersCache.set(sessionContext, virtualParameters) - return callback(null, sessionData) + return callback(null, sessionContext) ) -loadParameters = (sessionData, callback) -> - if not sessionData.toLoad?.length +loadParameters = (sessionContext, callback) -> + if not sessionContext.toLoad?.length return callback() - db.fetchDevice(sessionData.deviceId, sessionData.timestamp, sessionData.toLoad, (err, parameters, loaded) -> + db.fetchDevice(sessionContext.deviceId, sessionContext.timestamp, sessionContext.toLoad, (err, parameters, loaded) -> return callback(err) if err if not parameters? # Device not available in database, mark as new - sessionData.new = true - loaded = [[[], (1 << config.get('MAX_DEPTH', sessionData.deviceId)) - 1]] + sessionContext.new = true + loaded = [[[], (1 << config.get('MAX_DEPTH', sessionContext.deviceId)) - 1]] parameters = [] for p in loaded - path = sessionData.deviceData.paths.add(p[0]) + path = sessionContext.deviceData.paths.add(p[0]) if p[1] - l = sessionData.deviceData.loaded.get(path) | 0 - sessionData.deviceData.loaded.set(path, l | p[1]) + l = sessionContext.deviceData.loaded.get(path) | 0 + sessionContext.deviceData.loaded.set(path, l | p[1]) for p in parameters - path = sessionData.deviceData.paths.add(p[0]) + path = sessionContext.deviceData.paths.add(p[0]) - sessionData.deviceData.timestamps.set(path, p[1], 0) + sessionContext.deviceData.timestamps.set(path, p[1], 0) if p[2] - sessionData.deviceData.attributes.set(path, p[2], 0) + sessionContext.deviceData.attributes.set(path, p[2], 0) - delete sessionData.toLoad + delete sessionContext.toLoad return callback() ) -generateRpcId = (sessionData) -> - return sessionData.timestamp.toString(16) + - sessionData.cycle.toString(16).slice(-1) + - "0#{sessionData.rpcCount.toString(16)}".slice(-2) +generateRpcId = (sessionContext) -> + return sessionContext.timestamp.toString(16) + + sessionContext.cycle.toString(16).slice(-1) + + "0#{sessionContext.rpcCount.toString(16)}".slice(-2) -inform = (sessionData, rpcReq, callback) -> - timestamp = sessionData.timestamp + sessionData.iteration + 1 +inform = (sessionContext, rpcReq, callback) -> + timestamp = sessionContext.timestamp + sessionContext.iteration + 1 params = [] params.push([['DeviceID', 'Manufacturer'], timestamp, @@ -141,39 +142,39 @@ inform = (sessionData, rpcReq, callback) -> {object: [timestamp, 0], writable: [timestamp, 0], value: [timestamp, [timestamp, 'xsd:dateTime']]}]) for p in params - loadPath(sessionData, p[0]) + loadPath(sessionContext, p[0]) - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err - sessionData.deviceData.timestamps.revision = 1 - sessionData.deviceData.attributes.revision = 1 + sessionContext.deviceData.timestamps.revision = 1 + sessionContext.deviceData.attributes.revision = 1 toClear = null for p in params - toClear = device.set(sessionData.deviceData, p[0], p[1], p[2], toClear) + toClear = device.set(sessionContext.deviceData, p[0], p[1], p[2], toClear) - clear(sessionData, toClear, (err) -> - return callback(err, {type : 'InformResponse'}) + clear(sessionContext, toClear, (err) -> + return callback(err, {name: 'InformResponse'}) ) ) -transferComplete = (sessionData, rpcReq, callback) -> +transferComplete = (sessionContext, rpcReq, callback) -> commandKey = rpcReq.commandKey - operation = sessionData.operations[commandKey] + operation = sessionContext.operations[commandKey] if not operation? - return callback(null, {type : 'TransferCompleteResponse'}) + return callback(null, {name: 'TransferCompleteResponse'}) instance = operation.args.instance - delete sessionData.operations[commandKey] - sessionData.operationsTouched ?= {} - sessionData.operationsTouched[commandKey] = 1 + delete sessionContext.operations[commandKey] + sessionContext.operationsTouched ?= {} + sessionContext.operationsTouched[commandKey] = 1 if rpcReq.faultStruct? and rpcReq.faultStruct.faultCode != '0' - return revertDownloadParameters(sessionData, operation.args.instance, (err) -> + return revertDownloadParameters(sessionContext, operation.args.instance, (err) -> fault = { code: "cwmp.#{rpcReq.faultStruct.faultCode}" message: rpcReq.faultStruct.faultString @@ -181,81 +182,80 @@ transferComplete = (sessionData, rpcReq, callback) -> timestamp: operation.timestamp } - return callback(err, {type : 'TransferCompleteResponse'}, operation, fault) + return callback(err, {name: 'TransferCompleteResponse'}, operation, fault) ) - loadPath(sessionData, ['Downloads', instance, '*']) + loadPath(sessionContext, ['Downloads', instance, '*']) - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err toClear = null - timestamp = sessionData.timestamp + sessionData.iteration + 1 + timestamp = sessionContext.timestamp + sessionContext.iteration + 1 - p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastDownload']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'LastDownload']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [operation.timestamp, 'xsd:dateTime']]}, toClear) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastFileType']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'LastFileType']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [operation.args.fileType, 'xsd:string']]}, toClear) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastFileName']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'LastFileName']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [operation.args.fileName, 'xsd:string']]}, toClear) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastTargetFileName']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'LastTargetFileName']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [operation.args.targetFileName, 'xsd:string']]}, toClear) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'StartTime']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'StartTime']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [+rpcReq.startTime, 'xsd:dateTime']]}, toClear) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'CompleteTime']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'CompleteTime']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [+rpcReq.completeTime, 'xsd:dateTime']]}, toClear) - clear(sessionData, toClear, (err) -> - return callback(err, {type : 'TransferCompleteResponse'}, operation) + clear(sessionContext, toClear, (err) -> + return callback(err, {name: 'TransferCompleteResponse'}, operation) ) ) -revertDownloadParameters = (sessionData, instance, callback) -> - loadPath(sessionData, ['Downloads', instance, '*']) +revertDownloadParameters = (sessionContext, instance, callback) -> + loadPath(sessionContext, ['Downloads', instance, '*']) - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err - timestamp = sessionData.timestamp + sessionData.iteration + 1 + timestamp = sessionContext.timestamp + sessionContext.iteration + 1 - p = sessionData.deviceData.paths.add(['Downloads', instance, 'LastDownload']) - LastDownload = sessionData.deviceData.attributes.get(p) + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'LastDownload']) + LastDownload = sessionContext.deviceData.attributes.get(p) - p = sessionData.deviceData.paths.add(['Downloads', instance, 'Download']) - toClear = device.set(sessionData.deviceData, p, timestamp, + p = sessionContext.deviceData.paths.add(['Downloads', instance, 'Download']) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, [LastDownload?.value[1]?[0] or 0, 'xsd:dateTime']]}, toClear) - clear(sessionData, toClear, callback) + clear(sessionContext, toClear, callback) ) - -timeoutOperations = (sessionData, callback) -> +timeoutOperations = (sessionContext, callback) -> faults = [] operations = [] counter = 3 - for commandKey, operation of sessionData.operations - if operation.type isnt 'Download' - return callback(new Error("Unknown operation type #{operation.type}")) + for commandKey, operation of sessionContext.operations + if operation.name isnt 'Download' + return callback(new Error("Unknown operation name #{operation.name}")) - DOWNLOAD_TIMEOUT = config.get('DOWNLOAD_TIMEOUT', sessionData.deviceId) * 1000 + DOWNLOAD_TIMEOUT = config.get('DOWNLOAD_TIMEOUT', sessionContext.deviceId) * 1000 - if sessionData.timestamp > operation.timestamp + DOWNLOAD_TIMEOUT - delete sessionData.operations[commandKey] - sessionData.operationsTouched ?= {} - sessionData.operationsTouched[commandKey] = 1 + if sessionContext.timestamp > operation.timestamp + DOWNLOAD_TIMEOUT + delete sessionContext.operations[commandKey] + sessionContext.operationsTouched ?= {} + sessionContext.operationsTouched[commandKey] = 1 faults.push({ code: 'timeout' @@ -266,7 +266,7 @@ timeoutOperations = (sessionData, callback) -> operations.push(operation) counter += 2 - revertDownloadParameters(sessionData, operation.args.instance, (err) -> + revertDownloadParameters(sessionContext, operation.args.instance, (err) -> if err callback(err) if counter & 1 return counter = 0 @@ -277,78 +277,79 @@ timeoutOperations = (sessionData, callback) -> return callback(null, faults, operations) if (counter -= 2) == 1 -addProvisions = (sessionData, channel, provisions) -> - delete sessionData.syncState - delete sessionData.rpcRequest - sessionData.declarations = [] - sessionData.doneProvisions = 0 - if sessionData.revisions[sessionData.revisions.length - 1] > 0 - sessionData.deviceData.timestamps.collapse(1) - sessionData.deviceData.attributes.collapse(1) - sessionData.revisions = [0] - sessionData.extensionsCache = {} +addProvisions = (sessionContext, channel, provisions) -> + delete sessionContext.syncState + delete sessionContext.rpcRequest + sessionContext.declarations = [] + sessionContext.doneProvisions = 0 + if sessionContext.revisions[sessionContext.revisions.length - 1] > 0 + sessionContext.deviceData.timestamps.collapse(1) + sessionContext.deviceData.attributes.collapse(1) + sessionContext.revisions = [0] + sessionContext.extensionsCache = {} - if sessionData.iteration != sessionData.cycle * MAX_ITERATIONS - sessionData.cycle += 1 - sessionData.rpcCount = 0 - sessionData.iteration = sessionData.cycle * MAX_ITERATIONS + if sessionContext.iteration != sessionContext.cycle * MAX_ITERATIONS + sessionContext.cycle += 1 + sessionContext.rpcCount = 0 + sessionContext.iteration = sessionContext.cycle * MAX_ITERATIONS - sessionData.channels[channel] |= 0 + sessionContext.channels[channel] |= 0 for provision, i in provisions channels = [channel] # Remove duplicate provisions provisionStr = JSON.stringify(provision) - for p, j in sessionData.provisions + for p, j in sessionContext.provisions if JSON.stringify(p) == provisionStr - sessionData.provisions.splice(j, 1) - for c of sessionData.channels - channels.push(c) if sessionData.channels[c] & (1 << j) - a = sessionData.channels[c] >> (j + 1) - sessionData.channels[c] &= ((1 << j) - 1) - sessionData.channels[c] |= (a << j) + sessionContext.provisions.splice(j, 1) + for c of sessionContext.channels + channels.push(c) if sessionContext.channels[c] & (1 << j) + a = sessionContext.channels[c] >> (j + 1) + sessionContext.channels[c] &= ((1 << j) - 1) + sessionContext.channels[c] |= (a << j) for c in channels - sessionData.channels[channel] |= (1 << sessionData.provisions.length) - sessionData.provisions.push(provision) + sessionContext.channels[channel] |= (1 << sessionContext.provisions.length) + sessionContext.provisions.push(provision) -clearProvisions = (sessionData) -> - if sessionData.revisions[sessionData.revisions.length - 1] > 0 - sessionData.deviceData.timestamps.collapse(1) - sessionData.deviceData.attributes.collapse(1) +clearProvisions = (sessionContext) -> + if sessionContext.revisions[sessionContext.revisions.length - 1] > 0 + sessionContext.deviceData.timestamps.collapse(1) + sessionContext.deviceData.attributes.collapse(1) - if sessionData.iteration != sessionData.cycle * MAX_ITERATIONS - sessionData.cycle += 1 - sessionData.rpcCount = 0 - sessionData.iteration = sessionData.cycle * MAX_ITERATIONS + if sessionContext.iteration != sessionContext.cycle * MAX_ITERATIONS + sessionContext.cycle += 1 + sessionContext.rpcCount = 0 + sessionContext.iteration = sessionContext.cycle * MAX_ITERATIONS - delete sessionData.syncState - delete sessionData.rpcRequest - sessionData.provisions = [] - sessionData.virtualParameters = [] - sessionData.channels = {} - sessionData.declarations = [] - sessionData.doneProvisions = 0 - sessionData.revisions = [0] - sessionData.extensionsCache = {} + delete sessionContext.syncState + delete sessionContext.rpcRequest + sessionContext.provisions = [] + sessionContext.virtualParameters = [] + sessionContext.channels = {} + sessionContext.declarations = [] + sessionContext.doneProvisions = 0 + sessionContext.revisions = [0] + sessionContext.extensionsCache = {} -runProvisions = (sessionData, provisions, startRevision, endRevision, callback) -> +runProvisions = (sessionContext, provisions, startRevision, endRevision, callback) -> done = true allDeclarations = [] allClear = [] counter = 3 + allProvisions = provisionsCache.get(sessionContext) for provision in provisions - if not sessionData.cache.provisions[provision[0]]? + if not allProvisions[provision[0]]? if defaultProvisions[provision[0]] - done = defaultProvisions[provision[0]](sessionData, provision, allDeclarations, startRevision, endRevision) and done + done = defaultProvisions[provision[0]](sessionContext, provision, allDeclarations, startRevision, endRevision) and done continue counter += 2 - sandbox.run(sessionData.cache.provisions[provision[0]].script, - {args: provision.slice(1)}, sessionData.timestamp, sessionData.deviceData, - sessionData.extensionsCache, startRevision, endRevision, + sandbox.run(allProvisions[provision[0]].script, + {args: provision.slice(1)}, sessionContext.timestamp, sessionContext.deviceData, + sessionContext.extensionsCache, startRevision, endRevision, (err, _fault, _clear, _declarations, _done) -> if err or _fault callback(err, _fault) if counter & 1 @@ -370,18 +371,19 @@ runProvisions = (sessionData, provisions, startRevision, endRevision, callback) return callback(null, null, done, allDeclarations, allClear) -runVirtualParameters = (sessionData, provisions, startRevision, endRevision, callback) -> +runVirtualParameters = (sessionContext, provisions, startRevision, endRevision, callback) -> done = true virtualParameterUpdates = [] allDeclarations = [] allClear = [] counter = 3 + allVirtualParameters = virtualParametersCache.get(sessionContext) for provision in provisions counter += 2 globals = {TIMESTAMPS: provision[1], VALUES: provision[2]} - sandbox.run(sessionData.cache.virtualParameters[provision[0]].script, globals, - sessionData.timestamp, sessionData.deviceData, - sessionData.extensionsCache, startRevision, endRevision, + sandbox.run(allVirtualParameters[provision[0]].script, globals, + sessionContext.timestamp, sessionContext.deviceData, + sessionContext.extensionsCache, startRevision, endRevision, (err, _fault, _clear, _declarations, _done, _returnValue) -> if err or _fault callback(err, _fault) if counter & 1 @@ -402,9 +404,9 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal toClear = null if virtualParameterUpdates.length == provisions.length for vpu, i in virtualParameterUpdates - toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear) + toClear = commitVirtualParameter(sessionContext, provisions[i], vpu, toClear) - clear(sessionData, toClear, (err) -> + clear(sessionContext, toClear, (err) -> return callback(err, null, done, allDeclarations, allClear) ) ) @@ -413,16 +415,16 @@ runVirtualParameters = (sessionData, provisions, startRevision, endRevision, cal toClear = null if virtualParameterUpdates.length == provisions.length for vpu, i in virtualParameterUpdates - toClear = commitVirtualParameter(sessionData, provisions[i], vpu, toClear) + toClear = commitVirtualParameter(sessionContext, provisions[i], vpu, toClear) - clear(sessionData, toClear, (err) -> + clear(sessionContext, toClear, (err) -> return callback(err, null, done, allDeclarations, allClear) ) -runDeclarations = (sessionData, declarations) -> - sessionData.iteration += 2 - sessionData.syncState ?= { +runDeclarations = (sessionContext, declarations) -> + sessionContext.iteration += 2 + sessionContext.syncState ?= { refreshAttributes: { exist: new Set() object: new Set() @@ -448,26 +450,28 @@ runDeclarations = (sessionData, declarations) -> allDeclareAttributeTimestamps = new Map() allDeclareAttributeValues = new Map() + allVirtualParameters = virtualParametersCache.get(sessionContext) + for declaration, i in declarations path = common.addPathMeta(declaration[0]) if (path.alias | path.wildcard) & 1 or path[0] == 'VirtualParameters' - sessionData.deviceData.paths.add(['VirtualParameters']) + sessionContext.deviceData.paths.add(['VirtualParameters']) if (path.alias | path.wildcard) & 2 - sessionData.deviceData.paths.add(['VirtualParameters', '*']) - for k of sessionData.cache.virtualParameters - sessionData.deviceData.paths.add(['VirtualParameters', k]) + sessionContext.deviceData.paths.add(['VirtualParameters', '*']) + for k of allVirtualParameters + sessionContext.deviceData.paths.add(['VirtualParameters', k]) if (path.alias | path.wildcard) & 1 or path[0] == 'Reboot' - sessionData.deviceData.paths.add(['Reboot']) + sessionContext.deviceData.paths.add(['Reboot']) if (path.alias | path.wildcard) & 1 or path[0] == 'FactoryReset' - sessionData.deviceData.paths.add(['FactoryReset']) + sessionContext.deviceData.paths.add(['FactoryReset']) if path.alias aliasDecs = device.getAliasDeclarations(path, declaration[1] or 1) for ad in aliasDecs - p = sessionData.deviceData.paths.add(ad[0]) + p = sessionContext.deviceData.paths.add(ad[0]) allDeclareTimestamps.set(p, Math.max(ad[1] or 1, allDeclareTimestamps.get(p) or 0)) attrTrackers = [] if ad[2] @@ -477,9 +481,9 @@ runDeclarations = (sessionData, declarations) -> attrTrackers.push(k) cur[k] = Math.max(v, cur[k] or 0) - device.track(sessionData.deviceData, p, 'prerequisite', false, null, true, attrTrackers) + device.track(sessionContext.deviceData, p, 'prerequisite', false, null, true, attrTrackers) - unpacked = device.unpack(sessionData.deviceData, path) + unpacked = device.unpack(sessionContext.deviceData, path) for u in unpacked allDeclareTimestamps.set(u, Math.max(declaration[1] or 1, allDeclareTimestamps.get(u) or 0)) if declaration[2] @@ -494,7 +498,7 @@ runDeclarations = (sessionData, declarations) -> for k, v of declaration[4] cur[k] = v else - path = sessionData.deviceData.paths.add(path) + path = sessionContext.deviceData.paths.add(path) allDeclareTimestamps.set(path, Math.max(declaration[1] or 1, allDeclareTimestamps.get(path) or 0)) if declaration[2] cur = allDeclareAttributeTimestamps.get(path) @@ -509,7 +513,7 @@ runDeclarations = (sessionData, declarations) -> for k, v of declaration[4] cur[k] = v - device.track(sessionData.deviceData, path, 'prerequisite', false, null, true) + device.track(sessionContext.deviceData, path, 'prerequisite', false, null, true) if declaration[3]? if Array.isArray(declaration[3]) @@ -528,172 +532,174 @@ runDeclarations = (sessionData, declarations) -> keys = {} if ((path.wildcard | path.alias) & ((1 << (path.length - 1)) - 1)) == 0 - parent = sessionData.deviceData.paths.add(parent) - unpacked ?= device.unpack(sessionData.deviceData, path) - processInstances(sessionData, parent, unpacked, keys, minInstances, maxInstances) + parent = sessionContext.deviceData.paths.add(parent) + unpacked ?= device.unpack(sessionContext.deviceData, path) + processInstances(sessionContext, parent, unpacked, keys, minInstances, maxInstances) else - parentsUnpacked = device.unpack(sessionData.deviceData, parent) + parentsUnpacked = device.unpack(sessionContext.deviceData, parent) for parent in parentsUnpacked - parent = sessionData.deviceData.paths.add(parent) - processInstances(sessionData, parent, device.unpack(sessionData.deviceData, common.addPathMeta(parent.concat([path[parent.length]]))), keys, minInstances, maxInstances) + parent = sessionContext.deviceData.paths.add(parent) + processInstances(sessionContext, parent, device.unpack(sessionContext.deviceData, common.addPathMeta(parent.concat([path[parent.length]]))), keys, minInstances, maxInstances) - return processDeclarations(sessionData, allDeclareTimestamps, allDeclareAttributeTimestamps, allDeclareAttributeValues) + return processDeclarations(sessionContext, allDeclareTimestamps, allDeclareAttributeTimestamps, allDeclareAttributeValues) -rpcRequest = (sessionData, _declarations, callback) -> - if sessionData.rpcRequest? - return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest) +rpcRequest = (sessionContext, _declarations, callback) -> + if sessionContext.rpcRequest? + return callback(null, null, generateRpcId(sessionContext), sessionContext.rpcRequest) - if sessionData.virtualParameters.length == 0 and - sessionData.declarations.length == 0 and + if sessionContext.virtualParameters.length == 0 and + sessionContext.declarations.length == 0 and not _declarations?.length and - (sessionData.doneProvisions or sessionData.provisions.length == 0) + (sessionContext.doneProvisions or sessionContext.provisions.length == 0) return callback() - if sessionData.declarations.length <= sessionData.virtualParameters.length - inception = sessionData.declarations.length - revision = sessionData.revisions[inception] + 1 - sessionData.deviceData.timestamps.revision = revision - sessionData.deviceData.attributes.revision = revision + if sessionContext.declarations.length <= sessionContext.virtualParameters.length + inception = sessionContext.declarations.length + revision = sessionContext.revisions[inception] + 1 + sessionContext.deviceData.timestamps.revision = revision + sessionContext.deviceData.attributes.revision = revision if inception == 0 run = runProvisions - provisions = sessionData.provisions + provisions = sessionContext.provisions else run = runVirtualParameters - provisions = sessionData.virtualParameters[inception - 1] + provisions = sessionContext.virtualParameters[inception - 1] - return run(sessionData, provisions, sessionData.revisions[inception - 1] or 0, sessionData.revisions[inception], (err, fault, done, decs, toClear) -> + return run(sessionContext, provisions, sessionContext.revisions[inception - 1] or 0, sessionContext.revisions[inception], (err, fault, done, decs, toClear) -> return callback(err) if err if fault - fault.timestamp = sessionData.timestamp + fault.timestamp = sessionContext.timestamp return callback(null, fault) - sessionData.declarations.push(decs) - sessionData.doneProvisions |= 1 << inception if done or not decs.length + sessionContext.declarations.push(decs) + sessionContext.doneProvisions |= 1 << inception if done or not decs.length for d in decs for ad in device.getAliasDeclarations(d[0], 1) - loadPath(sessionData, ad[0]) + loadPath(sessionContext, ad[0]) - return clear(sessionData, toClear, (err) -> + return clear(sessionContext, toClear, (err) -> return callback(err) if err - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err - rpcRequest(sessionData, _declarations, callback) + rpcRequest(sessionContext, _declarations, callback) ) ) ) if _declarations?.length - delete sessionData.syncState - sessionData.declarations[0] ?= [] - sessionData.declarations[0] = sessionData.declarations[0].concat(_declarations) + delete sessionContext.syncState + sessionContext.declarations[0] ?= [] + sessionContext.declarations[0] = sessionContext.declarations[0].concat(_declarations) for d in _declarations for ad in device.getAliasDeclarations(d[0], 1) - loadPath(sessionData, ad[0]) + loadPath(sessionContext, ad[0]) - return loadParameters(sessionData, (err) -> + return loadParameters(sessionContext, (err) -> return callback(err) if err - return rpcRequest(sessionData, null, callback) + return rpcRequest(sessionContext, null, callback) ) - if not sessionData.declarations[0].length + if not sessionContext.declarations[0].length return callback() - if sessionData.rpcCount >= 255 or - sessionData.revisions.length >= 8 or - sessionData.cycle >= 16 or - sessionData.iteration >= MAX_ITERATIONS * (sessionData.cycle + 1) + if sessionContext.rpcCount >= 255 or + sessionContext.revisions.length >= 8 or + sessionContext.cycle >= 16 or + sessionContext.iteration >= MAX_ITERATIONS * (sessionContext.cycle + 1) fault = { code: 'endless_cycle' message: 'The provision seems to be repeating indefinitely' - timestamp: sessionData.timestamp + timestamp: sessionContext.timestamp } return callback(null, fault) - if (sessionData.syncState?.virtualParameterDeclarations?.length or 0) < sessionData.declarations.length - inception = sessionData.syncState?.virtualParameterDeclarations?.length or 0 - vpd = runDeclarations(sessionData, sessionData.declarations[inception]) - timestamp = sessionData.timestamp + sessionData.iteration + if (sessionContext.syncState?.virtualParameterDeclarations?.length or 0) < sessionContext.declarations.length + inception = sessionContext.syncState?.virtualParameterDeclarations?.length or 0 + vpd = runDeclarations(sessionContext, sessionContext.declarations[inception]) + timestamp = sessionContext.timestamp + sessionContext.iteration toClear = null + allVirtualParameters = virtualParametersCache.get(sessionContext) + vpd = vpd.filter((declaration) -> if declaration[0].length == 1 - if Object.keys(sessionData.cache.virtualParameters).length - toClear = device.set(sessionData.deviceData, declaration[0], timestamp, {object: [timestamp, 1], writable: [timestamp, 0]}, toClear) + if Object.keys(allVirtualParameters).length + toClear = device.set(sessionContext.deviceData, declaration[0], timestamp, {object: [timestamp, 1], writable: [timestamp, 0]}, toClear) else - toClear = device.set(sessionData.deviceData, declaration[0], timestamp, null, toClear) + toClear = device.set(sessionContext.deviceData, declaration[0], timestamp, null, toClear) else if declaration[0].length == 2 if declaration[0][1] == '*' - for k, v of sessionData.cache.virtualParameters - toClear = device.set(sessionData.deviceData, ['VirtualParameters', k], timestamp, {object: [timestamp, 0]}, toClear) - toClear = device.set(sessionData.deviceData, declaration[0], timestamp, null, toClear) - else if not (declaration[0][1] of sessionData.cache.virtualParameters) - toClear = device.set(sessionData.deviceData, declaration[0], timestamp, null, toClear) + for k, v of allVirtualParameters + toClear = device.set(sessionContext.deviceData, ['VirtualParameters', k], timestamp, {object: [timestamp, 0]}, toClear) + toClear = device.set(sessionContext.deviceData, declaration[0], timestamp, null, toClear) + else if not (declaration[0][1] of allVirtualParameters) + toClear = device.set(sessionContext.deviceData, declaration[0], timestamp, null, toClear) else return true return false ) - return clear(sessionData, toClear, (err) -> + return clear(sessionContext, toClear, (err) -> return callback(err) if err - sessionData.syncState.virtualParameterDeclarations[inception] = vpd - return rpcRequest(sessionData, null, callback) + sessionContext.syncState.virtualParameterDeclarations[inception] = vpd + return rpcRequest(sessionContext, null, callback) ) - if not sessionData.syncState? + if not sessionContext.syncState? return callback() - inception = sessionData.declarations.length - 1 + inception = sessionContext.declarations.length - 1 - provisions = generateGetVirtualParameterProvisions(sessionData, sessionData.syncState.virtualParameterDeclarations[inception]) + provisions = generateGetVirtualParameterProvisions(sessionContext, sessionContext.syncState.virtualParameterDeclarations[inception]) if not provisions - sessionData.rpcRequest = generateGetRpcRequest(sessionData) - if not sessionData.rpcRequest - if sessionData.deviceData.changes.has('prerequisite') - delete sessionData.syncState - device.clearTrackers(sessionData.deviceData, 'prerequisite') - return rpcRequest(sessionData, null, callback) + sessionContext.rpcRequest = generateGetRpcRequest(sessionContext) + if not sessionContext.rpcRequest + if sessionContext.deviceData.changes.has('prerequisite') + delete sessionContext.syncState + device.clearTrackers(sessionContext.deviceData, 'prerequisite') + return rpcRequest(sessionContext, null, callback) toClear = null - timestamp = sessionData.timestamp + sessionData.iteration + 1 + timestamp = sessionContext.timestamp + sessionContext.iteration + 1 # Update tags - sessionData.syncState.tags.forEach((v, p) -> - c = sessionData.deviceData.attributes.get(p) + sessionContext.syncState.tags.forEach((v, p) -> + c = sessionContext.deviceData.attributes.get(p) if v and not c? - toClear = device.set(sessionData.deviceData, p, timestamp, {object: [timestamp, false], writable: [timestamp, true], value: [timestamp, [true, 'xsd:boolean']]}, toClear) + toClear = device.set(sessionContext.deviceData, p, timestamp, {object: [timestamp, false], writable: [timestamp, true], value: [timestamp, [true, 'xsd:boolean']]}, toClear) else if c? and not v - toClear = device.set(sessionData.deviceData, p, timestamp, toClear) + toClear = device.set(sessionContext.deviceData, p, timestamp, toClear) noMoreTags = true - for p in sessionData.deviceData.paths.find(['Tags', '*'], false, true) - if sessionData.deviceData.attributes.has(p) + for p in sessionContext.deviceData.paths.find(['Tags', '*'], false, true) + if sessionContext.deviceData.attributes.has(p) noMoreTags = false break if noMoreTags - toClear = device.set(sessionData.deviceData, ['Tags'], timestamp, null, toClear) + toClear = device.set(sessionContext.deviceData, ['Tags'], timestamp, null, toClear) ) # Downloads index = null - sessionData.syncState.downloadsToCreate.forEach((instance) -> + sessionContext.syncState.downloadsToCreate.forEach((instance) -> if not index? index = 0 - for p in sessionData.deviceData.paths.find(['Downloads', '*'], false, true) - if +p[1] > index and sessionData.deviceData.attributes.has(p) + for p in sessionContext.deviceData.paths.find(['Downloads', '*'], false, true) + if +p[1] > index and sessionContext.deviceData.attributes.has(p) index = +p[1] ++ index - toClear = device.set(sessionData.deviceData, ['Downloads'], + toClear = device.set(sessionContext.deviceData, ['Downloads'], timestamp, {object: [timestamp, 1], writable: [timestamp, 1]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}"], + toClear = device.set(sessionContext.deviceData, ['Downloads', "#{index}"], timestamp, {object: [timestamp, 1], writable: [timestamp, 1]}, toClear) @@ -711,75 +717,75 @@ rpcRequest = (sessionData, _declarations, callback) -> } for k, v of params - toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}", k], + toClear = device.set(sessionContext.deviceData, ['Downloads', "#{index}", k], timestamp, {object: [timestamp, 0], writable: [timestamp, v.writable], value: [timestamp, v.value]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', "#{index}", '*'], + toClear = device.set(sessionContext.deviceData, ['Downloads', "#{index}", '*'], timestamp, null, toClear) ) - sessionData.syncState.downloadsToCreate.clear() + sessionContext.syncState.downloadsToCreate.clear() - sessionData.syncState.downloadsToDelete.forEach((instance) -> - toClear = device.set(sessionData.deviceData, instance, timestamp, null, toClear) - sessionData.syncState.downloadsValues.forEach((v, p) -> + sessionContext.syncState.downloadsToDelete.forEach((instance) -> + toClear = device.set(sessionContext.deviceData, instance, timestamp, null, toClear) + sessionContext.syncState.downloadsValues.forEach((v, p) -> if p[1] == instance[1] - sessionData.syncState.downloadsValues.delete(p) + sessionContext.syncState.downloadsValues.delete(p) ) ) - sessionData.syncState.downloadsToDelete.clear() + sessionContext.syncState.downloadsToDelete.clear() - sessionData.syncState.downloadsValues.forEach((v, p) -> - if attrs = sessionData.deviceData.attributes.get(p) + sessionContext.syncState.downloadsValues.forEach((v, p) -> + if attrs = sessionContext.deviceData.attributes.get(p) if attrs.writable?[1] and attrs.value? v = device.sanitizeParameterValue([v, attrs.value[1][1]]) if v[0] != attrs.value[1][0] - toClear = device.set(sessionData.deviceData, p, timestamp, {value: [timestamp, v]}, toClear) + toClear = device.set(sessionContext.deviceData, p, timestamp, {value: [timestamp, v]}, toClear) ) if toClear - return clear(sessionData, toClear, (err) -> + return clear(sessionContext, toClear, (err) -> return callback(err) if err - if sessionData.deviceData.changes.has('prerequisite') - delete sessionData.syncState - device.clearTrackers(sessionData.deviceData, 'prerequisite') - return rpcRequest(sessionData, null, callback) + if sessionContext.deviceData.changes.has('prerequisite') + delete sessionContext.syncState + device.clearTrackers(sessionContext.deviceData, 'prerequisite') + return rpcRequest(sessionContext, null, callback) ) - if sessionData.deviceData.changes.has('prerequisite') - delete sessionData.syncState - device.clearTrackers(sessionData.deviceData, 'prerequisite') - return rpcRequest(sessionData, null, callback) + if sessionContext.deviceData.changes.has('prerequisite') + delete sessionContext.syncState + device.clearTrackers(sessionContext.deviceData, 'prerequisite') + return rpcRequest(sessionContext, null, callback) - provisions = generateSetVirtualParameterProvisions(sessionData, sessionData.syncState.virtualParameterDeclarations[inception]) + provisions = generateSetVirtualParameterProvisions(sessionContext, sessionContext.syncState.virtualParameterDeclarations[inception]) if not provisions - sessionData.rpcRequest = generateSetRpcRequest(sessionData) + sessionContext.rpcRequest = generateSetRpcRequest(sessionContext) if provisions - sessionData.virtualParameters.push(provisions) - sessionData.revisions.push(sessionData.revisions[inception]) - return rpcRequest(sessionData, null, callback) + sessionContext.virtualParameters.push(provisions) + sessionContext.revisions.push(sessionContext.revisions[inception]) + return rpcRequest(sessionContext, null, callback) - if sessionData.rpcRequest - return callback(null, null, generateRpcId(sessionData), sessionData.rpcRequest) + if sessionContext.rpcRequest + return callback(null, null, generateRpcId(sessionContext), sessionContext.rpcRequest) - ++ sessionData.revisions[inception] - sessionData.declarations.pop() + ++ sessionContext.revisions[inception] + sessionContext.declarations.pop() - while sessionData.doneProvisions & (2 << (sessionData.virtualParameters.length - 1)) - doneProvisions = sessionData.virtualParameters.pop() - sessionData.revisions.pop() - sessionData.declarations.pop() - sessionData.doneProvisions &= (2 << sessionData.virtualParameters.length) - 1 - rev = sessionData.revisions[sessionData.revisions.length - 1] - sessionData.deviceData.timestamps.collapse(rev + 1) - sessionData.deviceData.attributes.collapse(rev + 1) - for k of sessionData.extensionsCache + while sessionContext.doneProvisions & (2 << (sessionContext.virtualParameters.length - 1)) + doneProvisions = sessionContext.virtualParameters.pop() + sessionContext.revisions.pop() + sessionContext.declarations.pop() + sessionContext.doneProvisions &= (2 << sessionContext.virtualParameters.length) - 1 + rev = sessionContext.revisions[sessionContext.revisions.length - 1] + sessionContext.deviceData.timestamps.collapse(rev + 1) + sessionContext.deviceData.attributes.collapse(rev + 1) + for k of sessionContext.extensionsCache if rev < Number(k.split(':', 1)[0]) - delete sessionData.extensionsCache[k] + delete sessionContext.extensionsCache[k] - sessionData.syncState.virtualParameterDeclarations.length = sessionData.declarations.length + sessionContext.syncState.virtualParameterDeclarations.length = sessionContext.declarations.length - return rpcRequest(sessionData, null, callback) + return rpcRequest(sessionContext, null, callback) # Simple algorithm to estimate GPN count in a set of patterns used @@ -822,14 +828,14 @@ estimateGpnCount = (patterns) -> return Math.round(count) -generateGetRpcRequest = (sessionData) -> - if not (syncState = sessionData.syncState)? +generateGetRpcRequest = (sessionContext) -> + if not (syncState = sessionContext.syncState)? return iter = syncState.refreshAttributes.exist.values() while (path = iter.next().value) found = false - for p in sessionData.deviceData.paths.find(path, false, true, 99) + for p in sessionContext.deviceData.paths.find(path, false, true, 99) if syncState.refreshAttributes.value.has(p) or syncState.refreshAttributes.object.has(p) or syncState.refreshAttributes.writable.has(p) or @@ -837,7 +843,7 @@ generateGetRpcRequest = (sessionData) -> found = true break if not found - p = sessionData.deviceData.paths.add(path.slice(0, -1)) + p = sessionContext.deviceData.paths.add(path.slice(0, -1)) syncState.gpn.add(p) f = 1 << p.length syncState.gpnPatterns.set(p, f | syncState.gpnPatterns.get(p)) @@ -846,7 +852,7 @@ generateGetRpcRequest = (sessionData) -> iter = syncState.refreshAttributes.object.values() while (path = iter.next().value) found = false - for p in sessionData.deviceData.paths.find(path, false, true, 99) + for p in sessionContext.deviceData.paths.find(path, false, true, 99) if syncState.refreshAttributes.value.has(p) or (p.length > path.length and (syncState.refreshAttributes.object.has(p) or @@ -854,7 +860,7 @@ generateGetRpcRequest = (sessionData) -> found = true break if not found - p = sessionData.deviceData.paths.add(path.slice(0, -1)) + p = sessionContext.deviceData.paths.add(path.slice(0, -1)) syncState.gpn.add(p) f = 1 << p.length syncState.gpnPatterns.set(p, f | syncState.gpnPatterns.get(p)) @@ -862,7 +868,7 @@ generateGetRpcRequest = (sessionData) -> iter = syncState.refreshAttributes.writable.values() while (path = iter.next().value) - p = sessionData.deviceData.paths.add(path.slice(0, -1)) + p = sessionContext.deviceData.paths.add(path.slice(0, -1)) syncState.gpn.add(p) f = 1 << p.length syncState.gpnPatterns.set(p, f | syncState.gpnPatterns.get(p)) @@ -870,17 +876,17 @@ generateGetRpcRequest = (sessionData) -> if syncState.gpn.size GET_PARAMETER_NAMES_DEPTH_THRESHOLD = - config.get('GET_PARAMETER_NAMES_DEPTH_THRESHOLD', sessionData.deviceId) + config.get('GET_PARAMETER_NAMES_DEPTH_THRESHOLD', sessionContext.deviceId) paths = Array.from(syncState.gpn.keys()).sort((a,b) -> b.length - a.length) path = paths.pop() - while path and not sessionData.deviceData.attributes.has(path) + while path and not sessionContext.deviceData.attributes.has(path) syncState.gpn.delete(path) path = paths.pop() if path patterns = [] - for p in sessionData.deviceData.paths.find(path, true, false, 99) + for p in sessionContext.deviceData.paths.find(path, true, false, 99) if v = syncState.gpnPatterns.get(p) patterns.push([p, (v >> path.length) << path.length]) @@ -894,59 +900,59 @@ generateGetRpcRequest = (sessionData) -> syncState.gpn.delete(path) else nextLevel = false - for p in sessionData.deviceData.paths.find(path, false, true, 99) + for p in sessionContext.deviceData.paths.find(path, false, true, 99) syncState.gpn.delete(p) return { - type: 'GetParameterNames' + name: 'GetParameterNames' parameterPath: path.concat('').join('.') nextLevel: nextLevel } if syncState.refreshAttributes.value.size TASK_PARAMETERS_BATCH_SIZE = - config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) + config.get('TASK_PARAMETERS_BATCH_SIZE', sessionContext.deviceId) parameterNames = [] iter = syncState.refreshAttributes.value.values() while (path = iter.next().value) and parameterNames.length < TASK_PARAMETERS_BATCH_SIZE syncState.refreshAttributes.value.delete(path) - if sessionData.deviceData.attributes.has(path) + if sessionContext.deviceData.attributes.has(path) parameterNames.push(path) return { - type: 'GetParameterValues' + name: 'GetParameterValues' parameterNames: (p.join('.') for p in parameterNames) } return null -generateSetRpcRequest = (sessionData) -> - if not (syncState = sessionData.syncState)? +generateSetRpcRequest = (sessionContext) -> + if not (syncState = sessionContext.syncState)? return - deviceData = sessionData.deviceData + deviceData = sessionContext.deviceData # Delete instance iter = syncState.instancesToDelete.values() while instances = iter.next().value if (instance = instances.values().next().value) and - sessionData.deviceData.attributes.has(instance) + sessionContext.deviceData.attributes.has(instance) return { - type: 'DeleteObject' + name: 'DeleteObject' objectName: instance.concat('').join('.') } # Create instance iter = syncState.instancesToCreate.entries() while pair = iter.next().value - if sessionData.deviceData.attributes.has(pair[0]) and + if sessionContext.deviceData.attributes.has(pair[0]) and instance = pair[1].values().next().value pair[1].delete(instance) return { - type: 'AddObject' + name: 'AddObject' objectName: pair[0].concat('').join('.') instanceValues: instance next: 'getInstanceKeys' @@ -954,12 +960,12 @@ generateSetRpcRequest = (sessionData) -> # Set values TASK_PARAMETERS_BATCH_SIZE = - config.get('TASK_PARAMETERS_BATCH_SIZE', sessionData.deviceId) + config.get('TASK_PARAMETERS_BATCH_SIZE', sessionContext.deviceId) parameterValues = [] syncState.spv.forEach((v, k) -> return if parameterValues.length >= TASK_PARAMETERS_BATCH_SIZE - attrs = sessionData.deviceData.attributes.get(k) + attrs = sessionContext.deviceData.attributes.get(k) if (curVal = attrs.value?[1])? and attrs.writable?[1] val = v.slice() val[1] = curVal[1] if not val[1]? @@ -972,26 +978,26 @@ generateSetRpcRequest = (sessionData) -> if parameterValues.length return { - type: 'SetParameterValues' + name: 'SetParameterValues' parameterList: ([p[0].join('.'), p[1], p[2]] for p in parameterValues) } # Reboot if syncState.reboot? - p = sessionData.deviceData.paths.get(['Reboot']) - if not (p? and sessionData.deviceData.attributes.get(p)?.value?[1][0] >= syncState.reboot) + p = sessionContext.deviceData.paths.get(['Reboot']) + if not (p? and sessionContext.deviceData.attributes.get(p)?.value?[1][0] >= syncState.reboot) delete syncState.reboot return { - type: 'Reboot' + name: 'Reboot' } # Factory reset if syncState.factoryReset? - p = sessionData.deviceData.paths.get(['FactoryReset']) - if not (p? and sessionData.deviceData.attributes.get(p)?.value?[1][0] >= syncState.factoryReset) + p = sessionContext.deviceData.paths.get(['FactoryReset']) + if not (p? and sessionContext.deviceData.attributes.get(p)?.value?[1][0] >= syncState.factoryReset) delete syncState.factoryReset return { - type: 'FactoryReset' + name: 'FactoryReset' } # Download @@ -1002,8 +1008,8 @@ generateSetRpcRequest = (sessionData) -> fileNamePath = deviceData.paths.get(pair[0].slice(0, -1).concat('FileName')) targetFileNamePath = deviceData.paths.get(pair[0].slice(0, -1).concat('TargetFileName')) return { - type: 'Download' - commandKey: generateRpcId(sessionData) + name: 'Download' + commandKey: generateRpcId(sessionContext) instance: pair[0][1] fileType: deviceData.attributes.get(fileTypePath)?.value?[1][0] fileName: deviceData.attributes.get(fileNamePath)?.value?[1][0] @@ -1013,7 +1019,7 @@ generateSetRpcRequest = (sessionData) -> return null -generateGetVirtualParameterProvisions = (sessionData, virtualParameterDeclarations) -> +generateGetVirtualParameterProvisions = (sessionContext, virtualParameterDeclarations) -> provisions = null for declaration in virtualParameterDeclarations if declaration[1] @@ -1024,11 +1030,11 @@ generateGetVirtualParameterProvisions = (sessionData, virtualParameterDeclaratio return provisions -generateSetVirtualParameterProvisions = (sessionData, virtualParameterDeclarations) -> +generateSetVirtualParameterProvisions = (sessionContext, virtualParameterDeclarations) -> provisions = null for declaration in virtualParameterDeclarations if declaration[2]?.value? - attrs = sessionData.deviceData.attributes.get(declaration[0]) + attrs = sessionContext.deviceData.attributes.get(declaration[0]) if (curVal = attrs.value?[1])? and attrs.writable?[1] val = declaration[2].value.slice() val[1] = curVal[1] if not val[1]? @@ -1042,12 +1048,12 @@ generateSetVirtualParameterProvisions = (sessionData, virtualParameterDeclaratio return provisions -processDeclarations = (sessionData, allDeclareTimestamps, allDeclareAttributeTimestamps, allDeclareAttributeValues) -> - deviceData = sessionData.deviceData - syncState = sessionData.syncState +processDeclarations = (sessionContext, allDeclareTimestamps, allDeclareAttributeTimestamps, allDeclareAttributeValues) -> + deviceData = sessionContext.deviceData + syncState = sessionContext.syncState - root = sessionData.deviceData.paths.add([]) - paths = sessionData.deviceData.paths.find([], false, true, 99) + root = sessionContext.deviceData.paths.add([]) + paths = sessionContext.deviceData.paths.find([], false, true, 99) paths.sort((a, b) -> if a.wildcard == b.wildcard return a.length - b.length @@ -1179,8 +1185,8 @@ processDeclarations = (sessionData, allDeclareTimestamps, allDeclareAttributeTim return virtualParameterDeclarations -loadPath = (sessionData, path, depth) -> - return true if sessionData.new +loadPath = (sessionContext, path, depth) -> + return true if sessionContext.new depth ?= (1 << path.length) - 1 # Trim trailing wildcards @@ -1189,32 +1195,32 @@ loadPath = (sessionData, path, depth) -> path = path.slice(0, trimWildcard) if trimWildcard < path.length for i in [0..path.length] by 1 - for sup in sessionData.deviceData.paths.find(path.slice(0, i), true, false) - v = sessionData.deviceData.loaded.get(sup) - depth &= depth ^ sessionData.deviceData.loaded.get(sup) + for sup in sessionContext.deviceData.paths.find(path.slice(0, i), true, false) + v = sessionContext.deviceData.loaded.get(sup) + depth &= depth ^ sessionContext.deviceData.loaded.get(sup) return true if depth == 0 - sessionData.toLoad ?= [] - sessionData.toLoad.push([path, depth]) + sessionContext.toLoad ?= [] + sessionContext.toLoad.push([path, depth]) return false -processInstances = (sessionData, parent, parameters, keys, minInstances, maxInstances) -> +processInstances = (sessionContext, parent, parameters, keys, minInstances, maxInstances) -> if parent[0] == 'Downloads' return if parent.length != 1 - instancesToDelete = sessionData.syncState.downloadsToDelete - instancesToCreate = sessionData.syncState.downloadsToCreate + instancesToDelete = sessionContext.syncState.downloadsToDelete + instancesToCreate = sessionContext.syncState.downloadsToCreate else - instancesToDelete = sessionData.syncState.instancesToDelete.get(parent) + instancesToDelete = sessionContext.syncState.instancesToDelete.get(parent) if not instancesToDelete? instancesToDelete = new Set() - sessionData.syncState.instancesToDelete.set(parent, instancesToDelete) + sessionContext.syncState.instancesToDelete.set(parent, instancesToDelete) - instancesToCreate = sessionData.syncState.instancesToCreate.get(parent) + instancesToCreate = sessionContext.syncState.instancesToCreate.get(parent) if not instancesToCreate? instancesToCreate = new InstanceSet() - sessionData.syncState.instancesToCreate.set(parent, instancesToCreate) + sessionContext.syncState.instancesToCreate.set(parent, instancesToCreate) counter = 0 for p in parameters @@ -1244,29 +1250,29 @@ processInstances = (sessionData, parent, parameters, keys, minInstances, maxInst return -clear = (sessionData, toClear, callback) -> +clear = (sessionContext, toClear, callback) -> return callback() if not toClear?.length - MAX_DEPTH = config.get('MAX_DEPTH', sessionData.deviceId) + MAX_DEPTH = config.get('MAX_DEPTH', sessionContext.deviceId) toClear.forEach((c) -> - loadPath(sessionData, c[0], ((1 << c[0].length) - 1) ^ ((1 << MAX_DEPTH) - 1)) + loadPath(sessionContext, c[0], ((1 << c[0].length) - 1) ^ ((1 << MAX_DEPTH) - 1)) ) - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err toClear.forEach((c) -> - device.clear(sessionData.deviceData, c[0], c[1], c[2]) + device.clear(sessionContext.deviceData, c[0], c[1], c[2]) ) return callback() ) -commitVirtualParameter = (sessionData, provision, update, toClear) -> +commitVirtualParameter = (sessionContext, provision, update, toClear) -> attributes = {} - timestamp = sessionData.timestamp + sessionData.iteration + timestamp = sessionContext.timestamp + sessionContext.iteration timestamp += 1 if provision[2]? if update.writable? @@ -1279,18 +1285,18 @@ commitVirtualParameter = (sessionData, provision, update, toClear) -> else if provision[1]?.value? or provision[2]?.value? throw new Error('Virtual parameter must provide declared attributes') - return device.set(sessionData.deviceData, ['VirtualParameters', provision[0]], timestamp, attributes, toClear) + return device.set(sessionContext.deviceData, ['VirtualParameters', provision[0]], timestamp, attributes, toClear) -rpcResponse = (sessionData, id, rpcRes, callback) -> - if id != generateRpcId(sessionData) +rpcResponse = (sessionContext, id, rpcRes, callback) -> + if id != generateRpcId(sessionContext) return callback(new Error('Request ID not recognized')) - ++ sessionData.rpcCount + ++ sessionContext.rpcCount - rpcReq = sessionData.rpcRequest + rpcReq = sessionContext.rpcRequest if not rpcReq.next? - sessionData.rpcRequest = null + sessionContext.rpcRequest = null else if rpcReq.next == 'getInstanceKeys' instanceNumber = rpcRes.instanceNumber parameterNames = [] @@ -1301,10 +1307,10 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> instanceValues[n] = v if parameterNames.length == 0 - sessionData.rpcRequest = null + sessionContext.rpcRequest = null else - sessionData.rpcRequest = { - type: 'GetParameterValues' + sessionContext.rpcRequest = { + name: 'GetParameterValues' parameterNames: parameterNames next: 'setInstanceKeys' instanceValues: instanceValues @@ -1317,31 +1323,31 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> [p[0]].concat(device.sanitizeParameterValue([rpcReq.instanceValues[p[0]], p[2]]))) if not parameterList.length - sessionData.rpcRequest = null + sessionContext.rpcRequest = null else - sessionData.rpcRequest = { - type: 'SetParameterValues' + sessionContext.rpcRequest = { + name: 'SetParameterValues' parameterList: parameterList } - timestamp = sessionData.timestamp + sessionData.iteration + timestamp = sessionContext.timestamp + sessionContext.iteration - revision = sessionData.revisions[sessionData.revisions.length - 1] + 1 - sessionData.deviceData.timestamps.revision = revision - sessionData.deviceData.attributes.revision = revision + revision = sessionContext.revisions[sessionContext.revisions.length - 1] + 1 + sessionContext.deviceData.timestamps.revision = revision + sessionContext.deviceData.attributes.revision = revision toClear = null - switch rpcRes.type + switch rpcRes.name when 'GetParameterValuesResponse' - return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterValues' + return callback(new Error('Response name does not match request name')) if rpcReq.name isnt 'GetParameterValues' for p in rpcRes.parameterList - toClear = device.set(sessionData.deviceData, common.parsePath(p[0]), timestamp, + toClear = device.set(sessionContext.deviceData, common.parsePath(p[0]), timestamp, {object: [timestamp, 0], value: [timestamp, p.slice(1)]}, toClear) when 'GetParameterNamesResponse' - return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'GetParameterNames' + return callback(new Error('Response name does not match request name')) if rpcReq.name isnt 'GetParameterNames' if rpcReq.parameterPath.endsWith('.') root = common.parsePath(rpcReq.parameterPath.slice(0, -1)) @@ -1389,73 +1395,73 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> ) if rpcReq.nextLevel - loadPath(sessionData, root, (1 << (root.length + 1)) - 1) + loadPath(sessionContext, root, (1 << (root.length + 1)) - 1) else - loadPath(sessionData, root, (1 << config.get('MAX_DEPTH', sessionData.deviceId)) - 1) + loadPath(sessionContext, root, (1 << config.get('MAX_DEPTH', sessionContext.deviceId)) - 1) - loadParameters(sessionData, (err) -> + loadParameters(sessionContext, (err) -> return callback(err) if err if root.length == 0 for n in ['DeviceID', 'Events', 'Tags', 'Reboot', 'FactoryReset', 'VirtualParameters'] - if p = sessionData.deviceData.paths.get([n]) - if sessionData.deviceData.attributes.has(p) - sessionData.deviceData.timestamps.set(p, timestamp) + if p = sessionContext.deviceData.paths.get([n]) + if sessionContext.deviceData.attributes.has(p) + sessionContext.deviceData.timestamps.set(p, timestamp) for p in params - toClear = device.set(sessionData.deviceData, p[0], p[1], p[2], toClear) + toClear = device.set(sessionContext.deviceData, p[0], p[1], p[2], toClear) - clear(sessionData, toClear, callback) + clear(sessionContext, toClear, callback) ) return when 'SetParameterValuesResponse' - return callback(new Error('Response type does not match request type')) if rpcReq.type isnt 'SetParameterValues' + return callback(new Error('Response name does not match request name')) if rpcReq.name isnt 'SetParameterValues' for p in rpcReq.parameterList - toClear = device.set(sessionData.deviceData, common.parsePath(p[0]), timestamp + 1, + toClear = device.set(sessionContext.deviceData, common.parsePath(p[0]), timestamp + 1, {object: [timestamp + 1, 0], writable: [timestamp + 1, 1], value: [timestamp + 1, p.slice(1)]}, toClear) when 'AddObjectResponse' - toClear = device.set(sessionData.deviceData, common.parsePath(rpcReq.objectName + rpcRes.instanceNumber), + toClear = device.set(sessionContext.deviceData, common.parsePath(rpcReq.objectName + rpcRes.instanceNumber), timestamp + 1, {object: [timestamp + 1, 1]}, toClear) when 'DeleteObjectResponse' - toClear = device.set(sessionData.deviceData, common.parsePath(rpcReq.objectName.slice(0, -1)), + toClear = device.set(sessionContext.deviceData, common.parsePath(rpcReq.objectName.slice(0, -1)), timestamp + 1, null, toClear) when 'RebootResponse' - toClear = device.set(sessionData.deviceData, common.parsePath('Reboot'), + toClear = device.set(sessionContext.deviceData, common.parsePath('Reboot'), timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) when 'FactoryResetResponse' - toClear = device.set(sessionData.deviceData, common.parsePath('FactoryReset'), + toClear = device.set(sessionContext.deviceData, common.parsePath('FactoryReset'), timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) when 'DownloadResponse' - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'Download'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'Download'], timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) if rpcRes.status == 0 - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastDownload'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'LastDownload'], timestamp + 1, {value: [timestamp + 1, [timestamp + 1, 'xsd:dateTime']]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastFileType'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'LastFileType'], timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastFileName'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'LastFileName'], timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'LastTargetFileName'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'LastTargetFileName'], timestamp + 1, {value: [timestamp + 1, [rpcReq.fileType, 'xsd:dateTime']]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'StartTime'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'StartTime'], timestamp + 1, {value: [timestamp + 1, [+rpcRes.startTime, 'xsd:dateTime']]}, toClear) - toClear = device.set(sessionData.deviceData, ['Downloads', rpcReq.instance, 'CompleteTime'], + toClear = device.set(sessionContext.deviceData, ['Downloads', rpcReq.instance, 'CompleteTime'], timestamp + 1, {value: [timestamp + 1, [+rpcRes.completeTime, 'xsd:dateTime']]}, toClear) else operation = { - type: 'Download' - timestamp: sessionData.timestamp - provisions: sessionData.provisions - channels: sessionData.channels + name: 'Download' + timestamp: sessionContext.timestamp + provisions: sessionContext.provisions + channels: sessionContext.channels retries: {} args: { instance: rpcReq.instance @@ -1465,21 +1471,21 @@ rpcResponse = (sessionData, id, rpcRes, callback) -> } } - for channel of sessionData.channels - if sessionData.retries[channel]? - operation.retries[channel] = sessionData.retries[channel] + for channel of sessionContext.channels + if sessionContext.retries[channel]? + operation.retries[channel] = sessionContext.retries[channel] - sessionData.operations[rpcReq.commandKey] = operation - sessionData.operationsTouched ?= {} - sessionData.operationsTouched[rpcReq.commandKey] = 1 + sessionContext.operations[rpcReq.commandKey] = operation + sessionContext.operationsTouched ?= {} + sessionContext.operationsTouched[rpcReq.commandKey] = 1 else - return callback(new Error('Response type not recognized')) + return callback(new Error('Response name not recognized')) - return clear(sessionData, toClear, callback) + return clear(sessionContext, toClear, callback) -rpcFault = (sessionData, id, faultResponse, callback) -> +rpcFault = (sessionContext, id, faultResponse, callback) -> # TODO Consider handling invalid parameter faults by automatically refreshing # relevant data model portions @@ -1487,33 +1493,31 @@ rpcFault = (sessionData, id, faultResponse, callback) -> code: "cwmp.#{faultResponse.detail.faultCode}" message: faultResponse.detail.faultString detail: faultResponse.detail - timestamp: sessionData.timestamp + timestamp: sessionContext.timestamp } - delete sessionData.syncState + delete sessionContext.syncState return callback(null, fault) -deserialize = (sessionDataString, callback) -> +deserialize = (sessionContextString, callback) -> cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> return callback(err) if err - sessionData = JSON.parse(sessionDataString) + sessionContext = JSON.parse(sessionContextString) - if sessionData.presetsHash? and sessionData.presetsHash != hash + if sessionContext.presetsHash? and sessionContext.presetsHash != hash return callback(new Error('Preset hash mismatch')) - sessionData.cache = { - provisions: provisions - virtualParameters: virtualParameters - } + provisionsCache.set(sessionContext, provisions) + virtualParametersCache.set(sessionContext, virtualParameters) - for d in sessionData.declarations + for d in sessionContext.declarations common.addPathMeta(d[0]) deviceData = initDeviceData() - for r in sessionData.deviceData + for r in sessionContext.deviceData path = deviceData.paths.add(r[0]) if r[1] @@ -1528,85 +1532,82 @@ deserialize = (sessionDataString, callback) -> if r[4] deviceData.attributes.setRevisions(path, r[4]) - sessionData.deviceData = deviceData + sessionContext.deviceData = deviceData - return callback(null, sessionData) + return callback(null, sessionContext) ) -serialize = (sessionData, callback) -> +serialize = (sessionContext, callback) -> deviceData = [] - for path in sessionData.deviceData.paths.find([], false, false, 99) + for path in sessionContext.deviceData.paths.find([], false, false, 99) e = [path] - e[1] = sessionData.deviceData.loaded.get(path) || 0 - e[2] = sessionData.deviceData.trackers.get(path) || null - e[3] = sessionData.deviceData.timestamps.getRevisions(path) || null - e[4] = sessionData.deviceData.attributes.getRevisions(path) || null + e[1] = sessionContext.deviceData.loaded.get(path) || 0 + e[2] = sessionContext.deviceData.trackers.get(path) || null + e[3] = sessionContext.deviceData.timestamps.getRevisions(path) || null + e[4] = sessionContext.deviceData.attributes.getRevisions(path) || null deviceData.push(e) - oldDeviceData = sessionData.deviceData - oldSyncState = sessionData.syncState - oldToLoad = sessionData.toLoad - oldCache = sessionData.cache + oldDeviceData = sessionContext.deviceData + oldSyncState = sessionContext.syncState + oldToLoad = sessionContext.toLoad - sessionData.deviceData = deviceData - delete sessionData.syncState - delete sessionData.toLoad - delete sessionData.cache + sessionContext.deviceData = deviceData + delete sessionContext.syncState + delete sessionContext.toLoad - sessionDataString = JSON.stringify(sessionData) + sessionContextString = JSON.stringify(sessionContext) - sessionData.deviceData = oldDeviceData - sessionData.syncState = oldSyncState - sessionData.toLoad = oldToLoad - sessionData.cache = oldCache + sessionContext.deviceData = oldDeviceData + sessionContext.syncState = oldSyncState + sessionContext.toLoad = oldToLoad - return callback(null, sessionDataString) + return callback(null, sessionContextString) -end = (sessionData, callback) -> +end = (sessionContext, callback) -> counter = 3 counter += 2 - db.saveDevice(sessionData.deviceId, sessionData.deviceData, sessionData.new, (err) -> + db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, (err) -> if err callback(err) if counter & 1 return counter = 0 - return callback(null, sessionData.new) if (counter -= 2) == 1 + return callback(null, sessionContext.new) if (counter -= 2) == 1 ) counter += 2 - db.redisClient.del("session_#{sessionData.id}", (err) -> + db.redisClient.del("session_#{sessionContext.id}", (err) -> if err callback(err) if counter & 1 return counter = 0 - return callback(null, sessionData.new) if (counter -= 2) == 1 + return callback(null, sessionContext.new) if (counter -= 2) == 1 ) - for k of sessionData.operationsTouched + for k of sessionContext.operationsTouched counter += 2 - if sessionData.operations[k]? - db.saveOperation(sessionData.deviceId, k, sessionData.operations[k], (err) -> + if sessionContext.operations[k]? + db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) -> if err callback(err) if counter & 1 return counter = 0 - return callback(null, sessionData.new) if (counter -= 2) == 1 + return callback(null, sessionContext.new) if (counter -= 2) == 1 ) else - db.deleteOperation(sessionData.deviceId, k, (err) -> + db.deleteOperation(sessionContext.deviceId, k, (err) -> if err callback(err) if counter & 1 return counter = 0 - return callback(null, sessionData.new) if (counter -= 2) == 1 + return callback(null, sessionContext.new) if (counter -= 2) == 1 ) - callback(null, sessionData.new) if (counter -= 2) == 1 + callback(null, sessionContext.new) if (counter -= 2) == 1 exports.init = init exports.timeoutOperations = timeoutOperations diff --git a/lib/soap.coffee b/lib/soap.coffee deleted file mode 100644 index 7a5503d..0000000 --- a/lib/soap.coffee +++ /dev/null @@ -1,464 +0,0 @@ -### -# Copyright 2013, 2014 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 . -# -# This file incorporates work covered by the following copyright and -# permission notice: -# -# Copyright 2013 Fanoos Telecom -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files (the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions: -# -# The above copyright notice and this permission notice shall be included in -# all copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -# THE SOFTWARE. -### - -config = require './config' -libxmljs = require 'libxmljs' - -SERVER_NAME = "GenieACS/#{require('../package.json').version}" - -NAMESPACES = { - '1.0' : { - 'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/', - 'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/', - 'xsd' : 'http://www.w3.org/2001/XMLSchema', - 'xsi' : 'http://www.w3.org/2001/XMLSchema-instance', - 'cwmp' : 'urn:dslforum-org:cwmp-1-0' - }, - '1.1' : { - 'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/', - 'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/', - 'xsd' : 'http://www.w3.org/2001/XMLSchema', - 'xsi' : 'http://www.w3.org/2001/XMLSchema-instance', - 'cwmp' : 'urn:dslforum-org:cwmp-1-1' - }, - '1.2' : { - 'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/', - 'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/', - 'xsd' : 'http://www.w3.org/2001/XMLSchema', - 'xsi' : 'http://www.w3.org/2001/XMLSchema-instance', - 'cwmp' : 'urn:dslforum-org:cwmp-1-2' - }, - '1.3' : { - 'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/', - 'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/', - 'xsd' : 'http://www.w3.org/2001/XMLSchema', - 'xsi' : 'http://www.w3.org/2001/XMLSchema-instance', - 'cwmp' : 'urn:dslforum-org:cwmp-1-2' - }, - '1.4' : { - 'soap-enc' : 'http://schemas.xmlsoap.org/soap/encoding/', - 'soap-env' : 'http://schemas.xmlsoap.org/soap/envelope/', - 'xsd' : 'http://www.w3.org/2001/XMLSchema', - 'xsi' : 'http://www.w3.org/2001/XMLSchema-instance', - 'cwmp' : 'urn:dslforum-org:cwmp-1-3' - } -} - -LIBXMLJS_PARSE_OPTIONS = { - 'nocdata' : true, - 'noent' : true, - 'recover' : config.get('XML_RECOVER'), - 'ignore_enc' : config.get('XML_IGNORE_ENC') -} - -LIBXMLJS_SAVE_OPTIONS = { - 'format' : config.get('XML_FORMAT'), - 'declaration' : not config.get('XML_NO_DECL'), - 'selfCloseEmpty' : not config.get('XML_NO_EMPTY') -} - -XML_IGNORE_NAMESPACE = config.get('XML_IGNORE_NAMESPACE') - - -# Workaround for devices that don't use correct namespace declarations or prefixes -if XML_IGNORE_NAMESPACE - libxmljs.Element.prototype.__find = libxmljs.Element.prototype.find - libxmljs.Element.prototype.find = (xpath, namespaces) -> - # Modify xpath queries to work regardless of element's namespace - p = xpath.replace(/([^\/:]+:)?([^\/]+)(\/|$)/g, '$2$3') \ - .replace(/([a-zA-Z0-9_-]+)([^\/]*)(\/|$)/g, "*[local-name()='$1']$2$3") - libxmljs.Element.prototype.__find.call(this, p, namespaces) - - -event = (xml) -> - e.text().trim() for e in xml.find('EventStruct/EventCode') - - -parameterInfoList = (xml) -> - infoList = [] - for e in xml.find('ParameterInfoStruct') - name = e.get('Name').text().trim() - writable = Boolean(JSON.parse(e.get('Writable').text())) - infoList.push([name, writable]) - return infoList - - -parameterValueList = (xml) -> - valueList = [] - for e in xml.find('ParameterValueStruct') - valueType = e.get('Value').attr('type').value().trim() - name = e.get('Name').text().trim() - value = e.get('Value').text().trim() - try - switch valueType - when 'xsd:boolean' - value = Boolean(JSON.parse(value)) - when 'xsd:unsignedInt' - value = JSON.parse(value) - when 'xsd:int' - value = JSON.parse(value) - catch err - value = null - valueList.push([name, value, valueType]) - return valueList - - -cpeGetParameterNames = (xml, methodRequest) -> - el = xml.node('cwmp:GetParameterNames') - el.node('ParameterPath').text(methodRequest.parameterPath) - el.node('NextLevel').text(+methodRequest.nextLevel) - - -cpeGetParameterNamesResponse = (xml) -> - {parameterList : parameterInfoList(xml.get('ParameterList'))} - - -cpeGetParameterValues = (xml, methodRequest) -> - el = xml.node('cwmp:GetParameterValues').node('ParameterNames') - el.attr({'soap-enc:arrayType' : "xsd:string[#{methodRequest.parameterNames.length}]"}) - for p in methodRequest.parameterNames - el.node('string').text(p) - - -cpeGetParameterValuesResponse = (xml) -> - {parameterList : parameterValueList(xml.get('ParameterList'))} - - -cpeSetParameterValues = (xml, methodRequest) -> - el = xml.node('cwmp:SetParameterValues') - paramList = el.node('ParameterList') - paramList.attr({'soap-enc:arrayType' : "cwmp:ParameterValueStruct[#{methodRequest.parameterList.length}]"}) - for i in methodRequest.parameterList - pvs = paramList.node('ParameterValueStruct') - pvs.node('Name').text(i[0]) - v = pvs.node('Value') - v.text(i[1]) - v.attr({'xsi:type' : i[2]}) if i[2]? - el.node('ParameterKey').text(methodRequest.parameterKey ? '') - - -cpeSetParameterValuesResponse = (xml) -> - {status : JSON.parse(xml.get('Status').text())} - - -cpeAddObject = (xml, methodRequest) -> - el = xml.node('cwmp:AddObject') - el.node('ObjectName').text(methodRequest.objectName) - el.node('ParameterKey').text(methodRequest.parameterKey ? '') - - -cpeAddObjectResponse = (xml) -> - { - instanceNumber : parseInt(xml.get('InstanceNumber').text()), - status : parseInt(xml.get('Status').text()) - } - - -cpeDeleteObject = (xml, methodRequest) -> - el = xml.node('cwmp:DeleteObject') - el.node('ObjectName').text(methodRequest.objectName) - el.node('ParameterKey').text(methodRequest.parameterKey ? '') - - -cpeDeleteObjectResponse = (xml) -> - {status : parseInt(xml.get('Status').text())} - - -cpeReboot = (xml, methodRequest) -> - el = xml.node('cwmp:Reboot') - el.node('CommandKey').text(methodRequest.commandKey ? '') - - -cpeRebootResponse = (xml) -> - {} - - -cpeFactoryReset = (xml, methodRequest) -> - el = xml.node('cwmp:FactoryReset') - - -cpeFactoryResetResponse = (xml, methodRequest) -> - {} - - -cpeDownload = (xml, methodRequest) -> - el = xml.node('cwmp:Download') - el.node('CommandKey').text(methodRequest.commandKey ? '') - el.node('FileType').text(methodRequest.fileType) - el.node('URL').text(methodRequest.url) - el.node('Username').text(methodRequest.username ? '') - el.node('Password').text(methodRequest.password ? '') - el.node('FileSize').text(methodRequest.fileSize ? '0') - el.node('TargetFileName').text(methodRequest.targetFileName ? '') - el.node('DelaySeconds').text(methodRequest.delaySeconds ? '0') - el.node('SuccessURL').text(methodRequest.successUrl ? '') - el.node('FailureURL').text(methodRequest.failureUrl ? '') - - -cpeDownloadResponse = (xml) -> - res = { - status : parseInt(xml.get('Status').text()), - } - - if res.status == 0 - res.startTime = Date.parse(xml.get('StartTime').text()) - res.completeTime = Date.parse(xml.get('CompleteTime').text()) - - return res - - -acsInform = (xml) -> - { - parameterList : parameterValueList(xml.get('ParameterList')), - deviceId : { - 'Manufacturer' : xml.get('DeviceId/Manufacturer').text(), - 'OUI' : xml.get('DeviceId/OUI').text(), - 'ProductClass' : xml.get('DeviceId/ProductClass').text(), - 'SerialNumber' : xml.get('DeviceId/SerialNumber').text() - } - event : event(xml.get('Event')), - retryCount : parseInt(xml.get('RetryCount').text()) - } - - -acsInformResponse = (xml, methodResponse) -> - xml.node('cwmp:InformResponse').node('MaxEnvelopes').text(1) - - -acsGetRPCMethods = (xml) -> - {} - - -acsGetRPCMethodsResponse = (xml, methodResponse) -> - el = xml.node('cwmp:GetRPCMethodsResponse').node('MethodList') - el.attr({'soap-enc:arrayType' : "xsd:string[#{methodResponse.methodList.length}]"}) - for m in methodResponse.methodList - el.node('string').text(m) - - -acsTransferComplete = (xml) -> - { - commandKey : xml.get('CommandKey').text(), - faultStruct : faultStruct(xml.get('FaultStruct')), - startTime : Date.parse(xml.get('StartTime').text()), - completeTime : Date.parse(xml.get('CompleteTime').text()) - } - - -acsTransferCompleteResponse = (xml, methodResponse) -> - xml.node('cwmp:TransferCompleteResponse').text('') - - -acsRequestDownload = (xml) -> - # TODO FileTypeArg - { - fileType : xml.get('FileType').text() - } - - -acsRequestDownloadResponse = (xml, methodResponse) -> - xml.node('cwmp:RequestDownloadResponse').text('') - - -faultStruct = (xml) -> - f = { - faultCode: xml.get('FaultCode').text(), - faultString: xml.get('FaultString').text() - } - - for e in xml.find('SetParameterValuesFault') - f.setParameterValuesFault ?= [] - f.setParameterValuesFault.push({ - parameterName: e.get('ParameterName').text(), - faultCode: e.get('FaultCode').text(), - faultString: e.get('FaultString').text() - }) - - return f - - -fault = (xml, cwmpVersion) -> - { - faultCode: xml.get('faultcode').text(), - faultString: xml.get('faultstring').text(), - detail: faultStruct(xml.get('detail/cwmp:Fault', NAMESPACES[cwmpVersion])) - } - - -exports.request = (httpRequest, cwmpVersion) -> - cwmpRequest = {cwmpVersion : cwmpVersion} - - data = httpRequest.getBody() - - if data.length > 0 - xml = libxmljs.parseXml(data, LIBXMLJS_PARSE_OPTIONS) - - if not cwmpRequest.cwmpVersion? - # cwmpVersion not passed, thus it's an inform request - methodElement = xml.get('/soap-env:Envelope/soap-env:Body/*', NAMESPACES['1.0']) - switch methodElement.namespace().href() - when 'urn:dslforum-org:cwmp-1-0' - cwmpRequest.cwmpVersion = '1.0' - when 'urn:dslforum-org:cwmp-1-1' - cwmpRequest.cwmpVersion = '1.1' - when 'urn:dslforum-org:cwmp-1-2' - cwmpRequest.sessionTimeout = try xml.get('/soap-env:Envelope/soap-env:Header/cwmp:sessionTimeout', NAMESPACES['1.2']).text() catch then null - if cwmpRequest.sessionTimeout? - cwmpRequest.cwmpVersion = '1.3' - else - cwmpRequest.cwmpVersion = '1.2' - when 'urn:dslforum-org:cwmp-1-3' - cwmpRequest.cwmpVersion = '1.4' - else - throw new Error('Unrecognized CWMP version') - - cwmpRequest.sessionTimeout ?= try xml.get('/soap-env:Envelope/soap-env:Header/cwmp:sessionTimeout', NAMESPACES[cwmpRequest.cwmpVersion]).text() catch then null - else - methodElement = xml.get('/soap-env:Envelope/soap-env:Body/cwmp:*', NAMESPACES[cwmpRequest.cwmpVersion]) - - cwmpRequest.id = try xml.get('/soap-env:Envelope/soap-env:Header/cwmp:ID', NAMESPACES[cwmpRequest.cwmpVersion]).text() catch then null - - if methodElement? and not (XML_IGNORE_NAMESPACE and methodElement.name() is 'Fault') - switch methodElement.name() - when 'Inform' - cwmpRequest.methodRequest = acsInform(methodElement) - cwmpRequest.methodRequest.type = 'Inform' - when 'GetRPCMethods' - cwmpRequest.methodRequest = acsGetRPCMethods(methodElement) - cwmpRequest.methodRequest.type = 'GetRPCMethods' - when 'TransferComplete' - cwmpRequest.methodRequest = acsTransferComplete(methodElement) - cwmpRequest.methodRequest.type = 'TransferComplete' - when 'RequestDownload' - cwmpRequest.methodRequest = acsRequestDownload(methodElement) - cwmpRequest.methodRequest.type = 'RequestDownload' - when 'GetParameterNamesResponse' - cwmpRequest.methodResponse = cpeGetParameterNamesResponse(methodElement) - cwmpRequest.methodResponse.type = 'GetParameterNamesResponse' - when 'GetParameterValuesResponse' - cwmpRequest.methodResponse = cpeGetParameterValuesResponse(methodElement) - cwmpRequest.methodResponse.type = 'GetParameterValuesResponse' - when 'SetParameterValuesResponse' - cwmpRequest.methodResponse = cpeSetParameterValuesResponse(methodElement) - cwmpRequest.methodResponse.type = 'SetParameterValuesResponse' - when 'AddObjectResponse' - cwmpRequest.methodResponse = cpeAddObjectResponse(methodElement) - cwmpRequest.methodResponse.type = 'AddObjectResponse' - when 'DeleteObjectResponse' - cwmpRequest.methodResponse = cpeDeleteObjectResponse(methodElement) - cwmpRequest.methodResponse.type = 'DeleteObjectResponse' - when 'RebootResponse' - cwmpRequest.methodResponse = cpeRebootResponse(methodElement) - cwmpRequest.methodResponse.type = 'RebootResponse' - when 'FactoryResetResponse' - cwmpRequest.methodResponse = cpeFactoryResetResponse(methodElement) - cwmpRequest.methodResponse.type = 'FactoryResetResponse' - when 'DownloadResponse' - cwmpRequest.methodResponse = cpeDownloadResponse(methodElement) - cwmpRequest.methodResponse.type = 'DownloadResponse' - else - throw Error('8000 Method not supported ' + methodElement.name()) - else - faultElement = xml.get('/soap-env:Envelope/soap-env:Body/soap-env:Fault', NAMESPACES[cwmpRequest.cwmpVersion]) - cwmpRequest.fault = fault(faultElement, cwmpRequest.cwmpVersion) - - return cwmpRequest - - -createSoapEnv = (cwmpVersion) -> - xml = libxmljs.Document() - env = xml.node('soap-env:Envelope') - for prefix, url of NAMESPACES[cwmpVersion] - env.defineNamespace prefix, url - return env - - -exports.response = (cwmpResponse) -> - headers = { - 'Server' : SERVER_NAME, - 'SOAPServer' : SERVER_NAME, - } - - if cwmpResponse?.methodResponse? - env = createSoapEnv(cwmpResponse.cwmpVersion) - header = env.node('soap-env:Header') - header.node('cwmp:ID').attr({'soap-env:mustUnderstand' : 1}).text(cwmpResponse.id) - body = env.node('soap-env:Body') - switch cwmpResponse.methodResponse.type - when 'InformResponse' - acsInformResponse(body, cwmpResponse.methodResponse) - when 'GetRPCMethodsResponse' - acsGetRPCMethodsResponse(body, cwmpResponse.methodResponse) - when 'TransferCompleteResponse' - acsTransferCompleteResponse(body, cwmpResponse.methodResponse) - when 'RequestDownloadResponse' - acsRequestDownloadResponse(body, cwmpResponse.methodResponse) - else - throw Error("Unknown method response type #{cwmpResponse.methodResponse.type}") - else if cwmpResponse?.methodRequest? - env = createSoapEnv(cwmpResponse.cwmpVersion) - header = env.node('soap-env:Header') - header.node('cwmp:ID').attr({'soap-env:mustUnderstand' : 1}).text(cwmpResponse.id) - body = env.node('soap-env:Body') - switch cwmpResponse.methodRequest.type - when 'GetParameterNames' - cpeGetParameterNames(body, cwmpResponse.methodRequest) - when 'GetParameterValues' - cpeGetParameterValues(body, cwmpResponse.methodRequest) - when 'SetParameterValues' - cpeSetParameterValues(body, cwmpResponse.methodRequest) - when 'AddObject' - cpeAddObject(body, cwmpResponse.methodRequest) - when 'DeleteObject' - cpeDeleteObject(body, cwmpResponse.methodRequest) - when 'Reboot' - cpeReboot(body, cwmpResponse.methodRequest) - when 'FactoryReset' - cpeFactoryReset(body, cwmpResponse.methodRequest) - when 'Download' - cpeDownload(body, cwmpResponse.methodRequest) - else - throw Error("Unknown method request #{cwmpResponse.methodRequest.type}") - - if env? - headers['Content-Type'] = 'text/xml; charset="utf-8"' - return {code: 200, headers: headers, data: new Buffer(env.doc().toString(LIBXMLJS_SAVE_OPTIONS))} - else - return {code: 204, headers: headers, data: new Buffer(0)} diff --git a/lib/soap.js b/lib/soap.js new file mode 100644 index 0000000..64e18a1 --- /dev/null +++ b/lib/soap.js @@ -0,0 +1,559 @@ +/** + * Copyright 2013-2017 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 libxmljs = require("libxmljs"); + +const config = require("./config"); + +const SERVER_NAME = `GenieACS/${require("../package.json").version}`; + +const NAMESPACES = { + "1.0": { + "soap-enc": "http://schemas.xmlsoap.org/soap/encoding/", + "soap-env": "http://schemas.xmlsoap.org/soap/envelope/", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "cwmp": "urn:dslforum-org:cwmp-1-0" + }, + "1.1": { + "soap-enc": "http://schemas.xmlsoap.org/soap/encoding/", + "soap-env": "http://schemas.xmlsoap.org/soap/envelope/", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "cwmp": "urn:dslforum-org:cwmp-1-1" + }, + "1.2": { + "soap-enc": "http://schemas.xmlsoap.org/soap/encoding/", + "soap-env": "http://schemas.xmlsoap.org/soap/envelope/", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "cwmp": "urn:dslforum-org:cwmp-1-2" + }, + "1.3": { + "soap-enc": "http://schemas.xmlsoap.org/soap/encoding/", + "soap-env": "http://schemas.xmlsoap.org/soap/envelope/", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "cwmp": "urn:dslforum-org:cwmp-1-2" + }, + "1.4": { + "soap-enc": "http://schemas.xmlsoap.org/soap/encoding/", + "soap-env": "http://schemas.xmlsoap.org/soap/envelope/", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "cwmp": "urn:dslforum-org:cwmp-1-3" + } +}; + +const LIBXMLJS_PARSE_OPTIONS = { + "nocdata": true, + "noent": true, + "recover": config.get("XML_RECOVER"), + "ignore_enc": config.get("XML_IGNORE_ENC") +}; + +const LIBXMLJS_SAVE_OPTIONS = { + "format": config.get("XML_FORMAT"), + "declaration": !config.get("XML_NO_DECL"), + "selfCloseEmpty": !config.get("XML_NO_EMPTY") +}; + +const XML_IGNORE_NAMESPACE = config.get("XML_IGNORE_NAMESPACE"); + +if (XML_IGNORE_NAMESPACE) { + libxmljs.Element.prototype.__find = libxmljs.Element.prototype.find; + libxmljs.Element.prototype.find = function(xpath, namespaces) { + const p = xpath.replace(/([^\/:]+:)?([^\/]+)(\/|$)/g, "$2$3").replace(/([a-zA-Z0-9_-]+)([^\/]*)(\/|$)/g, "*[local-name()=\"$1\"]$2$3"); + return libxmljs.Element.prototype.__find.call(this, p, namespaces); + }; +} + + +function parseBool(v) { + v = "" + v; + if (v === "true" || v === "TRUE" || v === "True" || v === "1") + return true; + else if (v === "false" || v === "FALSE" || v === "False" || v === "0") + return false; + else + throw new Error("Invalid boolean value"); +} + + +function event(xml) { + return xml.find("EventStruct/EventCode").map(e => e.text().trim()); +} + + +function parameterInfoList(xml) { + return xml.find("ParameterInfoStruct").map( + e => [e.get("Name").text(), parseBool(e.get("Writable").text())] + ); +} + + +function parameterValueList(xml) { + return xml.find("ParameterValueStruct").map(function(e) { + const valueElement = e.get("Value"); + const valueType = valueElement.attr("type").value(); + let value = valueElement.text(); + if (valueType === "xsd:boolean") + value = parseBool(value); + else if (valueType === "xsd:int" || valueType === "xsd:unsignedInt") + value = parseInt(value); + else if (valueType === "xsd:dateTime") + value = Date.parse(value); + + return [e.get("Name").text(), value, valueType]; + }); +} + + +function GetParameterNames(xml, methodRequest) { + const el = xml.node("cwmp:GetParameterNames"); + el.node("ParameterPath").text(methodRequest.parameterPath); + el.node("NextLevel").text(+methodRequest.nextLevel); +} + + +function GetParameterNamesResponse(xml) { + return { + name: "GetParameterNamesResponse", + parameterList: parameterInfoList(xml.get("ParameterList")) + }; +} + + +function GetParameterValues(xml, methodRequest) { + const el = xml.node("cwmp:GetParameterValues").node("ParameterNames"); + el.attr({ + "soap-enc:arrayType": `xsd:string[${methodRequest.parameterNames.length}]` + }); + + for (const p of methodRequest.parameterNames) + el.node("string").text(p); +} + + +function GetParameterValuesResponse(xml) { + return { + name: "GetParameterValuesResponse", + parameterList: parameterValueList(xml.get("ParameterList")) + }; +} + + +function SetParameterValues(xml, methodRequest) { + const el = xml.node("cwmp:SetParameterValues"); + const paramList = el.node("ParameterList"); + paramList.attr({ + "soap-enc:arrayType": `cwmp:ParameterValueStruct[${methodRequest.parameterList.length}]` + }); + + for (const p of methodRequest.parameterList) { + const pvs = paramList.node("ParameterValueStruct"); + pvs.node("Name").text(p[0]); + pvs.node("Value").attr({"xsi:type": p[2]}).text(p[1]); + } + + el.node("ParameterKey").text(methodRequest.parameterKey || ""); +} + + +function SetParameterValuesResponse(xml) { + return { + name: "SetParameterValuesResponse", + status: parseInt(xml.get("Status").text()) + }; +} + + +function AddObject(xml, methodRequest) { + const el = xml.node("cwmp:AddObject"); + el.node("ObjectName").text(methodRequest.objectName); + el.node("ParameterKey").text(methodRequest.parameterKey || ""); +} + + +function AddObjectResponse(xml) { + return { + name: "AddObjectResponse", + instanceNumber: parseInt(xml.get("InstanceNumber").text()), + status: parseInt(xml.get("Status").text()) + }; +} + + +function DeleteObject(xml, methodRequest) { + const el = xml.node("cwmp:DeleteObject"); + el.node("ObjectName").text(methodRequest.objectName); + el.node("ParameterKey").text(methodRequest.parameterKey || ""); +} + + +function DeleteObjectResponse(xml) { + return { + name: "DeleteObjectResponse", + status: parseInt(xml.get("Status").text()) + }; +} + + +function Reboot(xml, methodRequest) { + const el = xml.node("cwmp:Reboot"); + el.node("CommandKey").text(methodRequest.commandKey || ""); +} + + +function RebootResponse(xml) { + return { + name: "RebootResponse" + }; +} + + +function FactoryReset(xml, methodRequest) { + xml.node("cwmp:FactoryReset"); +} + + +function FactoryResetResponse(xml, methodRequest) { + return { + name: "FactoryResetResponse" + }; +} + + +function Download(xml, methodRequest) { + const el = xml.node("cwmp:Download"); + el.node("CommandKey").text(methodRequest.commandKey || ""); + el.node("FileType").text(methodRequest.fileType); + el.node("URL").text(methodRequest.url); + el.node("Username").text(methodRequest.username || ""); + el.node("Password").text(methodRequest.password || ""); + el.node("FileSize").text(methodRequest.fileSize || "0"); + el.node("TargetFileName").text(methodRequest.targetFileName || ""); + el.node("DelaySeconds").text(methodRequest.delaySeconds || "0"); + el.node("SuccessURL").text(methodRequest.successUrl || ""); + el.node("FailureURL").text(methodRequest.failureUrl || ""); +} + + +function DownloadResponse(xml) { + const res = { + name: "DownloadResponse", + status: parseInt(xml.get("Status").text()) + }; + + if (res.status === 0) { + res.startTime = Date.parse(xml.get("StartTime").text()); + res.completeTime = Date.parse(xml.get("CompleteTime").text()); + } + + return res; +} + + +function Inform(xml) { + return { + name: "Inform", + parameterList: parameterValueList(xml.get("ParameterList")), + deviceId: { + "Manufacturer": xml.get("DeviceId/Manufacturer").text(), + "OUI": xml.get("DeviceId/OUI").text(), + "ProductClass": xml.get("DeviceId/ProductClass").text(), + "SerialNumber": xml.get("DeviceId/SerialNumber").text() + }, + event: event(xml.get("Event")), + retryCount: parseInt(xml.get("RetryCount").text()) + }; +} + + +function InformResponse(xml, methodResponse) { + xml.node("cwmp:InformResponse").node("MaxEnvelopes").text(1); +} + + +function GetRPCMethods(xml) { + return {name: "GetRPCMethods"}; +} + + +function GetRPCMethodsResponse(xml, methodResponse) { + const el = xml.node("cwmp:GetRPCMethodsResponse").node("MethodList"); + el.attr({ + "soap-enc:arrayType": `xsd:string[${methodResponse.methodList.length}]` + }); + + for (const m of methodResponse.methodList) + el.node("string").text(m); +} + + +function TransferComplete(xml) { + return { + name: "TransferComplete", + commandKey: xml.get("CommandKey").text(), + faultStruct: faultStruct(xml.get("FaultStruct")), + startTime: Date.parse(xml.get("StartTime").text()), + completeTime: Date.parse(xml.get("CompleteTime").text()) + }; +} + + +function TransferCompleteResponse(xml, methodResponse) { + xml.node("cwmp:TransferCompleteResponse").text(""); +} + + +function RequestDownload(xml) { + return { + name: "RequestDownload", + fileType: xml.get("FileType").text() + }; +} + + +function RequestDownloadResponse(xml, methodResponse) { + xml.node("cwmp:RequestDownloadResponse").text(""); +} + + +function faultStruct(xml) { + const f = { + faultCode: xml.get("FaultCode").text(), + faultString: xml.get("FaultString").text() + }; + + for (const e of xml.find("SetParameterValuesFault")) { + if (!f.setParameterValuesFault) + f.setParameterValuesFault = []; + + f.setParameterValuesFault.push({ + parameterName: e.get("ParameterName").text(), + faultCode: e.get("FaultCode").text(), + faultString: e.get("FaultString").text() + }); + } + + return f; +} + + +function fault(xml, cwmpVersion) { + return { + faultCode: xml.get("faultcode").text(), + faultString: xml.get("faultstring").text(), + detail: faultStruct(xml.get("detail/cwmp:Fault", NAMESPACES[cwmpVersion])) + }; +} + + +function request(httpRequest, cwmpVersion) { + const rpc = { + cwmpVersion: cwmpVersion + }; + + const data = httpRequest.getBody(); + + if (!data.length) + return rpc; + + const xml = libxmljs.parseXml(data, LIBXMLJS_PARSE_OPTIONS); + + let methodElement; + + if (rpc.cwmpVersion) { + methodElement = xml.get("/soap-env:Envelope/soap-env:Body/cwmp:*", NAMESPACES[rpc.cwmpVersion]); + } + else { + // cwmpVersion not passed, thus it's an inform request + methodElement = xml.get("/soap-env:Envelope/soap-env:Body/*", NAMESPACES["1.0"]); + switch (methodElement.namespace().href()) { + case "urn:dslforum-org:cwmp-1-0": + rpc.cwmpVersion = "1.0"; + break; + case "urn:dslforum-org:cwmp-1-1": + rpc.cwmpVersion = "1.1"; + break; + case "urn:dslforum-org:cwmp-1-2": + const timeoutElement = xml.get("/soap-env:Envelope/soap-env:Header/cwmp:sessionTimeout", NAMESPACES["1.2"]); + if (timeoutElement) { + rpc.sessionTimeout = parseInt(timeoutElement.text()); + rpc.cwmpVersion = "1.3"; + } + else { + rpc.cwmpVersion = "1.2"; + } + break; + case "urn:dslforum-org:cwmp-1-3": + rpc.cwmpVersion = "1.4"; + break; + default: + throw new Error("Unrecognized CWMP version"); + } + + if (!rpc.sessionTimeout) { + const timeoutElement = xml.get("/soap-env:Envelope/soap-env:Header/cwmp:sessionTimeout", NAMESPACES[rpc.cwmpVersion]); + if (timeoutElement) + rpc.sessionTimeout = parseInt(timeoutElement.text()); + } + } + + rpc.id = xml.get("/soap-env:Envelope/soap-env:Header/cwmp:ID", NAMESPACES[rpc.cwmpVersion]).text(); + + if (methodElement && !(XML_IGNORE_NAMESPACE && methodElement.name() === "Fault")) { + switch (methodElement.name()) { + case "Inform": + rpc.cpeRequest = Inform(methodElement); + break; + case "GetRPCMethods": + rpc.cpeRequest = GetRPCMethods(methodElement); + break; + case "TransferComplete": + rpc.cpeRequest = TransferComplete(methodElement); + break; + case "RequestDownload": + rpc.cpeRequest = RequestDownload(methodElement); + break; + case "GetParameterNamesResponse": + rpc.cpeResponse = GetParameterNamesResponse(methodElement); + break; + case "GetParameterValuesResponse": + rpc.cpeResponse = GetParameterValuesResponse(methodElement); + break; + case "SetParameterValuesResponse": + rpc.cpeResponse = SetParameterValuesResponse(methodElement); + break; + case "AddObjectResponse": + rpc.cpeResponse = AddObjectResponse(methodElement); + break; + case "DeleteObjectResponse": + rpc.cpeResponse = DeleteObjectResponse(methodElement); + break; + case "RebootResponse": + rpc.cpeResponse = RebootResponse(methodElement); + break; + case "FactoryResetResponse": + rpc.cpeResponse = FactoryResetResponse(methodElement); + break; + case "DownloadResponse": + rpc.cpeResponse = DownloadResponse(methodElement); + break; + default: + throw new Error(`8000 Method not supported ${methodElement.name()}`); + } + } + else { + const faultElement = xml.get("/soap-env:Envelope/soap-env:Body/soap-env:Fault", NAMESPACES[rpc.cwmpVersion]); + rpc.cpeFault = fault(faultElement, rpc.cwmpVersion); + } + + return rpc; +} + + +function createSoapEnv(cwmpVersion) { + const xml = libxmljs.Document(); + const env = xml.node("soap-env:Envelope"); + + for (const prefix of Object.keys(NAMESPACES[cwmpVersion])) + env.defineNamespace(prefix, NAMESPACES[cwmpVersion][prefix]); + + return env; +} + + +function response(rpc) { + const headers = { + "Server": SERVER_NAME, + "SOAPServer": SERVER_NAME + }; + + if (!rpc) + return {code: 204, headers: headers, data: new Buffer(0)}; + + const env = createSoapEnv(rpc.cwmpVersion); + const header = env.node("soap-env:Header"); + header.node("cwmp:ID").attr({ + "soap-env:mustUnderstand": 1 + }).text(rpc.id); + + const body = env.node("soap-env:Body"); + + if (rpc.acsResponse) { + switch (rpc.acsResponse.name) { + case "InformResponse": + InformResponse(body, rpc.acsResponse); + break; + case "GetRPCMethodsResponse": + GetRPCMethodsResponse(body, rpc.acsResponse); + break; + case "TransferCompleteResponse": + TransferCompleteResponse(body, rpc.acsResponse); + break; + case "RequestDownloadResponse": + RequestDownloadResponse(body, rpc.acsResponse); + break; + default: + throw new Error(`Unknown method response type ${rpc.acsResponse.name}`); + } + } + else if (rpc.acsRequest) { + switch (rpc.acsRequest.name) { + case "GetParameterNames": + GetParameterNames(body, rpc.acsRequest); + break; + case "GetParameterValues": + GetParameterValues(body, rpc.acsRequest); + break; + case "SetParameterValues": + SetParameterValues(body, rpc.acsRequest); + break; + case "AddObject": + AddObject(body, rpc.acsRequest); + break; + case "DeleteObject": + DeleteObject(body, rpc.acsRequest); + break; + case "Reboot": + Reboot(body, rpc.acsRequest); + break; + case "FactoryReset": + FactoryReset(body, rpc.acsRequest); + break; + case "Download": + Download(body, rpc.acsRequest); + break; + default: + throw new Error(`Unknown method request ${rpc.acsRequest.name}`); + } + } + + headers["Content-Type"] = "text/xml; charset=\"utf-8\""; + return { + code: 200, + headers: headers, + data: new Buffer(env.doc().toString(LIBXMLJS_SAVE_OPTIONS)) + }; +} + + +exports.request = request; +exports.response = response;