From 819e8a52a37986dbd758ff5b5a35af10b7127de3 Mon Sep 17 00:00:00 2001 From: Zaid Abdulla Date: Thu, 27 Dec 2018 22:23:37 -0800 Subject: [PATCH] Load config from DB --- lib/cwmp.js | 116 +++++++++++++++++++++++++++++---------------- lib/db.js | 20 +++++++- lib/local-cache.js | 51 +++++++++++++++++++- lib/session.js | 81 ++++++++++++++++++++++--------- 4 files changed, 203 insertions(+), 65 deletions(-) diff --git a/lib/cwmp.js b/lib/cwmp.js index 3842735..2f9fb14 100644 --- a/lib/cwmp.js +++ b/lib/cwmp.js @@ -57,7 +57,7 @@ function writeResponse(sessionContext, res, close) { // Close connection after last request in session if (close) res.headers["Connection"] = "close"; - if (config.get("DEBUG", sessionContext.deviceId)) { + if (sessionContext.debug) { const dump = `# RESPONSE ${new Date(Date.now())}\n` + JSON.stringify(res.headers) + @@ -195,7 +195,12 @@ function inform(sessionContext, rpc) { cwmpVersion: sessionContext.cwmpVersion }); - const cookiesPath = config.get("COOKIES_PATH", sessionContext.deviceId); + const cookiesPath = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.cookiesPath", + sessionContext.configContext + ); + if (cookiesPath) { res.headers["Set-Cookie"] = `session=${ sessionContext.sessionId @@ -284,7 +289,11 @@ function applyPresets(sessionContext) { const blackList = {}; let whiteList = null; let whiteListProvisions = null; - const RETRY_DELAY = config.get("RETRY_DELAY", sessionContext.deviceId); + const RETRY_DELAY = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.retryDelay", + sessionContext.configContext + ); if (sessionContext.faults) { for (const [channel, fault] of Object.entries(sessionContext.faults)) { @@ -1062,7 +1071,7 @@ function listener(httpRequest, httpResponse) { logger.accessWarn(w); } - if (config.get("DEBUG", sessionContext.deviceId)) { + if (sessionContext.debug) { const dump = `# REQUEST ${new Date(Date.now())}\n` + JSON.stringify(httpRequest.headers) + @@ -1165,45 +1174,72 @@ function listener(httpRequest, httpResponse) { 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 void throwError(err, httpResponse); + localCache.getCurrentSnapshot((err, cacheSnapshot) => { + if (err) return void throwError(err, httpResponse); - _sessionContext.httpRequest = httpRequest; - _sessionContext.httpResponse = httpResponse; - _sessionContext.sessionId = crypto.randomBytes(8).toString("hex"); - httpRequest.connection.setTimeout(_sessionContext.timeout * 1000); + const configContext = { + id: deviceId, + serialNumber: rpc.cpeRequest.deviceId["SerialNumber"], + productClass: rpc.cpeRequest.deviceId["ProductClass"], + oui: rpc.cpeRequest.deviceId["OUI"] + }; - getDueTasksAndFaultsAndOperations( - deviceId, - _sessionContext.timestamp, - (err, dueTasks, faults, operations, cacheUntil) => { - if (err) return void throwError(err, httpResponse); - - _sessionContext.tasks = dueTasks; - _sessionContext.operations = operations; - _sessionContext.cacheUntil = cacheUntil; - _sessionContext.faults = faults; - _sessionContext.retries = {}; - for (const [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); - } + const sessionTimeout = + rpc.sessionTimeout || + localCache.getConfig( + cacheSnapshot, + "cwmp.sessionTimeout", + configContext ); - } - ); + + session.init( + deviceId, + rpc.cwmpVersion, + sessionTimeout, + (err, _sessionContext) => { + if (err) return void throwError(err, httpResponse); + + _sessionContext.cacheSnapshot = cacheSnapshot; + _sessionContext.configContext = configContext; + _sessionContext.debug = localCache.getConfig( + cacheSnapshot, + "cwmp.debug", + configContext + ); + + _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 void throwError(err, httpResponse); + + _sessionContext.tasks = dueTasks; + _sessionContext.operations = operations; + _sessionContext.cacheUntil = cacheUntil; + _sessionContext.faults = faults; + _sessionContext.retries = {}; + for (const [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); + } + ); + } + ); + }); }); }); } diff --git a/lib/db.js b/lib/db.js index 2f701e8..ff38aa2 100644 --- a/lib/db.js +++ b/lib/db.js @@ -21,6 +21,7 @@ const mongodb = require("mongodb"); const config = require("./config"); const common = require("./common"); +const expressionParser = require("./common/expression-parser"); let tasksCollection, devicesCollection, @@ -30,7 +31,8 @@ let tasksCollection, virtualParametersCollection, faultsCollection, filesCollection, - operationsCollection; + operationsCollection, + configCollection; function connect(callback) { mongodb.MongoClient.connect( @@ -58,6 +60,7 @@ function connect(callback) { exports.operationsCollection = operationsCollection = db.collection( "operations" ); + configCollection = db.collection("config"); callback(); } @@ -873,6 +876,20 @@ function getFiles(callback) { filesCollection.find().toArray(callback); } +function getConfig(callback) { + configCollection.find().toArray((err, res) => { + if (err) return void callback(err); + + callback( + null, + res.map(c => ({ + id: c["_id"], + value: expressionParser.parse(c["value"]) + })) + ); + }); +} + exports.connect = connect; exports.disconnect = disconnect; exports.fetchDevice = fetchDevice; @@ -890,3 +907,4 @@ exports.getVirtualParameters = getVirtualParameters; exports.getFiles = getFiles; exports.getDueTasks = getDueTasks; exports.getOperations = getOperations; +exports.getConfig = getConfig; diff --git a/lib/local-cache.js b/lib/local-cache.js index 078c7a5..90e9933 100644 --- a/lib/local-cache.js +++ b/lib/local-cache.js @@ -21,11 +21,13 @@ const vm = require("vm"); const crypto = require("crypto"); +const config = require("./config"); const db = require("./db"); const cache = require("./cache"); const query = require("./query"); const logger = require("./logger"); const scheduling = require("./scheduling"); +const expression = require("./common/expression"); const REFRESH = 3000; const EVICT_TIMEOUT = 60000; @@ -56,6 +58,10 @@ function computeHash(snapshot) { h.update(JSON.stringify(keys)); for (const k of keys) h.update(snapshot.virtualParameters[k].md5); + keys = Object.keys(snapshot.config).sort(); + h.update(JSON.stringify(keys)); + for (const k of keys) h.update(JSON.stringify(snapshot.config[k])); + return h.digest("hex"); } @@ -296,13 +302,27 @@ function refresh(callback) { }) ); + promises.push( + new Promise((resolve, reject) => { + db.getConfig((err, conf) => { + if (err) return void reject(err); + + const _config = {}; + for (const c of conf) _config[c.id] = c.value; + + resolve(_config); + }); + }) + ); + Promise.all(promises) .then(res => { const snapshot = { presets: res[0], provisions: res[1], virtualParameters: res[2], - files: res[3] + files: res[3], + config: res[4] }; if (currentSnapshot) { @@ -362,9 +382,38 @@ function getFiles(snapshotKey) { return snapshot.files; } +function getConfig(snapshotKey, key, context, now) { + const snapshot = snapshots.get(snapshotKey); + if (!snapshot) throw new Error("Cache snapshot does not exist"); + + const oldOpts = { + "cwmp.downloadTimeout": "DOWNLOAD_TIMEOUT", + "cwmp.debug": "DEBUG", + "cwmp.retryDelay": "RETRY_DELAY", + "cwmp.sessionTimeout": "SESSION_TIMEOUT", + "cwmp.connectionRequestTimeout": "CONNECTION_REQUEST_TIMEOUT", + "cwmp.gpnNextLevel": "GPN_NEXT_LEVEL", + "cwmp.gpvBatchSize": "GPV_BATCH_SIZE", + "cwmp.cookiesPath": "COOKIES_PATH", + "cwmp.datetimeMilliseconds": "DATETIME_MILLISECONDS", + "cwmp.booleanLiteral": "BOOLEAN_LITERAL", + "cwmp.connectionRequestAllowBasicAuth": + "CONNECTION_REQUEST_ALLOW_BASIC_AUTH", + "cwmp.maxCommitIterations": "MAX_COMMIT_ITERATIONS" + }; + + if (!(key in snapshot.config)) { + if (key in oldOpts) + return config.get(oldOpts[key], context ? context.id : null); + else return null; + } + return expression.evaluate(snapshot.config[key], context, now || Date.now()); +} + exports.getCurrentSnapshot = getCurrentSnapshot; exports.hasSnapshot = hasSnapshot; exports.getPresets = getPresets; exports.getProvisions = getProvisions; exports.getVirtualParameters = getVirtualParameters; exports.getFiles = getFiles; +exports.getConfig = getConfig; diff --git a/lib/session.js b/lib/session.js index af2c00a..1d4fa38 100644 --- a/lib/session.js +++ b/lib/session.js @@ -30,10 +30,6 @@ const InstanceSet = require("./instance-set"); const defaultProvisions = require("./default-provisions"); const gpnHeuristic = require("./gpn-heuristic"); -// Multiply by two because every iteration is two -// phases: read and update -const MAX_ITERATIONS = config.get("MAX_COMMIT_ITERATIONS") * 2; - function initDeviceData() { return { paths: new PathSet(), @@ -64,11 +60,7 @@ function init(deviceId, cwmpVersion, timeout, callback) { declarations: [] }; - localCache.getCurrentSnapshot((err, cacheSnapshot) => { - if (err) return void callback(err); - sessionContext.cacheSnapshot = cacheSnapshot; - callback(null, sessionContext); - }); + callback(null, sessionContext); } function loadParameters(sessionContext, callback) { @@ -435,7 +427,11 @@ function timeoutOperations(sessionContext, callback) { } const DOWNLOAD_TIMEOUT = - config.get("DOWNLOAD_TIMEOUT", sessionContext.deviceId) * 1000; + localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.downloadTimeout", + sessionContext.configContext + ) * 1000; if (sessionContext.timestamp > operation.timestamp + DOWNLOAD_TIMEOUT) { delete sessionContext.operations[commandKey]; @@ -465,6 +461,15 @@ function timeoutOperations(sessionContext, callback) { } function addProvisions(sessionContext, channel, provisions) { + // Multiply by two because every iteration is two + // phases: read and update + const MAX_ITERATIONS = + localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.maxCommitIterations", + sessionContext.configContext + ) * 2; + delete sessionContext.syncState; delete sessionContext.rpcRequest; sessionContext.declarations = []; @@ -509,6 +514,15 @@ function addProvisions(sessionContext, channel, provisions) { } function clearProvisions(sessionContext) { + // Multiply by two because every iteration is two + // phases: read and update + const MAX_ITERATIONS = + localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.maxCommitIterations", + sessionContext.configContext + ) * 2; + if (sessionContext.revisions[sessionContext.revisions.length - 1] > 0) { sessionContext.deviceData.timestamps.collapse(1); sessionContext.deviceData.attributes.collapse(1); @@ -1080,6 +1094,15 @@ function rpcRequest(sessionContext, _declarations, callback) { }); } + // Multiply by two because every iteration is two + // phases: read and update + const MAX_ITERATIONS = + localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.maxCommitIterations", + sessionContext.configContext + ) * 2; + if (sessionContext.iteration >= MAX_ITERATIONS * (sessionContext.cycle + 1)) { return void callback(null, { code: "too_many_commits", @@ -1504,10 +1527,12 @@ function generateGetRpcRequest(sessionContext) { syncState.refreshAttributes.writable.clear(); if (syncState.gpn.size) { - const GPN_NEXT_LEVEL = config.get( - "GPN_NEXT_LEVEL", - sessionContext.deviceId + const GPN_NEXT_LEVEL = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.gpnNextLevel", + sessionContext.configContext ); + const paths = Array.from(syncState.gpn.keys()).sort( (a, b) => b.length - a.length ); @@ -1561,10 +1586,12 @@ function generateGetRpcRequest(sessionContext) { } if (syncState.refreshAttributes.value.size) { - const GPV_BATCH_SIZE = config.get( - "GPV_BATCH_SIZE", - sessionContext.deviceId + const GPV_BATCH_SIZE = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.gpvBatchSize", + sessionContext.configContext ); + const parameterNames = []; for (const path of syncState.refreshAttributes.value) { syncState.refreshAttributes.value.delete(path); @@ -1620,14 +1647,22 @@ function generateSetRpcRequest(sessionContext) { } // Set values - const GPV_BATCH_SIZE = config.get("GPV_BATCH_SIZE", sessionContext.deviceId); - const DATETIME_MILLISECONDS = config.get( - "DATETIME_MILLISECONDS", - sessionContext.deviceId + const GPV_BATCH_SIZE = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.gpvBatchSize", + sessionContext.configContext ); - const BOOLEAN_LITERAL = config.get( - "BOOLEAN_LITERAL", - sessionContext.deviceId + + const DATETIME_MILLISECONDS = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.datetimeMilliseconds", + sessionContext.configContext + ); + + const BOOLEAN_LITERAL = localCache.getConfig( + sessionContext.cacheSnapshot, + "cwmp.booleanLiteral", + sessionContext.configContext ); const parameterValues = [];