rodrigo/genieacs
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Async/await let's go!

Browse Source
This commit is contained in:
Zaid Abdulla committed 2019-06-05 00:17:36 -07:00
1 parent 1e6d3acd1b
commit c3501debd5
20 files changed
+2617 -2916
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.eslintrc.json
+1 -1
View File
@@ -14,10 +14,10 @@
"handle-callback-err": "error", "handle-callback-err": "error",
"prefer-arrow-callback": "error", "prefer-arrow-callback": "error",
"curly": ["error", "multi-or-nest", "consistent"], "curly": ["error", "multi-or-nest", "consistent"],
"consistent-return": ["error", {"treatUndefinedAsUnspecified": true}],
"no-buffer-constructor": "error", "no-buffer-constructor": "error",
"prefer-const": ["error", {"destructuring": "all"}], "prefer-const": ["error", {"destructuring": "all"}],
"eqeqeq": ["error", "always", {"null": "ignore"}], "eqeqeq": ["error", "always", {"null": "ignore"}],
"no-return-await": "error",
"@typescript-eslint/no-explicit-any": 0, "@typescript-eslint/no-explicit-any": 0,
"@typescript-eslint/no-use-before-define": ["error", { "functions": false }] "@typescript-eslint/no-use-before-define": ["error", { "functions": false }]
} }
bin/genieacs-cwmp
+8 -6
View File
@@ -27,7 +27,7 @@ function exitWorkerGracefully(): void {
} }
function exitWorkerUngracefully(): void { function exitWorkerUngracefully(): void {
extensions.killAll(() => { extensions.killAll().then(() => {
process.exit(1); process.exit(1);
}); });
} }
@@ -97,10 +97,8 @@ if (!cluster.worker) {
.catch(exitWorkerUngracefully); .catch(exitWorkerUngracefully);
}); });
db.connect(err => { Promise.all([db.connect(), cache.connect()])
if (err) throw err; .then(() => {
cache.connect(err => {
if (err) throw err;
server.start( server.start(
SERVICE_PORT, SERVICE_PORT,
SERVICE_ADDRESS, SERVICE_ADDRESS,
@@ -108,6 +106,10 @@ if (!cluster.worker) {
cwmp.listener, cwmp.listener,
cwmp.onConnection cwmp.onConnection
); );
})
.catch(err => {
setTimeout(() => {
throw err;
});
}); });
});
} }
bin/genieacs-ext
+2 -2
View File
@@ -59,7 +59,7 @@ function errorToFault(err: Error): Fault {
process.on("uncaughtException", err => { process.on("uncaughtException", err => {
const fault = errorToFault(err); const fault = errorToFault(err);
jobs.forEach(jobId => { jobs.forEach(jobId => {
process.send([jobId, fault]); process.send([jobId, fault, null]);
}); });
jobs.clear(); jobs.clear();
process.disconnect(); process.disconnect();
@@ -81,7 +81,7 @@ process.on("message", message => {
code: "ext", code: "ext",
message: `No such function '${funcName}' in extension '${fileName}'` message: `No such function '${funcName}' in extension '${fileName}'`
}; };
process.send([message[0], fault]); process.send([message[0], fault, null]);
return; return;
} }
bin/genieacs-fs
+7 -5
View File
@@ -93,11 +93,13 @@ if (!cluster.worker) {
.catch(exitWorkerUngracefully); .catch(exitWorkerUngracefully);
}); });
db.connect(err => { Promise.all([db.connect(), cache.connect()])
if (err) throw err; .then(() => {
cache.connect(err => {
if (err) throw err;
server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, listener); server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, listener);
})
.catch(err => {
setTimeout(() => {
throw err;
});
}); });
});
} }
bin/genieacs-nbi
+8 -6
View File
@@ -27,7 +27,7 @@ function exitWorkerGracefully(): void {
} }
function exitWorkerUngracefully(): void { function exitWorkerUngracefully(): void {
extensions.killAll(() => { extensions.killAll().then(() => {
process.exit(1); process.exit(1);
}); });
} }
@@ -97,11 +97,13 @@ if (!cluster.worker) {
.catch(exitWorkerUngracefully); .catch(exitWorkerUngracefully);
}); });
db.connect(err => { Promise.all([db.connect(), cache.connect()])
if (err) throw err; .then(() => {
cache.connect(err => {
if (err) throw err;
server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, listener); server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, listener);
})
.catch(err => {
setTimeout(() => {
throw err;
});
}); });
});
} }
bin/genieacs-ui
+7 -5
View File
@@ -105,11 +105,13 @@ if (!cluster.worker) {
listener(req, res); listener(req, res);
}; };
db2.connect(err => { Promise.all([db2.connect(), cache.connect()])
if (err) throw err; .then(() => {
cache.connect(err => {
if (err) throw err;
server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, _listener); server.start(SERVICE_PORT, SERVICE_ADDRESS, ssl, _listener);
})
.catch(err => {
setTimeout(() => {
throw err;
});
}); });
});
} }
build/build.ts
+1
View File
@@ -31,6 +31,7 @@ const externals = [
"https", "https",
"zlib", "zlib",
"crypto", "crypto",
"util",
"mongodb", "mongodb",
"libxmljs", "libxmljs",
"vm", "vm",
lib/api-functions.ts
+138 -167
View File
@@ -29,9 +29,9 @@ import {
httpConnectionRequest, httpConnectionRequest,
udpConnectionRequest udpConnectionRequest
} from "./connection-request"; } from "./connection-request";
import { Expression } from "./types"; import { Expression, Task } from "./types";
export function connectionRequest(deviceId, callback): void { export async function connectionRequest(deviceId): Promise<void> {
const options = { const options = {
projection: { projection: {
_deviceId: 1, _deviceId: 1,
@@ -46,133 +46,119 @@ export function connectionRequest(deviceId, callback): void {
} }
}; };
db.devicesCollection.findOne({ _id: deviceId }, options, (err, device) => { const device = await db.devicesCollection.findOne({ _id: deviceId }, options);
if (err) return void callback(err); if (!device) throw new Error("No such device");
if (!device) return void callback(new Error("No such device"));
let managementServer, let managementServer,
connectionRequestUrl, connectionRequestUrl,
udpConnectionRequestAddress,
username,
password;
if (device.Device)
// TR-181 data model
managementServer = device.Device.ManagementServer;
// TR-098 data model
else managementServer = device.InternetGatewayDevice.ManagementServer;
if (managementServer.ConnectionRequestURL)
connectionRequestUrl = managementServer.ConnectionRequestURL._value;
if (managementServer.UDPConnectionRequestAddress) {
udpConnectionRequestAddress =
managementServer.UDPConnectionRequestAddress._value;
}
if (managementServer.ConnectionRequestUsername)
username = managementServer.ConnectionRequestUsername._value;
if (managementServer.ConnectionRequestPassword)
password = managementServer.ConnectionRequestPassword._value;
const context = {
id: device["_id"],
serialNumber: device["_deviceId"]["SerialNumber"],
productClass: device["_deviceId"]["ProductClass"],
oui: device["_deviceId"]["OUI"],
username: username || "",
password: password || ""
};
const snapshot = await getCurrentSnapshot();
const now = Date.now();
const UDP_CONNECTION_REQUEST_PORT = +getConfig(
snapshot,
"cwmp.udpConnectionRequestAuth",
context,
now
);
const CONNECTION_REQUEST_TIMEOUT = +getConfig(
snapshot,
"cwmp.connectionRequestTimeout",
context,
now
);
const CONNECTION_REQUEST_ALLOW_BASIC_AUTH = !!getConfig(
snapshot,
"cwmp.connectionRequestAllowBasicAuth",
context,
now
);
let authExp: Expression = getConfigExpression(
snapshot,
"cwmp.connectionRequestAuth"
);
if (authExp === undefined) {
authExp = [
"FUNC",
"AUTH",
["PARAM", "username"],
["PARAM", "password"]
] as Expression;
}
let udpProm;
if (udpConnectionRequestAddress) {
udpProm = udpConnectionRequest(
udpConnectionRequestAddress, udpConnectionRequestAddress,
username, authExp,
password; db.configCollection,
if (device.Device) UDP_CONNECTION_REQUEST_PORT
// TR-181 data model
managementServer = device.Device.ManagementServer;
// TR-098 data model
else managementServer = device.InternetGatewayDevice.ManagementServer;
if (managementServer.ConnectionRequestURL)
connectionRequestUrl = managementServer.ConnectionRequestURL._value;
if (managementServer.UDPConnectionRequestAddress) {
udpConnectionRequestAddress =
managementServer.UDPConnectionRequestAddress._value;
}
if (managementServer.ConnectionRequestUsername)
username = managementServer.ConnectionRequestUsername._value;
if (managementServer.ConnectionRequestPassword)
password = managementServer.ConnectionRequestPassword._value;
const context = {
id: device["_id"],
serialNumber: device["_deviceId"]["SerialNumber"],
productClass: device["_deviceId"]["ProductClass"],
oui: device["_deviceId"]["OUI"],
username: username || "",
password: password || ""
};
getCurrentSnapshot()
.then(snapshot => {
const now = Date.now();
const UDP_CONNECTION_REQUEST_PORT = +getConfig(
snapshot,
"cwmp.udpConnectionRequestAuth",
context,
now
);
const CONNECTION_REQUEST_TIMEOUT = +getConfig(
snapshot,
"cwmp.connectionRequestTimeout",
context,
now
);
const CONNECTION_REQUEST_ALLOW_BASIC_AUTH = !!getConfig(
snapshot,
"cwmp.connectionRequestAllowBasicAuth",
context,
now
);
let authExp: Expression = getConfigExpression(
snapshot,
"cwmp.connectionRequestAuth"
);
if (authExp === undefined) {
authExp = [
"FUNC",
"AUTH",
["PARAM", "username"],
["PARAM", "password"]
] as Expression;
}
let udpProm;
if (udpConnectionRequestAddress) {
udpProm = udpConnectionRequest(
udpConnectionRequestAddress,
authExp,
db.configCollection,
UDP_CONNECTION_REQUEST_PORT
);
}
httpConnectionRequest(
connectionRequestUrl,
authExp,
context,
CONNECTION_REQUEST_ALLOW_BASIC_AUTH,
CONNECTION_REQUEST_TIMEOUT
)
.then(callback)
.catch(err => {
if (udpProm) return void udpProm.then(callback).catch(callback);
callback(err);
});
})
.catch(callback);
});
}
export function watchTask(deviceId, taskId, timeout, callback): void {
setTimeout(() => {
db.tasksCollection.findOne(
{ _id: taskId },
{ projection: { _id: 1 } },
(err, task) => {
if (err) return void callback(err);
if (!task) return void callback(null, "completed");
const q = { _id: `${deviceId}:task_${taskId}` };
db.faultsCollection.findOne(
q,
{ projection: { _id: 1 } },
(err, fault) => {
if (err) return void callback(err);
if (fault) return void callback(null, "fault");
if ((timeout -= 500) <= 0) return void callback(null, "timeout");
watchTask(deviceId, taskId, timeout, callback);
}
);
}
); );
}, 500); }
try {
await httpConnectionRequest(
connectionRequestUrl,
authExp,
context,
CONNECTION_REQUEST_ALLOW_BASIC_AUTH,
CONNECTION_REQUEST_TIMEOUT
);
} catch (err) {
if (!udpProm) throw err;
await udpProm;
}
} }
function sanitizeTask(task, callback): void { export async function watchTask(deviceId, taskId, timeout): Promise<string> {
await new Promise(resolve => setTimeout(resolve, 500));
const task = await db.tasksCollection.findOne(
{ _id: taskId },
{ projection: { _id: 1 } }
);
if (!task) return "completed";
const q = { _id: `${deviceId}:task_${taskId}` };
const fault = await db.faultsCollection.findOne(q, {
projection: { _id: 1 }
});
if (fault) return "fault";
if ((timeout -= 500) <= 0) return "timeout";
return watchTask(deviceId, taskId, timeout);
}
function sanitizeTask(task): void {
task.timestamp = new Date(task.timestamp || Date.now()); task.timestamp = new Date(task.timestamp || Date.now());
if (task.expiry) { if (task.expiry) {
if (common.typeOf(task.expiry) === common.DATE_TYPE || isNaN(task.expiry)) if (common.typeOf(task.expiry) === common.DATE_TYPE || isNaN(task.expiry))
@@ -180,55 +166,40 @@ function sanitizeTask(task, callback): void {
else task.expiry = new Date(task.timestamp.getTime() + +task.expiry * 1000); else task.expiry = new Date(task.timestamp.getTime() + +task.expiry * 1000);
} }
callback(task); return task;
} }
export function insertTasks(tasks, callback): void { export async function insertTasks(tasks): Promise<Task[]> {
if (tasks && common.typeOf(tasks) !== common.ARRAY_TYPE) tasks = [tasks]; if (tasks && common.typeOf(tasks) !== common.ARRAY_TYPE) tasks = [tasks];
else if (!tasks || tasks.length === 0) else if (!tasks || tasks.length === 0) return tasks || [];
return void callback(null, tasks || []);
let counter = tasks.length;
for (const task of tasks) { for (const task of tasks) {
sanitizeTask(task, t => { sanitizeTask(task);
if (t.uniqueKey) { if (task.uniqueKey) {
db.tasksCollection.deleteOne( await db.tasksCollection.deleteOne({
{ device: t.device, uniqueKey: t.uniqueKey }, device: task.device,
() => {} uniqueKey: task.uniqueKey
); });
} }
if (--counter === 0) db.tasksCollection.insertMany(tasks, callback);
});
} }
await db.tasksCollection.insertMany(tasks);
return tasks;
} }
export function deleteDevice(deviceId, callback): void { export async function deleteDevice(deviceId): Promise<void> {
db.tasksCollection.deleteMany({ device: deviceId }, err => { await Promise.all([
if (err) return void callback(err); db.tasksCollection.deleteMany({ device: deviceId }),
db.devicesCollection.deleteOne({ _id: deviceId }, err => { db.devicesCollection.deleteOne({ _id: deviceId }),
if (err) return void callback(err); db.faultsCollection.deleteMany({
db.faultsCollection.deleteMany( _id: {
{ $regex: `^${common.escapeRegExp(deviceId)}\\:`
_id: { }
$regex: `^${common.escapeRegExp(deviceId)}\\:` }),
} db.operationsCollection.deleteMany({
}, _id: {
err => { $regex: `^${common.escapeRegExp(deviceId)}\\:`
if (err) return void callback(err); }
db.operationsCollection.deleteMany( }),
{ cache.del(`${deviceId}_tasks_faults_operations`)
_id: { ]);
$regex: `^${common.escapeRegExp(deviceId)}\\:`
}
},
err => {
if (err) return void callback(err);
cache.del(`${deviceId}_tasks_faults_operations`, callback);
}
);
}
);
});
});
} }
lib/cache.ts
+70 -108
View File
@@ -17,162 +17,124 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { MongoClient } from "mongodb"; import { MongoClient, Collection } from "mongodb";
import * as config from "./config"; import * as config from "./config";
const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL"); const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL");
let mongoClient; let mongoClient: MongoClient;
let mongoCollection; let mongoCollection: Collection;
let mongoTimeOffset = 0; let mongoTimeOffset = 0;
export function connect(callback): void { export async function connect(): Promise<void> {
const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL"); const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
mongoClient = await MongoClient.connect(MONGODB_CONNECTION_URL, {
MongoClient.connect( useNewUrlParser: true
MONGODB_CONNECTION_URL, });
{ useNewUrlParser: true }, const db = mongoClient.db();
(err, client) => { mongoCollection = db.collection("cache");
if (err) return void callback(err); await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
const now = Date.now();
const db = client.db(); const res = await db.command({ hostInfo: 1 });
mongoClient = client; mongoTimeOffset = res.system.currentTime.getTime() - now;
mongoCollection = db.collection("cache");
mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
const now = Date.now();
db.command({ hostInfo: 1 }, (err, res) => {
if (err) return void callback(err);
mongoTimeOffset = res.system.currentTime.getTime() - now;
callback();
});
}
);
} }
export function disconnect(): void { export async function disconnect(): Promise<void> {
if (mongoClient) mongoClient.close(); if (mongoClient) await mongoClient.close();
} }
export function get(key, callback): void { export async function get(key): Promise<any> {
const expire = new Date(Date.now() - mongoTimeOffset); const expire = new Date(Date.now() - mongoTimeOffset);
if (Array.isArray(key)) { if (Array.isArray(key)) {
mongoCollection.find({ _id: { $in: key } }).toArray((err, res) => { const res = await mongoCollection.find({ _id: { $in: key } }).toArray();
if (err) return void callback(err);
const indices = {}; const indices = {};
key.forEach((v, i) => { key.forEach((v, i) => {
indices[v] = i; indices[v] = i;
});
const values = [];
res.forEach(r => {
if (r["expire"] > expire) values[indices[r["_id"]]] = r["value"];
});
callback(null, values);
}); });
const values = [];
res.forEach(r => {
if (r["expire"] > expire) values[indices[r["_id"]]] = r["value"];
});
return values;
} else { } else {
mongoCollection.findOne({ _id: { $in: [key] } }, (err, res) => { const res = await mongoCollection.findOne({ _id: { $in: [key] } });
if (err || !res) return void callback(err); if (res && res["expire"] > expire) return res["value"];
return res;
if (res["expire"] > expire) return void callback(null, res["value"]);
callback();
});
} }
} }
export function del(key, callback): void { export async function del(key): Promise<void> {
if (Array.isArray(key)) if (Array.isArray(key))
mongoCollection.deleteMany({ _id: { $in: key } }, callback); await mongoCollection.deleteMany({ _id: { $in: key } });
else mongoCollection.deleteOne({ _id: key }, callback); else await mongoCollection.deleteOne({ _id: key });
} }
export function set( export async function set(
key: string, key: string,
value: string | number, value: string | number,
callback: (err?: Error) => void ttl: number = MAX_CACHE_TTL
): void; ): Promise<void> {
export function set( const expire = new Date(Date.now() - mongoTimeOffset + ttl * 1000);
key: string, await mongoCollection.replaceOne(
value: string | number,
ttl: number,
callback: (err?: Error) => void
): void;
export function set(
key: string,
value: string | number,
ttl: number | ((err?: Error) => void),
callback?: (err?: Error) => void
): void {
if (!callback && typeof ttl === "function") {
callback = ttl;
ttl = null;
}
if (!ttl) ttl = MAX_CACHE_TTL;
const expire = new Date(
Date.now() - mongoTimeOffset + (ttl as number) * 1000
);
mongoCollection.replaceOne(
{ _id: key }, { _id: key },
{ _id: key, value: value, expire: expire }, { _id: key, value: value, expire: expire },
{ upsert: true }, { upsert: true }
callback
); );
} }
export function pop(key, callback): void { export async function pop(key): Promise<any> {
mongoCollection.findOneAndDelete({ _id: key }, (err, res) => { const res = await mongoCollection.findOneAndDelete({ _id: key });
if (err || !res["value"]) return void callback(err);
if (res["value"]["expire"] > new Date(Date.now() - mongoTimeOffset)) if (
return void callback(null, res["value"]["value"]); res &&
res["value"] &&
+res["value"]["expire"] - (Date.now() - mongoTimeOffset)
)
return res["value"]["value"];
callback(); return null;
});
} }
export function lock(lockName, ttl, callback): void { export async function lock(lockName, ttl): Promise<Function> {
const token = Math.random() const token = Math.random()
.toString(36) .toString(36)
.slice(2); .slice(2);
function unlockOrExtend(extendTtl): void { async function unlockOrExtend(extendTtl): Promise<void> {
if (!extendTtl) { if (!extendTtl) {
mongoCollection.deleteOne({ _id: lockName, value: token }, (err, res) => { const res = await mongoCollection.deleteOne({
if (err || res["result"]["n"] !== 1) _id: lockName,
throw err || new Error("Lock expired"); value: token
}); });
if (res["result"]["n"] !== 1) throw new Error("Lock expired");
} else { } else {
const expire = new Date(Date.now() - mongoTimeOffset + extendTtl * 1000); const expire = new Date(Date.now() - mongoTimeOffset + extendTtl * 1000);
mongoCollection.updateOne( const res = await mongoCollection.updateOne(
{ _id: lockName, value: token }, { _id: lockName, value: token },
{ expire: expire }, { expire: expire }
(err, res) => {
if (err || res["result"]["n"] !== 1)
throw err || new Error("Lock expired");
}
); );
if (res["result"]["n"] !== 1) throw new Error("Lock expired");
} }
} }
const expireTest = new Date(Date.now() - mongoTimeOffset); const expireTest = new Date(Date.now() - mongoTimeOffset);
const expireSet = new Date(Date.now() - mongoTimeOffset + ttl * 1000); const expireSet = new Date(Date.now() - mongoTimeOffset + ttl * 1000);
mongoCollection.updateOne( try {
{ _id: lockName, expire: { $lte: expireTest } }, await mongoCollection.updateOne(
{ $set: { value: token, expire: expireSet } }, { _id: lockName, expire: { $lte: expireTest } },
{ upsert: true }, { $set: { value: token, expire: expireSet } },
err => { { upsert: true }
if (err && err.code === 11000) { );
return setTimeout(() => { } catch (err) {
lock(lockName, ttl, callback); if (err && err.code === 11000) {
}, 200); await new Promise(resolve => setTimeout(resolve, 200));
} return lock(lockName, ttl);
return callback(err, unlockOrExtend);
} }
); }
return unlockOrExtend;
} }
lib/cwmp.ts
+643 -774
View File
File diff suppressed because it is too large. Load diff
lib/db.ts
+392 -424
View File
@@ -17,59 +17,60 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { MongoClient, ObjectID } from "mongodb"; import { MongoClient, ObjectID, Collection, FindOneOptions } from "mongodb";
import { get } from "./config"; import { get } from "./config";
import { escapeRegExp } from "./common"; import { escapeRegExp } from "./common";
import { parse } from "./common/expression-parser"; import { parse } from "./common/expression-parser";
import { DeviceData, Attributes } from "./types"; import {
DeviceData,
Attributes,
SessionFault,
Task,
Operation,
Expression
} from "./types";
import Path from "./common/path"; import Path from "./common/path";
export let tasksCollection, export let tasksCollection: Collection,
devicesCollection, devicesCollection: Collection,
presetsCollection, presetsCollection: Collection,
objectsCollection, objectsCollection: Collection,
provisionsCollection, provisionsCollection: Collection,
virtualParametersCollection, virtualParametersCollection: Collection,
faultsCollection, faultsCollection: Collection,
filesCollection, filesCollection: Collection,
operationsCollection, operationsCollection: Collection,
permissionsCollection, permissionsCollection: Collection,
usersCollection, usersCollection: Collection,
configCollection; configCollection: Collection;
export let client; export let client: MongoClient;
export function connect(callback): void { export async function connect(): Promise<void> {
MongoClient.connect( client = await MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), {
"" + get("MONGODB_CONNECTION_URL"), useNewUrlParser: true
{ useNewUrlParser: true }, });
(err, _client) => {
if (err) return void callback(err);
client = _client;
const db = client.db();
tasksCollection = db.collection("tasks"); const db = client.db();
tasksCollection.createIndex({ device: 1, timestamp: 1 });
devicesCollection = db.collection("devices"); tasksCollection = db.collection("tasks");
presetsCollection = db.collection("presets"); await tasksCollection.createIndex({ device: 1, timestamp: 1 });
objectsCollection = db.collection("objects");
filesCollection = db.collection("fs.files");
provisionsCollection = db.collection("provisions");
virtualParametersCollection = db.collection("virtualParameters");
faultsCollection = db.collection("faults");
operationsCollection = db.collection("operations");
permissionsCollection = db.collection("permissions");
usersCollection = db.collection("users");
configCollection = db.collection("config");
callback(); devicesCollection = db.collection("devices");
} presetsCollection = db.collection("presets");
); objectsCollection = db.collection("objects");
filesCollection = db.collection("fs.files");
provisionsCollection = db.collection("provisions");
virtualParametersCollection = db.collection("virtualParameters");
faultsCollection = db.collection("faults");
operationsCollection = db.collection("operations");
permissionsCollection = db.collection("permissions");
usersCollection = db.collection("users");
configCollection = db.collection("config");
} }
export function disconnect(): void { export async function disconnect(): Promise<void> {
if (client) client.close(); if (client) await client.close();
} }
// Optimize projection by removing overlaps // Optimize projection by removing overlaps
@@ -93,12 +94,11 @@ function optimizeProjection(obj: {}): {} {
return obj; return obj;
} }
export function fetchDevice( export async function fetchDevice(
id: string, id: string,
timestamp: number, timestamp: number,
patterns: [Path, number][], patterns: [Path, number][]
callback ): Promise<[[Path, number, Attributes?][], [Path, number][]]> {
): void {
const MAX_DEPTH = +get("MAX_DEPTH", id); const MAX_DEPTH = +get("MAX_DEPTH", id);
if (!patterns || !patterns.length) if (!patterns || !patterns.length)
@@ -273,259 +273,257 @@ export function fetchDevice(
} }
} }
let options; let options: FindOneOptions;
if (!projection[""]) { if (!projection[""]) {
if (!Object.keys(projection).length) options = { projection: { _id: 1 } }; if (!Object.keys(projection).length) options = { projection: { _id: 1 } };
else options = { projection: projection }; else options = { projection: projection };
} }
devicesCollection.findOne({ _id: id }, options, (err, device) => { const device = await devicesCollection.findOne({ _id: id }, options);
if (err || !device) return void callback(err); if (!device) return [null, null];
function storeParams( function storeParams(
obj, obj,
path: string, path: string,
pathLength: number, pathLength: number,
ts, ts,
descendantsFetched, descendantsFetched,
projTree projTree
): void { ): void {
let thisFetched = false; let thisFetched = false;
if (descendantsFetched) { if (descendantsFetched) {
thisFetched = true; thisFetched = true;
} else if (projection[path.slice(0, -1)]) { } else if (projection[path.slice(0, -1)]) {
descendantsFetched = true; descendantsFetched = true;
if ( if (
pathLength && pathLength &&
projection[ projection[
path.slice(0, path.lastIndexOf(".", path.length - 2) + 1) + path.slice(0, path.lastIndexOf(".", path.length - 2) + 1) +
"_timestamp" "_timestamp"
] ]
) { ) {
thisFetched = true;
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << (pathLength - 1)) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
} else {
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
}
} else if (projection[path + "_writable"]) {
loaded.push([Path.parse(path.slice(0, -1)), 1 << (pathLength - 1)]);
thisFetched = true; thisFetched = true;
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << (pathLength - 1)) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
} else {
loaded.push([
Path.parse(path.slice(0, -1)),
((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)
]);
} }
} else if (projection[path + "_writable"]) {
loaded.push([Path.parse(path.slice(0, -1)), 1 << (pathLength - 1)]);
thisFetched = true;
}
if (obj["_timestamp"]) obj["_timestamp"] = +obj["_timestamp"]; if (obj["_timestamp"]) obj["_timestamp"] = +obj["_timestamp"];
// For compatibility with v1.0 database // For compatibility with v1.0 database
if (obj["_instance"] && obj["_object"] == null) obj["_object"] = true; if (obj["_instance"] && obj["_object"] == null) obj["_object"] = true;
if (thisFetched) { if (thisFetched) {
const attrs: Attributes = {}; const attrs: Attributes = {};
let t = obj["_timestamp"] || 1; let t = obj["_timestamp"] || 1;
if (ts > t) t = ts; if (ts > t) t = ts;
if (obj["_value"] != null) { if (obj["_value"] != null) {
attrs.value = [obj["_timestamp"] || 1, [obj["_value"], obj["_type"]]]; attrs.value = [obj["_timestamp"] || 1, [obj["_value"], obj["_type"]]];
if (obj["_type"] === "xsd:dateTime") if (obj["_type"] === "xsd:dateTime")
attrs.value[1][0] = +attrs.value[1][0]; attrs.value[1][0] = +attrs.value[1][0];
obj["_object"] = false; obj["_object"] = false;
}
if (obj["_writable"] != null)
attrs.writable = [ts || 1, obj["_writable"] ? 1 : 0];
if (obj["_object"] != null) attrs.object = [t, obj["_object"] ? 1 : 0];
res.push([Path.parse(path.slice(0, -1)), t, attrs]);
} }
if (obj["_writable"] != null)
attrs.writable = [ts || 1, obj["_writable"] ? 1 : 0];
for (const [k, v] of Object.entries(obj)) { if (obj["_object"] != null) attrs.object = [t, obj["_object"] ? 1 : 0];
if (!k.startsWith("_")) {
obj["_object"] = true;
storeParams(
v,
path + k + ".",
pathLength + 1,
obj["_timestamp"],
descendantsFetched,
projTree ? projTree[k] : null
);
if (projTree) delete projTree[k];
}
}
if (!descendantsFetched) { res.push([Path.parse(path.slice(0, -1)), t, attrs]);
if (projTree) { }
for (const k of Object.keys(projTree)) {
const p = Path.parse(path + k); for (const [k, v] of Object.entries(obj)) {
loaded.push([p, ((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)]); if (!k.startsWith("_")) {
if ((obj["_object"] || !pathLength) && obj["_timestamp"]) obj["_object"] = true;
res.push([p, obj["_timestamp"]]); storeParams(
} v,
} path + k + ".",
} else if ((obj["_object"] || !pathLength) && obj["_timestamp"]) { pathLength + 1,
res.push([Path.parse(path + "*"), obj["_timestamp"]]); obj["_timestamp"],
descendantsFetched,
projTree ? projTree[k] : null
);
if (projTree) delete projTree[k];
} }
} }
for (const [k, v] of Object.entries(device)) { if (!descendantsFetched) {
switch (k) { if (projTree) {
case "_lastInform": for (const k of Object.keys(projTree)) {
const p = Path.parse(path + k);
loaded.push([p, ((1 << pathLength) - 1) ^ ((1 << MAX_DEPTH) - 1)]);
if ((obj["_object"] || !pathLength) && obj["_timestamp"])
res.push([p, obj["_timestamp"]]);
}
}
} else if ((obj["_object"] || !pathLength) && obj["_timestamp"]) {
res.push([Path.parse(path + "*"), obj["_timestamp"]]);
}
}
for (const [k, v] of Object.entries(device)) {
switch (k) {
case "_lastInform":
res.push([
Path.parse("Events.Inform"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBoot":
res.push([
Path.parse("Events.1_BOOT"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBootstrap":
res.push([
Path.parse("Events.0_BOOTSTRAP"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_registered":
// Use current timestamp for registered event attribute timestamps
res.push([
Path.parse("Events.Registered"),
timestamp,
{
object: [timestamp, 0],
writable: [timestamp, 0],
value: [timestamp, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_id":
if (projection[""] || projection["_id"]) {
res.push([ res.push([
Path.parse("Events.Inform"), Path.parse("DeviceID.ID"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBoot":
res.push([
Path.parse("Events.1_BOOT"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_lastBootstrap":
res.push([
Path.parse("Events.0_BOOTSTRAP"),
+v,
{
object: [+v, 0],
writable: [+v, 0],
value: [+v, [+v, "xsd:dateTime"]]
}
]);
delete device[k];
break;
case "_registered":
// Use current timestamp for registered event attribute timestamps
res.push([
Path.parse("Events.Registered"),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 0], writable: [timestamp, 0],
value: [timestamp, [+v, "xsd:dateTime"]] value: [timestamp, [v as string, "xsd:string"]]
} }
]); ]);
delete device[k]; }
break;
case "_id":
if (projection[""] || projection["_id"]) {
res.push([
Path.parse("DeviceID.ID"),
timestamp,
{
object: [timestamp, 0],
writable: [timestamp, 0],
value: [timestamp, [v as string, "xsd:string"]]
}
]);
}
delete device[k]; delete device[k];
break; break;
case "_tags": case "_tags":
if ((v as string[]).length) { if ((v as string[]).length) {
res.push([ res.push([
Path.parse("Tags"), Path.parse("Tags"),
timestamp, timestamp,
{ object: [timestamp, 1], writable: [timestamp, 0] } { object: [timestamp, 1], writable: [timestamp, 0] }
]); ]);
} }
for (let t of v as string[]) { for (let t of v as string[]) {
t = t.replace(/[^a-zA-Z0-9-]+/g, "_"); t = t.replace(/[^a-zA-Z0-9-]+/g, "_");
res.push([ res.push([
Path.parse("Tags." + t), Path.parse("Tags." + t),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 1], writable: [timestamp, 1],
value: [timestamp, [true, "xsd:boolean"]] value: [timestamp, [true, "xsd:boolean"]]
} }
]); ]);
} }
delete device[k]; delete device[k];
break; break;
case "_deviceId": case "_deviceId":
if (v["_Manufacturer"] != null) { if (v["_Manufacturer"] != null) {
res.push([ res.push([
Path.parse("DeviceID.Manufacturer"), Path.parse("DeviceID.Manufacturer"),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 0], writable: [timestamp, 0],
value: [timestamp, [v["_Manufacturer"], "xsd:string"]] value: [timestamp, [v["_Manufacturer"], "xsd:string"]]
} }
]); ]);
} }
if (v["_OUI"] != null) { if (v["_OUI"] != null) {
res.push([ res.push([
Path.parse("DeviceID.OUI"), Path.parse("DeviceID.OUI"),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 0], writable: [timestamp, 0],
value: [timestamp, [v["_OUI"], "xsd:string"]] value: [timestamp, [v["_OUI"], "xsd:string"]]
} }
]); ]);
} }
if (v["_ProductClass"] != null) { if (v["_ProductClass"] != null) {
res.push([ res.push([
Path.parse("DeviceID.ProductClass"), Path.parse("DeviceID.ProductClass"),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 0], writable: [timestamp, 0],
value: [timestamp, [v["_ProductClass"], "xsd:string"]] value: [timestamp, [v["_ProductClass"], "xsd:string"]]
} }
]); ]);
} }
if (v["_SerialNumber"] != null) { if (v["_SerialNumber"] != null) {
res.push([ res.push([
Path.parse("DeviceID.SerialNumber"), Path.parse("DeviceID.SerialNumber"),
timestamp, timestamp,
{ {
object: [timestamp, 0], object: [timestamp, 0],
writable: [timestamp, 0], writable: [timestamp, 0],
value: [timestamp, [v["_SerialNumber"], "xsd:string"]] value: [timestamp, [v["_SerialNumber"], "xsd:string"]]
} }
]); ]);
} }
delete device[k]; delete device[k];
}
} }
}
storeParams(device, "", 0, 0, false, projectionTree); storeParams(device, "", 0, 0, false, projectionTree);
callback(null, res, loaded); return [res, loaded];
});
} }
export function saveDevice( export async function saveDevice(
deviceId: string, deviceId: string,
deviceData: DeviceData, deviceData: DeviceData,
isNew: boolean, isNew: boolean,
sessionTimestamp: number, sessionTimestamp: number
callback ): Promise<void> {
): void {
const update = { $set: {}, $unset: {}, $addToSet: {}, $pull: {} }; const update = { $set: {}, $unset: {}, $addToSet: {}, $pull: {} };
for (const diff of deviceData.timestamps.diff()) { for (const diff of deviceData.timestamps.diff()) {
@@ -737,7 +735,7 @@ export function saveDevice(
if (!Object.keys(v).length) delete update[k]; if (!Object.keys(v).length) delete update[k];
} }
if (!Object.keys(update).length) return void callback(); if (!Object.keys(update).length) return;
// Mongo doesn't allow $addToSet and $pull at the same time // Mongo doesn't allow $addToSet and $pull at the same time
let update2; let update2;
@@ -746,52 +744,41 @@ export function saveDevice(
delete update["$pull"]; delete update["$pull"];
} }
devicesCollection.updateOne( const result = await devicesCollection.updateOne({ _id: deviceId }, update, {
{ _id: deviceId }, upsert: isNew
update, });
{ upsert: isNew },
(err, result) => {
if (err || result.result.n !== 1) {
return void callback(
err || new Error(`Device ${deviceId} not found in database`)
);
}
if (update2) { if (result.result.n !== 1)
return void devicesCollection.updateOne( throw new Error(`Device ${deviceId} not found in database`);
{ _id: deviceId },
update2,
callback
);
}
callback(); if (update2) {
} await devicesCollection.updateOne({ _id: deviceId }, update2);
); return;
}
} }
export function getFaults(deviceId, callback): void { export async function getFaults(
faultsCollection deviceId
): Promise<{ [channel: string]: SessionFault }> {
const res = await faultsCollection
.find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } }) .find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
.toArray((err, res) => { .toArray();
if (err) return callback(err);
const faults = {}; const faults: { [channel: string]: SessionFault } = {};
for (const r of res) { for (const r of res) {
const channel = r._id.slice(deviceId.length + 1); const channel = r._id.slice(deviceId.length + 1);
delete r._id; delete r._id;
delete r.channel; delete r.channel;
delete r.device; delete r.device;
r.timestamp = +r.timestamp; r.timestamp = +r.timestamp;
r.provisions = JSON.parse(r.provisions); r.provisions = JSON.parse(r.provisions);
faults[channel] = r; faults[channel] = r;
} }
return callback(err, faults); return faults;
});
} }
export function saveFault(deviceId, channel, fault, callback): void { export async function saveFault(deviceId, channel, fault): Promise<void> {
const id = `${deviceId}:${channel}`; const id = `${deviceId}:${channel}`;
fault = Object.assign({}, fault); fault = Object.assign({}, fault);
fault._id = id; fault._id = id;
@@ -799,87 +786,79 @@ export function saveFault(deviceId, channel, fault, callback): void {
fault.channel = channel; fault.channel = channel;
fault.timestamp = new Date(fault.timestamp); fault.timestamp = new Date(fault.timestamp);
fault.provisions = JSON.stringify(fault.provisions); fault.provisions = JSON.stringify(fault.provisions);
faultsCollection.replaceOne({ _id: id }, fault, { upsert: true }, callback); await faultsCollection.replaceOne({ _id: id }, fault, { upsert: true });
} }
export function deleteFault(deviceId, channel, callback): void { export async function deleteFault(deviceId, channel): Promise<void> {
faultsCollection.deleteOne({ _id: `${deviceId}:${channel}` }, callback); await faultsCollection.deleteOne({ _id: `${deviceId}:${channel}` });
} }
export function getDueTasks(deviceId, timestamp, callback): void { export async function getDueTasks(
deviceId,
timestamp
): Promise<[Task[], number]> {
const cur = tasksCollection.find({ device: deviceId }).sort(["timestamp"]); const cur = tasksCollection.find({ device: deviceId }).sort(["timestamp"]);
const tasks = []; const tasks = [] as Task[];
let f; for await (const task of cur) {
cur.next( if (task.timestamp) task.timestamp = +task.timestamp;
(f = (err, task) => { if (task.expiry) task.expiry = +task.expiry;
if (err) return void callback(err); if (task.timestamp >= timestamp) return [tasks, +task.timestamp];
task._id = String(task._id);
if (!task) return void callback(null, tasks, null); tasks.push(task);
if (task.timestamp) task.timestamp = +task.timestamp; // For API compatibility
if (task.expiry) task.expiry = +task.expiry; if (task.name === "download" && task.file) {
if (task.timestamp >= timestamp) let q;
return void callback(null, tasks, +task.timestamp); if (ObjectID.isValid(task.file))
task._id = String(task._id); q = { _id: { $in: [task.file, new ObjectID(task.file)] } };
else q = { _id: task.file };
tasks.push(task); const res = await filesCollection.find(q).toArray();
// For API compatibility if (res[0]) {
if (task.name === "download" && task.file) { if (!task.fileType) task.fileType = res[0].metadata.fileType;
let q;
if (ObjectID.isValid(task.file))
q = { _id: { $in: [task.file, new ObjectID(task.file)] } };
else q = { _id: task.file };
filesCollection.find(q).toArray((err, res) => { if (!task.fileName)
if (err) return void callback(err); task.fileName = res[0].filename || res[0]._id.toString();
if (res[0]) {
if (!task.fileType) task.fileType = res[0].metadata.fileType;
if (!task.fileName)
task.fileName = res[0].filename || res[0]._id.toString();
}
cur.next(f);
});
} else {
cur.next(f);
} }
}) }
); }
return [tasks, null];
} }
export function clearTasks(deviceId, taskIds, callback): void { export async function clearTasks(deviceId, taskIds): Promise<void> {
tasksCollection.deleteMany( await tasksCollection.deleteMany({
{ _id: { $in: taskIds.map(id => new ObjectID(id)) } }, _id: { $in: taskIds.map(id => new ObjectID(id)) }
callback });
);
} }
export function getOperations(deviceId, callback): void { export async function getOperations(
operationsCollection deviceId
): Promise<{ [commandKey: string]: Operation }> {
const res = await operationsCollection
.find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } }) .find({ _id: { $regex: `^${escapeRegExp(deviceId)}\\:` } })
.toArray((err, res) => { .toArray();
if (err) return void callback(err);
const operations = {}; const operations: { [commandKey: string]: Operation } = {};
for (const r of res) { for (const r of res) {
const commandKey = r._id.slice(deviceId.length + 1); const commandKey = r._id.slice(deviceId.length + 1);
delete r._id; delete r._id;
r.timestamp = +r.timestamp; r.timestamp = +r.timestamp;
if (r.args) r.args = JSON.parse(r.args); if (r.args) r.args = JSON.parse(r.args);
r.provisions = JSON.parse(r.provisions); r.provisions = JSON.parse(r.provisions);
r.retries = JSON.parse(r.retries); r.retries = JSON.parse(r.retries);
operations[commandKey] = r; operations[commandKey] = r;
} }
return operations;
callback(err, operations);
});
} }
export function saveOperation(deviceId, commandKey, operation, callback): void { export async function saveOperation(
deviceId,
commandKey,
operation
): Promise<void> {
const id = `${deviceId}:${commandKey}`; const id = `${deviceId}:${commandKey}`;
operation = Object.assign({}, operation); operation = Object.assign({}, operation);
operation._id = id; operation._id = id;
@@ -887,75 +866,64 @@ export function saveOperation(deviceId, commandKey, operation, callback): void {
operation.provisions = JSON.stringify(operation.provisions); operation.provisions = JSON.stringify(operation.provisions);
operation.retries = JSON.stringify(operation.retries); operation.retries = JSON.stringify(operation.retries);
operation.args = JSON.stringify(operation.args); operation.args = JSON.stringify(operation.args);
operationsCollection.replaceOne( await operationsCollection.replaceOne({ _id: id }, operation, {
{ _id: id }, upsert: true
operation,
{ upsert: true },
callback
);
}
export function deleteOperation(deviceId, commandKey, callback): void {
operationsCollection.deleteOne(
{ _id: `${deviceId}:${commandKey}` },
callback
);
}
export function getPresets(callback): void {
presetsCollection.find().toArray(callback);
}
export function getObjects(callback): void {
objectsCollection.find().toArray(callback);
}
export function getProvisions(callback): void {
provisionsCollection.find().toArray(callback);
}
export function getVirtualParameters(callback): void {
virtualParametersCollection.find().toArray(callback);
}
export function getFiles(callback): void {
filesCollection.find().toArray(callback);
}
export function getConfig(callback): void {
configCollection.find().toArray((err, res) => {
if (err) return void callback(err);
callback(
null,
res.map(c => ({
id: c["_id"],
value: parse(c["value"])
}))
);
}); });
} }
export function getPermissions( export async function deleteOperation(deviceId, commandKey): Promise<void> {
callback: ( await operationsCollection.deleteOne({ _id: `${deviceId}:${commandKey}` });
err?: Error,
permissions?: {
role: string;
resource: string;
access: number;
filter: string;
validate: string;
}[]
) => void
): void {
permissionsCollection.find().toArray(callback);
} }
export function getUsers( export async function getPresets(): Promise<{}[]> {
callback: ( return presetsCollection.find().toArray();
err?: Error, }
users?: { _id: string; password: string; salt: string; roles: string }[]
) => void export async function getObjects(): Promise<{}[]> {
): void { return objectsCollection.find().toArray();
usersCollection.find().toArray(callback); }
export async function getProvisions(): Promise<{}[]> {
return provisionsCollection.find().toArray();
}
export async function getVirtualParameters(): Promise<{}[]> {
return virtualParametersCollection.find().toArray();
}
export function getFiles(): Promise<{}[]> {
return filesCollection.find().toArray();
}
export async function getConfig(): Promise<
{ id: string; value: Expression }[]
> {
const res = await configCollection.find().toArray();
return res.map(c => ({
id: c["_id"],
value: parse(c["value"])
}));
}
interface Permission {
role: string;
resource: string;
access: number;
filter: string;
validate: string;
}
export async function getPermissions(): Promise<Permission[]> {
return permissionsCollection.find().toArray();
}
interface User {
_id: string;
password: string;
salt: string;
roles: string;
}
export async function getUsers(): Promise<User[]> {
return usersCollection.find().toArray();
} }
lib/extensions.ts
+70 -65
View File
@@ -17,92 +17,97 @@
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>. * along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/ */
import { spawn } from "child_process"; import { spawn, ChildProcess } from "child_process";
import * as crypto from "crypto"; import * as crypto from "crypto";
import * as config from "./config"; import * as config from "./config";
import { Fault } from "./types";
const TIMEOUT = +config.get("EXT_TIMEOUT"); const TIMEOUT = +config.get("EXT_TIMEOUT");
const processes = {}; const processes: { [script: string]: ChildProcess } = {};
const jobs = new Map(); const jobs = new Map();
function messageHandler(message): void { function messageHandler(message): void {
const func = jobs.get(message[0]); const func = jobs.get(message[0]);
if (func) { if (func) {
jobs.delete(message[0]); jobs.delete(message[0]);
func(null, message[1], message[2]); func({ fault: message[1], value: message[2] });
} }
} }
export function run(args, callback): boolean { export function run(args): Promise<{ fault: Fault; value: any }> {
const scriptName = args[0]; return new Promise(resolve => {
const scriptName = args[0];
const id = crypto.randomBytes(8).toString("hex"); const id = crypto.randomBytes(8).toString("hex");
jobs.set(id, callback); jobs.set(id, resolve);
if (!processes[scriptName]) { if (!processes[scriptName]) {
const p = spawn("bin/genieacs-ext", [scriptName], { const p = spawn("bin/genieacs-ext", [scriptName], {
stdio: ["inherit", "inherit", "inherit", "ipc"] stdio: ["inherit", "inherit", "inherit", "ipc"]
}); });
processes[scriptName] = p; processes[scriptName] = p;
p.on("error", err => { p.on("error", err => {
if (processes[scriptName] === p) { if (processes[scriptName] === p) {
if (jobs.delete(id)) if (jobs.delete(id)) {
callback(null, { code: err.name, message: err.message }); resolve({
fault: { code: err.name, message: err.message },
value: null
});
}
kill(processes[scriptName]); kill(processes[scriptName]);
delete processes[scriptName]; delete processes[scriptName];
} }
}); });
p.on("disconnect", () => { p.on("disconnect", () => {
if (processes[scriptName] === p) { if (processes[scriptName] === p) delete processes[scriptName];
kill(processes[scriptName]); });
delete processes[scriptName];
}
});
p.on("message", messageHandler); p.on("message", messageHandler);
}
setTimeout(() => {
if (jobs.delete(id))
callback(null, { code: "timeout", message: "Extension timed out" });
}, TIMEOUT);
if (!processes[scriptName].connected) return false;
return processes[scriptName].send([id, args.slice(1)]);
}
function kill(process, callback?: () => void): void {
if (process.signalCode !== null || process.exitCode !== null) return;
const timeToKill = Date.now() + 5000;
process.kill();
const t = setInterval(() => {
if (process.signalCode !== null || process.exitCode !== null) {
clearInterval(t);
if (callback) callback();
} else if (Date.now() > timeToKill) {
process.kill("SIGKILL");
clearInterval(t);
if (callback) callback();
} }
}, 100);
setTimeout(() => {
if (jobs.delete(id)) {
resolve({
fault: { code: "timeout", message: "Extension timed out" },
value: null
});
}
}, TIMEOUT);
if (!processes[scriptName].connected) return false;
return processes[scriptName].send([id, args.slice(1)]);
});
} }
export function killAll(callback?): void { function kill(process: ChildProcess): Promise<void> {
let counter = 1; return new Promise(resolve => {
for (const n in processes) { const timeToKill = Date.now() + 5000;
++counter;
kill(processes[n], () => { process.kill();
if (--counter === 0 && callback) callback();
}); const t = setInterval(() => {
delete processes[n]; if (!process.connected) {
} clearInterval(t);
if (--counter === 0 && callback) callback(); resolve();
} else if (Date.now() > timeToKill) {
process.kill("SIGKILL");
clearInterval(t);
resolve();
}
}, 100);
});
}
export async function killAll(): Promise<void> {
await Promise.all(
Object.entries(processes).map(([k, p]) => {
delete processes[k];
return kill(p);
})
);
} }
lib/local-cache.ts
+309 -374
View File
@@ -108,391 +108,326 @@ function flattenObject(src, prefix = "", dst = {}): {} {
return dst; return dst;
} }
function refresh(callback): void { async function fetchPresets(): Promise<Preset[]> {
const res = await db.getPresets();
let objects = await db.getObjects();
objects = objects.map(obj => {
// Flatten object
obj = flattenObject(obj);
// If no keys are defined, consider all parameters as keys to keep the
// same behavior from v1.0
if (!obj["_keys"] || !obj["_keys"].length)
obj["_keys"] = Object.keys(obj).filter(k => !k.startsWith("_"));
return obj;
});
res.sort((a, b) => {
if (a["weight"] === b["weight"])
return a["_id"] > b["_id"] ? 1 : a["_id"] < b["_id"] ? -1 : 0;
else return a["weight"] - b["weight"];
});
const presets = [] as Preset[];
for (const preset of res) {
let schedule: { md5: string; duration: number; schedule: any } = null;
if (preset["schedule"]) {
const parts = preset["schedule"].trim().split(/\s+/);
schedule = {
md5: crypto
.createHash("md5")
.update(preset["schedule"])
.digest("hex"),
duration: null,
schedule: null
};
try {
schedule.duration = +parts.shift() * 1000;
schedule.schedule = scheduling.parseCron(parts.join(" "));
} catch (err) {
logger.warn({
message: "Invalid preset schedule",
preset: preset["_id"],
schedule: preset["schedule"]
});
schedule.schedule = false;
}
}
const events = preset["events"] || {};
let precondition;
try {
precondition = parse(preset["precondition"]);
} catch (error) {
precondition = mongoQueryToFilter(JSON.parse(preset["precondition"]));
}
const _provisions = preset["provisions"] || [];
// Generate provisions from the old configuration format
for (const c of preset["configurations"]) {
switch (c.type) {
case "age":
_provisions.push(["refresh", c.name, c.age]);
break;
case "value":
_provisions.push(["value", c.name, c.value]);
break;
case "add_tag":
_provisions.push(["tag", c.tag, true]);
break;
case "delete_tag":
_provisions.push(["tag", c.tag, false]);
break;
case "provision":
_provisions.push([c.name].concat(c.args || []));
break;
case "add_object":
for (const obj of objects) {
if (obj["_id"] === c.object) {
const alias = obj["_keys"]
.map(k => `${k}:${JSON.stringify(obj[k])}`)
.join(",");
const p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 1]);
for (const k in obj) {
if (!k.startsWith("_") && !(obj["_keys"].indexOf(k) !== -1))
_provisions.push(["value", `${p}.${k}`, obj[k]]);
}
}
}
break;
case "delete_object":
for (const obj of objects) {
if (obj["_id"] === c.object) {
const alias = obj["_keys"]
.map(k => `${k}:${JSON.stringify(obj[k])}`)
.join(",");
const p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 0]);
}
}
break;
default:
throw new Error(`Unknown configuration type ${c.type}`);
}
}
presets.push({
name: preset["_id"],
channel: (preset["channel"] as string) || "default",
schedule: schedule,
events: events,
precondition: precondition,
provisions: _provisions
});
}
return presets;
}
async function fetchProvisions(): Promise<{}> {
const res = await db.getProvisions();
const provisions = {};
for (const r of res) {
provisions[r["_id"]] = {};
provisions[r["_id"]].md5 = crypto
.createHash("md5")
.update(r["script"])
.digest("hex");
provisions[r["_id"]].script = new vm.Script(
`"use strict";(function(){\n${r["script"]}\n})();`,
{ filename: r["_id"], lineOffset: -1, timeout: 50 }
);
}
return provisions;
}
async function fetchVirtualParameters(): Promise<{}> {
const res = await db.getVirtualParameters();
const virtualParameters = {};
for (const r of res) {
virtualParameters[r["_id"]] = {};
virtualParameters[r["_id"]].md5 = crypto
.createHash("md5")
.update(r["script"])
.digest("hex");
virtualParameters[r["_id"]].script = new vm.Script(
`"use strict";(function(){\n${r["script"]}\n})();`,
{ filename: r["_id"], lineOffset: -1, timeout: 50 }
);
}
return virtualParameters;
}
async function fetchPermissions(): Promise<Permissions> {
const perms = await db.getPermissions();
const permissions: Permissions = {};
for (const p of perms) {
if (!permissions[p.role]) permissions[p.role] = {};
if (!permissions[p.role][p.access]) permissions[p.role][p.access] = {};
permissions[p.role][p.access][p.resource] = {
access: p.access,
filter: parse(p.filter || "true")
};
if (p.validate)
permissions[p.role][p.access][p.resource].validate = parse(p.validate);
}
return permissions;
}
async function fetchFiles(): Promise<{}> {
const res = await db.getFiles();
const files = {};
for (const r of res) {
const id = r["filename"] || r["_id"].toString();
files[id] = {};
files[id].length = r["length"];
files[id].md5 = r["md5"];
files[id].contentType = r["contentType"];
}
return files;
}
async function fetchUsers(): Promise<{}> {
const _users = await db.getUsers();
const users = {};
for (const user of _users) {
users[user._id] = {
password: user.password,
salt: user.salt,
roles: user.roles.split(",").map(s => s.trim())
};
}
return users;
}
async function fetchConfig(): Promise<[{}, {}]> {
const conf = await db.getConfig();
conf.sort((a, b) => (a.id > b.id ? 1 : a.id < b.id ? -1 : 0));
const ui = {
filters: {},
device: {},
index: {},
overview: { charts: {}, groups: {} }
};
const _config = {};
for (const c of conf) {
// Evaluate expressions to simplify them
const val = expression.evaluate(c.value);
_config[c.id] = val;
if (c.id.startsWith("ui.")) {
const keys = c.id.split(".");
// remove the first key(ui)
keys.shift();
let ref = ui;
while (keys.length > 1) {
const k = keys.shift();
if (typeof ref[k] !== "object") ref[k] = {};
ref = ref[k];
}
ref[keys[0]] = val;
}
}
if (!Object.keys(ui["index"]).length) {
ui["index"] = {
"0": {
label: "ID",
parameter: ["PARAM", "DeviceID.ID"]
}
};
}
return [_config, ui];
}
async function refresh(): Promise<void> {
if (!nextRefresh) { if (!nextRefresh) {
return void setTimeout(() => { await new Promise(resolve => setTimeout(resolve, 20));
refresh(callback); await refresh();
}, 20); return;
} }
nextRefresh = 0; nextRefresh = 0;
const now = Date.now(); const now = Date.now();
cache.get("presets_hash", (err, dbHash) => { const dbHash = await cache.get("presets_hash");
if (err) return void callback(err);
if (currentSnapshot && dbHash === currentSnapshot) { if (currentSnapshot && dbHash === currentSnapshot) {
nextRefresh = now + (REFRESH - (now % REFRESH)); nextRefresh = now + (REFRESH - (now % REFRESH));
return void callback(); return;
}
cache.lock("presets_hash_lock", 3, (err, unlockOrExtend) => {
if (err) return void callback(err);
const promises = [];
promises.push(
new Promise((resolve, reject) => {
db.getPresets((err, res) => {
if (err) return void reject(err);
db.getObjects((err, objects) => {
if (err) return void reject(err);
objects = objects.map(obj => {
// Flatten object
obj = flattenObject(obj);
// If no keys are defined, consider all parameters as keys to keep the
// same behavior from v1.0
if (!obj["_keys"] || !obj["_keys"].length) {
obj["_keys"] = Object.keys(obj).filter(
k => !k.startsWith("_")
);
}
return obj;
});
res.sort((a, b) => {
if (a.weight === b.weight) return a._id > b._id;
else return a.weight - b.weight;
});
const presets = [] as Preset[];
for (const preset of res) {
let schedule = null;
if (preset.schedule) {
const parts = preset.schedule.trim().split(/\s+/);
schedule = {
md5: crypto
.createHash("md5")
.update(preset.schedule)
.digest("hex")
};
try {
schedule.duration = +parts.shift() * 1000;
schedule.schedule = scheduling.parseCron(parts.join(" "));
} catch (err) {
logger.warn({
message: "Invalid preset schedule",
preset: preset._id,
schedule: preset.schedule
});
schedule.schedule = false;
}
}
const events = preset.events || {};
let precondition;
try {
precondition = parse(preset.precondition);
} catch (error) {
precondition = mongoQueryToFilter(
JSON.parse(preset.precondition)
);
}
const _provisions = preset.provisions || [];
// Generate provisions from the old configuration format
for (const c of preset.configurations) {
switch (c.type) {
case "age":
_provisions.push(["refresh", c.name, c.age]);
break;
case "value":
_provisions.push(["value", c.name, c.value]);
break;
case "add_tag":
_provisions.push(["tag", c.tag, true]);
break;
case "delete_tag":
_provisions.push(["tag", c.tag, false]);
break;
case "provision":
_provisions.push([c.name].concat(c.args || []));
break;
case "add_object":
for (const obj of objects) {
if (obj["_id"] === c.object) {
const alias = obj["_keys"]
.map(k => `${k}:${JSON.stringify(obj[k])}`)
.join(",");
const p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 1]);
for (const k in obj) {
if (
!k.startsWith("_") &&
!(obj["_keys"].indexOf(k) !== -1)
)
_provisions.push(["value", `${p}.${k}`, obj[k]]);
}
}
}
break;
case "delete_object":
for (const obj of objects) {
if (obj["_id"] === c.object) {
const alias = obj["_keys"]
.map(k => `${k}:${JSON.stringify(obj[k])}`)
.join(",");
const p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 0]);
}
}
break;
default:
return void reject(
new Error(`Unknown configuration type ${c.type}`)
);
}
}
presets.push({
name: preset._id,
channel: preset.channel || "default",
schedule: schedule,
events: events,
precondition: precondition,
provisions: _provisions
});
}
resolve(presets);
});
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getProvisions((err, res) => {
if (err) return void reject(err);
const provisions = {};
for (const r of res) {
provisions[r._id] = {};
provisions[r._id].md5 = crypto
.createHash("md5")
.update(r.script)
.digest("hex");
provisions[r._id].script = new vm.Script(
`"use strict";(function(){\n${r.script}\n})();`,
{ filename: r._id, lineOffset: -1, timeout: 50 }
);
}
resolve(provisions);
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getVirtualParameters((err, res) => {
if (err) return void reject(err);
const virtualParameters = {};
for (const r of res) {
virtualParameters[r._id] = {};
virtualParameters[r._id].md5 = crypto
.createHash("md5")
.update(r.script)
.digest("hex");
virtualParameters[r._id].script = new vm.Script(
`"use strict";(function(){\n${r.script}\n})();`,
{ filename: r._id, lineOffset: -1, timeout: 50 }
);
}
resolve(virtualParameters);
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getFiles((err, res) => {
if (err) return void reject(err);
const files = {};
for (const r of res) {
const id = r.filename || r._id.toString();
files[id] = {};
files[id].length = r.length;
files[id].md5 = r.md5;
files[id].contentType = r.contentType;
}
resolve(files);
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getPermissions((err, perms) => {
if (err) return void reject(err);
const permissions: Permissions = {};
for (const p of perms) {
if (!permissions[p.role]) permissions[p.role] = {};
if (!permissions[p.role][p.access])
permissions[p.role][p.access] = {};
permissions[p.role][p.access][p.resource] = {
access: p.access,
filter: parse(p.filter || "true")
};
if (p.validate) {
permissions[p.role][p.access][p.resource].validate = parse(
p.validate
);
}
}
resolve(permissions);
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getUsers((err, _users) => {
if (err) return void reject(err);
const users = {};
for (const user of _users) {
users[user._id] = {
password: user.password,
salt: user.salt,
roles: user.roles.split(",").map(s => s.trim())
};
}
resolve(users);
});
})
);
promises.push(
new Promise((resolve, reject) => {
db.getConfig((err, conf) => {
if (err) return void reject(err);
conf.sort((a, b) => (a.id > b.id ? 1 : a.id < b.id ? -1 : 0));
const ui = {
filters: {},
device: {},
index: {},
overview: { charts: {}, groups: {} }
};
const _config = {};
for (const c of conf) {
// Evaluate expressions to simplify them
const val = expression.evaluate(c.value);
_config[c.id] = val;
if (c.id.startsWith("ui.")) {
const keys = c.id.split(".");
// remove the first key(ui)
keys.shift();
let ref = ui;
while (keys.length > 1) {
const k = keys.shift();
if (typeof ref[k] !== "object") ref[k] = {};
ref = ref[k];
}
ref[keys[0]] = val;
}
}
if (!Object.keys(ui["index"]).length) {
ui["index"] = {
"0": {
label: "ID",
parameter: ["PARAM", "DeviceID.ID"]
}
};
}
resolve([_config, ui]);
});
})
);
Promise.all(promises)
.then(res => {
const snapshot = {
presets: res[0],
provisions: res[1],
virtualParameters: res[2],
files: res[3],
permissions: res[4],
users: res[5],
config: res[6][0],
ui: res[6][1]
};
if (currentSnapshot) {
const h = currentSnapshot;
const s = snapshots.get(h);
setTimeout(() => {
if (snapshots.get(h) === s) snapshots.delete(h);
}, EVICT_TIMEOUT).unref();
}
currentSnapshot = computeHash(snapshot);
snapshots.set(currentSnapshot, snapshot);
cache.set("presets_hash", currentSnapshot, 300, err => {
unlockOrExtend(0);
if (err) return void callback(err);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback();
});
})
.catch(callback);
});
});
}
interface GetCurrentSnapshotCallback {
(err: null, currentSnapshot: string): void;
(err: Error, currentSnapshot?: null): void;
}
export function getCurrentSnapshot(callback: GetCurrentSnapshotCallback);
export function getCurrentSnapshot(): Promise<string>;
export function getCurrentSnapshot(
callback?: GetCurrentSnapshotCallback
): Promise<string> | void {
if (Date.now() < nextRefresh) {
if (callback) return callback(null, currentSnapshot);
return Promise.resolve(currentSnapshot);
} }
if (callback) { const unlockOrExtend = await cache.lock("presets_hash_lock", 3);
return refresh(err => {
callback(err, currentSnapshot); const res = await Promise.all([
}); fetchPresets(),
fetchProvisions(),
fetchVirtualParameters(),
fetchFiles(),
fetchPermissions(),
fetchUsers(),
fetchConfig()
]);
const snapshot = {
presets: res[0],
provisions: res[1],
virtualParameters: res[2],
files: res[3],
permissions: res[4],
users: res[5],
config: res[6][0],
ui: res[6][1]
};
if (currentSnapshot) {
const h = currentSnapshot;
const s = snapshots.get(h);
setTimeout(() => {
if (snapshots.get(h) === s) snapshots.delete(h);
}, EVICT_TIMEOUT).unref();
} }
return new Promise((resolve, reject) => { currentSnapshot = computeHash(snapshot);
refresh(err => { snapshots.set(currentSnapshot, snapshot);
if (err) return void reject(err); await cache.set("presets_hash", currentSnapshot, 300);
resolve(currentSnapshot); await unlockOrExtend(0);
});
}); nextRefresh = now + (REFRESH - (now % REFRESH));
}
export async function getCurrentSnapshot(): Promise<string> {
if (Date.now() > nextRefresh) await refresh();
return currentSnapshot;
} }
export function hasSnapshot(hash): boolean { export function hasSnapshot(hash): boolean {
lib/nbi.ts
+220 -145
View File
@@ -91,24 +91,34 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.presetsCollection.deleteOne({ _id: presetName }, err => { db.presetsCollection.deleteOne({ _id: presetName }, err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
}); });
} else { } else {
response.writeHead(405, { Allow: "PUT, DELETE" }); response.writeHead(405, { Allow: "PUT, DELETE" });
@@ -128,24 +138,34 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.objectsCollection.deleteOne({ _id: objectName }, err => { db.objectsCollection.deleteOne({ _id: objectName }, err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
}); });
} else { } else {
response.writeHead(405, { Allow: "PUT, DELETE" }); response.writeHead(405, { Allow: "PUT, DELETE" });
@@ -176,24 +196,34 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
db.provisionsCollection.deleteOne({ _id: provisionName }, err => { db.provisionsCollection.deleteOne({ _id: provisionName }, err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
}); });
} else { } else {
response.writeHead(405, { Allow: "PUT, DELETE" }); response.writeHead(405, { Allow: "PUT, DELETE" });
@@ -224,12 +254,17 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} else if (request.method === "DELETE") { } else if (request.method === "DELETE") {
@@ -238,12 +273,17 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del("presets_hash", err => { cache
if (err) return void throwError(err, response); .del("presets_hash")
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} else { } else {
@@ -258,7 +298,6 @@ export function listener(request, response): void {
db.devicesCollection.updateOne( db.devicesCollection.updateOne(
{ _id: deviceId }, { _id: deviceId },
{ $addToSet: { _tags: tag } }, { $addToSet: { _tags: tag } },
{ safe: true },
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
response.writeHead(200); response.writeHead(200);
@@ -269,7 +308,6 @@ export function listener(request, response): void {
db.devicesCollection.updateOne( db.devicesCollection.updateOne(
{ _id: deviceId }, { _id: deviceId },
{ $pull: { _tags: tag } }, { $pull: { _tags: tag } },
{ safe: true },
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
@@ -296,21 +334,31 @@ export function listener(request, response): void {
return void db.tasksCollection.deleteOne({ _id: objId }, err => { return void db.tasksCollection.deleteOne({ _id: objId }, err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del(`${deviceId}_tasks_faults_operations`, err => { cache
if (err) return void throwError(err, response); .del(`${deviceId}_tasks_faults_operations`)
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
}); });
} }
cache.del(`${deviceId}_tasks_faults_operations`, err => { cache
if (err) return void throwError(err, response); .del(`${deviceId}_tasks_faults_operations`)
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
}); });
} else { } else {
response.writeHead(405, { Allow: "DELETE" }); response.writeHead(405, { Allow: "DELETE" });
@@ -324,83 +372,95 @@ export function listener(request, response): void {
if (body.length) { if (body.length) {
const task = JSON.parse(body.toString()); const task = JSON.parse(body.toString());
task.device = deviceId; task.device = deviceId;
apiFunctions.insertTasks(task, err => { apiFunctions
if (err) return void throwError(err, response); .insertTasks(task)
.then(() => {
cache
.del(`${deviceId}_tasks_faults_operations`)
.then(() => {
if (urlParts.query.connection_request != null) {
apiFunctions
.connectionRequest(deviceId)
.then(() => {
const taskTimeout =
(urlParts.query.timeout &&
parseInt(urlParts.query.timeout as string)) ||
config.get("DEVICE_ONLINE_THRESHOLD", deviceId);
cache.del(`${deviceId}_tasks_faults_operations`, err => { apiFunctions
if (err) return void throwError(err, response); .watchTask(deviceId, task._id, taskTimeout)
.then(status => {
if (status === "timeout") {
response.writeHead(
202,
"Task queued but not processed",
{
"Content-Type": "application/json"
}
);
response.end(JSON.stringify(task));
} else if (status === "fault") {
db.tasksCollection.findOne(
{ _id: task._id },
(err, task2) => {
if (err)
return void throwError(err, response);
if (urlParts.query.connection_request != null) { response.writeHead(202, "Task faulted", {
apiFunctions.connectionRequest(deviceId, err => { "Content-Type": "application/json"
if (err) { });
response.writeHead(202, err.message, { response.end(JSON.stringify(task2));
"Content-Type": "application/json" }
}); );
response.end(JSON.stringify(task)); } else {
return; response.writeHead(200, {
} "Content-Type": "application/json"
});
const taskTimeout = response.end(JSON.stringify(task));
(urlParts.query.timeout && }
parseInt(urlParts.query.timeout as string)) || })
config.get("DEVICE_ONLINE_THRESHOLD", deviceId); .catch(err => {
setTimeout(() => {
apiFunctions.watchTask( throwError(err, response);
deviceId,
task._id,
taskTimeout,
(err, status) => {
if (err) return void throwError(err, response);
if (status === "timeout") {
response.writeHead(
202,
"Task queued but not processed",
{
"Content-Type": "application/json"
}
);
response.end(JSON.stringify(task));
} else if (status === "fault") {
db.tasksCollection.findOne(
{ _id: task._id },
(err, task2) => {
if (err) return void throwError(err, response);
response.writeHead(202, "Task faulted", {
"Content-Type": "application/json"
}); });
response.end(JSON.stringify(task2)); });
} })
); .catch(err => {
} else { response.writeHead(202, err.message, {
response.writeHead(200, {
"Content-Type": "application/json" "Content-Type": "application/json"
}); });
response.end(JSON.stringify(task)); response.end(JSON.stringify(task));
} });
} } else {
); response.writeHead(202, {
"Content-Type": "application/json"
});
response.end(JSON.stringify(task));
}
})
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
}); });
} else { })
response.writeHead(202, { .catch(err => {
"Content-Type": "application/json" setTimeout(() => {
}); throwError(err, response);
response.end(JSON.stringify(task)); });
}
}); });
});
} else if (urlParts.query.connection_request != null) { } else if (urlParts.query.connection_request != null) {
// No task, send connection request only // No task, send connection request only
apiFunctions.connectionRequest(deviceId, err => { apiFunctions
if (err) { .connectionRequest(deviceId)
.then(() => {
response.writeHead(200);
response.end();
})
.catch(err => {
response.writeHead(504); response.writeHead(504);
response.end(`${err.name}: ${err.message}`); response.end(`${err.name}: ${err.message}`);
return; });
}
response.writeHead(200);
response.end();
});
} else { } else {
response.writeHead(400); response.writeHead(400);
response.end(); response.end();
@@ -438,12 +498,17 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del(`${deviceId}_tasks_faults_operations`, err => { cache
if (err) return void throwError(err, response); .del(`${deviceId}_tasks_faults_operations`)
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} }
@@ -468,12 +533,17 @@ export function listener(request, response): void {
err => { err => {
if (err) return void throwError(err, response); if (err) return void throwError(err, response);
cache.del(`${deviceId}_tasks_faults_operations`, err => { cache
if (err) return void throwError(err, response); .del(`${deviceId}_tasks_faults_operations`)
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} }
); );
} }
@@ -547,12 +617,17 @@ export function listener(request, response): void {
const deviceId = querystring.unescape( const deviceId = querystring.unescape(
DELETE_DEVICE_REGEX.exec(urlParts.pathname)[1] DELETE_DEVICE_REGEX.exec(urlParts.pathname)[1]
); );
apiFunctions.deleteDevice(deviceId, err => { apiFunctions
if (err) return void throwError(err, response); .deleteDevice(deviceId)
.then(() => {
response.writeHead(200); response.writeHead(200);
response.end(); response.end();
}); })
.catch(err => {
setTimeout(() => {
throwError(err, response);
});
});
} else if (QUERY_REGEX.test(urlParts.pathname)) { } else if (QUERY_REGEX.test(urlParts.pathname)) {
let collectionName = QUERY_REGEX.exec(urlParts.pathname)[1]; let collectionName = QUERY_REGEX.exec(urlParts.pathname)[1];
lib/sandbox.ts
+71 -67
View File
@@ -24,7 +24,7 @@ import * as extensions from "./extensions";
import * as logger from "./logger"; import * as logger from "./logger";
import * as scheduling from "./scheduling"; import * as scheduling from "./scheduling";
import Path from "./common/path"; import Path from "./common/path";
import { Fault, SessionContext } from "./types"; import { Fault, SessionContext, ScriptResult } from "./types";
// Used for throwing to exit user script and commit // Used for throwing to exit user script and commit
const COMMIT = Symbol(); const COMMIT = Symbol();
@@ -38,25 +38,35 @@ const context = vm.createContext();
let state; let state;
const runningExtensions = new WeakMap(); const runningExtensions = new WeakMap<
function runExtension(sessionContext, key, extCall, callback): void { SessionContext,
Map<string, Promise<Fault>>
>();
function runExtension(sessionContext, key, extCall): Promise<Fault> {
let re = runningExtensions.get(sessionContext); let re = runningExtensions.get(sessionContext);
if (!re) { if (!re) {
re = {}; re = new Map<string, Promise<Fault>>();
runningExtensions.set(sessionContext, re); runningExtensions.set(sessionContext, re);
} }
if (re[key]) { let prom = re.get(key);
re[key].push(callback); if (!prom) {
} else { re.set(
re[key] = [callback]; key,
extensions.run(extCall, (err, fault, res) => { (prom = new Promise((resolve, reject) => {
const callbacks = re[key]; extensions
delete re[key]; .run(extCall)
if (!err && !fault) sessionContext.extensionsCache[key] = res; .then(({ fault, value }) => {
for (const c of callbacks) c(err, fault); re.delete(key);
}); if (!fault) sessionContext.extensionsCache[key] = value;
resolve(fault);
})
.catch(reject);
}))
);
} }
return prom;
} }
class SandboxDate { class SandboxDate {
@@ -343,15 +353,14 @@ function errorToFault(err: Error): Fault {
return fault; return fault;
} }
export function run( export async function run(
script: vm.Script, script: vm.Script,
globals, globals: {},
sessionContext: SessionContext, sessionContext: SessionContext,
startRevision: number, startRevision: number,
maxRevision: number, maxRevision: number,
callback,
extCounter = 0 extCounter = 0
): void { ): Promise<ScriptResult> {
state = { state = {
sessionContext: sessionContext, sessionContext: sessionContext,
revision: startRevision, revision: startRevision,
@@ -374,61 +383,56 @@ export function run(
ret = script.runInContext(context, { displayErrors: false }); ret = script.runInContext(context, { displayErrors: false });
status = 0; status = 0;
} catch (err) { } catch (err) {
if (err === COMMIT) status = 1; if (err === COMMIT) {
else if (err === EXT) status = 2; status = 1;
else return void callback(null, errorToFault(err)); } else if (err === EXT) {
status = 2;
} else {
return {
fault: errorToFault(err),
clear: null,
declare: null,
done: false,
returnValue: null
};
}
} }
const _state = state; const _state = state;
const args = Array.from(arguments); let fault;
// Need to maintain a counter of ext() calls to avoid calling certain await Promise.all(
// functions (e.g. log()) multiple times as calling ext() results in Object.entries(_state.extensions).map(async ([k, v]) => {
// re-execution of script without revision being incremented. fault = (await runExtension(_state.sessionContext, k, v)) || fault;
args[6] = 0 - (state.extCounter - extCounter); })
);
let counter = 3; if (fault) {
for (const key of Object.keys(_state.extensions)) { return {
counter += 2; fault: fault,
runExtension( clear: null,
_state.sessionContext, declare: null,
key, done: false,
_state.extensions[key], returnValue: null
(err, fault) => { };
if (err || fault) { }
if (counter & 1) callback(err, fault);
return void (counter = 0); if (status === 2) {
} return run(
if ((counter -= 2) === 1) { script,
if (status === 2) { globals,
return void run.apply(null, args); sessionContext,
} else { startRevision,
return void callback( maxRevision,
null, extCounter - _state.extCounter
null,
_state.clear,
_state.declarations,
status === 0,
ret
);
}
}
}
); );
} }
if ((counter -= 2) === 1) { return {
if (status === 2) { fault: null,
run.apply(null, args); clear: _state.clear,
} else { declare: _state.declarations,
callback( done: status === 0,
null, returnValue: ret
null, };
_state.clear,
_state.declarations,
status === 0,
ret
);
}
}
} }
lib/session.ts
+538 -595
View File
File diff suppressed because it is too large. Load diff
lib/soap.ts
+5 -15
View File
@@ -21,7 +21,6 @@ import { parseXml, Document, Element, ParserOptions } from "libxmljs";
import * as config from "./config"; import * as config from "./config";
import { version as VERSION } from "../package.json"; import { version as VERSION } from "../package.json";
import { import {
CpeResponse,
CpeGetResponse, CpeGetResponse,
CpeSetResponse, CpeSetResponse,
InformRequest, InformRequest,
@@ -29,7 +28,9 @@ import {
CpeRequest, CpeRequest,
FaultStruct, FaultStruct,
SpvFault, SpvFault,
CpeFault CpeFault,
SoapMessage,
TransferCompleteRequest
} from "./types"; } from "./types";
const SERVER_NAME = `GenieACS/${VERSION}`; const SERVER_NAME = `GenieACS/${VERSION}`;
@@ -444,7 +445,7 @@ function GetRPCMethodsResponse(xml, methodResponse): void {
for (const m of methodResponse.methodList) el.node("string").text(m); for (const m of methodResponse.methodList) el.node("string").text(m);
} }
function TransferComplete(xml): AcsResponse { function TransferComplete(xml): TransferCompleteRequest {
let commandKey, _faultStruct, startTime, completeTime; let commandKey, _faultStruct, startTime, completeTime;
for (const c of xml.childNodes()) { for (const c of xml.childNodes()) {
switch (c.name()) { switch (c.name()) {
@@ -545,18 +546,7 @@ function fault(xml): CpeFault {
return { faultCode, faultString, detail }; return { faultCode, faultString, detail };
} }
export function request( export function request(data, cwmpVersion, warn): SoapMessage {
data,
cwmpVersion,
warn
): {
id: string;
cwmpVersion: string;
sessionTimeout: number;
cpeRequest?: CpeRequest;
cpeFault?: CpeFault;
cpeResponse?: CpeResponse;
} {
warnings = warn; warnings = warn;
const rpc = { const rpc = {
lib/types.ts
+26 -2
View File
@@ -16,6 +16,7 @@ export interface Fault {
message: string; message: string;
stack?: string; stack?: string;
}; };
timestamp?: number;
} }
export interface SessionFault extends Fault { export interface SessionFault extends Fault {
@@ -255,14 +256,20 @@ export interface InformRequest extends CpeRequest {
parameterList: [string, string | number | boolean, string][]; parameterList: [string, string | number | boolean, string][];
} }
export interface AcsResponse { export interface TransferCompleteRequest extends CpeRequest {
name: string; name: "TransferComplete";
commandKey?: string; commandKey?: string;
faultStruct?: FaultStruct; faultStruct?: FaultStruct;
startTime?: number; startTime?: number;
completeTime?: number; completeTime?: number;
} }
export interface AcsResponse {
name: string;
commandKey?: string;
faultStruct?: FaultStruct;
}
export interface QueryOptions { export interface QueryOptions {
projection?: any; projection?: any;
skip?: number; skip?: number;
@@ -308,3 +315,20 @@ export type PermissionSet = {
filter: Expression; filter: Expression;
}; };
}[]; }[];
export interface SoapMessage {
id: string;
cwmpVersion: string;
sessionTimeout: number;
cpeRequest?: CpeRequest;
cpeFault?: CpeFault;
cpeResponse?: CpeResponse;
}
export interface ScriptResult {
fault: Fault;
clear: Clear[];
declare: Declaration[];
done: boolean;
returnValue: any;
}
lib/ui/api-functions.ts
+100 -145
View File
@@ -11,144 +11,110 @@ import {
deleteDevice deleteDevice
} from "../api-functions"; } from "../api-functions";
function deleteFault(id): Promise<void> { async function deleteFault(id): Promise<void> {
const deviceId = id.split(":", 1)[0]; const deviceId = id.split(":", 1)[0];
const channel = id.slice(deviceId.length + 1); const channel = id.slice(deviceId.length + 1);
return new Promise((resolve, reject) => {
Promise.all([ await Promise.all([
db.deleteFault(id), db.deleteFault(id),
channel.startsWith("task_") channel.startsWith("task_")
? db.deleteTask(new ObjectID(channel.slice(5))) ? db.deleteTask(new ObjectID(channel.slice(5)))
: null : null
]) ]);
.then(() => {
del(`${deviceId}_tasks_faults_operations`, err => { await del(`${deviceId}_tasks_faults_operations`);
err ? reject(err) : resolve();
});
})
.catch(reject);
});
} }
export function deleteResource(resource, id): Promise<void> { export async function deleteResource(resource, id): Promise<void> {
return new Promise((resolve, reject) => { switch (resource) {
let promise; case "devices":
await deleteDevice(id);
break;
switch (resource) { case "files":
case "devices": await db.deleteFile(id);
promise = new Promise((res, rej) => { break;
deleteDevice(id, err => {
if (err) return void rej(err);
res();
});
});
break;
case "files": case "faults":
promise = db.deleteFile(id); await deleteFault(id);
break; break;
case "faults": case "provisions":
promise = deleteFault(id); await db.deleteProvision(id);
break; break;
case "provisions": case "presets":
promise = db.deleteProvision(id); await db.deletePreset(id);
break; break;
case "presets": case "virtualParameters":
promise = db.deletePreset(id); await db.deleteVirtualParameter(id);
break; break;
case "virtualParameters": case "config":
promise = db.deleteVirtualParameter(id); await db.deleteConfig(id);
break; break;
case "config": case "permissions":
promise = db.deleteConfig(id); await db.deletePermission(id);
break; break;
case "permissions": case "users":
promise = db.deletePermission(id); await db.deleteUser(id);
break; break;
}
case "users": await del("presets_hash");
promise = db.deleteUser(id);
break;
}
promise
.then(() => {
del("presets_hash", err => {
if (err) return void reject(err);
resolve();
});
})
.catch(reject);
});
} }
export function postTasks( export async function postTasks(
deviceId, deviceId,
tasks, tasks,
timeout timeout
): Promise<{ connectionRequest: string; tasks: any[] }> { ): Promise<{ connectionRequest: string; tasks: any[] }> {
return new Promise((resolve, reject) => { for (const task of tasks) task.device = deviceId;
for (const task of tasks) task.device = deviceId;
insertTasks(tasks, err => { tasks = await insertTasks(tasks);
if (err) return void reject(err); const statuses = tasks.map(t => {
const statuses = tasks.map(t => { return { _id: t._id, status: "pending" };
return { _id: t._id, status: "pending" };
});
del(`${deviceId}_tasks_faults_operations`, err => {
if (err) return void reject(err);
connectionRequest(deviceId, err => {
if (err) {
return void resolve({
connectionRequest: err.message,
tasks: statuses
});
}
const sample = tasks[tasks.length - 1];
// Waiting for session to finish or timeout
watchTask(deviceId, sample._id, timeout, err => {
if (err) return void reject(err);
const promises2 = [];
for (const s of statuses) {
promises2.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id]));
promises2.push(
db.query("faults", [
"=",
["PARAM", "_id"],
`${deviceId}:task_${s._id}`
])
);
}
Promise.all(promises2)
.then(res => {
for (const [i, r] of statuses.entries()) {
if (res[i * 2].length === 0) {
r.status = "done";
} else if (res[i * 2 + 1].length === 1) {
r.status = "fault";
r.fault = res[i * 2 + 1][0];
}
db.deleteTask(r._id);
}
resolve({ connectionRequest: "OK", tasks: statuses });
})
.catch(reject);
});
});
});
});
}); });
await del(`${deviceId}_tasks_faults_operations`);
try {
await connectionRequest(deviceId);
} catch (err) {
return {
connectionRequest: err.message,
tasks: statuses
};
}
const sample = tasks[tasks.length - 1];
// Waiting for session to finish or timeout
await watchTask(deviceId, sample._id, timeout);
const promises = [];
for (const s of statuses) {
promises.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id]));
promises.push(
db.query("faults", ["=", ["PARAM", "_id"], `${deviceId}:task_${s._id}`])
);
}
const res = await Promise.all(promises);
for (const [i, r] of statuses.entries()) {
if (res[i * 2].length === 0) {
r.status = "done";
} else if (res[i * 2 + 1].length === 1) {
r.status = "fault";
r.fault = res[i * 2 + 1][0];
}
db.deleteTask(r._id);
}
return { connectionRequest: "OK", tasks: statuses };
} }
interface PingResponse { interface PingResponse {
@@ -184,35 +150,24 @@ export function ping(host): Promise<PingResponse> {
}); });
} }
export function putResource(resource, id, data): Promise<void> { export async function putResource(resource, id, data): Promise<void> {
return new Promise((resolve, reject) => { if (resource === "presets") {
let promise; await db.putPreset(id, data);
} else if (resource === "provisions") {
await db.putProvision(id, data);
} else if (resource === "virtualParameters") {
await db.putVirtualParameter(id, data);
} else if (resource === "config") {
await db.putConfig(id, data);
} else if (resource === "permissions") {
await db.putPermission(id, data);
} else if (resource === "users") {
delete data.password;
delete data.salt;
await db.putUser(id, data);
}
if (resource === "presets") { await del("presets_hash");
promise = db.putPreset(id, data);
} else if (resource === "provisions") {
promise = db.putProvision(id, data);
} else if (resource === "virtualParameters") {
promise = db.putVirtualParameter(id, data);
} else if (resource === "config") {
promise = db.putConfig(id, data);
} else if (resource === "permissions") {
promise = db.putPermission(id, data);
} else if (resource === "users") {
delete data.password;
delete data.salt;
promise = db.putUser(id, data);
}
promise
.then(() => {
del("presets_hash", err => {
if (err) return void reject(err);
resolve();
});
})
.catch(reject);
});
} }
export function authSimple(snapshot, username, password): Promise<string[]> { export function authSimple(snapshot, username, password): Promise<string[]> {
lib/ui/api.ts
+1 -10
View File
@@ -546,16 +546,7 @@ router.put("/users/:id/password", async (ctx, next) => {
const password = await hashPassword(newPassword, salt); const password = await hashPassword(newPassword, salt);
await db.putUser(username, { password, salt }); await db.putUser(username, { password, salt });
function delCache(): Promise<void> { await del("presets_hash");
return new Promise((resolve, reject) => {
del("presets_hash", err => {
if (err) return void reject(err);
resolve();
});
});
}
await delCache();
logger.accessInfo(log); logger.accessInfo(log);
ctx.body = ""; ctx.body = "";