Fix XMPP stream not getting closed properly

Always bind resource to XMPP stream
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.
2023-02-06 14:48:00 -08:00 · 2021-01-05 18:05:19 -08:00 · 2020-12-08 02:08:25 -08:00 · 2020-12-08 02:08:25 -08:00 · 2020-12-08 02:08:25 -08:00 · 2020-12-08 02:08:25 -08:00
23 changed files with 1109 additions and 404 deletions
@@ -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);
    })
@@ -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([
-    db.disconnect(),
-    cache.disconnect(),
-    cluster.worker.disconnect(),
-  ]).catch(exitWorkerUngracefully);
+  Promise.all([db.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);
    })
@@ -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);
    })
@@ -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 db2 from "../lib/db";
-import * as cache from "../lib/cache";
+import * as db from "../lib/db";
 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);
    })
@@ -51,6 +51,8 @@ const externals = [
  "tls",
  "http",
  "https",
+  "net",
+  "events",
  "zlib",
  "crypto",
  "util",
@@ -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 {
-    await httpConnectionRequest(
+  let status;
+
+  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> {
-  await new Promise((resolve) => setTimeout(resolve, 500));
-
-  const task = await db.tasksCollection.findOne(
-    { _id: taskId },
-    { projection: { _id: 1 } }
+): Promise<boolean> {
+  const now = Date.now();
+  const device = await db.devicesCollection.findOne(
+    { _id: deviceId },
+    { projection: { _lastInform: 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}` };
-  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);
+export async function awaitSessionEnd(
+  deviceId: string,
+  timeout: number
+): Promise<boolean> {
+  const now = Date.now();
+  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 awaitSessionEnd(deviceId, timeout);
 }

 function sanitizeTask(task): void {
@@ -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 mongoCollection: Collection;
-let mongoTimeOffset = 0;
+let cacheCollection: Collection;

-export async function connect(): Promise<void> {
-  const MONGODB_CONNECTION_URL = "" + config.get("MONGODB_CONNECTION_URL");
-  clientPromise = MongoClient.connect(MONGODB_CONNECTION_URL, {
-    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;
+onConnect(async (db) => {
+  cacheCollection = db.collection("cache");
+  await cacheCollection.createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
 });

-    const values = [];
-    res.forEach((r) => {
-      if (r["expire"] > expire) values[indices[r["_id"]]] = r["value"];
-    });
-
-    return values;
-  } else {
-    const res = await mongoCollection.findOne({ _id: { $in: [key] } });
-    if (res && res["expire"] > expire) return res["value"];
+export async function get(key: string): Promise<string> {
+  const res = await cacheCollection.findOne({ _id: key });
+  if (res) return res["value"];
  return null;
 }
-}

-export async function del(key: string | string[]): Promise<void> {
-  if (Array.isArray(key))
-    await mongoCollection.deleteMany({ _id: { $in: key } });
-  else await mongoCollection.deleteOne({ _id: key });
+export async function del(key: string): Promise<void> {
+  await cacheCollection.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);
-  await mongoCollection.replaceOne(
+  const timestamp = new Date();
+  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 });
-
-  if (
-    res &&
-    res["value"] &&
-    +res["value"]["expire"] - (Date.now() - mongoTimeOffset)
-  )
-    return res["value"]["value"];
-
+  const res = await cacheCollection.findOneAndDelete({ _id: key });
+  if (res && res["value"]) return res["value"]["value"];
  return null;
 }

-export async function lock(
+export async function acquireLock(
  lockName: string,
-  ttl: number
-): Promise<(extendTtl?: number) => Promise<void>> {
-  const token = Math.random().toString(36).slice(2);
+  ttl: number,
+  timeout = 0,
+  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> {
-    if (!extendTtl) {
-      const res = await mongoCollection.deleteOne({
+  return token;
+}
+
+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;
 }
@@ -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);
@@ -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 } = {};
@@ -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();
-        reject(new Error("Device is offline"));
+        req.destroy();
+        reject(err);
        if (_debug) debug.outgoingHttpRequestError(req, deviceId, options, err);
      })
-      .on("socket", (socket) => {
-        socket.setTimeout(timeout);
-        socket.on("timeout", () => {
-          req.abort();
-        });
+      .on("timeout", () => {
+        req.destroy();
      });
  });
 }
@@ -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)
-          throw new Error("Basic HTTP authentication not allowed");
+        if (!allowBasicAuth) return "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)
-      res = await httpGet(opts, timeout, _debug, deviceId);
-    if (res.statusCode === 0) throw new Error("Device is offline");
-    if (res.statusCode === 200 || res.statusCode === 204) return;
+      if (authHeader) {
+        try {
+          res = await httpGet(opts, _debug, deviceId);
+        } 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(
-        `Unexpected response code from device: ${res.statusCode}`
-      );
+      return `Connection request error: Unexpected status code ${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";
+}
@@ -196,17 +196,23 @@ async function writeResponse(
      sessionContext: sessionContext,
      message: "Connection dropped",
    });
+    await endSession(sessionContext);
  } else if (close) {
-    const isNew = await endSession(sessionContext);
-    if (isNew) {
-      logger.accessInfo({
-        sessionContext: sessionContext,
-        message: "New device registered",
-      });
-    }
+    session.clearProvisions(sessionContext);
+    await endSession(sessionContext);
  } 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(
@@ -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();
@@ -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`);
+}
@@ -20,11 +20,19 @@
 import * as url from "url";
 import * as querystring from "querystring";
 import { IncomingMessage, ServerResponse } from "http";
-import { GridFSBucket } from "mongodb";
-import * as db from "./db";
+import { Collection, GridFSBucket } from "mongodb";
+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 = bucket.openDownloadStreamByName(filename);
+      const downloadStream = filesBucket.openDownloadStreamByName(filename);
      downloadStream.pipe(response);

      logger.accessInfo(log);
@@ -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));
 }
@@ -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));
-            return void db.tasksCollection.deleteOne({ _id: objId }, (err) => {
+            const objId = new ObjectId(channel.slice(5));
+            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(() => {
-              cache
-                .del(`${deviceId}_tasks_faults_operations`)
-                .then(() => {
-                  if (urlParts.query.connection_request != null) {
-                    apiFunctions
-                      .connectionRequest(deviceId)
-                      .then(() => {
-                        const taskTimeout =
+            .then(async () => {
+              const lastInform = Date.now();
+
+              const socketTimeout: number = request.socket["timeout"];
+
+              // Disable socket timeout while waiting for session
+              if (socketTimeout) request.socket.setTimeout(0);
+
+              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(
-                            "DEVICE_ONLINE_THRESHOLD",
-                            deviceId
-                          ) as number);
+                (config.get("DEVICE_ONLINE_THRESHOLD", deviceId) as number);

-                        apiFunctions
-                          .watchTask(deviceId, task._id, taskTimeout)
-                          .then((status) => {
-                            if (status === "timeout") {
-                              response.writeHead(
-                                202,
-                                "Task queued but not processed",
-                                {
+              const sessionStarted = await apiFunctions.awaitSessionStart(
+                deviceId,
+                lastInform,
+                onlineThreshold
+              );
+              if (!sessionStarted) {
+                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(
-            { _id: new mongodb.ObjectID(taskId) },
+          collections.tasks.findOne(
+            { _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(
-                { _id: new mongodb.ObjectID(taskId) },
+              collections.tasks.deleteOne(
+                { _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(
-            { _id: new mongodb.ObjectID(taskId) },
+          collections.tasks.findOne(
+            { _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());
-        bucket.delete((filename as unknown) as mongodb.ObjectId, () => {
-          const uploadStream = bucket.openUploadStreamWithId(
+        filesBucket.delete((filename as unknown) as ObjectId, () => {
+          const uploadStream = filesBucket.openUploadStreamWithId(
            filename,
            filename,
            {
@@ -591,8 +641,7 @@ export function listener(
          });
        });
      } else if (request.method === "DELETE") {
-        const bucket = new mongodb.GridFSBucket(db.client.db());
-        bucket.delete((filename as unknown) as mongodb.ObjectId, (err) => {
+        filesBucket.delete((filename as unknown) as 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,
          });
@@ -163,6 +163,7 @@ export interface SessionContext {
  cacheUntil?: number;
  syncState?: SyncState;
  lastActivity?: number;
+  extendLock?: number;
  rpcRequest?: AcsRequest;
  operationsTouched?: { [commandKey: string]: 1 | 0 };
  provisionsRet?: any[];
@@ -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" };
  });
-
-  await del(`${deviceId}_tasks_faults_operations`);
-
-  try {
-    await connectionRequest(deviceId, device);
-  } catch (err) {
+  const lastInform = Date.now();
+  const notInSession = await awaitSessionEnd(deviceId, 30000);
+  if (!notInSession) {
    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
-  await watchTask(deviceId, sample._id, timeout);
+  const status = await connectionRequest(deviceId, device);
+
+  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) {
@@ -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);
@@ -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 {
-  client
-    .db()
-    .collection("cache")
-    .createIndex({ expire: 1 }, { expireAfterSeconds: 0 });
-}
+let db: Db;

-function getClient(): Promise<MongoClient> {
-  if (!clientPromise) {
-    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);
-        }
-      );
+onConnect(async (_db) => {
+  db = _db;
 });
-  }
-
-  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()
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
+    const collection = 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")
-                doc = mongodbFunctions.flattenDevice(doc);
+          if (resource === "devices") 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()
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
+    const collection = 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 = client.db().collection("devices");
+  const collection = 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()
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
+    const collection = 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()
-      .then((client) => {
-        const collection = client
-          .db()
-          .collection(RESOURCE_COLLECTION[resource] || resource);
+    const collection = 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()
-      .then((client) => {
-        const collection = client.db().collection("users");
+    const collection = 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()
-      .then((client) => {
-        const bucket = new GridFSBucket(client.db());
+    const bucket = new GridFSBucket(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()
-      .then((client) => {
-        const bucket = new GridFSBucket(client.db());
+    const bucket = new GridFSBucket(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();
-}
@@ -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
@@ -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 }));
+      });
+    });
+  }
+}
@@ -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)];
-  filterToMongoQuery(q);
+  if (Array.isArray(q) && q[0] === "CASE") {
+    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 {
Author	SHA1	Message	Date
Zaid Humaimidi	b5f689b014	Fix XMPP stream not getting closed properly	2023-02-06 14:48:00 -08:00
Zaid Abdulla	ccfbcddeda	Always bind resource to XMPP stream 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.	2021-01-05 18:05:19 -08:00
Zaid Abdulla	081201cd2f	Only send UDP con req if STUNEnable is true Because even when STUN is not enabled, the param UDPConnectionRequestAddress will still hold IP of the listening interface which we can't reach.	2020-12-08 02:08:25 -08:00
Zaid Abdulla	0e0380917f	XMPP fixes	2020-12-08 02:08:25 -08:00
Zaid Abdulla	79ede00ef2	Implement XMPP connection request	2020-12-08 02:08:25 -08:00
Zaid Abdulla	82ed742a67	Convenience function to send IQ stanza messages	2020-12-08 02:08:25 -08:00
Zaid Abdulla	53de5d01e7	Initial XmppClient implementation	2020-12-08 02:08:25 -08:00
Zaid Abdulla	d29448e3a7	Fix parsing XML attrs with colon, equal chars	2020-12-08 02:07:34 -08:00
Zaid Abdulla	a91dbe5021	Handle multi-line XML declaration	2020-12-08 02:06:55 -08:00
Zaid Abdulla	63cd8dd2de	Save fault when session terminates unsuccessfully	2020-12-08 00:27:29 -08:00
Zaid Abdulla	bded263adf	Refactor HTTP connection requests	2020-12-04 15:33:35 -08:00
Zaid Abdulla	fa1e40e370	Read session lock to detect task completion	2020-11-30 18:11:51 -08:00
Zaid Abdulla	30bea8ac57	Save CPE changes in case of session error/timeout	2020-11-19 14:18:05 -08:00
Zaid Abdulla	d787f66e3f	Reimplement locking w/o relying on hostInfo cmd Because the hostInfo MongoDB command may not be allowed in some environments. Also, in sharded deployments the clock skew may differ between different shards so using hostInfo is not enough. This new implementation simply assumes the clocks are in sync within a certain range and throws an error if clocks are found to be out of sync.	2020-11-18 15:06:49 -08:00
Zaid Abdulla	a9b0e3a1f7	Use a single MongoDB client	2020-11-18 00:01:41 -08:00
Zaid Abdulla	bd55d6f512	Better support for LIKE op in query minimization	2020-11-16 15:23:38 -08:00
Zaid Abdulla	e64001cddc	Accept primitive expressions in filter component	2020-11-16 15:20:48 -08:00