diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee index 36b525c..a839e6c 100644 --- a/lib/cwmp.coffee +++ b/lib/cwmp.coffee @@ -40,6 +40,7 @@ util = require 'util' zlib = require 'zlib' +crypto = require 'crypto' mongodb = require 'mongodb' config = require './config' @@ -94,15 +95,12 @@ inform = (currentRequest, cwmpRequest) -> cwmpVersion : currentRequest.sessionData.cwmpVersion }) - session.save(currentRequest.sessionData, (err, sessionId) -> - throw err if err - if !!cookiesPath = config.get('COOKIES_PATH', currentRequest.sessionData.deviceId) - res.headers['Set-Cookie'] = "session=#{sessionId}; Path=#{cookiesPath}" - else - res.headers['Set-Cookie'] = "session=#{sessionId}" + if !!cookiesPath = config.get('COOKIES_PATH', currentRequest.sessionData.deviceId) + res.headers['Set-Cookie'] = "session=#{currentRequest.sessionData.sessionId}; Path=#{cookiesPath}" + else + res.headers['Set-Cookie'] = "session=#{currentRequest.sessionData.sessionId}" - writeResponse(currentRequest, res) - ) + writeResponse(currentRequest, res) ) @@ -268,6 +266,8 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> if not rpcRequest? session.end(currentRequest.sessionData, (err, isNew) -> throw err if err + + delete currentSessions.get(currentRequest.httpRequest.connection)[currentRequest.sessionData.sessionId] if isNew util.log("#{currentRequest.sessionData.deviceId}: New device registered") @@ -293,10 +293,7 @@ sendRpcRequest = (currentRequest, id, rpcRequest) -> cwmpVersion : currentRequest.sessionData.cwmpVersion }) - session.save(currentRequest.sessionData, (err) -> - throw err if err - writeResponse(currentRequest, res) - ) + writeResponse(currentRequest, res) getSession = (httpRequest, callback) -> @@ -307,9 +304,38 @@ getSession = (httpRequest, callback) -> return callback() if not sessionId? - session.load(sessionId, (err, sessionData) -> - throw err if err - return callback(sessionId, sessionData) + sessionData = currentSessions.get(httpRequest.connection)?[sessionId] + + if sessionData? + return callback(null, sessionData) + + 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) -> + return callback(err) if err + currentSessions.get(httpRequest.connection)[sessionId] = sessionData + callback(null, sessionData) + ) + ) + , 100 + ) + + +currentSessions = new WeakMap() + +# When socket closes, store active sessions in redis +onConnection = (socket) -> + currentSessions.set(socket, {}) + socket.on('close', () -> + sessions = currentSessions.get(socket) + for sessionId, sessionData of sessions + session.serialize(sessionData, (err, sessionDataString) -> + throw err if err + db.redisClient.setex("session_#{sessionId}", sessionData.timeout, sessionDataString, (err) -> + throw err if err + ) + ) ) @@ -352,7 +378,8 @@ listener = (httpRequest, httpResponse) -> stream.on('end', () -> cwmpRequest = null - getSession(httpRequest, f = (sessionId, sessionData) -> + getSession(httpRequest, f = (err, sessionData) -> + throw err if err cwmpRequest ?= soap.request(httpRequest, sessionData?.cwmpVersion) if not sessionData? if cwmpRequest.methodRequest?.type isnt 'Inform' @@ -364,6 +391,10 @@ listener = (httpRequest, httpResponse) -> return session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionData) -> throw err if err + sessionData.sessionId = crypto.randomBytes(8).toString('hex') + + currentSessions.get(httpRequest.connection)[sessionData.sessionId] = sessionData + httpRequest.connection.setTimeout(sessionData.timeout * 1000) db.getDueTasksAndFaults(deviceId, sessionData.timestamp, (err, dueTasks, faults) -> @@ -401,10 +432,7 @@ listener = (httpRequest, httpResponse) -> methodResponse : {type : 'GetRPCMethodsResponse', methodList : ['Inform', 'GetRPCMethods', 'TransferComplete', 'RequestDownload']}, cwmpVersion : currentRequest.sessionData.cwmpVersion }) - session.save(currentRequest.sessionData, (err) -> - throw err if err - writeResponse(currentRequest, res) - ) + writeResponse(currentRequest, res) else if cwmpRequest.methodRequest.type is 'TransferComplete' throw new Error('ACS method not supported') else if cwmpRequest.methodResponse? @@ -469,3 +497,4 @@ listener = (httpRequest, httpResponse) -> exports.listener = listener +exports.onConnection = onConnection diff --git a/lib/server.coffee b/lib/server.coffee index cbed29b..7d308f0 100644 --- a/lib/server.coffee +++ b/lib/server.coffee @@ -30,6 +30,7 @@ port = config.get("#{service.toUpperCase()}_PORT") useHttps = config.get("#{service.toUpperCase()}_SSL") serviceListener = require("./#{service}").listener +onConnection = require("./#{service}").onConnection server = null @@ -77,6 +78,9 @@ if useHttps else server = require('http').createServer(listener) +if onConnection? + server.on('connection', onConnection) + db.connect((err) -> throw err if err server.listen(port, networkInterface) diff --git a/lib/session.coffee b/lib/session.coffee index 31ee95c..7091987 100644 --- a/lib/session.coffee +++ b/lib/session.coffee @@ -15,8 +15,6 @@ # along with GenieACS. If not, see . ### -crypto = require 'crypto' - config = require './config' common = require './common' db = require './db' @@ -1090,54 +1088,45 @@ rpcFault = (sessionData, id, faultResponse, callback) -> return callback(null, faultResponse) -load = (id, callback) -> - db.redisClient.get("session_#{id}", (err, sessionDataString) -> - return callback(err) if err or not sessionDataString? +deserialize = (sessionDataString, callback) -> + cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> + return callback(err) if err - cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) -> - return callback(err) if err + sessionData = JSON.parse(sessionDataString) - sessionData = JSON.parse(sessionDataString) + if sessionData.presetsHash? and sessionData.presetsHash != hash + return callback(new Error('Preset hash mismatch')) - if sessionData.presetsHash? and sessionData.presetsHash != hash - return callback(new Error('Preset hash mismatch')) + sessionData.cache = { + provisions: provisions + virtualParameters: virtualParameters + } - sessionData.cache = { - provisions: provisions - virtualParameters: virtualParameters - } + for d in sessionData.declarations + common.addPathMeta(d[0]) - for d in sessionData.declarations - common.addPathMeta(d[0]) + deviceData = initDeviceData() + keys = ['exist', 'object', 'writable', 'value'] + for r in sessionData.deviceData + path = deviceData.paths.add(r[0]) - deviceData = initDeviceData() - keys = ['exist', 'object', 'writable', 'value'] - for r in sessionData.deviceData - path = deviceData.paths.add(r[0]) + if r[1] + deviceData.loaded.set(path, r[1]) - if r[1] - deviceData.loaded.set(path, r[1]) + for k, i in keys + if r[2 + i]? + deviceData.timestamps[k].setRevisions(path, r[2 + i]) - for k, i in keys - if r[2 + i]? - deviceData.timestamps[k].setRevisions(path, r[2 + i]) + if r[2 + keys.length + i]? + deviceData.values[k].setRevisions(path, r[2 + keys.length + i]) - if r[2 + keys.length + i]? - deviceData.values[k].setRevisions(path, r[2 + keys.length + i]) + sessionData.deviceData = deviceData - sessionData.deviceData = deviceData - - return callback(null, sessionData) - ) + return callback(null, sessionData) ) -save = (sessionData, callback) -> - delete sessionData.syncState - delete sessionData.toLoad - delete sessionData.cache - - sessionData.id ?= crypto.randomBytes(8).toString('hex') +serialize = (sessionData, callback) -> deviceData = [] keys = ['exist', 'object', 'writable', 'value'] iter = sessionData.deviceData.paths.find([], 99) @@ -1154,15 +1143,24 @@ save = (sessionData, callback) -> deviceData.push(e) - oldDeviceData = sessionData.deviceData.deviceData - sessionData.deviceData = deviceData - sessionDataString = JSON.stringify(sessionData) - sessionData.deviceData = oldDeviceData + oldDeviceData = sessionData.deviceData + oldSyncState = sessionData.syncState + oldToLoad = sessionData.toLoad + oldCache = sessionData.cache - db.redisClient.setex("session_#{sessionData.id}", sessionData.timeout, sessionDataString, (err) -> - return callback(err, sessionData.id) - ) - return + sessionData.deviceData = deviceData + delete sessionData.syncState + delete sessionData.toLoad + delete sessionData.cache + + sessionDataString = JSON.stringify(sessionData) + + sessionData.deviceData = oldDeviceData + sessionData.syncState = oldSyncState + sessionData.toLoad = sessionData.toLoad + sessionData.cache = sessionData.cache + + return callback(null, sessionDataString) end = (sessionData, callback) -> @@ -1182,5 +1180,5 @@ exports.rpcRequest = rpcRequest exports.rpcResponse = rpcResponse exports.rpcFault = rpcFault exports.end = end -exports.save = save -exports.load = load +exports.serialize = serialize +exports.deserialize = deserialize