Fix XMPP stream not getting closed properly

The XMPP server may not flag the bind feature as required but we're
still required to bind resource to the stream in order to be able to
send IQ stanzas to other clients.

bin/genieacs-cwmp.ts

@@ -24,7 +24,6 @@ import * as server from "../lib/server";
 import * as cwmp from "../lib/cwmp";
 import * as db from "../lib/db";
 import * as extensions from "../lib/extensions";
-import * as cache from "../lib/cache";
 import { version as VERSION } from "../package.json";
 
 logger.init("cwmp", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
   setTimeout(exitWorkerUngracefully, 5000).unref();
   Promise.all([
     db.disconnect(),
-    cache.disconnect(),
     extensions.killAll(),
     cluster.worker.disconnect(),
   ]).catch(exitWorkerUngracefully);
@@ -99,7 +97,8 @@ if (!cluster.worker) {
     server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully);
   });
 
-  const initPromise = Promise.all([db.connect(), cache.connect()])
+  const initPromise = db
+    .connect()
     .then(() => {
       server.start(options, cwmp.listener);
     })

bin/genieacs-fs.ts

@@ -23,7 +23,6 @@ import * as cluster from "../lib/cluster";
 import * as server from "../lib/server";
 import { listener } from "../lib/fs";
 import * as db from "../lib/db";
-import * as cache from "../lib/cache";
 import { version as VERSION } from "../package.json";
 
 logger.init("fs", VERSION);
