966 lines
31 KiB
CoffeeScript
966 lines
31 KiB
CoffeeScript
###
|
|
# 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 <http://www.gnu.org/licenses/>.
|
|
###
|
|
|
|
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
|