diff --git a/.gitignore b/.gitignore
index a4e333f..c8a7b48 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,5 @@
*~
lib/*.map
-lib/cwmp.js
lib/session.js
node_modules
debug
diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee
deleted file mode 100644
index 20b9734..0000000
--- a/lib/cwmp.coffee
+++ /dev/null
@@ -1,965 +0,0 @@
-###
-# Copyright 2013-2017 Zaid Abdulla
-#
-# This file is part of GenieACS.
-#
-# 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 .
-###
-
-util = require 'util'
-zlib = require 'zlib'
-crypto = require 'crypto'
-
-config = require './config'
-common = require './common'
-soap = require './soap'
-session = require './session'
-query = require './query'
-device = require './device'
-cache = require './cache'
-localCache = require './local-cache'
-db = require './db'
-logger = require './logger'
-scheduling = require './scheduling'
-
-
-MAX_CYCLES = 4
-
-MAX_CONCURRENT_REQUESTS = config.get('MAX_CONCURRENT_REQUESTS')
-stats = {
- concurrentRequests: 0
- totalRequests: 0
- droppedRequests: 0
- initiatedSessions: 0
-}
-
-throwError = (err, httpResponse) ->
- if httpResponse
- currentSessions.delete(httpResponse.connection)
- httpResponse.writeHead(500, {'Connection' : 'close'})
- httpResponse.end("#{err.name}: #{err.message}")
- stats.concurrentRequests -= 1
-
- throw err
-
-
-writeResponse = (sessionContext, res, close) ->
- # Close connection after last request in session
- res.headers['Connection'] = 'close' if close
-
- 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/#{sessionContext.deviceId}.dump", dump, (err) ->
- throwError(err) if err
- )
-
- finish = () ->
- stats.concurrentRequests -= 1
- if sessionContext.httpRequest.connection.destroyed
- logger.accessError({
- sessionContext: sessionContext
- message: 'Connection dropped'
- })
- else if close
- endSession(sessionContext, (err, isNew) ->
- return throwError(err) if err
-
- if isNew
- logger.accessInfo({
- sessionContext: sessionContext
- message: 'New device registered'
- })
- )
- else
- sessionContext.lastActivity = Date.now()
- currentSessions.set(sessionContext.httpRequest.connection, sessionContext)
-
- # respond using the same content-encoding as the request
- 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
- when 'deflate'
- res.headers['Content-Encoding'] = 'deflate'
- compress = zlib.deflate
-
- if compress?
- compress(res.data, (err, data) ->
- return throwError(err, sessionContext.httpResponse) if err
- res.headers['Content-Length'] = data.length
- sessionContext.httpResponse.writeHead(res.code, res.headers)
- sessionContext.httpResponse.end(data)
- finish()
- )
- else
- res.headers['Content-Length'] = res.data.length
- sessionContext.httpResponse.writeHead(res.code, res.headers)
- sessionContext.httpResponse.end(res.data)
- finish()
-
-
-recordFault = (sessionContext, fault, provisions, channels) ->
- if not provisions
- provisions = sessionContext.provisions
- channels = sessionContext.channels
-
- faults = sessionContext.faults
-
- for channel of channels
- provs = sessionContext.faults[channel]?.provisions or []
- faults[channel] = Object.assign({provisions: provs}, fault)
- if channel.startsWith('task_')
- taskId = channel.slice(5)
- for t in sessionContext.tasks
- faults[channel].expiry = t.expiry if t._id == taskId and t.expiry
-
- if sessionContext.retries[channel]?
- ++ sessionContext.retries[channel]
- else
- sessionContext.retries[channel] = 0
- if Object.keys(channels).length != 1
- faults[channel].retryNow = true
-
- if channels[channel] == 0
- faults[channel].precondition = true
-
- sessionContext.faultsTouched ?= {}
- sessionContext.faultsTouched[channel] = true
-
- logger.accessWarn({
- sessionContext: sessionContext
- message: 'Channel has faulted'
- fault: fault
- channel: channel
- retries: sessionContext.retries[channel]
- })
-
- for provision, i in provisions
- for channel of channels
- if (channels[channel] >> i) & 1
- faults[channel].provisions.push(provision)
-
- for channel of channels
- provs = faults[channel].provisions
- faults[channel].provisions = []
- appendProvisions(faults[channel].provisions, provs)
-
- session.clearProvisions(sessionContext)
- return
-
-
-inform = (sessionContext, rpc) ->
- session.inform(sessionContext, rpc.cpeRequest, (err, acsResponse) ->
- return throwError(err, sessionContext.httpResponse) if err
- res = soap.response({
- id: rpc.id,
- acsResponse: acsResponse,
- cwmpVersion : sessionContext.cwmpVersion
- })
-
- if !!cookiesPath = config.get('COOKIES_PATH', sessionContext.deviceId)
- res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}; Path=#{cookiesPath}"
- else
- res.headers['Set-Cookie'] = "session=#{sessionContext.sessionId}"
-
- writeResponse(sessionContext, res)
- )
-
-
-transferComplete = (sessionContext, rpc) ->
- session.transferComplete(sessionContext, rpc.cpeRequest, (err, acsResponse, operation, fault) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if not operation?
- logger.accessWarn({
- sessionContext: sessionContext
- message: 'Unrecognized command key'
- rpc: rpc
- })
-
- if fault
- for k, v of operation.retries
- sessionContext.retries[k] = v
-
- recordFault(sessionContext, fault, operation.provisions, operation.channels)
-
- res = soap.response({
- id : rpc.id,
- acsResponse : acsResponse,
- cwmpVersion : sessionContext.cwmpVersion
- })
- writeResponse(sessionContext, res)
- )
-
-
-# Append providions and remove duplicates
-appendProvisions = (original, toAppend) ->
- modified = false
- stringified = new WeakMap()
- for p, i in original
- stringified.set(p, JSON.stringify(p))
-
- for p, i in toAppend by -1
- s = JSON.stringify(p)
- for pp, j in original by -1
- ss = stringified.get(pp)
- if s == ss
- if (not p) or j >= original.length - (toAppend.length - i)
- p = null
- else
- original.splice(j, 1)
- modified = true
- if p
- original.splice(original.length - (toAppend.length - i) + 1, 0, p)
- stringified.set(p, s)
- modified = true
-
- return modified
-
-
-applyPresets = (sessionContext) ->
- localCache.getPresets((err, presetsHash, presets) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- # Filter presets based on existing faults
- blackList = {}
- whiteList = null
- whiteListProvisions = null
-
- RETRY_DELAY = config.get('RETRY_DELAY', sessionContext.deviceId)
-
- if sessionContext.faults?
- for channel, fault of sessionContext.faults
- if fault.retryNow
- retryTimestamp = 0
- else
- retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, sessionContext.retries[channel])) * 1000
-
- if retryTimestamp <= sessionContext.timestamp
- whiteList = channel
- whiteListProvisions = fault.provisions
- break
-
- blackList[channel] = if fault.precondition then 1 else 2
-
- sessionContext.deviceData.timestamps.revision = 1
- sessionContext.deviceData.attributes.revision = 1
-
- deviceEvents = {}
- 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 = {}
- filteredPresets = []
-
- for preset in presets
- if whiteList?
- continue if preset.channel != whiteList
- else if blackList[preset.channel] == 1
- continue
-
- eventsMatch = true
- for k, v of preset.events
- if (!v) != (!deviceEvents[k.replace(' ', '_')])
- eventsMatch = false
- break
-
- continue if not eventsMatch
-
- if preset.schedule
- continue if not preset.schedule.schedule
- r = scheduling.cron(sessionContext.timestamp, preset.schedule.schedule)
- if not (r[0] + preset.schedule.duration > sessionContext.timestamp)
- continue
-
- filteredPresets.push(preset)
- for k of preset.precondition
- sessionContext.channels[preset.channel] = 0
- p = k.split(/([^a-zA-Z0-9\-\_\.].*)/, 1)[0]
- parameters[p] = common.parsePath(p)
-
- declarations = []
- for k, v of parameters
- declarations.push([v, 1, {value: 1}])
-
- session.rpcRequest(sessionContext, declarations, (err, fault, id, acsRequest) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if fault
- recordFault(sessionContext, fault)
- session.clearProvisions(sessionContext)
- return applyPresets(sessionContext)
-
- if acsRequest
- return sendAcsRequest(sessionContext, id, acsRequest)
-
- session.clearProvisions(sessionContext)
-
- for k, v of parameters
- 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(sessionContext, whiteList, whiteListProvisions)
-
- appendProvisionsToFaults = {}
- for p in filteredPresets
- if query.testFilter(parameters, p.precondition)
- if blackList[p.channel] == 2
- appendProvisionsToFaults[p.channel] =
- (appendProvisionsToFaults[p.channel] or []).concat(p.provisions)
- else
- session.addProvisions(sessionContext, p.channel, p.provisions)
-
- for channel, provisions of appendProvisionsToFaults
- if appendProvisions(sessionContext.faults[channel].provisions, provisions)
- sessionContext.faultsTouched ?= {}
- sessionContext.faultsTouched[channel] = true
-
- # Don't increment when processing individual channels (e.g. after fault)
- sessionContext.presetCycles = (sessionContext.presetCycles or 0) + 1 if not whiteList?
-
- if sessionContext.presetCycles > MAX_CYCLES
- fault = {
- code: 'preset_loop'
- message: 'The presets are stuck in an endless configuration loop'
- timestamp: sessionContext.timestamp
- }
- recordFault(sessionContext, fault)
- # No need to save retryNow
- for c of sessionContext.faults
- delete sessionContext.faults[c].retryNow
- session.clearProvisions(sessionContext)
- return sendAcsRequest(sessionContext)
-
- sessionContext.deviceData.timestamps.dirty = 0;
- sessionContext.deviceData.attributes.dirty = 0;
-
- session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if fault
- recordFault(sessionContext, fault)
- session.clearProvisions(sessionContext)
- return applyPresets(sessionContext)
-
- 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(sessionContext)
-
- if sessionContext.deviceData.timestamps.dirty > 1 or
- sessionContext.deviceData.attributes.dirty > 1
- return applyPresets(sessionContext)
-
- sendAcsRequest(sessionContext, id, acsRequest)
- )
- )
- )
-
-
-nextRpc = (sessionContext) ->
- session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if fault
- recordFault(sessionContext, fault)
- session.clearProvisions(sessionContext)
- return nextRpc(sessionContext)
-
- if acsRequest
- return sendAcsRequest(sessionContext, id, acsRequest)
-
- 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)
- sessionContext.doneTasks ?= []
- sessionContext.doneTasks.push(taskId)
- for t, j in sessionContext.tasks
- if t._id == taskId
- sessionContext.tasks.splice(j, 1)
- break
-
- session.clearProvisions(sessionContext)
-
- # Clear expired tasks
- sessionContext.tasks = sessionContext.tasks.filter((task) ->
- return true if not (task.expiry <= sessionContext.timestamp)
-
- logger.accessInfo({
- sessionContext: sessionContext
- message: 'Task expired'
- task: task
- })
- sessionContext.doneTasks ?= []
- sessionContext.doneTasks.push(task._id)
- channel = "task_#{task._id}"
- if channel of sessionContext.faults
- delete sessionContext.faults[channel]
- sessionContext.faultsTouched ?= {}
- sessionContext.faultsTouched[channel] = true
-
- return false
- )
-
- task = sessionContext.tasks.find((task) -> !sessionContext.faults["task_#{task._id}"])
-
- return applyPresets(sessionContext) if not task
-
- switch task.name
- when 'getParameterValues'
- # Set channel in case params array is empty
- sessionContext.channels["task_#{task._id}"] = 0
- for p in task.parameterNames
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['refresh', p]])
- when 'setParameterValues'
- # Set channel in case params array is empty
- sessionContext.channels["task_#{task._id}"] = 0
- for p in task.parameterValues
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['value', p[0], p[1]]])
- when 'refreshObject'
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['refresh', task.objectName]])
- when 'reboot'
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['reboot']])
- when 'factoryReset'
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['reset']])
- when 'download'
- 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(sessionContext, "task_#{task._id}",
- [['instances', "#{task.objectName}.[#{alias}]", "+1"]]);
- when 'deleteObject'
- session.addProvisions(sessionContext, "task_#{task._id}",
- [['instances', task.objectName, 0]]);
- else
- return throwError(new Error('Task name not recognized'), sessionContext.httpResponse)
-
- return nextRpc(sessionContext)
- )
-
-
-endSession = (sessionContext, callback) ->
- saveCache = sessionContext.cacheUntil?
-
- counter = 3
-
- counter += 2
- db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, sessionContext.timestamp, (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
-
- for k of sessionContext.operationsTouched
- counter += 2
- saveCache = true
- 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, sessionContext.new) if (counter -= 2) == 1
- )
- else
- db.deleteOperation(sessionContext.deviceId, k, (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
-
- if sessionContext.doneTasks?.length
- counter += 2
- saveCache = true
- db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
-
- for k of sessionContext.faultsTouched
- counter += 2
- saveCache = true
- if sessionContext.faults[k]
- sessionContext.faults[k].retries = sessionContext.retries[k]
- db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
- else
- db.deleteFault(sessionContext.deviceId, k, (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
-
- if saveCache
- counter += 2
- cacheDueTasksAndFaultsAndOperations(sessionContext.deviceId, sessionContext.tasks, sessionContext.faults, sessionContext.operations, sessionContext.cacheUntil, (err) ->
- if err
- callback(err) if counter & 1
- return counter = 0
-
- return callback(null, sessionContext.new) if (counter -= 2) == 1
- )
-
- callback(null, sessionContext.new) if (counter -= 2) == 1
-
-
-sendAcsRequest = (sessionContext, id, acsRequest) ->
- if not acsRequest?
- return writeResponse(sessionContext, soap.response(null), true)
-
- if acsRequest.name is 'Download'
- if not acsRequest.url?
- FS_PORT = config.get('FS_PORT')
- FS_HOSTNAME = config.get('FS_HOSTNAME')
- FS_SSL = config.get('FS_SSL')
- acsRequest.url = if FS_SSL then 'https://' else 'http://'
- acsRequest.url += FS_HOSTNAME
- acsRequest.url += ":#{FS_PORT}" if FS_PORT != 80
- acsRequest.url += "/#{encodeURI(acsRequest.fileName)}"
-
- if not acsRequest.fileSize?
- return localCache.getFiles((err, hash, files) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if acsRequest.fileName of files
- acsRequest.fileSize = files[acsRequest.fileName].length
- else
- acsRequest.fileSize = 0
-
- return sendAcsRequest(sessionContext, id, acsRequest)
- )
-
- rpc = {
- id : id,
- acsRequest : acsRequest,
- cwmpVersion : sessionContext.cwmpVersion
- }
-
- logger.accessInfo({
- sessionContext: sessionContext
- message: 'ACS request'
- rpc: rpc
- })
-
- res = soap.response(rpc)
-
- writeResponse(sessionContext, res)
-
-
-getSession = (connection, sessionId, callback) ->
- sessionContext = currentSessions.get(connection)
-
- if sessionContext
- currentSessions.delete(connection)
- return callback(null, sessionContext)
-
- return callback() if not sessionId
-
- setTimeout(() ->
- cache.pop("session_#{sessionId}", (err, sessionContextString) ->
- return callback(err) if err or not sessionContextString
- session.deserialize(sessionContextString, callback)
- )
- , 100)
-
-
-currentSessions = new WeakMap()
-
-# When socket closes, store active sessions in cache
-onConnection = (socket) ->
- # The property remoteAddress may be undefined after the connection is
- # closed, unless we read it at least once (caching?)
- socket.remoteAddress
-
- socket.on('close', () ->
- sessionContext = currentSessions.get(socket)
- return if not sessionContext
- currentSessions.delete(socket)
- now = Date.now()
-
- lastActivity = sessionContext.lastActivity
- timeoutMsg = logger.flatten({
- sessionContext: sessionContext
- message: 'Session timeout'
- sessionTimestamp: sessionContext.timestamp
- })
-
- timeout = (sessionContext.lastActivity + sessionContext.timeout * 1000) - now
- if timeout <= 0
- return logger.accessError(timeoutMsg)
-
- setTimeout(() ->
- cache.get("session_#{sessionContext.sessionId}", (err, sessionContextString) ->
- return throwError(err) if err
- return if not sessionContextString
- session.deserialize(sessionContextString, (err, sessionContext) ->
- return throwError(err) if err
- if sessionContext.lastActivity == lastActivity
- logger.accessError(timeoutMsg)
- )
- )
- , timeout + 1000).unref()
-
- session.serialize(sessionContext, (err, sessionContextString) ->
- return throwError(err) if err
- cache.set("session_#{sessionContext.sessionId}", sessionContextString, Math.ceil(timeout / 1000) + 3, (err) ->
- return throwError(err) if err
- )
- )
- )
-
-setInterval(() ->
- if stats.droppedRequests
- logger.warn({
- message: 'Worker overloaded'
- droppedRequests: stats.droppedRequests
- totalRequests: stats.totalRequests
- initiatedSessions: stats.initiatedSessions
- pid: process.pid
- })
-
- stats.totalRequests = 0
- stats.droppedRequests = 0
- stats.initiatedSessions = 0
-, 10000).unref()
-
-
-getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
- cache.get("#{deviceId}_tasks_faults_operations", (err, res) ->
- return callback(err) if err
-
- if res
- res = JSON.parse(res)
- return callback(null, res.tasks or [], res.faults or {}, res.operations or {})
-
- faults = tasks = operations = cacheUntil = null
-
- db.getFaults(deviceId, (err, _faults) ->
- if err
- callback?(err)
- return callback = null
-
- faults = _faults
- if tasks? and operations?
- return callback(null, tasks, faults, operations, cacheUntil)
- )
-
- db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) ->
- if err
- callback?(err)
- return callback = null
-
- tasks = dueTasks
- cacheUntil = nextTimestamp or 0
-
- if faults? and operations?
- return callback(null, tasks, faults, operations, cacheUntil)
- )
-
- db.getOperations(deviceId, (err, _operations) ->
- if err
- callback?(err)
- return callback = null
-
- operations = _operations
-
- if faults? and tasks?
- return callback(null, tasks, faults, operations, cacheUntil)
- )
-
- if faults? and tasks? and operations?
- return callback(null, tasks, faults, operations, cacheUntil)
- )
-
-
-cacheDueTasksAndFaultsAndOperations = (deviceId, tasks, faults, operations, cacheUntil, callback) ->
- v = {}
- v.tasks = tasks if tasks.length
- v.faults = faults if Object.keys(faults).length
- v.operations = operations if Object.keys(operations).length
-
- if cacheUntil
- ttl = Math.trunc((Date.now() - cacheUntil) / 1000)
- else
- ttl = config.get('MAX_CACHE_TTL', deviceId)
-
- cache.set("#{deviceId}_tasks_faults_operations", JSON.stringify(v), ttl, callback)
-
-
-processRequest = (sessionContext, rpc) ->
- if rpc.cpeRequest?
- if rpc.cpeRequest.name is 'Inform'
- logger.accessInfo({
- sessionContext: sessionContext
- message: 'Inform'
- rpc: rpc
- })
- 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']}
- cwmpVersion : sessionContext.cwmpVersion
- })
- writeResponse(sessionContext, res)
- else
- 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 rpc.cpeFault
- logger.accessWarn({
- sessionContext: sessionContext
- message: 'CPE fault'
- rpc:rpc
- })
- session.rpcFault(sessionContext, rpc.id, rpc.cpeFault, (err, fault) ->
- return throwError(err, sessionContext.httpResponse) if err
-
- if fault
- recordFault(sessionContext, fault)
- session.clearProvisions(sessionContext)
- nextRpc(sessionContext)
- )
- else # CPE sent empty response
- 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
- sessionContext.retries[k] = v
-
- recordFault(sessionContext, fault, operations[i].provisions, operations[i].channels)
-
- nextRpc(sessionContext)
- )
-
-
-listener = (httpRequest, httpResponse) ->
- stats.totalRequests += 1
- if httpRequest.method != 'POST'
- httpResponse.writeHead 405, {'Allow': 'POST', 'Connection': 'close'}
- httpResponse.end('405 Method Not Allowed')
- return
-
- sessionId = null
- # Separation by comma is important as some devices don't comform to standard
- COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g
- while match = COOKIE_REGEX.exec(httpRequest.headers.cookie)
- sessionId = match[2] if match[1] == 'session'
-
- # If overloaded, ask CPE to retry in 60 seconds
- if not sessionId and stats.concurrentRequests > MAX_CONCURRENT_REQUESTS
- httpResponse.writeHead(503, {'Retry-after': 60, 'Connection': 'close'})
- httpResponse.end('503 Service Unavailable')
- stats.droppedRequests += 1
- return
-
- if httpRequest.headers['content-encoding']?
- switch httpRequest.headers['content-encoding']
- when 'gzip'
- stream = httpRequest.pipe(zlib.createGunzip())
- when 'deflate'
- stream = httpRequest.pipe(zlib.createInflate())
- else
- httpResponse.writeHead(415, {'Connection': 'close'})
- httpResponse.end('415 Unsupported Media Type')
- return
- else
- stream = httpRequest
-
- stats.concurrentRequests += 1
- httpRequest.on('aborted', () ->
- stats.concurrentRequests -= 1
- # In some cases event end can be emitted after aborted event
- httpRequest.removeAllListeners('end')
- )
-
- chunks = []
- bytes = 0
-
- stream.on('data', (chunk) ->
- chunks.push(chunk)
- bytes += chunk.length
- )
-
- stream.on('end', () ->
- body = new Buffer(bytes)
- offset = 0
- chunks.forEach((chunk) ->
- chunk.copy(body, offset, 0, chunk.length)
- offset += chunk.length
- )
-
- parsedRpc = (sessionContext, rpc, parseWarnings) ->
- for w in parseWarnings
- w.sessionContext = sessionContext
- w.rpc = rpc;
- logger.accessWarn(w)
-
- if config.get('DEBUG', sessionContext.deviceId)
- dump = "# REQUEST #{new Date(Date.now())}\n" + JSON.stringify(httpRequest.headers) + "\n#{body}\n\n"
- require('fs').appendFile("./debug/#{sessionContext.deviceId}.dump", dump, (err) ->
- return throwError(err) if err
- )
-
- processRequest(sessionContext, rpc)
-
- getSession(httpRequest.connection, sessionId, (err, sessionContext) ->
- return throwError(err, httpResponse) if err
-
- if sessionContext
- sessionContext.httpRequest = httpRequest
- sessionContext.httpResponse = httpResponse
-
- if sessionContext.sessionId != sessionId or
- sessionContext.lastActivity + sessionContext.timeout * 1000 < Date.now()
- logger.accessError({message: 'Invalid session', sessionContext: sessionContext})
- httpResponse.writeHead(400, {'Connection': 'close'})
- httpResponse.end('Invalid session')
- stats.concurrentRequests -= 1
- return
- else if stats.concurrentRequests > MAX_CONCURRENT_REQUESTS
- # Check again just in case device included old session ID from the previous session
- httpResponse.writeHead(503, {'Retry-after': 60, 'Connection': 'close'})
- httpResponse.end('503 Service Unavailable')
- stats.droppedRequests += 1
- stats.concurrentRequests -= 1
- return
-
- parseWarnings = []
- try
- rpc = soap.request(body, sessionContext?.cwmpVersion, parseWarnings)
- catch err
- logger.accessError({
- message: 'XML parse error'
- parseError: err.message.trim()
- sessionContext: sessionContext or {httpRequest: httpRequest, httpResponse: httpResponse}
- })
- httpResponse.writeHead(400, {'Connection': 'close'})
- httpResponse.end(err.message)
- stats.concurrentRequests -= 1
- return
-
- if sessionContext
- if rpc.cpeRequest?.name is 'Inform' or
- (!sessionContext.rpcRequest ^ !(rpc.cpeResponse or rpc.cpeFault))
- logger.accessError({
- message: 'Bad session state'
- sessionContext: sessionContext
- })
- httpResponse.writeHead(400, {'Connection': 'close'})
- httpResponse.end('Bad session state')
- stats.concurrentRequests -= 1
- return
-
- return parsedRpc(sessionContext, rpc, parseWarnings)
-
- if rpc.cpeRequest?.name isnt 'Inform'
- logger.accessError({
- message: 'Invalid session'
- sessionContext: sessionContext or {httpRequest: httpRequest, httpResponse: httpResponse}
- })
- httpResponse.writeHead(400, {'Connection': 'close'})
- httpResponse.end('Invalid session')
- stats.concurrentRequests -= 1
- return
-
- stats.initiatedSessions += 1
- 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
-
- sessionContext.httpRequest = httpRequest
- sessionContext.httpResponse = httpResponse
- sessionContext.sessionId = crypto.randomBytes(8).toString('hex')
- httpRequest.connection.setTimeout(sessionContext.timeout * 1000)
-
- getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations, cacheUntil) ->
- return throwError(err, httpResponse) if err
-
- sessionContext.tasks = dueTasks
- sessionContext.faults = faults
- sessionContext.retries = {}
- for k, v of faults
- sessionContext.retries[k] = v.retries
- sessionContext.operations = operations
- sessionContext.cacheUntil = cacheUntil
-
- # Delete expired faults
- for k, v of sessionContext.faults
- if v.expiry >= sessionContext.timestamp
- delete sessionContext.faults[k]
- sessionContext.faultsTouched ?= {}
- sessionContext.faultsTouched[k] = true
-
- parsedRpc(sessionContext, rpc, parseWarnings)
- )
- )
- )
- )
-
-
-exports.listener = listener
-exports.onConnection = onConnection
diff --git a/lib/cwmp.js b/lib/cwmp.js
new file mode 100644
index 0000000..735bbbe
--- /dev/null
+++ b/lib/cwmp.js
@@ -0,0 +1,1222 @@
+/**
+ * Copyright 2013-2018 Zaid Abdulla
+ *
+ * This file is part of GenieACS.
+ *
+ * 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 zlib = require("zlib");
+const crypto = require("crypto");
+const fs = require("fs");
+const config = require("./config");
+const common = require("./common");
+const soap = require("./soap");
+const session = require("./session");
+const query = require("./query");
+const device = require("./device");
+const cache = require("./cache");
+const localCache = require("./local-cache");
+const db = require("./db");
+const logger = require("./logger");
+const scheduling = require("./scheduling");
+
+const MAX_CYCLES = 4;
+const MAX_CONCURRENT_REQUESTS = config.get("MAX_CONCURRENT_REQUESTS");
+
+const stats = {
+ concurrentRequests: 0,
+ totalRequests: 0,
+ droppedRequests: 0,
+ initiatedSessions: 0
+};
+
+function throwError(err, httpResponse) {
+ if (httpResponse) {
+ currentSessions.delete(httpResponse.connection);
+ httpResponse.writeHead(500, { Connection: "close" });
+ httpResponse.end(`${err.name}: ${err.message}`);
+ stats.concurrentRequests -= 1;
+ }
+ throw err;
+}
+
+function writeResponse(sessionContext, res, close) {
+ // Close connection after last request in session
+ if (close) res.headers["Connection"] = "close";
+
+ if (config.get("DEBUG", sessionContext.deviceId)) {
+ const dump =
+ `# RESPONSE ${new Date(Date.now())}\n` +
+ JSON.stringify(res.headers) +
+ "\n" +
+ res.data +
+ "\n\n";
+ fs.appendFile(`./debug/${sessionContext.deviceId}.dump`, dump, err => {
+ if (err) return throwError(err);
+ });
+ }
+
+ function finish() {
+ stats.concurrentRequests -= 1;
+ if (sessionContext.httpRequest.connection.destroyed) {
+ logger.accessError({
+ sessionContext: sessionContext,
+ message: "Connection dropped"
+ });
+ } else if (close) {
+ endSession(sessionContext, (err, isNew) => {
+ if (err) return throwError(err);
+
+ if (isNew) {
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "New device registered"
+ });
+ }
+ });
+ } else {
+ sessionContext.lastActivity = Date.now();
+ currentSessions.set(
+ sessionContext.httpRequest.connection,
+ sessionContext
+ );
+ }
+ }
+
+ let compress;
+ // Respond using the same content-encoding as the request
+ if (
+ sessionContext.httpRequest.headers["content-encoding"] &&
+ res.data.length > 0
+ ) {
+ switch (sessionContext.httpRequest.headers["content-encoding"]) {
+ case "gzip":
+ res.headers["Content-Encoding"] = "gzip";
+ compress = zlib.gzip;
+ break;
+ case "deflate":
+ res.headers["Content-Encoding"] = "deflate";
+ compress = zlib.deflate;
+ }
+ }
+
+ if (compress) {
+ compress(res.data, (err, data) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ res.headers["Content-Length"] = data.length;
+ sessionContext.httpResponse.writeHead(res.code, res.headers);
+ sessionContext.httpResponse.end(data);
+ finish();
+ });
+ } else {
+ res.headers["Content-Length"] = res.data.length;
+ sessionContext.httpResponse.writeHead(res.code, res.headers);
+ sessionContext.httpResponse.end(res.data);
+ finish();
+ }
+}
+
+function recordFault(sessionContext, fault, provisions, channels) {
+ if (!provisions) {
+ provisions = sessionContext.provisions;
+ channels = sessionContext.channels;
+ }
+
+ const faults = sessionContext.faults;
+ for (const channel of Object.keys(channels)) {
+ const provs = sessionContext.faults[channel]
+ ? sessionContext.faults[channel].provisions
+ : [];
+ faults[channel] = Object.assign({ provisions: provs }, fault);
+ if (channel.startsWith("task_")) {
+ const taskId = channel.slice(5);
+ for (const t of sessionContext.tasks)
+ if (t._id === taskId && t.expiry) faults[channel].expiry = t.expiry;
+ }
+
+ if (sessionContext.retries[channel] != null) {
+ ++sessionContext.retries[channel];
+ } else {
+ sessionContext.retries[channel] = 0;
+ if (Object.keys(channels).length !== 1) faults[channel].retryNow = true;
+ }
+
+ if (channels[channel] === 0) faults[channel].precondition = true;
+
+ if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
+ sessionContext.faultsTouched[channel] = true;
+
+ logger.accessWarn({
+ sessionContext: sessionContext,
+ message: "Channel has faulted",
+ fault: fault,
+ channel: channel,
+ retries: sessionContext.retries[channel]
+ });
+ }
+
+ for (let i = 0; i < provisions.length; ++i) {
+ for (const channel of Object.keys(channels)) {
+ if ((channels[channel] >> i) & 1)
+ faults[channel].provisions.push(provisions[i]);
+ }
+ }
+
+ for (const channel of Object.keys(channels)) {
+ const provs = faults[channel].provisions;
+ faults[channel].provisions = [];
+ appendProvisions(faults[channel].provisions, provs);
+ }
+
+ session.clearProvisions(sessionContext);
+}
+
+function inform(sessionContext, rpc) {
+ session.inform(sessionContext, rpc.cpeRequest, (err, acsResponse) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ const res = soap.response({
+ id: rpc.id,
+ acsResponse: acsResponse,
+ cwmpVersion: sessionContext.cwmpVersion
+ });
+
+ const cookiesPath = config.get("COOKIES_PATH", sessionContext.deviceId);
+ if (cookiesPath) {
+ res.headers["Set-Cookie"] = `session=${
+ sessionContext.sessionId
+ }; Path=${cookiesPath}`;
+ } else {
+ res.headers["Set-Cookie"] = `session=${sessionContext.sessionId}`;
+ }
+
+ writeResponse(sessionContext, res);
+ });
+}
+
+function transferComplete(sessionContext, rpc) {
+ session.transferComplete(
+ sessionContext,
+ rpc.cpeRequest,
+ (err, acsResponse, operation, fault) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ if (!operation) {
+ logger.accessWarn({
+ sessionContext: sessionContext,
+ message: "Unrecognized command key",
+ rpc: rpc
+ });
+ }
+
+ if (fault) {
+ Object.assign(sessionContext.retries, operation.retries);
+ recordFault(
+ sessionContext,
+ fault,
+ operation.provisions,
+ operation.channels
+ );
+ }
+
+ const res = soap.response({
+ id: rpc.id,
+ acsResponse: acsResponse,
+ cwmpVersion: sessionContext.cwmpVersion
+ });
+
+ writeResponse(sessionContext, res);
+ }
+ );
+}
+
+// Append providions and remove duplicates
+function appendProvisions(original, toAppend) {
+ let modified = false;
+ const stringified = new WeakMap();
+
+ for (const p of original) stringified.set(p, JSON.stringify(p));
+
+ for (let i = toAppend.length - 1; i >= 0; --i) {
+ let p = toAppend[i];
+ const s = JSON.stringify(p);
+ for (let j = original.length - 1; j >= 0; --j) {
+ const ss = stringified.get(original[j]);
+ if (s === ss) {
+ if (!p || j >= original.length - (toAppend.length - i)) {
+ p = null;
+ } else {
+ original.splice(j, 1);
+ modified = true;
+ }
+ }
+ }
+
+ if (p) {
+ original.splice(original.length - (toAppend.length - i) + 1, 0, p);
+ stringified.set(p, s);
+ modified = true;
+ }
+ }
+
+ return modified;
+}
+
+function applyPresets(sessionContext) {
+ const deviceData = sessionContext.deviceData;
+ localCache.getPresets((err, presetsHash, presets) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ // Filter presets based on existing faults
+ const blackList = {};
+ let whiteList = null;
+ let whiteListProvisions = null;
+ const RETRY_DELAY = config.get("RETRY_DELAY", sessionContext.deviceId);
+
+ if (sessionContext.faults) {
+ for (let [channel, fault] of Object.entries(sessionContext.faults)) {
+ let retryTimestamp = 0;
+ if (!fault.retryNow) {
+ retryTimestamp =
+ fault.timestamp +
+ RETRY_DELAY * Math.pow(2, sessionContext.retries[channel]) * 1000;
+ }
+
+ if (retryTimestamp <= sessionContext.timestamp) {
+ whiteList = channel;
+ whiteListProvisions = fault.provisions;
+ break;
+ }
+
+ blackList[channel] = fault.precondition ? 1 : 2;
+ }
+ }
+
+ deviceData.timestamps.revision = 1;
+ deviceData.attributes.revision = 1;
+
+ const deviceEvents = {};
+ for (const p of deviceData.paths.find(["Events", "*"], false, true)) {
+ const attrs = deviceData.attributes.get(p);
+ if (attrs && attrs.value && attrs.value[1][0] >= sessionContext.timestamp)
+ deviceEvents[p[1]] = true;
+ }
+
+ const parameters = {};
+ const filteredPresets = [];
+
+ for (const preset of presets) {
+ if (whiteList != null) {
+ if (preset.channel !== whiteList) continue;
+ } else if (blackList[preset.channel] === 1) {
+ continue;
+ }
+
+ let eventsMatch = true;
+ for (let [k, v] of Object.entries(preset.events)) {
+ if (!v !== !deviceEvents[k.replace(/\s+/g, "_")]) {
+ eventsMatch = false;
+ break;
+ }
+ }
+
+ if (!eventsMatch) continue;
+
+ if (preset.schedule && preset.schedule.schedule) {
+ const r = scheduling.cron(
+ sessionContext.timestamp,
+ preset.schedule.schedule
+ );
+ if (!(r[0] + preset.schedule.duration > sessionContext.timestamp))
+ continue;
+ }
+
+ filteredPresets.push(preset);
+ for (let k of Object.keys(preset.precondition)) {
+ sessionContext.channels[preset.channel] = 0;
+ const p = k.split(/([^a-zA-Z0-9\-_.].*)/, 1)[0];
+ parameters[p] = common.parsePath(p);
+ }
+ }
+
+ const declarations = [];
+ for (let v of Object.values(parameters))
+ declarations.push([v, 1, { value: 1 }]);
+
+ session.rpcRequest(
+ sessionContext,
+ declarations,
+ (err, flt, reqId, acsReq) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ if (flt) {
+ recordFault(sessionContext, flt);
+ session.clearProvisions(sessionContext);
+ return applyPresets(sessionContext);
+ }
+
+ if (acsReq) return sendAcsRequest(sessionContext, reqId, acsReq);
+
+ session.clearProvisions(sessionContext);
+
+ const parameterValues = {};
+ for (let [k, v] of Object.entries(parameters)) {
+ const unpacked = device.unpack(deviceData, v);
+ if (!unpacked[0]) continue;
+ const attrs = deviceData.attributes.get(unpacked[0]);
+ if (attrs && attrs.value && attrs.value[1])
+ parameterValues[k] = attrs.value[1][0];
+ }
+
+ if (whiteList != null)
+ session.addProvisions(sessionContext, whiteList, whiteListProvisions);
+
+ const appendProvisionsToFaults = {};
+ for (const p of filteredPresets) {
+ if (query.testFilter(parameterValues, p.precondition)) {
+ if (blackList[p.channel] === 2) {
+ appendProvisionsToFaults[p.channel] = (
+ appendProvisionsToFaults[p.channel] || []
+ ).concat(p.provisions);
+ } else {
+ session.addProvisions(sessionContext, p.channel, p.provisions);
+ }
+ }
+ }
+
+ for (let [channel, provisions] of Object.entries(
+ appendProvisionsToFaults
+ )) {
+ if (
+ appendProvisions(
+ sessionContext.faults[channel].provisions,
+ provisions
+ )
+ ) {
+ if (!sessionContext.faultsTouched)
+ sessionContext.faultsTouched = {};
+ sessionContext.faultsTouched[channel] = true;
+ }
+ }
+
+ // Don't increment when processing a single channel (e.g. after fault)
+ if (whiteList == null)
+ sessionContext.presetCycles = (sessionContext.presetCycles || 0) + 1;
+
+ if (sessionContext.presetCycles > MAX_CYCLES) {
+ const fault = {
+ code: "preset_loop",
+ message: "The presets are stuck in an endless configuration loop",
+ timestamp: sessionContext.timestamp
+ };
+ recordFault(sessionContext, fault);
+ // No need to save retryNow
+ for (const f of Object.values(sessionContext.faults))
+ delete f.retryNow;
+ session.clearProvisions(sessionContext);
+ return sendAcsRequest(sessionContext);
+ }
+
+ deviceData.timestamps.dirty = 0;
+ deviceData.attributes.dirty = 0;
+ session.rpcRequest(
+ sessionContext,
+ null,
+ (err, fault, id, acsRequest) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ if (fault) {
+ recordFault(sessionContext, fault);
+ session.clearProvisions(sessionContext);
+ return applyPresets(sessionContext);
+ }
+
+ if (!acsRequest) {
+ for (const channel of Object.keys(sessionContext.channels)) {
+ if (sessionContext.faults[channel]) {
+ delete sessionContext.faults[channel];
+ if (!sessionContext.faultsTouched)
+ sessionContext.faultsTouched = {};
+ sessionContext.faultsTouched[channel] = true;
+ }
+ }
+
+ if (whiteList != null) return applyPresets(sessionContext);
+
+ if (
+ sessionContext.deviceData.timestamps.dirty > 1 ||
+ sessionContext.deviceData.attributes.dirty > 1
+ )
+ return applyPresets(sessionContext);
+ }
+
+ sendAcsRequest(sessionContext, id, acsRequest);
+ }
+ );
+ }
+ );
+ });
+}
+
+function nextRpc(sessionContext) {
+ session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ if (fault) {
+ recordFault(sessionContext, fault);
+ session.clearProvisions(sessionContext);
+ return nextRpc(sessionContext);
+ }
+
+ if (acsRequest) return sendAcsRequest(sessionContext, id, acsRequest);
+
+ for (let [channel, flags] of Object.entries(sessionContext.channels)) {
+ if (flags && sessionContext.faults[channel]) {
+ delete sessionContext.faults[channel];
+ if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
+
+ sessionContext.faultsTouched[channel] = true;
+ }
+ if (channel.startsWith("task_")) {
+ const taskId = channel.slice(5);
+ if (!sessionContext.doneTasks) sessionContext.doneTasks = [];
+ sessionContext.doneTasks.push(taskId);
+
+ for (let j = 0; j < sessionContext.tasks.length; ++j) {
+ if (sessionContext.tasks[j]._id === taskId) {
+ sessionContext.tasks.splice(j, 1);
+ break;
+ }
+ }
+ }
+ }
+
+ session.clearProvisions(sessionContext);
+
+ // Clear expired tasks
+ sessionContext.tasks = sessionContext.tasks.filter(task => {
+ if (!(task.expiry <= sessionContext.timestamp)) return true;
+
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "Task expired",
+ task: task
+ });
+
+ if (!sessionContext.doneTasks) sessionContext.doneTasks = [];
+ sessionContext.doneTasks.push(task._id);
+
+ const channel = `task_${task._id}`;
+ if (sessionContext.faults[channel]) {
+ delete sessionContext.faults[channel];
+ if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
+ sessionContext.faultsTouched[channel] = true;
+ }
+
+ return false;
+ });
+
+ const task = sessionContext.tasks.find(
+ t => !sessionContext.faults[`task_${t._id}`]
+ );
+
+ if (!task) return applyPresets(sessionContext);
+
+ let alias;
+
+ switch (task.name) {
+ case "getParameterValues":
+ // Set channel in case params array is empty
+ sessionContext.channels[`task_${task._id}`] = 0;
+ for (const p of task.parameterNames) {
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["refresh", p]
+ ]);
+ }
+
+ break;
+ case "setParameterValues":
+ // Set channel in case params array is empty
+ sessionContext.channels[`task_${task._id}`] = 0;
+ for (const p of task.parameterValues) {
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["value", p[0], p[1]]
+ ]);
+ }
+
+ break;
+ case "refreshObject":
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["refresh", task.objectName]
+ ]);
+ break;
+ case "reboot":
+ session.addProvisions(sessionContext, `task_${task._id}`, [["reboot"]]);
+ break;
+ case "factoryReset":
+ session.addProvisions(sessionContext, `task_${task._id}`, [["reset"]]);
+ break;
+ case "download":
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["download", task.fileType, task.fileName, task.targetFileName]
+ ]);
+ break;
+ case "addObject":
+ alias = (task.parameterValues || [])
+ .map(p => `${p[0]}:${JSON.stringify(p[1])}`)
+ .join(",");
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["instances", `${task.objectName}.[${alias}]`, "+1"]
+ ]);
+ break;
+ case "deleteObject":
+ session.addProvisions(sessionContext, `task_${task._id}`, [
+ ["instances", task.objectName, 0]
+ ]);
+ break;
+ default:
+ return throwError(
+ new Error("Task name not recognized"),
+ sessionContext.httpResponse
+ );
+ }
+ nextRpc(sessionContext);
+ });
+}
+
+function endSession(sessionContext, callback) {
+ let saveCache = sessionContext.cacheUntil != null;
+ let counter = 3;
+
+ counter += 2;
+ db.saveDevice(
+ sessionContext.deviceId,
+ sessionContext.deviceData,
+ sessionContext.new,
+ sessionContext.timestamp,
+ err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) return callback(null, sessionContext.new);
+ }
+ );
+
+ if (sessionContext.operationsTouched) {
+ for (const k of Object.keys(sessionContext.operationsTouched)) {
+ counter += 2;
+ saveCache = true;
+ if (sessionContext.operations[k]) {
+ db.saveOperation(
+ sessionContext.deviceId,
+ k,
+ sessionContext.operations[k],
+ err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+ }
+ );
+ } else {
+ db.deleteOperation(sessionContext.deviceId, k, err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+ });
+ }
+ }
+ }
+
+ if (sessionContext.doneTasks && sessionContext.doneTasks.length) {
+ counter += 2;
+ saveCache = true;
+ db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+ });
+ }
+
+ if (sessionContext.faultsTouched) {
+ for (const k of Object.keys(sessionContext.faultsTouched)) {
+ counter += 2;
+ saveCache = true;
+ if (sessionContext.faults[k]) {
+ sessionContext.faults[k].retries = sessionContext.retries[k];
+ db.saveFault(
+ sessionContext.deviceId,
+ k,
+ sessionContext.faults[k],
+ err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+ }
+ );
+ } else {
+ db.deleteFault(sessionContext.deviceId, k, err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+ });
+ }
+ }
+ }
+
+ if (saveCache) {
+ counter += 2;
+ cacheDueTasksAndFaultsAndOperations(
+ sessionContext.deviceId,
+ sessionContext.tasks,
+ sessionContext.faults,
+ sessionContext.operations,
+ sessionContext.cacheUntil,
+ err => {
+ if (err) {
+ if (counter & 1) callback(err);
+ return (counter = 0);
+ }
+ if ((counter -= 2) === 1) return callback(null, sessionContext.new);
+ }
+ );
+ }
+ if ((counter -= 2) === 1) callback(null, sessionContext.new);
+}
+
+function sendAcsRequest(sessionContext, id, acsRequest) {
+ if (!acsRequest)
+ return writeResponse(sessionContext, soap.response(null), true);
+
+ if (acsRequest.name === "Download") {
+ if (!acsRequest.url) {
+ const FS_PORT = config.get("FS_PORT");
+ const FS_HOSTNAME = config.get("FS_HOSTNAME");
+ const FS_SSL = config.get("FS_SSL");
+ acsRequest.url = FS_SSL ? "https://" : "http://";
+ acsRequest.url += FS_HOSTNAME;
+ if (FS_PORT !== 80) acsRequest.url += ":" + FS_PORT;
+ acsRequest.url += "/" + encodeURI(acsRequest.fileName);
+ }
+
+ if (acsRequest.fileSize == null) {
+ return localCache.getFiles((err, hash, files) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ if (files[acsRequest.fileName])
+ acsRequest.fileSize = files[acsRequest.fileName].length;
+ else acsRequest.fileSize = 0;
+
+ return sendAcsRequest(sessionContext, id, acsRequest);
+ });
+ }
+ }
+
+ const rpc = {
+ id: id,
+ acsRequest: acsRequest,
+ cwmpVersion: sessionContext.cwmpVersion
+ };
+
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "ACS request",
+ rpc: rpc
+ });
+
+ const res = soap.response(rpc);
+ writeResponse(sessionContext, res);
+}
+
+function getSession(connection, sessionId, callback) {
+ const sessionContext = currentSessions.get(connection);
+ if (sessionContext) {
+ currentSessions.delete(connection);
+ return callback(null, sessionContext);
+ }
+
+ if (!sessionId) return callback();
+
+ setTimeout(() => {
+ cache.pop(`session_${sessionId}`, (err, sessionContextString) => {
+ if (err || !sessionContextString) return callback(err);
+ session.deserialize(sessionContextString, callback);
+ });
+ }, 100);
+}
+
+const currentSessions = new WeakMap();
+
+// When socket closes, store active sessions in cache
+function onConnection(socket) {
+ // The property remoteAddress may be undefined after the connection is
+ // closed, unless we read it at least once (caching?)
+ socket.remoteAddress;
+
+ socket.on("close", () => {
+ const sessionContext = currentSessions.get(socket);
+ if (!sessionContext) return;
+ currentSessions.delete(socket);
+ const now = Date.now();
+
+ const lastActivity = sessionContext.lastActivity;
+ const timeoutMsg = logger.flatten({
+ sessionContext: sessionContext,
+ message: "Session timeout",
+ sessionTimestamp: sessionContext.timestamp
+ });
+
+ const timeout =
+ sessionContext.lastActivity + sessionContext.timeout * 1000 - now;
+ if (timeout <= 0) return logger.accessError(timeoutMsg);
+
+ setTimeout(() => {
+ return cache.get(
+ `session_${sessionContext.sessionId}`,
+ (err, sessionContextString) => {
+ if (err) return throwError(err);
+
+ if (!sessionContextString) return;
+
+ session.deserialize(sessionContextString, (err, _sessionContext) => {
+ if (err) return throwError(err);
+ if (_sessionContext.lastActivity === lastActivity)
+ logger.accessError(timeoutMsg);
+ });
+ }
+ );
+ }, timeout + 1000).unref();
+
+ session.serialize(sessionContext, (err, sessionContextString) => {
+ if (err) return throwError(err);
+ cache.set(
+ `session_${sessionContext.sessionId}`,
+ sessionContextString,
+ Math.ceil(timeout / 1000) + 3,
+ err => {
+ if (err) return throwError(err);
+ }
+ );
+ });
+ });
+}
+
+setInterval(() => {
+ if (stats.droppedRequests) {
+ logger.warn({
+ message: "Worker overloaded",
+ droppedRequests: stats.droppedRequests,
+ totalRequests: stats.totalRequests,
+ initiatedSessions: stats.initiatedSessions,
+ pid: process.pid
+ });
+ }
+
+ stats.totalRequests = 0;
+ stats.droppedRequests = 0;
+ stats.initiatedSessions = 0;
+}, 10000).unref();
+
+function getDueTasksAndFaultsAndOperations(deviceId, timestamp, callback) {
+ cache.get(`${deviceId}_tasks_faults_operations`, (err, res) => {
+ if (err) return callback(err);
+
+ if (res) {
+ res = JSON.parse(res);
+ return callback(
+ null,
+ res.tasks || [],
+ res.faults || {},
+ res.operations || {}
+ );
+ }
+
+ let faults, tasks, operations, cacheUntil;
+
+ db.getFaults(deviceId, (err, _faults) => {
+ if (err) {
+ if (callback) callback(err);
+ return (callback = null);
+ }
+ faults = _faults;
+ if (tasks && operations)
+ callback(null, tasks, faults, operations, cacheUntil);
+ });
+
+ db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) => {
+ if (err) {
+ if (callback) callback(err);
+ return (callback = null);
+ }
+ tasks = dueTasks;
+ cacheUntil = nextTimestamp || 0;
+ if (faults && operations)
+ callback(null, tasks, faults, operations, cacheUntil);
+ });
+
+ db.getOperations(deviceId, (err, _operations) => {
+ if (err) {
+ if (callback) callback(err);
+ return (callback = null);
+ }
+ operations = _operations;
+ if (faults && tasks)
+ callback(null, tasks, faults, operations, cacheUntil);
+ });
+
+ if (faults && tasks && operations)
+ callback(null, tasks, faults, operations, cacheUntil);
+ });
+}
+
+function cacheDueTasksAndFaultsAndOperations(
+ deviceId,
+ tasks,
+ faults,
+ operations,
+ cacheUntil,
+ callback
+) {
+ const v = {};
+ if (tasks.length) v.tasks = tasks;
+ if (Object.keys(faults).length) v.faults = faults;
+ if (Object.keys(operations).length) v.operations = operations;
+
+ let ttl;
+ if (cacheUntil) ttl = Math.trunc((Date.now() - cacheUntil) / 1000);
+ else ttl = config.get("MAX_CACHE_TTL", deviceId);
+
+ cache.set(
+ `${deviceId}_tasks_faults_operations`,
+ JSON.stringify(v),
+ ttl,
+ callback
+ );
+}
+
+function processRequest(sessionContext, rpc) {
+ if (rpc.cpeRequest) {
+ if (rpc.cpeRequest.name === "Inform") {
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "Inform",
+ rpc: rpc
+ });
+ inform(sessionContext, rpc);
+ } else if (rpc.cpeRequest.name === "TransferComplete") {
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "CPE request",
+ rpc: rpc
+ });
+ transferComplete(sessionContext, rpc);
+ } else if (rpc.cpeRequest.name === "GetRPCMethods") {
+ logger.accessInfo({
+ sessionContext: sessionContext,
+ message: "CPE request",
+ rpc: rpc
+ });
+ const res = soap.response({
+ id: rpc.id,
+ acsResponse: {
+ name: "GetRPCMethodsResponse",
+ methodList: ["Inform", "GetRPCMethods", "TransferComplete"]
+ },
+ cwmpVersion: sessionContext.cwmpVersion
+ });
+ writeResponse(sessionContext, res);
+ } else {
+ return throwError(
+ new Error("ACS method not supported"),
+ sessionContext.httpResponse
+ );
+ }
+ } else if (rpc.cpeResponse) {
+ session.rpcResponse(sessionContext, rpc.id, rpc.cpeResponse, err => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+ nextRpc(sessionContext);
+ });
+ } else if (rpc.cpeFault) {
+ logger.accessWarn({
+ sessionContext: sessionContext,
+ message: "CPE fault",
+ rpc: rpc
+ });
+
+ session.rpcFault(sessionContext, rpc.id, rpc.cpeFault, (err, fault) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+ if (fault) {
+ recordFault(sessionContext, fault);
+ session.clearProvisions(sessionContext);
+ }
+ nextRpc(sessionContext);
+ });
+ } else {
+ // CPE sent empty response
+ session.timeoutOperations(sessionContext, (err, faults, operations) => {
+ if (err) return throwError(err, sessionContext.httpResponse);
+
+ for (let i of faults) {
+ for (let [k, v] of Object.entries(operations[i].retries))
+ sessionContext.retries[k] = v;
+
+ recordFault(
+ sessionContext,
+ faults[i],
+ operations[i].provisions,
+ operations[i].channels
+ );
+ }
+
+ nextRpc(sessionContext);
+ });
+ }
+}
+
+function listener(httpRequest, httpResponse) {
+ stats.totalRequests += 1;
+
+ if (httpRequest.method !== "POST") {
+ httpResponse.writeHead(405, {
+ Allow: "POST",
+ Connection: "close"
+ });
+ httpResponse.end("405 Method Not Allowed");
+ return;
+ }
+
+ let sessionId;
+ // Separation by comma is important as some devices don't comform to standard
+ const COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g;
+ let match;
+ while ((match = COOKIE_REGEX.exec(httpRequest.headers.cookie)))
+ if (match[1] === "session") sessionId = match[2];
+
+ // If overloaded, ask CPE to retry in 60 seconds
+ if (!sessionId && stats.concurrentRequests > MAX_CONCURRENT_REQUESTS) {
+ httpResponse.writeHead(503, {
+ "Retry-after": 60,
+ Connection: "close"
+ });
+ httpResponse.end("503 Service Unavailable");
+ stats.droppedRequests += 1;
+ return;
+ }
+
+ let stream = httpRequest;
+ if (httpRequest.headers["content-encoding"]) {
+ switch (httpRequest.headers["content-encoding"]) {
+ case "gzip":
+ stream = httpRequest.pipe(zlib.createGunzip());
+ break;
+ case "deflate":
+ stream = httpRequest.pipe(zlib.createInflate());
+ break;
+ default:
+ httpResponse.writeHead(415, { Connection: "close" });
+ httpResponse.end("415 Unsupported Media Type");
+ return;
+ }
+ }
+
+ stats.concurrentRequests += 1;
+ httpRequest.on("aborted", () => {
+ stats.concurrentRequests -= 1;
+ // In some cases event end can be emitted after aborted event
+ httpRequest.removeAllListeners("end");
+ });
+
+ const chunks = [];
+ let bytes = 0;
+ stream.on("data", chunk => {
+ chunks.push(chunk);
+ bytes += chunk.length;
+ });
+
+ stream.on("end", () => {
+ const body = new Buffer(bytes);
+ let offset = 0;
+ for (const chunk of chunks) {
+ chunk.copy(body, offset, 0, chunk.length);
+ offset += chunk.length;
+ }
+
+ function parsedRpc(sessionContext, rpc, parseWarnings) {
+ for (const w of parseWarnings) {
+ w.sessionContext = sessionContext;
+ w.rpc = rpc;
+ logger.accessWarn(w);
+ }
+
+ if (config.get("DEBUG", sessionContext.deviceId)) {
+ const dump =
+ `# REQUEST ${new Date(Date.now())}\n` +
+ JSON.stringify(httpRequest.headers) +
+ "\n" +
+ body +
+ "\n\n";
+ fs.appendFile(`./debug/${sessionContext.deviceId}.dump`, dump, err => {
+ if (err) return throwError(err);
+ });
+ }
+
+ processRequest(sessionContext, rpc);
+ }
+
+ getSession(httpRequest.connection, sessionId, (err, sessionContext) => {
+ if (err) return throwError(err, httpResponse);
+
+ if (sessionContext) {
+ sessionContext.httpRequest = httpRequest;
+ sessionContext.httpResponse = httpResponse;
+ if (
+ sessionContext.sessionId !== sessionId ||
+ sessionContext.lastActivity + sessionContext.timeout * 1000 <
+ Date.now()
+ ) {
+ logger.accessError({
+ message: "Invalid session",
+ sessionContext: sessionContext
+ });
+
+ httpResponse.writeHead(400, { Connection: "close" });
+ httpResponse.end("Invalid session");
+ stats.concurrentRequests -= 1;
+ return;
+ }
+ } else if (stats.concurrentRequests > MAX_CONCURRENT_REQUESTS) {
+ // Check again just in case device included old session ID
+ // from the previous session
+ httpResponse.writeHead(503, { "Retry-after": 60, Connection: "close" });
+ httpResponse.end("503 Service Unavailable");
+ stats.droppedRequests += 1;
+ stats.concurrentRequests -= 1;
+ return;
+ }
+
+ let parseWarnings = [];
+ let rpc;
+ try {
+ rpc = soap.request(
+ body,
+ sessionContext ? sessionContext.cwmpVersion : null,
+ parseWarnings
+ );
+ } catch (error) {
+ logger.accessError({
+ message: "XML parse error",
+ parseError: error.message.trim(),
+ sessionContext: sessionContext || {
+ httpRequest: httpRequest,
+ httpResponse: httpResponse
+ }
+ });
+ httpResponse.writeHead(400, { Connection: "close" });
+ httpResponse.end(error.message);
+ stats.concurrentRequests -= 1;
+ return;
+ }
+
+ if (sessionContext) {
+ if (
+ (rpc.cpeRequest && rpc.cpeRequest.name === "Inform") ||
+ !sessionContext.rpcRequest ^ !(rpc.cpeResponse || rpc.cpeFault)
+ ) {
+ logger.accessError({
+ message: "Bad session state",
+ sessionContext: sessionContext
+ });
+ httpResponse.writeHead(400, { Connection: "close" });
+ httpResponse.end("Bad session state");
+ stats.concurrentRequests -= 1;
+ return;
+ }
+ return parsedRpc(sessionContext, rpc, parseWarnings);
+ }
+
+ if (!(rpc.cpeRequest && rpc.cpeRequest.name === "Inform")) {
+ logger.accessError({
+ message: "Invalid session",
+ sessionContext: sessionContext || {
+ httpRequest: httpRequest,
+ httpResponse: httpResponse
+ }
+ });
+ httpResponse.writeHead(400, { Connection: "close" });
+ httpResponse.end("Invalid session");
+ stats.concurrentRequests -= 1;
+ return;
+ }
+
+ stats.initiatedSessions += 1;
+ const deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId);
+
+ session.init(
+ deviceId,
+ rpc.cwmpVersion,
+ rpc.sessionTimeout || config.get("SESSION_TIMEOUT", deviceId),
+ (err, _sessionContext) => {
+ if (err) return throwError(err, httpResponse);
+
+ _sessionContext.httpRequest = httpRequest;
+ _sessionContext.httpResponse = httpResponse;
+ _sessionContext.sessionId = crypto.randomBytes(8).toString("hex");
+ httpRequest.connection.setTimeout(_sessionContext.timeout * 1000);
+
+ getDueTasksAndFaultsAndOperations(
+ deviceId,
+ _sessionContext.timestamp,
+ (err, dueTasks, faults, operations, cacheUntil) => {
+ if (err) return throwError(err, httpResponse);
+
+ _sessionContext.tasks = dueTasks;
+ _sessionContext.operations = operations;
+ _sessionContext.cacheUntil = cacheUntil;
+ _sessionContext.faults = faults;
+ _sessionContext.retries = {};
+ for (let [k, v] of Object.entries(_sessionContext.faults)) {
+ if (v.expiry >= _sessionContext.timestamp) {
+ // Delete expired faults
+ delete _sessionContext.faults[k];
+ if (!_sessionContext.faultsTouched)
+ _sessionContext.faultsTouched = {};
+ _sessionContext.faultsTouched[k] = true;
+ } else {
+ _sessionContext.retries[k] = v.retries;
+ }
+ }
+ parsedRpc(_sessionContext, rpc, parseWarnings);
+ }
+ );
+ }
+ );
+ });
+ });
+}
+
+exports.listener = listener;
+exports.onConnection = onConnection;