@@ -33,11 +32,9 @@ const SERVICE_PORT = config.get("FS_PORT") as number;
 
 function exitWorkerGracefully(): void {
   setTimeout(exitWorkerUngracefully, 5000).unref();
-  Promise.all([
+  Promise.all([db.disconnect(), cluster.worker.disconnect()]).catch(
-    db.disconnect(),
+    exitWorkerUngracefully
-    cache.disconnect(),
+  );
-    cluster.worker.disconnect(),
-  ]).catch(exitWorkerUngracefully);
 }
 
 function exitWorkerUngracefully(): void {
@@ -100,7 +97,8 @@ if (!cluster.worker) {
     listener(req, res);
   };
 
-  const initPromise = Promise.all([db.connect(), cache.connect()])
+  const initPromise = db
+    .connect()
     .then(() => {
       server.start(options, _listener);
     })

bin/genieacs-nbi.ts

@@ -24,7 +24,6 @@ import * as server from "../lib/server";
 import { listener } from "../lib/nbi";
 import * as db from "../lib/db";
 import * as extensions from "../lib/extensions";
-import * as cache from "../lib/cache";
 import { version as VERSION } from "../package.json";
 
 logger.init("nbi", VERSION);
@@ -36,7 +35,6 @@ function exitWorkerGracefully(): void {
   setTimeout(exitWorkerUngracefully, 5000).unref();
   Promise.all([
     db.disconnect(),
-    cache.disconnect(),
     extensions.killAll(),
     cluster.worker.disconnect(),
   ]).catch(exitWorkerUngracefully);
@@ -104,7 +102,8 @@ if (!cluster.worker) {
     listener(req, res);
   };
 
-  const initPromise = Promise.all([db.connect(), cache.connect()])
+  const initPromise = db
+    .connect()
     .then(() => {
       server.start(options, _listener);
     })

bin/genieacs-ui.ts

@@ -23,9 +23,7 @@ import * as cluster from "../lib/cluster";
 import * as server from "../lib/server";
 import { listener } from "../lib/ui";
 import * as extensions from "../lib/extensions";
-import * as db from "../lib/ui/db";
+import * as db from "../lib/db";
-import * as db2 from "../lib/db";
-import * as cache from "../lib/cache";
 import { version as VERSION } from "../package.json";
 
 logger.init("ui", VERSION);
@@ -37,8 +35,6 @@ function exitWorkerGracefully(): void {
   setTimeout(exitWorkerUngracefully, 5000).unref();
   Promise.all([
     db.disconnect(),
-    db2.disconnect(),
-    cache.disconnect(),
     extensions.killAll(),
     cluster.worker.disconnect(),
   ]).catch(exitWorkerUngracefully);
@@ -106,7 +102,8 @@ if (!cluster.worker) {
     listener(req, res);
   };
 
-  const initPromise = Promise.all([db2.connect(), cache.connect()])
+  const initPromise = db
+    .connect()
     .then(() => {
       server.start(options, _listener);
     })

build/build.ts

@@ -51,6 +51,8 @@ const externals = [
   "tls",
   "http",
   "https",
+  "net",
+  "events",
   "zlib",
   "crypto",
   "util",

lib/api-functions.ts

@@ -29,22 +29,31 @@ import {
 import {
   httpConnectionRequest,
   udpConnectionRequest,
+  xmppConnectionRequest,
 } from "./connection-request";
 import { Expression, Task } from "./types";
 import { flattenDevice } from "./mongodb-functions";
 import { evaluate } from "./common/expression";
+import * as config from "../lib/config";
+
+const XMPP_CONFIGURED = !!config.get("XMPP_JID");
 
 export async function connectionRequest(
   deviceId: string,
   device?: Record<string, { value?: [boolean | number | string, string] }>
-): Promise<void> {
+): Promise<string> {
   if (!device) {
     const res = await db.devicesCollection.findOne({ _id: deviceId });
     if (!res) throw new Error("No such device");
     device = flattenDevice(res);
   }
 
-  let connectionRequestUrl, udpConnectionRequestAddress, username, password;
+  let connectionRequestUrl,
+    udpConnectionRequestAddress,
+    stunEnable,
+    connReqJabberId,
+    username,
+    password;
 
   if (device["InternetGatewayDevice.ManagementServer.ConnectionRequestURL"]) {
     connectionRequestUrl = (device[
@@ -55,6 +64,12 @@ export async function connectionRequest(
         "InternetGatewayDevice.ManagementServer.UDPConnectionRequestAddress"
       ] || {}
     ).value || [""])[0];
+    stunEnable = ((
+      device["InternetGatewayDevice.ManagementServer.STUNEnable"] || {}
+    ).value || [""])[0];
+    connReqJabberId = ((
+      device["InternetGatewayDevice.ManagementServer.ConnReqJabberID"] || {}
+    ).value || [""])[0];
     username = ((
       device[
         "InternetGatewayDevice.ManagementServer.ConnectionRequestUsername"
@@ -72,6 +87,10 @@ export async function connectionRequest(
     udpConnectionRequestAddress = ((
       device["Device.ManagementServer.UDPConnectionRequestAddress"] || {}
     ).value || [""])[0];
+    stunEnable = ((device["Device.ManagementServer.STUNEnable"] || {})
+      .value || [""])[0];
+    connReqJabberId = ((device["Device.ManagementServer.ConnReqJabberID"] || {})
+      .value || [""])[0];
     username = ((
       device["Device.ManagementServer.ConnectionRequestUsername"] || {}
     ).value || [""])[0];
@@ -146,7 +165,7 @@ export async function connectionRequest(
   const debug = !!getConfig(snapshot, "cwmp.debug", {}, now, evalCallback);
 
   let udpProm;
-  if (udpConnectionRequestAddress) {
+  if (udpConnectionRequestAddress && +stunEnable) {
     udpProm = udpConnectionRequest(
       udpConnectionRequestAddress,
       authExp,
@@ -156,8 +175,18 @@ export async function connectionRequest(
     );
   }
 
-  try {
+  let status;
-    await httpConnectionRequest(
+
+  if (connReqJabberId && XMPP_CONFIGURED) {
+    status = await xmppConnectionRequest(
+      connReqJabberId,
+      authExp,
+      CONNECTION_REQUEST_TIMEOUT,
+      debug,
+      deviceId
+    );
+  } else {
+    status = await httpConnectionRequest(
       connectionRequestUrl,
       authExp,
       CONNECTION_REQUEST_ALLOW_BASIC_AUTH,
@@ -165,34 +194,47 @@ export async function connectionRequest(
       debug,
       deviceId
     );
-  } catch (err) {
+
-    if (!udpProm) throw err;
+    if (udpProm) {
       await udpProm;
+      return "";
     }
   }
 
-export async function watchTask(
+  return status;
+}
+
+export async function awaitSessionStart(
   deviceId: string,
-  taskId: string,
+  lastInform: number,
   timeout: number
-): Promise<string> {
+): Promise<boolean> {
-  await new Promise((resolve) => setTimeout(resolve, 500));
+  const now = Date.now();
-
+  const device = await db.devicesCollection.findOne(
-  const task = await db.tasksCollection.findOne(
+    { _id: deviceId },
-    { _id: taskId },
+    { projection: { _lastInform: 1 } }
-    { projection: { _id: 1 } }
   );
-  if (!task) return "completed";
+  const li = (device["_lastInform"] as Date).getTime();
+  if (li > lastInform) return true;
+  const token = await cache.get(`cwmp_session_${deviceId}`);
+  if (token) return true;
+  if (timeout < 500) return false;
+  await new Promise((resolve) => setTimeout(resolve, 500));
+  timeout -= Date.now() - now;
+  return awaitSessionStart(deviceId, lastInform, timeout);
+}
 
-  const q = { _id: `${deviceId}:task_${taskId}` };
+export async function awaitSessionEnd(
-  const fault = await db.faultsCollection.findOne(q, {
+  deviceId: string,
-    projection: { _id: 1 },
+  timeout: number
-  });
+): Promise<boolean> {
-  if (fault) return "fault";
+  const now = Date.now();
-
+  const token = await cache.get(`cwmp_session_${deviceId}`);
-  if ((timeout -= 500) <= 0) return "timeout";
+  if (!token) return true;
-
+  if (timeout < 500) return false;
-  return watchTask(deviceId, taskId, timeout);
+  await new Promise((resolve) => setTimeout(resolve, 500));
+  timeout -= Date.now() - now;
+  return awaitSessionEnd(deviceId, timeout);
 }
 
 function sanitizeTask(task): void {

lib/cache.ts

@@ -17,126 +17,102 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import { MongoClient, Collection } from "mongodb";
+import { Collection } from "mongodb";
+import { onConnect } from "./db";
 import * as config from "./config";
 
+const CLOCK_SKEW_TOLERANCE = 30000;
 const MAX_CACHE_TTL = +config.get("MAX_CACHE_TTL");
 
-let clientPromise: Promise<MongoClient>;
+let cacheCollection: Collection;
-let mongoCollection: Collection;
-let mongoTimeOffset = 0;
 
-export async function connect(): Promise<void> {
+onConnect(async (db) => {
-  const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
+  cacheCollection = db.collection("cache");
-  clientPromise = MongoClient.connect(MONGODB_CONNECTION_URL, {
+  await cacheCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-    useNewUrlParser: true,
-    useUnifiedTopology: true,
-  });
-  const db = (await clientPromise).db();
-  mongoCollection = db.collection("cache");
-  await mongoCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-  const now = Date.now();
-  const res = await db.command({ hostInfo: 1 });
-  mongoTimeOffset = res.system.currentTime.getTime() - now;
-}
-
-export async function disconnect(): Promise<void> {
-  if (clientPromise) await (await clientPromise).close();
-}
-
-export async function get(key: string | string[]): Promise<any> {
-  const expire = new Date(Date.now() - mongoTimeOffset);
-  if (Array.isArray(key)) {
-    const res = await mongoCollection.find({ _id: { $in: key } }).toArray();
-
-    const indices = {};
-    key.forEach((v, i) => {
-      indices[v] = i;
 });
 
-    const values = [];
+export async function get(key: string): Promise<string> {
-    res.forEach((r) => {
+  const res = await cacheCollection.findOne({ _id: key });
-      if (r["expire"] > expire) values[indices[r["_id"]]] = r["value"];
+  if (res) return res["value"];
-    });
-
-    return values;
-  } else {
-    const res = await mongoCollection.findOne({ _id: { $in: [key] } });
-    if (res && res["expire"] > expire) return res["value"];
   return null;
 }
-}
 
-export async function del(key: string | string[]): Promise<void> {
+export async function del(key: string): Promise<void> {
-  if (Array.isArray(key))
+  await cacheCollection.deleteOne({ _id: key });
-    await mongoCollection.deleteMany({ _id: { $in: key } });
-  else await mongoCollection.deleteOne({ _id: key });
 }
 
 export async function set(
   key: string,
-  value: string | number,
+  value: string,
   ttl: number = MAX_CACHE_TTL
 ): Promise<void> {
-  const expire = new Date(Date.now() - mongoTimeOffset + ttl * 1000);
+  const timestamp = new Date();
-  await mongoCollection.replaceOne(
+  const expire = new Date(
+    timestamp.getTime() + CLOCK_SKEW_TOLERANCE + ttl * 1000
+  );
+  await cacheCollection.replaceOne(
     { _id: key },
-    { _id: key, value: value, expire: expire },
+    { _id: key, value, expire, timestamp },
     { upsert: true }
   );
 }
 
 export async function pop(key: string): Promise<any> {
-  const res = await mongoCollection.findOneAndDelete({ _id: key });
+  const res = await cacheCollection.findOneAndDelete({ _id: key });
-
+  if (res && res["value"]) return res["value"]["value"];
-  if (
-    res &&
-    res["value"] &&
-    +res["value"]["expire"] - (Date.now() - mongoTimeOffset)
-  )
-    return res["value"]["value"];
-
   return null;
 }
 
-export async function lock(
+export async function acquireLock(
   lockName: string,
-  ttl: number
+  ttl: number,
-): Promise<(extendTtl?: number) => Promise<void>> {
+  timeout = 0,
-  const token = Math.random().toString(36).slice(2);
+  token = Math.random().toString(36).slice(2)
+): Promise<string> {
+  try {
+    const now = Date.now();
+    const r = await cacheCollection.findOneAndUpdate(
+      { _id: lockName, token },
+      {
+        $set: {
+          value: token,
+          expire: new Date(now + ttl + CLOCK_SKEW_TOLERANCE),
+        },
+        $currentDate: { timestamp: true },
+      },
+      { upsert: true, returnOriginal: false }
+    );
+    const v = r.value;
+    if (Math.abs(v["timestamp"].getTime() - now) > CLOCK_SKEW_TOLERANCE)
+      throw new Error("Database clock skew too great");
+  } catch (err) {
+    if (err.code === 11000) {
+      if (timeout > 0) {
+        return new Promise((resolve, reject) => {
+          const w = 50 + Math.random() * 50;
+          setTimeout(() => {
+            acquireLock(lockName, ttl, timeout - w, token).then(
+              resolve,
+              reject
+            );
+          }, w);
+        });
+      }
+      throw new Error("Failed to acquire lock");
+    }
+    throw err;
+  }
 
-  async function unlockOrExtend(extendTtl?: number): Promise<void> {
+  return token;
-    if (!extendTtl) {
+}
-      const res = await mongoCollection.deleteOne({
+
+export async function releaseLock(
+  lockName: string,
+  token: string
+): Promise<void> {
+  const res = await cacheCollection.deleteOne({
     _id: lockName,
     value: token,
   });
   if (res["result"]["n"] !== 1) throw new Error("Lock expired");
-    } else {
-      const expire = new Date(Date.now() - mongoTimeOffset + extendTtl * 1000);
-      const res = await mongoCollection.updateOne(
-        { _id: lockName, value: token },
-        { expire: expire }
-      );
-      if (res["result"]["n"] !== 1) throw new Error("Lock expired");
-    }
-  }
-
-  const expireTest = new Date(Date.now() - mongoTimeOffset);
-  const expireSet = new Date(Date.now() - mongoTimeOffset + ttl * 1000);
-
-  try {
-    await mongoCollection.updateOne(
-      { _id: lockName, expire: { $lte: expireTest } },
-      { $set: { value: token, expire: expireSet } },
-      { upsert: true }
-    );
-  } catch (err) {
-    if (err && err.code === 11000) {
-      await new Promise((resolve) => setTimeout(resolve, 200));
-      return lock(lockName, ttl);
-    }
-  }
-
-  return unlockOrExtend;
 }

lib/common/boolean-expression.ts

@@ -362,6 +362,10 @@ class VarSynth extends BoolExprSynth {
         exp = exp.slice();
         exp[0] = ">";
         this.negate = true;
+      } else if (op === "NOT LIKE") {
+        exp = exp.slice();
+        exp[0] = "LIKE";
+        this.negate = true;
       }
     }
     this.expStr = JSON.stringify(exp);
@@ -809,6 +813,7 @@ function sopToExpression(
       if (Array.isArray(expr) && expr[0] === "NOT" && Array.isArray(expr[1])) {
         const e: Expression[] = expr[1];
         if (e[0] === "IS NULL") expr = ["IS NOT NULL", ...e.slice(1)];
+        else if (e[0] === "LIKE") expr = ["NOT LIKE", ...e.slice(1)];
         else if (e[0] === "=") expr = ["<>", ...e.slice(1)];
         else if (e[0] === "<>") expr = ["=", ...e.slice(1)];
         else if (e[0] === ">") expr = ["<=", ...e.slice(1)];
@@ -830,7 +835,6 @@ function sopToExpression(
 function generateDcSetAndIsNull(
   variables: Map<number, Expression>
 ): { dcSet: number[][]; isNull: Map<number, number> } {
-  // TODO support LIKE and NOT LIKE operators
   const relations: Map<
     string,
     { op: string; rhs: boolean | number | string; var: number }[]
@@ -843,6 +847,11 @@ function generateDcSetAndIsNull(
       let r = relations.get(lhs);
       if (!r) relations.set(lhs, (r = []));
       r.push({ op: op, rhs: e[2], var: n });
+    } else if (op === "LIKE" && !Array.isArray(e[2]) && !Array.isArray(e[3])) {
+      const lhs = JSON.stringify(e[1]);
+      let r = relations.get(lhs);
+      if (!r) relations.set(lhs, (r = []));
+      r.push({ op: "", rhs: null, var: n });
     } else {
       const lhs = JSON.stringify(e);
       let r = relations.get(lhs);

lib/config.ts

@@ -101,6 +101,9 @@ const options = {
 
   // Should probably never be changed
   DEVICE_ONLINE_THRESHOLD: { type: "int", default: 4000 },
+
+  XMPP_JID: { type: "string", default: "" },
+  XMPP_PASSWORD: { type: "string", default: "" },
 };
 
 const allConfig: { [name: string]: string | number } = {};

lib/connection-request.ts

@@ -26,6 +26,10 @@ import { Expression } from "./types";
 import * as auth from "./auth";
 import * as extensions from "./extensions";
 import * as debug from "./debug";
+import XmppClient from "./xmpp-client";
+import * as config from "../lib/config";
+import { encodeEntities, parseAttrs, Element } from "./xml-parser";
+import * as logger from "../lib/logger";
 
 async function extractAuth(
   exp: Expression,
@@ -62,7 +66,6 @@ async function extractAuth(
 
 function httpGet(
   options: http.RequestOptions,
-  timeout,
   _debug: boolean,
   deviceId: string
 ): Promise<{ statusCode: number; headers: http.IncomingHttpHeaders }> {
@@ -77,15 +80,12 @@ function httpGet(
         }
       })
       .on("error", (err) => {
-        req.abort();
+        req.destroy();
-        reject(new Error("Device is offline"));
+        reject(err);
         if (_debug) debug.outgoingHttpRequestError(req, deviceId, options, err);
       })
-      .on("socket", (socket) => {
+      .on("timeout", () => {
-        socket.setTimeout(timeout);
+        req.destroy();
-        socket.on("timeout", () => {
-          req.abort();
-        });
       });
   });
 }
@@ -97,14 +97,15 @@ export async function httpConnectionRequest(
   timeout: number,
   _debug: boolean,
   deviceId: string
-): Promise<void> {
+): Promise<string> {
   const options: http.RequestOptions = parse(address);
   if (options.protocol !== "http:")
-    throw new Error("Invalid connection request URL or protocol");
+    return "Invalid connection request URL or protocol";
 
   options.agent = new http.Agent({
     maxSockets: 1,
     keepAlive: true,
+    timeout: timeout,
   });
 
   let authHeader: Record<string, string>;
@@ -115,8 +116,7 @@ export async function httpConnectionRequest(
     let opts = options;
     if (authHeader) {
       if (authHeader["method"] === "Basic") {
-        if (!allowBasicAuth)
+        if (!allowBasicAuth) return "Basic HTTP authentication not allowed";
-          throw new Error("Basic HTTP authentication not allowed");
 
         opts = Object.assign(
           {
@@ -143,17 +143,34 @@ export async function httpConnectionRequest(
           options
         );
       } else {
-        throw new Error("Unrecognized auth method");
+        return "Unrecognized auth method";
       }
     }
 
-    let res = await httpGet(opts, timeout, _debug, deviceId);
+    let res: { statusCode: number; headers: http.IncomingHttpHeaders };
-
+    try {
+      res = await httpGet(opts, _debug, deviceId);
+    } catch (err) {
       // Workaround for some devices unexpectedly closing the connection
-    if (res.statusCode === 0 && authHeader)
+      if (authHeader) {
-      res = await httpGet(opts, timeout, _debug, deviceId);
+        try {
-    if (res.statusCode === 0) throw new Error("Device is offline");
+          res = await httpGet(opts, _debug, deviceId);
-    if (res.statusCode === 200 || res.statusCode === 204) return;
+        } catch (err) {
+          return `Connection request error: ${err.message}`;
+        }
+      }
+
+      if (err["code"] === "ECONNRESET" || err["code"] === "ECONNREFUSED")
+        return "Device is offline";
+
+      return `Connection request error: ${err.message}`;
+    }
+
+    if (res.statusCode === 200 || res.statusCode === 204) return "";
+
+    // When a Connection Request is received for the Virtual CWMP Device and the Proxied
+    // Device is offline the CPE Proxier MUST respond with an HTTP 503 failure
+    if (res.statusCode === 503) return "Device is offline";
 
     if (res.statusCode === 401 && res.headers["www-authenticate"]) {
       authHeader = auth.parseWwwAuthenticateHeader(
@@ -161,12 +178,11 @@ export async function httpConnectionRequest(
       );
       [username, password, authExp] = await extractAuth(authExp, false);
     } else {
-      throw new Error(
+      return `Connection request error: Unexpected status code ${res.statusCode}`;
-        `Unexpected response code from device: ${res.statusCode}`
-      );
     }
   }
-  throw new Error("Incorrect connection request credentials");
+
+  return "Connection request error: Incorrect connection request credentials";
 }
 
 export async function udpConnectionRequest(
@@ -222,3 +238,84 @@ export async function udpConnectionRequest(
   }
   client.close();
 }
+
+const XMPP_JID = config.get("XMPP_JID") as string;
+const XMPP_PASSWORD = config.get("XMPP_PASSWORD") as string;
+const XMPP_RESOURCE = crypto.randomBytes(8).toString("hex");
+
+let xmppClient: XmppClient;
+
+function xmppClientOnError(err: Error): void {
+  xmppClient = null;
+  logger.error({
+    message: "XMPP exception",
+    exception: err,
+    pid: process.pid,
+  });
+}
+
+function xmppClientOnClose(): void {
+  xmppClient = null;
+}
+
+export async function xmppConnectionRequest(
+  jid: string,
+  authExp: Expression,
+  timeout: number,
+  _debug: boolean,
+  deviceId: string
+): Promise<string> {
+  if (!xmppClient) {
+    const [host, username] = XMPP_JID.split("@").reverse();
+    xmppClient = await XmppClient.connect({
+      host,
+      username,
+      resource: XMPP_RESOURCE,
+      password: XMPP_PASSWORD,
+      timeout: 120000,
+    });
+    xmppClient.on("error", xmppClientOnError);
+    xmppClient.on("close", xmppClientOnClose);
+    xmppClient.unref();
+  }
+
+  let username: string;
+  let password: string;
+
+  [username, password, authExp] = await extractAuth(authExp, null);
+  while (username != null && password != null) {
+    const msg = `<connectionRequest xmlns="urn:broadband-forum-org:cwmp:xmppConnReq-1-0"><username>${encodeEntities(
+      username
+    )}</username><password>${encodeEntities(
+      password
+    )}</password></connectionRequest>`;
+    let res: Element, rawRes: string, rawReq: string;
+    try {
+      ({ res, rawRes, rawReq } = await xmppClient.sendIqStanza(
+        XMPP_JID,
+        jid,
+        "get",
+        msg,
+        timeout
+      ));
+    } catch (err) {
+      return err.message;
+    }
+    if (_debug) {
+      debug.outgoingXmppStanza(deviceId, rawReq);
+      debug.incomingXmppStanza(deviceId, rawRes);
+    }
+    const attrs = parseAttrs(res.attrs);
+    const type = attrs.find((a) => a.name === "type");
+    if (type && type.value === "result") return "";
+    const error = res.children.find((c) => c.name === "error");
+    if (!error || !error.children[0])
+      return "Unexpected XMPP connection request response";
+    if (error.children[0].name === "service-unavailable")
+      return "Device is offline";
+    if (error.children[0].name !== "not-authorized")
+      return "Unexpected XMPP connection request response";
+    [username, password, authExp] = await extractAuth(authExp, null);
+  }
+  return "Incorrect connection request credentials";
+}

lib/cwmp.ts

@@ -196,17 +196,23 @@ async function writeResponse(
       sessionContext: sessionContext,
       message: "Connection dropped",
     });
+    await endSession(sessionContext);
   } else if (close) {
-    const isNew = await endSession(sessionContext);
+    session.clearProvisions(sessionContext);
-    if (isNew) {
+    await endSession(sessionContext);
-      logger.accessInfo({
-        sessionContext: sessionContext,
-        message: "New device registered",
-      });
-    }
   } else {
-    sessionContext.lastActivity = Date.now();
+    const now = Date.now();
+    sessionContext.lastActivity = now;
     currentSessions.set(connection, sessionContext);
+    if (now >= sessionContext.extendLock) {
+      sessionContext.extendLock = now + 10000;
+      await cache.acquireLock(
+        `cwmp_session_${sessionContext.deviceId}`,
+        sessionContext.timeout + 10000,
+        0,
+        sessionContext.sessionId
+      );
+    }
   }
 }
 
@@ -726,9 +732,20 @@ async function nextRpc(sessionContext: SessionContext): Promise<void> {
   return nextRpc(sessionContext);
 }
 
-async function endSession(sessionContext: SessionContext): Promise<boolean> {
+async function endSession(sessionContext: SessionContext): Promise<void> {
   let saveCache = sessionContext.cacheUntil != null;
 
+  if (sessionContext.provisions.length) {
+    const fault = {
+      code: "session_terminated",
+      message: "The TR-069 session was unsuccessfully terminated",
+      timestamp: sessionContext.timestamp,
+    };
+    recordFault(sessionContext, fault);
+    // No need to save retryNow
+    for (const f of Object.values(sessionContext.faults)) delete f.retryNow;
+  }
+
   const promises = [];
 
   promises.push(
@@ -791,7 +808,16 @@ async function endSession(sessionContext: SessionContext): Promise<boolean> {
   }
 
   await Promise.all(promises);
-  return sessionContext.new;
+  await cache.releaseLock(
+    `cwmp_session_${sessionContext.deviceId}`,
+    sessionContext.sessionId
+  );
+  if (sessionContext.new) {
+    logger.accessInfo({
+      sessionContext: sessionContext,
+      message: "New device registered",
+    });
+  }
 }
 
 async function sendAcsRequest(
@@ -878,15 +904,21 @@ export function onConnection(socket: Socket): void {
     const now = Date.now();
 
     const lastActivity = sessionContext.lastActivity;
-    const timeoutMsg = logger.flatten({
+    const timeoutMsg = {
       sessionContext: sessionContext,
       message: "Session timeout",
       sessionTimestamp: sessionContext.timestamp,
-    });
+    };
 
     const timeout =
       sessionContext.lastActivity + sessionContext.timeout * 1000 - now;
-    if (timeout <= 0) return void logger.accessError(timeoutMsg);
+
+    if (timeout <= 0) {
+      logger.accessError(timeoutMsg);
+      // TODO it's possible that lock would have already been expired
+      await endSession(sessionContext);
+      return;
+    }
 
     setTimeout(async () => {
       const sessionContextString = await cache.get(
@@ -894,8 +926,10 @@ export function onConnection(socket: Socket): void {
       );
       if (!sessionContextString) return;
       const _sessionContext = await session.deserialize(sessionContextString);
-      if (_sessionContext.lastActivity === lastActivity)
+      if (_sessionContext.lastActivity === lastActivity) {
         logger.accessError(timeoutMsg);
+        await endSession(sessionContext);
+      }
     }, timeout + 1000).unref();
 
     if (sessionContext.state === 0) return;
@@ -1092,6 +1126,37 @@ async function processRequest(
       }
     }
 
+    try {
+      sessionContext.extendLock = sessionContext.timestamp + 10000;
+      await cache.acquireLock(
+        `cwmp_session_${sessionContext.deviceId}`,
+        sessionContext.timeout + 10000,
+        0,
+        sessionContext.sessionId
+      );
+    } catch (err) {
+      logger.accessError({
+        message: "CPE already in session",
+        sessionContext: sessionContext,
+      });
+
+      const _body = "CPE already in session";
+      sessionContext.httpResponse.setHeader(
+        "Content-Length",
+        Buffer.byteLength(_body)
+      );
+      sessionContext.httpResponse.writeHead(400, { Connection: "close" });
+      if (sessionContext.debug) {
+        debug.outgoingHttpResponse(
+          sessionContext.httpResponse,
+          sessionContext.deviceId,
+          _body
+        );
+      }
+      sessionContext.httpResponse.end(_body);
+      return;
+    }
+
     sessionContext.state = 1;
     sessionContext.authState = 2;
 
@@ -1120,6 +1185,7 @@ async function processRequest(
         sessionContext.authState = 1;
         return responseUnauthorized(sessionContext, false);
       } else {
+        await endSession(sessionContext);
         return responseUnauthorized(sessionContext, true);
       }
     }
@@ -1343,6 +1409,7 @@ async function listenerAsync(
         );
       }
       httpResponse.end(_body);
+      await endSession(sessionContext);
       return;
     }
     if (newConnection && sessionContext.authState !== 1)
@@ -1393,6 +1460,7 @@ async function listenerAsync(
         );
         debug.outgoingHttpResponse(httpResponse, sessionContext.deviceId, msg);
       }
+      await endSession(sessionContext);
     } else {
       const cacheSnapshot = await localCache.getCurrentSnapshot();
       const d = !!localCache.getConfig(
@@ -1449,6 +1517,7 @@ async function listenerAsync(
           error.message
         );
       }
+      await endSession(sessionContext);
     } else {
       const cacheSnapshot = await localCache.getCurrentSnapshot();
       const d = !!localCache.getConfig(

lib/db.ts

@@ -17,7 +17,7 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import { MongoClient, ObjectID, Collection } from "mongodb";
+import { MongoClient, ObjectID, Collection, Db } from "mongodb";
 import { get } from "./config";
 import { encodeTag, escapeRegExp } from "./common";
 import { parse } from "./common/expression-parser";
@@ -45,7 +45,6 @@ export let tasksCollection: Collection,
   configCollection: Collection;
 
 let clientPromise: Promise<MongoClient>;
-export let client: MongoClient;
 
 function compareAccessLists(list1: string[], list2: string[]): boolean {
   if (list1.length !== list2.length) return false;
@@ -53,15 +52,24 @@ function compareAccessLists(list1: string[], list2: string[]): boolean {
   return true;
 }
 
+const onConnectCallbacks: ((db: Db) => Promise<void>)[] = [];
+
+export function onConnect(callback: (db: Db) => Promise<void>): void {
+  onConnectCallbacks.push(callback);
+}
+
 export async function connect(): Promise<void> {
   clientPromise = MongoClient.connect("" + get("MONGODB_CONNECTION_URL"), {
     useNewUrlParser: true,
     useUnifiedTopology: true,
   });
 
-  client = await clientPromise;
+  const client = await clientPromise;
   const db = client.db();
+  await Promise.all(onConnectCallbacks.map((c) => c(db)));
+}
 
+onConnect(async (db) => {
   tasksCollection = db.collection("tasks");
   await tasksCollection.createIndex({ device: 1, timestamp: 1 });
 
@@ -76,7 +84,7 @@ export async function connect(): Promise<void> {
   permissionsCollection = db.collection("permissions");
   usersCollection = db.collection("users");
   configCollection = db.collection("config");
-}
+});
 
 export async function disconnect(): Promise<void> {
   if (clientPromise) await (await clientPromise).close();

lib/debug.ts

@@ -202,3 +202,37 @@ export function outgoingUdpMessage(
     appendFileSync(DEBUG_FILE, JSON.stringify(msg) + "\n");
   else throw new Error(`Unrecognized DEBUG_FORMAT option`);
 }
+
+export function outgoingXmppStanza(deviceId: string, body: string): void {
+  if (!DEBUG_FILE) return;
+  const now = new Date();
+  const msg = {
+    event: "outgoing XMPP stanza",
+    timestamp: now,
+    deviceId: deviceId,
+    body: body,
+  };
+
+  if (DEBUG_FORMAT === "yaml")
+    appendFileSync(DEBUG_FILE, "---\n" + yaml.stringify(msg));
+  else if (DEBUG_FORMAT === "json")
+    appendFileSync(DEBUG_FILE, JSON.stringify(msg) + "\n");
+  else throw new Error(`Unrecognized DEBUG_FORMAT option`);
+}
+
+export function incomingXmppStanza(deviceId: string, body: string): void {
+  if (!DEBUG_FILE) return;
+  const now = new Date();
+  const msg = {
+    event: "incoming XMPP stanza",
+    timestamp: now,
+    deviceId: deviceId,
+    body: body,
+  };
+
+  if (DEBUG_FORMAT === "yaml")
+    appendFileSync(DEBUG_FILE, "---\n" + yaml.stringify(msg));
+  else if (DEBUG_FORMAT === "json")
+    appendFileSync(DEBUG_FILE, JSON.stringify(msg) + "\n");
+  else throw new Error(`Unrecognized DEBUG_FORMAT option`);
+}

lib/fs.ts

@@ -20,11 +20,19 @@
 import * as url from "url";
 import * as querystring from "querystring";
 import { IncomingMessage, ServerResponse } from "http";
-import { GridFSBucket } from "mongodb";
+import { Collection, GridFSBucket } from "mongodb";
-import * as db from "./db";
+import { onConnect } from "./db";
 import * as logger from "./logger";
 import { getRequestOrigin } from "./forwarded";
 
+let filesCollection: Collection;
+let filesBucket: GridFSBucket;
+
+onConnect(async (db) => {
+  filesCollection = db.collection("fs.files");
+  filesBucket = new GridFSBucket(db);
+});
+
 export function listener(
   request: IncomingMessage,
   response: ServerResponse
@@ -39,7 +47,7 @@ export function listener(
       remoteAddress: getRequestOrigin(request).remoteAddress,
     };
 
-    db.filesCollection.findOne({ _id: filename }, (err, file) => {
+    filesCollection.findOne({ _id: filename }, (err, file) => {
       if (err) throw err;
 
       if (!file) {
@@ -55,8 +63,7 @@ export function listener(
         "Content-Length": file.length,
       });
 
-      const bucket = new GridFSBucket(db.client.db());
+      const downloadStream = filesBucket.openDownloadStreamByName(filename);
-      const downloadStream = bucket.openDownloadStreamByName(filename);
       downloadStream.pipe(response);
 
       logger.accessInfo(log);

lib/local-cache.ts

@@ -398,7 +398,7 @@ async function refresh(): Promise<void> {
     return;
   }
 
-  const unlockOrExtend = await cache.lock("presets_hash_lock", 3);
+  const lockToken = await cache.acquireLock("presets_hash_lock", 4000, 4000);
 
   const res = await Promise.all([
     fetchPresets(),
@@ -432,7 +432,7 @@ async function refresh(): Promise<void> {
   currentSnapshot = computeHash(snapshot);
   snapshots.set(currentSnapshot, snapshot);
   await cache.set("presets_hash", currentSnapshot, 300);
-  await unlockOrExtend(0);
+  await cache.releaseLock("presets_hash_lock", lockToken);
 
   nextRefresh = now + (REFRESH - (now % REFRESH));
 }

lib/nbi.ts

@@ -18,11 +18,11 @@
  */
 
 import * as url from "url";
-import * as mongodb from "mongodb";
+import { Collection, GridFSBucket, ObjectId } from "mongodb";
 import * as querystring from "querystring";
 import * as vm from "vm";
 import * as config from "./config";
-import * as db from "./db";
+import { onConnect } from "./db";
 import * as query from "./query";
 import * as apiFunctions from "./api-functions";
 import { IncomingMessage, ServerResponse } from "http";
@@ -43,6 +43,23 @@ const PROVISIONS_REGEX = /^\/provisions\/([a-zA-Z0-9\-_%]+)\/?$/;
 const VIRTUAL_PARAMETERS_REGEX = /^\/virtual_parameters\/([a-zA-Z0-9\-_%]+)\/?$/;
 const FAULTS_REGEX = /^\/faults\/([a-zA-Z0-9\-_%:]+)\/?$/;
 
+const collections = {
+  tasks: null as Collection,
+  devices: null as Collection,
+  presets: null as Collection,
+  objects: null as Collection,
+  "fs.files": null as Collection,
+  provisions: null as Collection,
+  virtualParameters: null as Collection,
+  faults: null as Collection,
+};
+let filesBucket: GridFSBucket;
+
+onConnect(async (db) => {
+  for (const k of Object.keys(collections)) collections[k] = db.collection(k);
+  filesBucket = new GridFSBucket(db);
+});
+
 function throwError(err: Error, httpResponse?: ServerResponse): never {
   if (httpResponse) {
     httpResponse.writeHead(500, { Connection: "close" });
@@ -85,7 +102,7 @@ export function listener(
       if (request.method === "PUT") {
         const preset = JSON.parse(body.toString());
         preset._id = presetName;
-        db.presetsCollection.replaceOne(
+        collections.presets.replaceOne(
           { _id: presetName },
           preset,
           { upsert: true },
@@ -106,7 +123,7 @@ export function listener(
           }
         );
       } else if (request.method === "DELETE") {
-        db.presetsCollection.deleteOne({ _id: presetName }, (err) => {
+        collections.presets.deleteOne({ _id: presetName }, (err) => {
           if (err) return void throwError(err, response);
 
           cache
@@ -132,7 +149,7 @@ export function listener(
       if (request.method === "PUT") {
         const object = JSON.parse(body.toString());
         object._id = objectName;
-        db.objectsCollection.replaceOne(
+        collections.objects.replaceOne(
           { _id: objectName },
           object,
           { upsert: true },
@@ -153,7 +170,7 @@ export function listener(
           }
         );
       } else if (request.method === "DELETE") {
-        db.objectsCollection.deleteOne({ _id: objectName }, (err) => {
+        collections.objects.deleteOne({ _id: objectName }, (err) => {
           if (err) return void throwError(err, response);
 
           cache
@@ -190,7 +207,7 @@ export function listener(
           return;
         }
 
-        db.provisionsCollection.replaceOne(
+        collections.provisions.replaceOne(
           { _id: provisionName },
           object,
           { upsert: true },
@@ -211,7 +228,7 @@ export function listener(
           }
         );
       } else if (request.method === "DELETE") {
-        db.provisionsCollection.deleteOne({ _id: provisionName }, (err) => {
+        collections.provisions.deleteOne({ _id: provisionName }, (err) => {
           if (err) return void throwError(err, response);
 
           cache
@@ -248,7 +265,7 @@ export function listener(
           return;
         }
 
-        db.virtualParametersCollection.replaceOne(
+        collections.virtualParameters.replaceOne(
           { _id: virtualParameterName },
           object,
           { upsert: true },
@@ -269,7 +286,7 @@ export function listener(
           }
         );
       } else if (request.method === "DELETE") {
-        db.virtualParametersCollection.deleteOne(
+        collections.virtualParameters.deleteOne(
           { _id: virtualParameterName },
           (err) => {
             if (err) return void throwError(err, response);
@@ -296,7 +313,7 @@ export function listener(
       const deviceId = querystring.unescape(r[1]);
       const tag = querystring.unescape(r[2]);
       if (request.method === "POST") {
-        db.devicesCollection.updateOne(
+        collections.devices.updateOne(
           { _id: deviceId },
           { $addToSet: { _tags: tag } },
           (err) => {
@@ -306,7 +323,7 @@ export function listener(
           }
         );
       } else if (request.method === "DELETE") {
-        db.devicesCollection.updateOne(
+        collections.devices.updateOne(
           { _id: deviceId },
           { $pull: { _tags: tag } },
           (err) => {
@@ -327,12 +344,12 @@ export function listener(
         );
         const deviceId = faultId.split(":", 1)[0];
         const channel = faultId.slice(deviceId.length + 1);
-        db.faultsCollection.deleteOne({ _id: faultId }, (err) => {
+        collections.faults.deleteOne({ _id: faultId }, (err) => {
           if (err) return void throwError(err, response);
 
           if (channel.startsWith("task_")) {
-            const objId = new mongodb.ObjectID(channel.slice(5));
+            const objId = new ObjectId(channel.slice(5));
-            return void db.tasksCollection.deleteOne({ _id: objId }, (err) => {
+            return void collections.tasks.deleteOne({ _id: objId }, (err) => {
               if (err) return void throwError(err, response);
 
               cache
@@ -375,47 +392,106 @@ export function listener(
           task.device = deviceId;
           apiFunctions
             .insertTasks(task)
-            .then(() => {
+            .then(async () => {
-              cache
+              const lastInform = Date.now();
-                .del(`${deviceId}_tasks_faults_operations`)
+
-                .then(() => {
+              const socketTimeout: number = request.socket["timeout"];
-                  if (urlParts.query.connection_request != null) {
+
-                    apiFunctions
+              // Disable socket timeout while waiting for session
-                      .connectionRequest(deviceId)
+              if (socketTimeout) request.socket.setTimeout(0);
-                      .then(() => {
+
-                        const taskTimeout =
+              const notInSession = await apiFunctions.awaitSessionEnd(
+                deviceId,
+                30000
+              );
+              await cache.del(`${deviceId}_tasks_faults_operations`);
+              if (urlParts.query.connection_request == null) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
+                response.writeHead(202, {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              if (!notInSession) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const status = await apiFunctions.connectionRequest(deviceId);
+
+              if (status) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
+                response.writeHead(202, status, {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const onlineThreshold =
                 (urlParts.query.timeout &&
                   parseInt(urlParts.query.timeout as string)) ||
-                          (config.get(
+                (config.get("DEVICE_ONLINE_THRESHOLD", deviceId) as number);
-                            "DEVICE_ONLINE_THRESHOLD",
-                            deviceId
-                          ) as number);
 
-                        apiFunctions
+              const sessionStarted = await apiFunctions.awaitSessionStart(
-                          .watchTask(deviceId, task._id, taskTimeout)
+                deviceId,
-                          .then((status) => {
+                lastInform,
-                            if (status === "timeout") {
+                onlineThreshold
-                              response.writeHead(
+              );
-                                202,
+              if (!sessionStarted) {
-                                "Task queued but not processed",
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
-                                {
+                response.writeHead(202, "Task queued but not processed", {
                   "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const sessionEnded = await apiFunctions.awaitSessionEnd(
+                deviceId,
+                120000
+              );
+              if (!sessionEnded) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(task));
+                return;
+              }
+
+              const prom1 = collections.tasks.findOne(
+                { _id: task._id },
+                { projection: { _id: 1 } }
+              );
+              const prom2 = collections.faults.findOne(
+                { _id: `${deviceId}:task_${task._id}` },
+                {
+                  projection: { _id: 1 },
                 }
               );
-                              response.end(JSON.stringify(task));
-                            } else if (status === "fault") {
-                              db.tasksCollection.findOne(
-                                { _id: task._id },
-                                (err, task2) => {
-                                  if (err)
-                                    return void throwError(err, response);
 
+              const [t, f] = await Promise.all([prom1, prom2]);
+
+              // Restore socket timeout
+              if (socketTimeout) request.socket.setTimeout(socketTimeout);
+
+              if (f) {
                 response.writeHead(202, "Task faulted", {
                   "Content-Type": "application/json",
                 });
-                                  response.end(JSON.stringify(task2));
+                response.end(JSON.stringify(t || task));
-                                }
+              } else if (t) {
-                              );
+                response.writeHead(202, "Task queued but not processed", {
+                  "Content-Type": "application/json",
+                });
+                response.end(JSON.stringify(t));
               } else {
                 response.writeHead(200, {
                   "Content-Type": "application/json",
@@ -428,31 +504,6 @@ export function listener(
                 throwError(err, response);
               });
             });
-                      })
-                      .catch((err) => {
-                        response.writeHead(202, err.message, {
-                          "Content-Type": "application/json",
-                        });
-                        response.end(JSON.stringify(task));
-                      });
-                  } else {
-                    response.writeHead(202, {
-                      "Content-Type": "application/json",
-                    });
-                    response.end(JSON.stringify(task));
-                  }
-                })
-                .catch((err) => {
-                  setTimeout(() => {
-                    throwError(err, response);
-                  });
-                });
-            })
-            .catch((err) => {
-              setTimeout(() => {
-                throwError(err, response);
-              });
-            });
         } else if (urlParts.query.connection_request != null) {
           // No task, send connection request only
           apiFunctions
@@ -479,8 +530,8 @@ export function listener(
       const action = r[2];
       if (!action || action === "/") {
         if (request.method === "DELETE") {
-          db.tasksCollection.findOne(
+          collections.tasks.findOne(
-            { _id: new mongodb.ObjectID(taskId) },
+            { _id: new ObjectId(taskId) },
             { projection: { device: 1 } },
             (err, task) => {
               if (err) return void throwError(err, response);
@@ -492,12 +543,12 @@ export function listener(
               }
 
               const deviceId = task.device;
-              db.tasksCollection.deleteOne(
+              collections.tasks.deleteOne(
-                { _id: new mongodb.ObjectID(taskId) },
+                { _id: new ObjectId(taskId) },
                 (err) => {
                   if (err) return void throwError(err, response);
 
-                  db.faultsCollection.deleteOne(
+                  collections.faults.deleteOne(
                     { _id: `${deviceId}:task_${taskId}` },
                     (err) => {
                       if (err) return void throwError(err, response);
@@ -525,14 +576,14 @@ export function listener(
         }
       } else if (action === "/retry") {
         if (request.method === "POST") {
-          db.tasksCollection.findOne(
+          collections.tasks.findOne(
-            { _id: new mongodb.ObjectID(taskId) },
+            { _id: new ObjectId(taskId) },
             { projection: { device: 1 } },
             (err, task) => {
               if (err) return void throwError(err, response);
 
               const deviceId = task.device;
-              db.faultsCollection.deleteOne(
+              collections.faults.deleteOne(
                 { _id: `${deviceId}:task_${taskId}` },
                 (err) => {
                   if (err) return void throwError(err, response);
@@ -571,9 +622,8 @@ export function listener(
           productClass: request.headers.productclass,
           version: request.headers.version,
         };
-        const bucket = new mongodb.GridFSBucket(db.client.db());
+        filesBucket.delete((filename as unknown) as ObjectId, () => {
-        bucket.delete((filename as unknown) as mongodb.ObjectId, () => {
+          const uploadStream = filesBucket.openUploadStreamWithId(
-          const uploadStream = bucket.openUploadStreamWithId(
             filename,
             filename,
             {
@@ -591,8 +641,7 @@ export function listener(
           });
         });
       } else if (request.method === "DELETE") {
-        const bucket = new mongodb.GridFSBucket(db.client.db());
+        filesBucket.delete((filename as unknown) as ObjectId, (err) => {
-        bucket.delete((filename as unknown) as mongodb.ObjectId, (err) => {
           if (err) {
             if (err.message.startsWith("FileNotFound")) {
               response.writeHead(404);
@@ -669,7 +718,7 @@ export function listener(
         return;
       }
 
-      const collection = db[`${collectionName}Collection`];
+      const collection = collections[collectionName];
       if (!collection) {
         response.writeHead(404);
         response.end("404 Not Found");
@@ -693,7 +742,7 @@ export function listener(
           break;
         case "tasks":
           q = query.sanitizeQueryTypes(q, {
-            _id: (v) => new mongodb.ObjectID(v as string),
+            _id: (v) => new ObjectId(v as string),
             timestamp: (v) => new Date(v as number),
             retries: Number,
           });

lib/types.ts

@@ -163,6 +163,7 @@ export interface SessionContext {
   cacheUntil?: number;
   syncState?: SyncState;
   lastActivity?: number;
+  extendLock?: number;
   rpcRequest?: AcsRequest;
   operationsTouched?: { [commandKey: string]: 1 | 0 };
   provisionsRet?: any[];

lib/ui/api-functions.ts

@@ -24,7 +24,8 @@ import { getUsers } from "../local-cache";
 import { hashPassword } from "../auth";
 import {
   insertTasks,
-  watchTask,
+  awaitSessionEnd,
+  awaitSessionStart,
   connectionRequest,
   deleteDevice,
 } from "../api-functions";
@@ -104,23 +105,45 @@ export async function postTasks(
   const statuses = tasks.map((t) => {
     return { _id: t._id, status: "pending" };
   });
-
+  const lastInform = Date.now();
-  await del(`${deviceId}_tasks_faults_operations`);
+  const notInSession = await awaitSessionEnd(deviceId, 30000);
-
+  if (!notInSession) {
-  try {
-    await connectionRequest(deviceId, device);
-  } catch (err) {
     await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
     return {
-      connectionRequest: err.message,
+      connectionRequest: "Session took too long to complete",
       tasks: statuses,
     };
   }
 
-  const sample = tasks[tasks.length - 1];
+  await del(`${deviceId}_tasks_faults_operations`);
 
-  // Waiting for session to finish or timeout
+  const status = await connectionRequest(deviceId, device);
-  await watchTask(deviceId, sample._id, timeout);
+
+  if (status) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: status,
+      tasks: statuses,
+    };
+  }
+
+  const sessionStarted = await awaitSessionStart(deviceId, lastInform, timeout);
+  if (!sessionStarted) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: "No contact from CPE",
+      tasks: statuses,
+    };
+  }
+
+  const sessionEnded = await awaitSessionEnd(deviceId, 120000);
+  if (!sessionEnded) {
+    await Promise.all(statuses.map((t) => db.deleteTask(new ObjectID(t._id))));
+    return {
+      connectionRequest: "Session took too long to complete",
+      tasks: statuses,
+    };
+  }
 
   const promises = [];
   for (const s of statuses) {

lib/ui/api.ts

@@ -531,6 +531,11 @@ router.post("/devices/:id/tasks", async (ctx) => {
     }
   ) as number;
 
+  const socketTimeout: number = ctx.socket["timeout"];
+
+  // Disable socket timeout while waiting for postTasks()
+  if (socketTimeout) ctx.socket.setTimeout(0);
+
   const res = await apiFunctions.postTasks(
     ctx.params.id,
     ctx.request.body,
@@ -538,6 +543,9 @@ router.post("/devices/:id/tasks", async (ctx) => {
     device
   );
 
+  // Restore socket timeout
+  if (socketTimeout) ctx.socket.setTimeout(socketTimeout);
+
   log.tasks = res.tasks.map((t) => t._id).join(",");
 
   logger.accessInfo(log);

lib/ui/db.ts

@@ -17,85 +17,24 @@
  * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-import { MongoClient, GridFSBucket, ObjectID } from "mongodb";
+import { Db, GridFSBucket, ObjectID } from "mongodb";
 import { Script } from "vm";
-import * as config from "../config";
+import { onConnect } from "../db";
 import * as mongodbFunctions from "../mongodb-functions";
 import * as expression from "../common/expression";
 import { QueryOptions, Expression } from "../types";
 import { Readable } from "stream";
 import { minimize } from "../common/boolean-expression";
 
-const CACHE_TTL = 300000;
-
-let clientPromise: Promise<MongoClient>;
-
 const RESOURCE_COLLECTION = {
   files: "fs.files",
 };
 
-function ensureIndexes(client): void {
+let db: Db;
-  client
-    .db()
-    .collection("cache")
-    .createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-}
 
-function getClient(): Promise<MongoClient> {
+onConnect(async (_db) => {
-  if (!clientPromise) {
+  db = _db;
-    clientPromise = new Promise((resolve, reject) => {
-      const CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
-      MongoClient.connect(
-        CONNECTION_URL,
-        { useNewUrlParser: true, useUnifiedTopology: true },
-        (err, client) => {
-          if (err) return void reject(err);
-          ensureIndexes(client);
-          resolve(client);
-        }
-      );
 });
-  }
-
-  return clientPromise;
-}
-
-export function cache<T>(
-  key: string,
-  valueGetter: () => Promise<T>,
-  ttl: number
-): Promise<T> {
-  return new Promise((resolve, reject) => {
-    getClient()
-      .then((client) => {
-        const collection = client.db().collection("cache");
-        collection.findOne({ _id: key }, (err, doc) => {
-          if (err) return void reject(err);
-          if (doc != null) return void resolve(JSON.parse(doc.value) as T);
-          valueGetter()
-            .then((res) => {
-              const expire = Date.now() + (ttl || CACHE_TTL);
-              const cacheDoc = {
-                _id: key,
-                value: JSON.stringify(res),
-                expire: new Date(expire),
-              };
-              collection.updateOne(
-                { _id: key },
-                { $set: cacheDoc },
-                { upsert: true },
-                (err) => {
-                  if (err) reject(err);
-                  else resolve(res);
-                }
-              );
-            })
-            .catch(reject);
-        });
-      })
-      .catch(reject);
-  });
-}
 
 export function query(
   resource: string,
@@ -123,16 +62,12 @@ export function query(
     }
 
     q = mongodbFunctions.filterToMongoQuery(filter);
-  } else if (filter != null && !filter) {
+  } else if (!filter) {
     return Promise.resolve([]);
   }
 
   return new Promise((resolve, reject) => {
-    getClient()
+    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
     const cursor = collection.find(q);
     if (options.projection) {
       let projection = options.projection;
@@ -182,8 +117,7 @@ export function query(
     } else {
       cursor.forEach(
         (doc) => {
-              if (resource === "devices")
+          if (resource === "devices") doc = mongodbFunctions.flattenDevice(doc);
-                doc = mongodbFunctions.flattenDevice(doc);
           else if (resource === "faults")
             doc = mongodbFunctions.flattenFault(doc);
           else if (resource === "tasks")
@@ -200,8 +134,6 @@ export function query(
         }
       );
     }
-      })
-      .catch(reject);
   });
 }
 
@@ -218,22 +150,16 @@ export function count(resource: string, filter: Expression): Promise<number> {
     else if (resource === "faults")
       filter = mongodbFunctions.processFaultsFilter(filter);
     q = mongodbFunctions.filterToMongoQuery(filter);
-  } else if (filter != null && !filter) {
+  } else if (!filter) {
     return Promise.resolve(0);
   }
 
   return new Promise((resolve, reject) => {
-    getClient()
+    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
     collection.find(q).count((err, c) => {
       if (err) reject(err);
       else resolve(c);
     });
-      })
-      .catch(reject);
   });
 }
 
@@ -250,8 +176,7 @@ export async function updateDeviceTags(
     else pull.push(tag);
   }
 
-  const client = await getClient();
+  const collection = db.collection("devices");
-  const collection = client.db().collection("devices");
   const object = {};
 
   if (add && add.length) object["$addToSet"] = { _tags: { $each: add } };
@@ -262,17 +187,11 @@ export async function updateDeviceTags(
 
 function putResource(resource, id, object): Promise<void> {
   return new Promise((resolve, reject) => {
-    getClient()
+    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
     collection.replaceOne({ _id: id }, object, { upsert: true }, (err) => {
       if (err) return void reject(err);
       resolve();
     });
-      })
-      .catch(reject);
   });
 }
 
@@ -281,17 +200,11 @@ function deleteResource(
   id: string | ObjectID
 ): Promise<void> {
   return new Promise((resolve, reject) => {
-    getClient()
+    const collection = db.collection(RESOURCE_COLLECTION[resource] || resource);
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
     collection.deleteOne({ _id: id }, (err) => {
       if (err) return void reject(err);
       resolve();
     });
-      })
-      .catch(reject);
   });
 }
 
@@ -368,9 +281,7 @@ export function putUser(
   object: Record<string, unknown>
 ): Promise<void> {
   return new Promise((resolve, reject) => {
-    getClient()
+    const collection = db.collection("users");
-      .then((client) => {
-        const collection = client.db().collection("users");
     // update instead of replace to keep the password if not set by user
     collection.updateOne(
       { _id: id },
@@ -381,8 +292,6 @@ export function putUser(
         resolve();
       }
     );
-      })
-      .catch(reject);
   });
 }
 
@@ -396,9 +305,7 @@ export function putFile(
   contentStream: Readable
 ): Promise<void> {
   return new Promise((resolve, reject) => {
-    getClient()
+    const bucket = new GridFSBucket(db);
-      .then((client) => {
-        const bucket = new GridFSBucket(client.db());
     const uploadStream = bucket.openUploadStreamWithId(filename, filename, {
       metadata: metadata,
     });
@@ -406,22 +313,16 @@ export function putFile(
     contentStream.on("error", reject);
     uploadStream.on("finish", resolve);
     contentStream.pipe(uploadStream);
-      })
-      .catch(reject);
   });
 }
 
 export function deleteFile(filename: string): Promise<void> {
   return new Promise((resolve, reject) => {
-    getClient()
+    const bucket = new GridFSBucket(db);
-      .then((client) => {
-        const bucket = new GridFSBucket(client.db());
     bucket.delete(filename as any, (err) => {
       if (err) return void reject(err);
       resolve();
     });
-      })
-      .catch(reject);
   });
 }
 
@@ -432,7 +333,3 @@ export function deleteFault(id: string): Promise<void> {
 export function deleteTask(id: ObjectID): Promise<void> {
   return deleteResource("tasks", id);
 }
-
-export async function disconnect(): Promise<void> {
-  if (clientPromise) await (await clientPromise).close();
-}

lib/xml-parser.ts

@@ -57,9 +57,9 @@ export function parseXmlDeclaration(buffer: Buffer): Attribute[] {
   for (const enc of encodings) {
     let str = buffer.toString(enc, 0, 150);
     if (str.startsWith("<?xml")) {
-      str = str.split("\n")[0].trim();
+      str = str.slice(0, str.indexOf("?>"));
       try {
-        return parseAttrs(str.slice(5, -2));
+        return parseAttrs(str.slice(5));
       } catch (err) {
         // Ignore
       }
@@ -104,10 +104,12 @@ export function parseAttrs(string: string): Attribute[] {
         continue;
 
       case CHAR_COLON:
+        if (state) continue;
         if (idx >= colonIdx) colonIdx = i;
         continue;
 
       case CHAR_EQUAL:
+        if (state) continue;
         if (name) throw new Error(`Unexpected character at ${i}`);
         name = string.slice(idx, i).trim();
         // TODO validate name

lib/xmpp-client.ts

@@ -0,0 +1,481 @@
+/**
+ * Copyright 2013-2020  GenieACS Inc.
+ *
+ * This file is part of GenieACS.
+ *
+ * GenieACS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * GenieACS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with GenieACS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+import * as net from "net";
+import * as tls from "tls";
+import { EventEmitter } from "events";
+import { createHash, createHmac, randomBytes } from "crypto";
+import { parseXml, Element, parseAttrs } from "./xml-parser";
+
+function encodeBase64(str: string): string {
+  return Buffer.from(str).toString("base64");
+}
+
+function decodeBase64(str: string): string {
+  return Buffer.from(str, "base64").toString();
+}
+
+function detectStreamTag(data: string): number {
+  const i1 = data.indexOf("<stream:stream");
+  if (i1 < 0) throw new Error("Cannot detect opening stream tag");
+  const i2 = data.indexOf(">", i1);
+  if (i2 < 0) throw new Error("Cannot detect opening stream tag");
+  return i2 + 1;
+}
+
+function xmppStream<T>(
+  socket: net.Socket,
+  callback: Generator<void, T, Element>
+): Promise<T> {
+  return new Promise((resolve, reject) => {
+    const onError = (err: Error): void => {
+      socket.removeListener("error", onError);
+      reject(err);
+    };
+    socket.on("error", onError);
+
+    const onData = (chunk: Buffer): void => {
+      try {
+        const str = chunk.toString("utf8");
+        const xml = parseXml(str);
+        const { value, done } = callback.next(xml.children[0]);
+        if (done) {
+          socket.removeListener("error", onError);
+          socket.removeListener("data", onData);
+          resolve(value as T);
+        }
+      } catch (err) {
+        socket.removeListener("error", onError);
+        socket.removeListener("data", onData);
+        reject(err);
+      }
+    };
+
+    socket.once("data", (chunk: Buffer) => {
+      try {
+        const str = chunk.toString("utf8");
+        const i = detectStreamTag(str);
+        const streamTagStr = str.slice(0, i);
+        const xml = parseXml(streamTagStr + "</stream:stream>");
+        callback.next(xml.children[0]);
+        chunk = chunk.slice(Buffer.byteLength(streamTagStr));
+        if (chunk.length) onData(chunk);
+        socket.on("data", onData);
+      } catch (err) {
+        socket.removeListener("error", onError);
+        socket.removeListener("data", onData);
+        reject(err);
+      }
+    });
+
+    try {
+      callback.next();
+    } catch (err) {
+      socket.removeListener("error", onError);
+      socket.removeListener("data", onData);
+      reject(err);
+    }
+  });
+}
+
+const INT_1 = Buffer.from([0, 0, 0, 1]);
+const saltedPasswordCache = {
+  password: "",
+  iterationCount: 0,
+  saltBase64: "",
+  salted: Buffer.allocUnsafe(0),
+};
+
+function saltPassword(
+  password: string,
+  saltBase64: string,
+  iteractionCount: number
+): Buffer {
+  if (
+    password === saltedPasswordCache.password &&
+    saltBase64 === saltedPasswordCache.saltBase64 &&
+    iteractionCount === saltedPasswordCache.iterationCount
+  )
+    return saltedPasswordCache.salted;
+
+  const hi = createHmac("sha1", password)
+    .update(Buffer.concat([Buffer.from(saltBase64, "base64"), INT_1]))
+    .digest();
+  let hi2: Buffer = hi;
+  for (let i = 1; i < iteractionCount; ++i) {
+    hi2 = createHmac("sha1", password).update(hi2).digest();
+    for (const [j, b] of hi2.entries()) hi[j] ^= b;
+  }
+
+  saltedPasswordCache.saltBase64 = saltBase64;
+  saltedPasswordCache.password = password;
+  saltedPasswordCache.iterationCount = iteractionCount;
+  saltedPasswordCache.salted = hi;
+
+  return hi;
+}
+
+function* loginPlain(
+  socket: net.Socket,
+  username: string,
+  password: string
+): Generator<void, void, Element> {
+  socket.write(
+    `<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="PLAIN">${encodeBase64(
+      `\x00${username}\x00${password}`
+    )}</auth>`
+  );
+  const res1 = yield;
+
+  if (
+    res1.name === "failure" &&
+    res1.children.some((c) => c.name === "not-authorized")
+  )
+    throw new Error("Not authorized");
+
+  if (res1.name !== "success")
+    throw new Error(`Unexpected response ${res1.name}`);
+}
+
+function* loginScram(
+  socket: net.Socket,
+  username: string,
+  password: string
+): Generator<void, void, Element> {
+  const cnonce = randomBytes(8).toString("base64");
+  const gs2Header = "n,,";
+  const clientFirstMessageBare = `n=${username},r=${cnonce}`;
+  const clientFirstMessage = gs2Header + clientFirstMessageBare;
+  socket.write(
+    `<auth xmlns="urn:ietf:params:xml:ns:xmpp-sasl" mechanism="SCRAM-SHA-1">${encodeBase64(
+      clientFirstMessage
+    )}</auth>`
+  );
+  const res1 = yield;
+  if (res1.name !== "challenge")
+    throw new Error(`Unexpected element ${res1.name}`);
+  const serverFirstMessage = decodeBase64(res1.text);
+
+  let iterationCount: number;
+  let saltBase64: string;
+  let nonce: string;
+  for (const s of serverFirstMessage.split(",")) {
+    if (s.startsWith("i=")) iterationCount = parseInt(s.slice(2));
+    else if (s.startsWith("s=")) saltBase64 = s.slice(2);
+    else if (s.startsWith("r=")) nonce = s.slice(2);
+  }
+  if (iterationCount == null || isNaN(iterationCount))
+    throw new Error("Invalid iteration count");
+  if (saltBase64 == null) throw new Error("Missing salt");
+  if (nonce == null) throw new Error("Missing nonce");
+
+  const saltedPassword = saltPassword(password, saltBase64, iterationCount);
+  const clientKey = createHmac("sha1", saltedPassword)
+    .update("Client Key")
+    .digest();
+  const storedKey = createHash("sha1").update(clientKey).digest();
+
+  const clientFinalMessageWithoutProof = `c=${encodeBase64(
+    gs2Header
+  )},r=${nonce}`;
+  const authMessage = `${clientFirstMessageBare},${serverFirstMessage},${clientFinalMessageWithoutProof}`;
+  const clientSignature = createHmac("sha1", storedKey)
+    .update(authMessage)
+    .digest();
+
+  const clientProof = Buffer.from(clientKey);
+  for (const [i, b] of clientSignature.entries()) clientProof[i] ^= b;
+
+  const clientFinalMessage = `${clientFinalMessageWithoutProof},p=${clientProof.toString(
+    "base64"
+  )}`;
+  socket.write(
+    `<response xmlns="urn:ietf:params:xml:ns:xmpp-sasl">${encodeBase64(
+      clientFinalMessage
+    )}</response>`
+  );
+  const res2 = yield;
+
+  if (
+    res2.name === "failure" &&
+    res2.children.some((c) => c.name === "not-authorized")
+  )
+    throw new Error("Not authorized");
+
+  if (res2.name !== "success")
+    throw new Error(`Unexpected response ${res2.name}`);
+
+  const serverKey = createHmac("sha1", saltedPassword)
+    .update("Server Key")
+    .digest();
+  const serverSignature = createHmac("sha1", serverKey)
+    .update(authMessage)
+    .digest("base64");
+
+  if (!decodeBase64(res2.text).endsWith(serverSignature))
+    throw new Error("Invalid server signature");
+}
+
+function* starttls(socket: net.Socket): Generator<void, void, Element> {
+  socket.write("<starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'/>");
+  const res1 = yield;
+  if (res1.name !== "proceed") throw new Error("Failed to initiate STARTTLS");
+}
+
+function* bind(
+  socket: net.Socket,
+  resource: string
+): Generator<void, void, Element> {
+  const id = randomBytes(8).toString("base64");
+  socket.write(
+    `<iq id='${id}' type='set'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'><resource>${resource}</resource></bind></iq>`
+  );
+  const res1 = yield;
+  if (res1.name !== "iq") throw new Error(`Unexpected element ${res1.name}`);
+  const attrs1 = parseAttrs(res1.attrs);
+  const idAttr = attrs1.find((a) => a.name === "id");
+  if (!idAttr || idAttr.value !== id) throw new Error("Invalid ID");
+  const typeAttr = attrs1.find((a) => a.name === "type");
+  if (!typeAttr) throw new Error("Missing type attribute");
+  if (typeAttr.value !== "result") throw new Error("Cannot bind to resource");
+}
+
+const STATUS_RESTART_STREAM = 1;
+const STATUS_STARTTLS = 2;
+
+function* init(
+  socket: net.Socket,
+  host: string,
+  username: string,
+  password: string,
+  resource: string
+): Generator<void, number, Element> {
+  socket.write(
+    `<?xml version='1.0'?><stream:stream from='${username}@${host}' to='${host}' version='1.0' xml:lang='en' xmlns='jabber:client' xmlns:stream='http://etherx.jabber.org/streams'>`
+  );
+  const open = yield;
+  if (open.name !== "stream:stream")
+    throw new Error(`Unexpected element ${open.name}`);
+  const features = yield;
+  if (features.name !== "stream:features")
+    throw new Error(`Unexpected element ${features.name}`);
+  for (const feature of features.children) {
+    if (feature.name === "starttls") {
+      if (feature.children.some((c) => c.name === "required")) {
+        yield* starttls(socket);
+        return STATUS_STARTTLS;
+      }
+    } else if (feature.name === "mechanisms") {
+      const mechanisms: Set<string> = new Set(
+        feature.children.map((c) => c.text)
+      );
+      if (mechanisms.has("PLAIN")) {
+        yield* loginPlain(socket, username, password);
+        return STATUS_RESTART_STREAM;
+      } else if (mechanisms.has("SCRAM-SHA-1")) {
+        yield* loginScram(socket, username, password);
+        return STATUS_RESTART_STREAM;
+      } else {
+        throw new Error("No supported SASL method");
+      }
+    } else if (feature.name === "bind") {
+      yield* bind(socket, resource);
+      return 0;
+    }
+  }
+  return 0;
+}
+
+function upgradeTls(socket: net.Socket, host: string): Promise<tls.TLSSocket> {
+  return new Promise((resolve, reject) => {
+    socket.on("error", reject);
+    const newSocket = tls.connect({ socket, host }, () => {
+      socket.removeListener("error", reject);
+      resolve(newSocket);
+    });
+  });
+}
+
+interface XmppClientOptions {
+  host: string;
+  port?: number;
+  username?: string;
+  password?: string;
+  resource?: string;
+  timeout?: number;
+}
+
+export default class XmppClient extends EventEmitter {
+  private _socket: net.Socket;
+  private _host: string;
+  private _username: string;
+  private _resource: string;
+  private _iqStanzaCallbacks: Map<
+    string,
+    (err: Error, r?: { rawRes: string; res: Element }) => void
+  >;
+
+  private constructor() {
+    super();
+    this._socket = null;
+    this._host = null;
+    this._username = null;
+    this._resource = null;
+    this._iqStanzaCallbacks = new Map();
+  }
+
+  static connect(opts: XmppClientOptions): Promise<XmppClient> {
+    return new Promise((resolve, reject) => {
+      let socket = new net.Socket();
+      socket.on("error", reject);
+
+      socket.connect(opts.port || 5222, opts.host, async () => {
+        socket.removeListener("error", reject);
+        try {
+          let status = 1;
+          while (status) {
+            if (status === STATUS_STARTTLS)
+              socket = await upgradeTls(socket, opts.host);
+            status = await xmppStream(
+              socket,
+              init(
+                socket,
+                opts.host,
+                opts.username,
+                opts.password,
+                opts.resource
+              )
+            );
+          }
+
+          const client = new XmppClient();
+          client._socket = socket;
+          client._host = opts.host;
+          client._username = opts.username;
+          client._resource = opts.resource;
+          socket.on("data", client._onData.bind(client));
+          socket.on("error", client._onError.bind(client));
+          if (opts.timeout)
+            socket.setTimeout(opts.timeout, client.close.bind(client));
+          resolve(client);
+        } catch (err) {
+          socket.destroy();
+          reject(err);
+        }
+      });
+    });
+  }
+
+  close(): void {
+    this._socket.end("</stream:stream>");
+    this._socket.removeAllListeners("data");
+    this._socket.removeAllListeners("error");
+    this.emit("close");
+  }
+
+  ref(): void {
+    this._socket.ref();
+  }
+
+  unref(): void {
+    this._socket.unref();
+  }
+
+  get host(): string {
+    return this._host;
+  }
+
+  get username(): string {
+    return this._username;
+  }
+
+  get resource(): string {
+    return this._resource;
+  }
+
+  private _onData(chunk: Buffer): void {
+    try {
+      let close = false;
+      let str = chunk.toString("utf8");
+      if (str.endsWith("</stream:stream>")) {
+        str = str.slice(0, -16);
+        close = true;
+      }
+      const xml = parseXml(str);
+      const idx = xml.children.map((c) => c.bodyIndex);
+      for (const [i, c] of xml.children.entries()) {
+        const s = str.slice(idx[i], idx[i + 1]);
+        if (c.name === "iq") {
+          const attrs = parseAttrs(c.attrs);
+          const id = attrs.find((a) => a.name === "id");
+          if (id) {
+            const cb = this._iqStanzaCallbacks.get(id.value);
+            if (cb) cb(null, { rawRes: s, res: c });
+          }
+        }
+        this.emit("stanza", c, s);
+      }
+      if (close) {
+        this._socket.removeAllListeners("data");
+        this._socket.removeAllListeners("error");
+        this.emit("close");
+      }
+    } catch (err) {
+      this._socket.removeAllListeners("data");
+      this._socket.removeAllListeners("error");
+      this.emit("error", err);
+    }
+  }
+
+  private _onError(err: Error): void {
+    this._socket.end();
+    this.emit("error", err);
+    for (const cb of this._iqStanzaCallbacks.values()) cb(err);
+  }
+
+  send(msg: string): void {
+    this._socket.write(msg);
+  }
+
+  sendIqStanza(
+    from: string,
+    to: string,
+    type: string,
+    body: string,
+    timeout = 3000
+  ): Promise<{ rawReq: string; rawRes: string; res: Element }> {
+    return new Promise((resolve, reject) => {
+      const id = randomBytes(8).toString("base64");
+      const rawReq = `<iq from="${from}" to="${to}" id="${id}" type="${type}">${body}</iq>`;
+      this.send(rawReq);
+      const t = setTimeout(() => {
+        this._iqStanzaCallbacks.delete(id);
+        reject(
+          new Error("Did not receive IQ stanza response in a timely manner")
+        );
+      }, timeout);
+      this._iqStanzaCallbacks.set(id, (err, r) => {
+        this._iqStanzaCallbacks.delete(id);
+        clearTimeout(t);
+        if (err) reject(err);
+        else resolve(Object.assign(r, { rawReq }));
+      });
+    });
+  }
+}

ui/filter-component.ts

@@ -44,8 +44,12 @@ const getAutocomplete = memoize((resource) => {
 
 function validateQuery(q: Expression): void {
   q = normalize(q);
-  if (Array.isArray(q) && q[0] === "CASE") q = ["AND", ...q.slice(1)];
+  if (Array.isArray(q) && q[0] === "CASE") {
-  filterToMongoQuery(q);
+    q = q.slice(1).filter((a) => Array.isArray(a));
+    if (q.length > 1) q = ["AND", ...q];
+    else q = q[0];
+  }
+  if (Array.isArray(q)) filterToMongoQuery(q);
 }
 
 function parseFilter(resource, f): Expression {