Add more upload file types

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.

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 {
@@ -97,10 +94,13 @@ if (!cluster.worker) {
 
   const _listener = (req, res): void => {
     if (stopping) res.setHeader("Connection", "close");
-    listener(req, res);
+    listener(req, res).catch((err) => {
+      throw err;
+    });
   };
 
-  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);
     })

lib/api-functions.ts

@@ -37,7 +37,7 @@ import { evaluate } from "./common/expression";
 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");
@@ -156,8 +156,7 @@ export async function connectionRequest(
     );
   }
 
-  try {
+  const status = await httpConnectionRequest(
-    await httpConnectionRequest(
     connectionRequestUrl,
     authExp,
     CONNECTION_REQUEST_ALLOW_BASIC_AUTH,
@@ -165,34 +164,46 @@ 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 {
@@ -322,5 +333,6 @@ export async function deleteDevice(deviceId: string): Promise<void> {
       },
     }),
     cache.del(`${deviceId}_tasks_faults_operations`),
+    db.deleteDeviceUploads(deviceId),
   ]);
 }

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/connection-request.ts

@@ -62,7 +62,6 @@ async function extractAuth(
 
 function httpGet(
   options: http.RequestOptions,
-  timeout,
   _debug: boolean,
   deviceId: string
 ): Promise<{ statusCode: number; headers: http.IncomingHttpHeaders }> {
@@ -77,15 +76,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 +93,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 +112,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 +139,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 +174,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(

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
+      );
+    }
   }
 }
 
@@ -699,6 +705,11 @@ async function nextRpc(sessionContext: SessionContext): Promise<void> {
         ["download", task.fileType, task.fileName, task.targetFileName || ""],
       ]);
       break;
+    case "upload":
+      session.addProvisions(sessionContext, `task_${task._id}`, [
+        ["upload", task.fileType, task.fileName],
+      ]);
+      break;
     case "addObject":
       alias = (task.parameterValues || [])
         .map((p) => `${p[0]}:${JSON.stringify(p[1])}`)
@@ -726,9 +737,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 +813,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(
@@ -824,6 +855,21 @@ async function sendAcsRequest(
     }
   }
 
+  if (acsRequest.name === "Upload") {
+    let prefix = "" + config.get("FS_URL_PREFIX");
+
+    if (!prefix) {
+      const FS_PORT = +config.get("FS_PORT");
+      const ssl = !!config.get("FS_SSL_CERT");
+      const origin = getRequestOrigin(sessionContext.httpRequest);
+      let hostname = origin.localAddress;
+      if (origin.host) [hostname] = origin.host.split(":", 1);
+      prefix = (ssl ? "https" : "http") + `://${hostname}:${FS_PORT}/`;
+    }
+
+    acsRequest.url = prefix + encodeURI(acsRequest.fileName);
+  }
+
   const rpc = {
     id: id,
     acsRequest: acsRequest,
@@ -878,15 +924,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 +946,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 +1146,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 +1205,7 @@ async function processRequest(
         sessionContext.authState = 1;
         return responseUnauthorized(sessionContext, false);
       } else {
+        await endSession(sessionContext);
         return responseUnauthorized(sessionContext, true);
       }
     }
@@ -1343,6 +1429,7 @@ async function listenerAsync(
         );
       }
       httpResponse.end(_body);
+      await endSession(sessionContext);
       return;
     }
     if (newConnection && sessionContext.authState !== 1)
@@ -1393,6 +1480,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 +1537,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, GridFSBucket } from "mongodb";
 import { get } from "./config";
 import { encodeTag, escapeRegExp } from "./common";
 import { parse } from "./common/expression-parser";
@@ -39,13 +39,14 @@ export let tasksCollection: Collection,
   virtualParametersCollection: Collection,
   faultsCollection: Collection,
   filesCollection: Collection,
+  uploadsCollection: Collection,
   operationsCollection: Collection,
   permissionsCollection: Collection,
   usersCollection: Collection,
-  configCollection: Collection;
+  configCollection: Collection,
+  uploadsBucket: GridFSBucket;
 
 let clientPromise: Promise<MongoClient>;
-export let client: MongoClient;
 
 function compareAccessLists(list1: string[], list2: string[]): boolean {
   if (list1.length !== list2.length) return false;
@@ -53,15 +54,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 });
 
@@ -69,6 +79,7 @@ export async function connect(): Promise<void> {
   presetsCollection = db.collection("presets");
   objectsCollection = db.collection("objects");
   filesCollection = db.collection("fs.files");
+  uploadsCollection = db.collection("uploads.files");
   provisionsCollection = db.collection("provisions");
   virtualParametersCollection = db.collection("virtualParameters");
   faultsCollection = db.collection("faults");
@@ -76,7 +87,8 @@ export async function connect(): Promise<void> {
   permissionsCollection = db.collection("permissions");
   usersCollection = db.collection("users");
   configCollection = db.collection("config");
-}
+  uploadsBucket = new GridFSBucket(db, { bucketName: "uploads" });
+});
 
 export async function disconnect(): Promise<void> {
   if (clientPromise) await (await clientPromise).close();
@@ -431,6 +443,17 @@ export async function saveDevice(
 
         break;
       default:
+        if (
+          diff[0].segments[0] === "Uploads" &&
+          diff[0].segments[2] === "LastFileName" &&
+          value1 &&
+          value1 !== value2
+        ) {
+          uploadsBucket.delete(value1 as any, () => {
+            // Ignore error due to mising files
+          });
+        }
+
         if (!diff[2]) {
           update["$unset"][path.toString()] = 1;
           continue;
@@ -780,3 +803,14 @@ interface User {
 export async function getUsers(): Promise<User[]> {
   return usersCollection.find().toArray();
 }
+
+export async function deleteDeviceUploads(deviceId: string): Promise<void> {
+  const files = await uploadsCollection
+    .find({
+      _id: {
+        $regex: `^${escapeRegExp(deviceId)}\\/`,
+      },
+    })
+    .toArray();
+  await Promise.all(files.map((f) => uploadsBucket.delete(f["_id"])));
+}

lib/default-provisions.ts

@@ -229,6 +229,44 @@ export function download(
   return true;
 }
 
+export function upload(
+  sessionContext: SessionContext,
+  provision: (string | number | boolean)[],
+  declarations: Declaration[]
+): boolean {
+  if (
+    provision.length !== 3 ||
+    typeof provision[1] !== "string" ||
+    typeof provision[2] !== "string"
+  )
+    throw new Error("Invalid arguments");
+
+  const alias = [
+    `FileType:${JSON.stringify(provision[1] || "")}`,
+    `FileName:${JSON.stringify(provision[2] || "")}`,
+  ].join(",");
+
+  declarations.push({
+    path: Path.parse(`Uploads.[${alias}]`),
+    pathGet: 1,
+    pathSet: 1,
+    attrGet: null,
+    attrSet: null,
+    defer: true,
+  });
+
+  declarations.push({
+    path: Path.parse(`Uploads.[${alias}].Upload`),
+    pathGet: 1,
+    pathSet: null,
+    attrGet: { value: 1 },
+    attrSet: { value: [sessionContext.timestamp] },
+    defer: true,
+  });
+
+  return true;
+}
+
 export function instances(
   sessionContext: SessionContext,
   provision: (string | number | boolean)[],

lib/fs.ts

@@ -20,15 +20,42 @@
 import * as url from "url";
 import * as querystring from "querystring";
 import { IncomingMessage, ServerResponse } from "http";
-import { GridFSBucket } from "mongodb";
+import { Collection, GridFSBucket, ObjectId } from "mongodb";
-import * as db from "./db";
+import { onConnect, getOperations } from "./db";
 import * as logger from "./logger";
 import { getRequestOrigin } from "./forwarded";
 
-export function listener(
+let filesCollection: Collection;
+let filesBucket: GridFSBucket;
+let uploadsBucket: GridFSBucket;
+
+onConnect(async (db) => {
+  filesCollection = db.collection("fs.files");
+  filesBucket = new GridFSBucket(db);
+  uploadsBucket = new GridFSBucket(db, { bucketName: "uploads" });
+});
+
+async function canUpload(
+  deviceId: string,
+  fileName: string,
+  timeout = Date.now() + 5000
+): Promise<boolean> {
+  const operations = Object.values(await getOperations(deviceId));
+  for (const operation of operations) {
+    if (operation.name === "Upload" && operation.args.fileName === fileName)
+      return true;
+  }
+
+  if (Date.now() >= timeout) return false;
+  // Need to wait and retry in case upload was initiated before session was closed
+  await new Promise((resolve) => setTimeout(resolve, 1000));
+  return canUpload(deviceId, fileName, timeout);
+}
+
+export async function listener(
   request: IncomingMessage,
   response: ServerResponse
-): void {
+): Promise<void> {
   const urlParts = url.parse(request.url, true);
   if (request.method === "GET") {
     const filename = querystring.unescape(urlParts.pathname.substring(1));
@@ -39,9 +66,7 @@ export function listener(
       remoteAddress: getRequestOrigin(request).remoteAddress,
     };
 
-    db.filesCollection.findOne({ _id: filename }, (err, file) => {
+    const file = await filesCollection.findOne({ _id: filename });
-      if (err) throw err;
-
     if (!file) {
       response.writeHead(404);
       response.end();
@@ -55,12 +80,44 @@ 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);
+  } else if (request.method === "PUT") {
+    const [, deviceId, ...filePath] = urlParts.pathname
+      .split("/")
+      .map(querystring.unescape);
+
+    const fileName = `${deviceId}/${filePath.join("/")}`;
+
+    if (!(await canUpload(deviceId, fileName))) {
+      response.writeHead(403);
+      response.end("403 Forbidden");
+      return;
+    }
+
+    try {
+      await uploadsBucket.delete((fileName as unknown) as ObjectId);
+    } catch (err) {
+      // File not found, do nothing
+    }
+    const uploadStream = uploadsBucket.openUploadStreamWithId(
+      fileName,
+      fileName
+    );
+
+    uploadStream.on("finish", () => {
+      response.writeHead(200);
+      response.end();
     });
+
+    uploadStream.on("error", (err) => {
+      response.writeHead(500);
+      response.end(err.message);
+    });
+
+    request.pipe(uploadStream);
   } else {
     response.writeHead(405, { Allow: "GET" });
     response.end("405 Method Not Allowed");

lib/init.ts

@@ -108,6 +108,7 @@ export async function seed(options: Record<string, boolean>): Promise<void> {
       { role: "admin", resource: "devices", access: 3, validate: "true" },
       { role: "admin", resource: "faults", access: 3, validate: "true" },
       { role: "admin", resource: "files", access: 3, validate: "true" },
+      { role: "admin", resource: "uploads", access: 3, validate: "true" },
       { role: "admin", resource: "presets", access: 3, validate: "true" },
       { role: "admin", resource: "provisions", access: 3, validate: "true" },
       { role: "admin", resource: "config", access: 3, validate: "true" },

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/mongodb-functions.ts

@@ -640,6 +640,14 @@ export function flattenFile(
   return f;
 }
 
+export function flattenUpload(
+  file: Record<string, unknown>
+): Record<string, unknown> {
+  const f = {};
+  f["_id"] = file["_id"];
+  return f;
+}
+
 export function preProcessPreset(
   data: Record<string, unknown>
 ): Record<string, unknown> {

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/session.ts

@@ -53,6 +53,7 @@ import {
   AddObject,
   DeleteObject,
   Download,
+  Upload,
   Reboot,
   FactoryReset,
   AddObjectResponse,
@@ -292,14 +293,15 @@ export async function transferComplete(
     };
   }
 
-  const instance = operation.args.instance;
-
   delete sessionContext.operations[commandKey];
   if (!sessionContext.operationsTouched) sessionContext.operationsTouched = {};
   sessionContext.operationsTouched[commandKey] = 1;
 
   if (rpcReq.faultStruct && rpcReq.faultStruct.faultCode !== "0") {
+    if (operation.name === "Download")
       revertDownloadParameters(sessionContext, operation.args.instance);
+    else if (operation.name === "Upload")
+      revertUploadParameters(sessionContext, operation.args.instance);
 
     const fault: Fault = {
       code: `cwmp.${rpcReq.faultStruct.faultCode}`,
@@ -318,6 +320,44 @@ export async function transferComplete(
   let toClear = null;
   const timestamp = sessionContext.timestamp + sessionContext.iteration + 1;
 
+  if (operation.name === "Download") {
+    toClear = commitDownloadOperation(
+      sessionContext,
+      operation,
+      rpcReq,
+      timestamp,
+      toClear
+    );
+  } else if (operation.name === "Upload") {
+    toClear = commitUploadOperation(
+      sessionContext,
+      operation,
+      rpcReq,
+      timestamp,
+      toClear
+    );
+  }
+
+  if (toClear) {
+    for (const c of toClear)
+      device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
+  }
+
+  return {
+    acsResponse: { name: "TransferCompleteResponse" },
+    operation: operation,
+    fault: null,
+  };
+}
+
+function commitDownloadOperation(
+  sessionContext: SessionContext,
+  operation: Operation,
+  rpcReq: TransferCompleteRequest,
+  timestamp: number,
+  toClear: Clear[]
+): Clear[] {
+  const instance = operation.args.instance;
   let p;
 
   p = sessionContext.deviceData.paths.add(
@@ -386,16 +426,110 @@ export async function transferComplete(
     toClear
   );
 
+  return toClear;
+}
+
+function commitUploadOperation(
+  sessionContext: SessionContext,
+  operation: Operation,
+  rpcReq: TransferCompleteRequest,
+  timestamp: number,
+  toClear: Clear[]
+): Clear[] {
+  const instance = operation.args.instance;
+
+  let p;
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.LastUpload`)
+  );
+  toClear = device.set(
+    sessionContext.deviceData,
+    p,
+    timestamp,
+    { value: [timestamp, [operation.timestamp, "xsd:dateTime"]] },
+    toClear
+  );
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.LastFileType`)
+  );
+  toClear = device.set(
+    sessionContext.deviceData,
+    p,
+    timestamp,
+    { value: [timestamp, [operation.args.fileType, "xsd:string"]] },
+    toClear
+  );
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.LastFileName`)
+  );
+  toClear = device.set(
+    sessionContext.deviceData,
+    p,
+    timestamp,
+    { value: [timestamp, [operation.args.fileName, "xsd:string"]] },
+    toClear
+  );
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.StartTime`)
+  );
+  toClear = device.set(
+    sessionContext.deviceData,
+    p,
+    timestamp,
+    { value: [timestamp, [+rpcReq.startTime, "xsd:dateTime"]] },
+    toClear
+  );
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.CompleteTime`)
+  );
+  toClear = device.set(
+    sessionContext.deviceData,
+    p,
+    timestamp,
+    { value: [timestamp, [+rpcReq.completeTime, "xsd:dateTime"]] },
+    toClear
+  );
+
+  return toClear;
+}
+
+function revertUploadParameters(
+  sessionContext: SessionContext,
+  instance
+): void {
+  const timestamp = sessionContext.timestamp + sessionContext.iteration + 1;
+
+  let p;
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.LastUpload`)
+  );
+
+  const lastUpload = sessionContext.deviceData.attributes.get(p);
+
+  p = sessionContext.deviceData.paths.add(
+    Path.parse(`Uploads.${instance}.Upload`)
+  );
+
+  const toClear = device.set(sessionContext.deviceData, p, timestamp, {
+    value: [
+      timestamp,
+      [
+        lastUpload && lastUpload.value[1] ? lastUpload.value[1][0] : 0,
+        "xsd:dateTime",
+      ],
+    ],
+  });
+
   if (toClear) {
     for (const c of toClear)
       device.clear(sessionContext.deviceData, c[0], c[1], c[2], c[3]);
   }
-
-  return {
-    acsResponse: { name: "TransferCompleteResponse" },
-    operation: operation,
-    fault: null,
-  };
 }
 
 function revertDownloadParameters(
@@ -445,9 +579,7 @@ export async function timeoutOperations(
   for (const [commandKey, operation] of Object.entries(
     sessionContext.operations
   )) {
-    if (operation.name !== "Download")
+    if (operation.name === "Download") {
-      throw new Error(`Unknown operation name ${operation.name}`);
-
       const DOWNLOAD_TIMEOUT =
         +localCache.getConfig(
           sessionContext.cacheSnapshot,
@@ -473,6 +605,37 @@ export async function timeoutOperations(
 
         revertDownloadParameters(sessionContext, operation.args.instance);
       }
+    } else if (operation.name === "Upload") {
+      const UPLOAD_TIMEOUT =
+        +(
+          localCache.getConfig(
+            sessionContext.cacheSnapshot,
+            "cwmp.uploadTimeout",
+            {},
+            sessionContext.timestamp,
+            (e) => configContextCallback(sessionContext, e)
+          ) || 3600
+        ) * 1000;
+
+      if (sessionContext.timestamp > operation.timestamp + UPLOAD_TIMEOUT) {
+        delete sessionContext.operations[commandKey];
+        if (!sessionContext.operationsTouched)
+          sessionContext.operationsTouched = {};
+        sessionContext.operationsTouched[commandKey] = 1;
+
+        faults.push({
+          code: "timeout",
+          message: "Upload operation timed out",
+          timestamp: operation.timestamp,
+        });
+
+        operations.push(operation);
+
+        revertUploadParameters(sessionContext, operation.args.instance);
+      }
+    } else {
+      throw new Error(`Unknown operation name ${operation.name}`);
+    }
   }
 
   return { faults, operations };
@@ -793,6 +956,10 @@ function runDeclarations(
       downloadsToCreate: new InstanceSet(),
       downloadsValues: new Map(),
       downloadsDownload: new Map(),
+      uploadsToDelete: new Set(),
+      uploadsToCreate: new InstanceSet(),
+      uploadsValues: new Map(),
+      uploadsUpload: new Map(),
       reboot: 0,
       factoryReset: 0,
     };
@@ -1304,7 +1471,7 @@ export async function rpcRequest(
       }
 
       // Downloads
-      let index;
+      let index: number;
       for (const instance of sessionContext.syncState.downloadsToCreate) {
         if (index == null) {
           index = 0;
@@ -1426,6 +1593,9 @@ export async function rpcRequest(
         }
       }
 
+      // Uploads
+      toClear = syncUploads(sessionContext, timestamp, toClear);
+
       if (toClear || sessionContext.deviceData.changes.has("prerequisite")) {
         if (toClear) {
           for (const c of toClear)
@@ -1503,6 +1673,129 @@ export async function rpcRequest(
   return rpcRequest(sessionContext, null);
 }
 
+function syncUploads(
+  sessionContext: SessionContext,
+  timestamp: number,
+  toClear: Clear[]
+): Clear[] {
+  let index: number;
+  for (const instance of sessionContext.syncState.uploadsToCreate) {
+    if (index == null) {
+      index = 0;
+      for (const p of sessionContext.deviceData.paths.find(
+        Path.parse("Uploads.*"),
+        false,
+        true
+      )) {
+        if (
+          +p.segments[1] > index &&
+          sessionContext.deviceData.attributes.has(p)
+        )
+          index = +p.segments[1];
+      }
+    }
+
+    ++index;
+
+    toClear = device.set(
+      sessionContext.deviceData,
+      Path.parse("Uploads"),
+      timestamp,
+      { object: [timestamp, 1], writable: [timestamp, 1] },
+      toClear
+    );
+
+    toClear = device.set(
+      sessionContext.deviceData,
+      Path.parse(`Uploads.${index}`),
+      timestamp,
+      { object: [timestamp, 1], writable: [timestamp, 1] },
+      toClear
+    );
+
+    const params = {
+      FileType: {
+        writable: 1,
+        value: [instance.FileType || "", "xsd:string"],
+      },
+      FileName: {
+        writable: 1,
+        value: [instance.FileName || "", "xsd:string"],
+      },
+      Upload: {
+        writable: 1,
+        value: [instance.Upload || 0, "xsd:dateTime"],
+      },
+      LastFileType: { writable: 0, value: ["", "xsd:string"] },
+      LastFileName: { writable: 0, value: ["", "xsd:string"] },
+      LastTargetFileName: { writable: 0, value: ["", "xsd:string"] },
+      LastUpload: { writable: 0, value: [0, "xsd:dateTime"] },
+      StartTime: { writable: 0, value: [0, "xsd:dateTime"] },
+      CompleteTime: { writable: 0, value: [0, "xsd:dateTime"] },
+    };
+
+    for (const [k, v] of Object.entries(params)) {
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${index}.${k}`),
+        timestamp,
+        {
+          object: [timestamp, 0],
+          writable: [timestamp, v.writable as 0 | 1],
+          value: [timestamp, v.value as [string | number | boolean, string]],
+        },
+        toClear
+      );
+    }
+
+    toClear = device.set(
+      sessionContext.deviceData,
+      Path.parse(`Uploads.${index}.*`),
+      timestamp,
+      null,
+      toClear
+    );
+  }
+
+  sessionContext.syncState.uploadsToCreate.clear();
+
+  for (const instance of sessionContext.syncState.uploadsToDelete) {
+    toClear = device.set(
+      sessionContext.deviceData,
+      instance,
+      timestamp,
+      null,
+      toClear
+    );
+    for (const p of sessionContext.syncState.uploadsValues.keys()) {
+      if (p.segments[1] === instance.segments[1])
+        sessionContext.syncState.uploadsValues.delete(p);
+    }
+  }
+
+  sessionContext.syncState.uploadsToDelete.clear();
+
+  for (const [p, v] of sessionContext.syncState.uploadsValues) {
+    const attrs = sessionContext.deviceData.attributes.get(p);
+    if (attrs) {
+      if (attrs.writable && attrs.writable[1] && attrs.value) {
+        const val = device.sanitizeParameterValue([v, attrs.value[1][1]]);
+        if (val[0] !== attrs.value[1][0]) {
+          toClear = device.set(
+            sessionContext.deviceData,
+            p,
+            timestamp,
+            { value: [timestamp, val] },
+            toClear
+          );
+        }
+      }
+    }
+  }
+
+  return toClear;
+}
+
 function generateGetRpcRequest(
   sessionContext: SessionContext
 ): GetParameterNames | GetParameterValues | GetParameterAttributes {
@@ -1728,6 +2021,7 @@ function generateSetRpcRequest(
   | FactoryReset
   | Reboot
   | Download
+  | Upload
 ) & { next?: string } {
   const syncState = sessionContext.syncState;
   if (!syncState) return null;
@@ -1906,6 +2200,36 @@ function generateSetRpcRequest(
     }
   }
 
+  // Uploads
+  for (const [p, t] of syncState.uploadsUpload) {
+    const attrs = deviceData.attributes.get(p);
+    if (!(attrs && attrs.value && t <= attrs.value[1][0])) {
+      const fileTypeAttrs = deviceData.attributes.get(
+        deviceData.paths.get(p.slice(0, -1).concat(Path.parse("FileType")))
+      );
+      const fileNameAttrs = deviceData.attributes.get(
+        deviceData.paths.get(p.slice(0, -1).concat(Path.parse("FileName")))
+      );
+
+      if (fileNameAttrs && fileNameAttrs.value) {
+        return {
+          name: "Upload",
+          commandKey: generateRpcId(sessionContext),
+          instance: p.segments[1] as string,
+          fileType: fileTypeAttrs
+            ? fileTypeAttrs.value
+              ? (fileTypeAttrs.value[1][0] as string)
+              : null
+            : null,
+          fileName:
+            sessionContext.deviceId +
+            "/" +
+            (fileNameAttrs.value[1][0] as string),
+        };
+      }
+    }
+  }
+
   // Reboot
   if (syncState.reboot) {
     const p = sessionContext.deviceData.paths.get(Path.parse("Reboot"));
@@ -2187,6 +2511,26 @@ function processDeclarations(
           }
         }
         break;
+      case "Uploads":
+        if (
+          currentPath.length === 3 &&
+          currentPath.wildcard === 0 &&
+          declareAttributeValues &&
+          declareAttributeValues.value
+        ) {
+          if (currentPath.segments[2] === "Upload") {
+            syncState.uploadsUpload.set(
+              currentPath,
+              declareAttributeValues.value[0]
+            );
+          } else {
+            syncState.uploadsValues.set(
+              currentPath,
+              declareAttributeValues.value[0]
+            );
+          }
+        }
+        break;
       case "VirtualParameters":
         if (currentPath.length <= 2) {
           let d;
@@ -2357,6 +2701,10 @@ function processInstances(
     if (parent.length !== 1) return;
     instancesToDelete = sessionContext.syncState.downloadsToDelete;
     instancesToCreate = sessionContext.syncState.downloadsToCreate;
+  } else if (parent.segments[0] === "Uploads") {
+    if (parent.length !== 1) return;
+    instancesToDelete = sessionContext.syncState.uploadsToDelete;
+    instancesToCreate = sessionContext.syncState.uploadsToCreate;
   } else {
     instancesToDelete = sessionContext.syncState.instancesToDelete.get(parent);
     if (instancesToDelete == null) {
@@ -2578,6 +2926,7 @@ export async function rpcResponse(
         "FactoryReset",
         "VirtualParameters",
         "Downloads",
+        "Uploads",
       ]) {
         const p = sessionContext.deviceData.paths.get(Path.parse(n));
         if (p && sessionContext.deviceData.attributes.has(p))
@@ -2788,6 +3137,84 @@ export async function rpcResponse(
           operation.retries[channel] = sessionContext.retries[channel];
       }
 
+      sessionContext.operations[rpcReq.commandKey] = operation;
+      if (!sessionContext.operationsTouched)
+        sessionContext.operationsTouched = {};
+      sessionContext.operationsTouched[rpcReq.commandKey] = 1;
+    }
+  } else if (rpcRes.name === "UploadResponse") {
+    if (rpcReq.name !== "Upload")
+      throw new Error("Response name does not match request name");
+
+    toClear = device.set(
+      sessionContext.deviceData,
+      Path.parse(`Uploads.${rpcReq.instance}.Upload`),
+      timestamp + 1,
+      { value: [timestamp + 1, [sessionContext.timestamp, "xsd:dateTime"]] },
+      toClear
+    );
+
+    if (rpcRes.status === 0) {
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${rpcReq.instance}.LastUpload`),
+        timestamp + 1,
+        {
+          value: [timestamp + 1, [sessionContext.timestamp, "xsd:dateTime"]],
+        },
+        toClear
+      );
+
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${rpcReq.instance}.LastFileType`),
+        timestamp + 1,
+        { value: [timestamp + 1, [rpcReq.fileType, "xsd:string"]] },
+        toClear
+      );
+
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${rpcReq.instance}.LastFileName`),
+        timestamp + 1,
+        { value: [timestamp + 1, [rpcReq.fileType, "xsd:string"]] },
+        toClear
+      );
+
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${rpcReq.instance}.StartTime`),
+        timestamp + 1,
+        { value: [timestamp + 1, [+rpcRes.startTime, "xsd:dateTime"]] },
+        toClear
+      );
+
+      toClear = device.set(
+        sessionContext.deviceData,
+        Path.parse(`Uploads.${rpcReq.instance}.CompleteTime`),
+        timestamp + 1,
+        { value: [timestamp + 1, [+rpcRes.completeTime, "xsd:dateTime"]] },
+        toClear
+      );
+    } else {
+      const operation = {
+        name: "Upload",
+        timestamp: sessionContext.timestamp,
+        provisions: sessionContext.provisions,
+        channels: sessionContext.channels,
+        retries: {},
+        args: {
+          instance: rpcReq.instance,
+          fileType: rpcReq.fileType,
+          fileName: rpcReq.fileName,
+        },
+      };
+
+      for (const channel of Object.keys(sessionContext.channels)) {
+        if (sessionContext.retries[channel] != null)
+          operation.retries[channel] = sessionContext.retries[channel];
+      }
+
       sessionContext.operations[rpcReq.commandKey] = operation;
       if (!sessionContext.operationsTouched)
         sessionContext.operationsTouched = {};

lib/soap.ts

@@ -45,6 +45,7 @@ import {
   RebootResponse,
   FactoryResetResponse,
   DownloadResponse,
+  UploadResponse,
   GetRPCMethodsRequest,
   RequestDownloadRequest,
   AcsResponse,
@@ -496,6 +497,24 @@ function Download(methodRequest): string {
   )}</FailureURL></cwmp:Download>`;
 }
 
+function Upload(methodRequest): string {
+  return `<cwmp:Upload><CommandKey>${
+    methodRequest.commandKey || ""
+  }</CommandKey><FileType>${methodRequest.fileType}</FileType><URL>${
+    methodRequest.url
+  }</URL><Username>${encodeEntities(
+    methodRequest.username || ""
+  )}</Username><Password>${encodeEntities(
+    methodRequest.password || ""
+  )}</Password><DelaySeconds>${
+    methodRequest.delaySeconds || "0"
+  }</DelaySeconds><SuccessURL>${encodeEntities(
+    methodRequest.successUrl || ""
+  )}</SuccessURL><FailureURL>${encodeEntities(
+    methodRequest.failureUrl || ""
+  )}</FailureURL></cwmp:Upload>`;
+}
+
 function DownloadResponse(xml: Element): DownloadResponse {
   let status: number, startTime: number, completeTime: number;
   for (const c of xml.children) {
@@ -544,6 +563,35 @@ function DownloadResponse(xml: Element): DownloadResponse {
   };
 }
 
+function UploadResponse(xml: Element): UploadResponse {
+  let status: number, startTime: number, completeTime: number;
+  for (const c of xml.children) {
+    switch (c.localName) {
+      case "Status":
+        status = parseInt(c.text);
+        break;
+      case "StartTime":
+        startTime = Date.parse(c.text);
+        break;
+      case "CompleteTime":
+        completeTime = Date.parse(c.text);
+        break;
+    }
+  }
+
+  if (!(status >= 0)) {
+    warnings.push({ message: "Invalid UploadResponse status" });
+    status = 0;
+  }
+
+  return {
+    name: "UploadResponse",
+    status: status,
+    startTime: startTime,
+    completeTime: completeTime,
+  };
+}
+
 function Inform(xml: Element): InformRequest {
   let retryCount: number, evnt: string[];
   let parameterList: [Path, string | number | boolean, string][];
@@ -927,6 +975,9 @@ export function request(
     case "DownloadResponse":
       rpc.cpeResponse = DownloadResponse(methodElement);
       break;
+    case "UploadResponse":
+      rpc.cpeResponse = UploadResponse(methodElement);
+      break;
     case "Fault":
       rpc.cpeFault = fault(methodElement);
       break;
@@ -1022,6 +1073,9 @@ export function response(rpc: {
       case "Download":
         body = Download(rpc.acsRequest);
         break;
+      case "Upload":
+        body = Upload(rpc.acsRequest);
+        break;
       default:
         throw new Error(
           `Unknown method request ${(rpc.acsRequest as AcsRequest).name}`

lib/types.ts

@@ -121,6 +121,10 @@ export interface SyncState {
   downloadsToCreate: InstanceSet;
   downloadsValues: Map<Path, string | number>;
   downloadsDownload: Map<Path, number>;
+  uploadsToDelete: Set<Path>;
+  uploadsToCreate: InstanceSet;
+  uploadsValues: Map<Path, string | number>;
+  uploadsUpload: Map<Path, number>;
   reboot: number;
   factoryReset: number;
 }
@@ -163,6 +167,7 @@ export interface SessionContext {
   cacheUntil?: number;
   syncState?: SyncState;
   lastActivity?: number;
+  extendLock?: number;
   rpcRequest?: AcsRequest;
   operationsTouched?: { [commandKey: string]: 1 | 0 };
   provisionsRet?: any[];
@@ -192,7 +197,7 @@ export interface Operation {
     instance: string;
     fileType: string;
     fileName: string;
-    targetFileName: string;
+    targetFileName?: string;
   };
 }
 
@@ -206,7 +211,8 @@ export type AcsRequest =
   | DeleteObject
   | FactoryReset
   | Reboot
-  | Download;
+  | Download
+  | Upload;
 
 export interface GetParameterNames {
   name: "GetParameterNames";
@@ -276,6 +282,20 @@ export interface Download {
   failureUrl?: string;
 }
 
+export interface Upload {
+  name: "Upload";
+  commandKey: string;
+  instance: string;
+  fileType: string;
+  fileName?: string;
+  url?: string;
+  username?: string;
+  password?: string;
+  delaySecods?: number;
+  successUrl?: string;
+  failureUrl?: string;
+}
+
 export interface SpvFault {
   parameterName: string;
   faultCode: string;
@@ -304,7 +324,8 @@ export type CpeResponse =
   | DeleteObjectResponse
   | RebootResponse
   | FactoryResetResponse
-  | DownloadResponse;
+  | DownloadResponse
+  | UploadResponse;
 
 export interface GetParameterNamesResponse {
   name: "GetParameterNamesResponse";
@@ -356,6 +377,13 @@ export interface DownloadResponse {
   completeTime?: number;
 }
 
+export interface UploadResponse {
+  name: "UploadResponse";
+  status: number;
+  startTime?: number;
+  completeTime?: number;
+}
+
 export type CpeRequest =
   | InformRequest
   | TransferCompleteRequest

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";
@@ -57,6 +58,10 @@ export async function deleteResource(
       await db.deleteFile(id);
       break;
 
+    case "uploads":
+      await db.deleteUpload(id);
+      break;
+
     case "faults":
       await deleteFault(id);
       break;
@@ -104,23 +109,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

@@ -49,6 +49,7 @@ const RESOURCE_IDS = {
   presets: "_id",
   provisions: "_id",
   files: "_id",
+  uploads: "_id",
   virtualParameters: "_id",
   config: "_id",
   permissions: "_id",
@@ -62,6 +63,7 @@ const resources = {
   presets: 0 | RESOURCE_DELETE | RESOURCE_PUT,
   provisions: 0 | RESOURCE_DELETE | RESOURCE_PUT,
   files: 0 | RESOURCE_DELETE,
+  uploads: 0 | RESOURCE_DELETE,
   virtualParameters: 0 | RESOURCE_DELETE | RESOURCE_PUT,
   config: 0 | RESOURCE_DELETE | RESOURCE_PUT,
   permissions: 0 | RESOURCE_DELETE | RESOURCE_PUT,
@@ -361,7 +363,7 @@ for (const [resource, flags] of Object.entries(resources)) {
   });
 
   if (flags & RESOURCE_DELETE) {
-    router.delete(`/${resource}/:id`, async (ctx) => {
+    router.delete(`/${resource}/:id(.+)`, async (ctx) => {
       const authorizer: Authorizer = ctx.state.authorizer;
       const log = {
         message: `Delete ${resource}`,
@@ -478,6 +480,41 @@ router.put("/files/:id", async (ctx) => {
   ctx.body = "";
 });
 
+router.get("/uploads/blob/:id(.+)", async (ctx) => {
+  const authorizer: Authorizer = ctx.state.authorizer;
+  const resource = "uploads";
+  const id = ctx.params.id;
+
+  const log = {
+    message: `Download ${resource} blob`,
+    context: ctx,
+    id: id,
+  };
+  1;
+
+  const filter = and(authorizer.getFilter(resource, 2), [
+    "=",
+    ["PARAM", RESOURCE_IDS[resource]],
+    id,
+  ]);
+
+  if (!authorizer.hasAccess(resource, 2)) {
+    logUnauthorizedWarning(log);
+    return void (ctx.status = 403);
+  }
+
+  const res = await db.query(resource, filter);
+  if (!res[0]) return void (ctx.status = 404);
+
+  ctx.type = "application/octet-stream";
+  ctx.attachment(id.replace(/\//g, "_"));
+  ctx.body = new stream.PassThrough();
+
+  const downloadStream = await db.getUploadBlob(id);
+  downloadStream.pipe(ctx.body);
+  logger.accessInfo(log);
+});
+
 router.post("/devices/:id/tasks", async (ctx) => {
   const authorizer: Authorizer = ctx.state.authorizer;
   const log = {
@@ -531,6 +568,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 +580,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,25 @@
  * 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",
+  uploads: "uploads.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 +63,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;
@@ -177,13 +113,14 @@ export function query(
           docs = docs.map((d) => mongodbFunctions.flattenPreset(d));
         else if (resource === "files")
           docs = docs.map((d) => mongodbFunctions.flattenFile(d));
+        else if (resource === "uploads")
+          docs = docs.map((d) => mongodbFunctions.flattenUpload(d));
         return resolve(docs);
       });
     } 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")
@@ -192,6 +129,8 @@ export function query(
             doc = mongodbFunctions.flattenPreset(doc);
           else if (resource === "files")
             doc = mongodbFunctions.flattenFile(doc);
+          else if (resource === "uploads")
+            doc = mongodbFunctions.flattenUpload(doc);
           callback(doc);
         },
         (err) => {
@@ -200,8 +139,6 @@ export function query(
         }
       );
     }
-      })
-      .catch(reject);
   });
 }
 
@@ -218,22 +155,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 +181,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 +192,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 +205,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 +286,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 +297,6 @@ export function putUser(
         resolve();
       }
     );
-      })
-      .catch(reject);
   });
 }
 
@@ -396,9 +310,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,25 +318,29 @@ 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);
   });
 }
 
+export async function deleteUpload(filename: string): Promise<void> {
+  const bucket = new GridFSBucket(db, { bucketName: "uploads" });
+  await bucket.delete(filename as any);
+}
+
+export async function getUploadBlob(filename: string): Promise<Readable> {
+  const bucket = new GridFSBucket(db, { bucketName: "uploads" });
+  return bucket.openDownloadStreamByName(filename);
+}
+
 export function deleteFault(id: string): Promise<void> {
   return deleteResource("faults", id);
 }
@@ -432,7 +348,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

ui/components.ts

@@ -39,6 +39,7 @@ import tags from "./components/tags";
 import ping from "./components/ping";
 import deviceLink from "./components/device-link";
 import longTextComponent from "./long-text-component";
+import deviceUploads from "./components/device-uploads";
 
 const comps = {
   parameter,
@@ -54,6 +55,7 @@ const comps = {
   ping,
   "device-link": deviceLink,
   "long-text": longTextComponent,
+  "device-uploads": deviceUploads,
 };
 
 const contextifiedComponents = new WeakMap<ComponentTypes, ComponentTypes>();

ui/components/device-actions.ts

@@ -22,6 +22,7 @@ import { m } from "../components";
 import * as taskQueue from "../task-queue";
 import * as notifications from "../notifications";
 import * as store from "../store";
+import * as overlay from "../overlay";
 
 const component: ClosureComponent = (): Component => {
   return {
@@ -78,6 +79,30 @@ const component: ClosureComponent = (): Component => {
         )
       );
 
+      buttons.push(
+        m(
+          "button.primary",
+          {
+            title: "Fetch files from device",
+            onclick: () => {
+              const deviceId = device["DeviceID.ID"].value[0];
+              overlay.open(() => {
+                const dev = store.fetch("devices", [
+                  "=",
+                  ["PARAM", "DeviceID.ID"],
+                  deviceId,
+                ]);
+                return [
+                  m("h3", `Uploads from ${deviceId}`),
+                  m("device-uploads", { device: dev.value[0] }),
+                ];
+              });
+            },
+          },
+          "Uploads"
+        )
+      );
+
       buttons.push(
         m(
           "button.primary",

ui/components/device-uploads.ts

@@ -0,0 +1,155 @@
+/**
+ * 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 { ClosureComponent, Component } from "mithril";
+import { m } from "../components";
+import * as store from "../store";
+import * as notifications from "../notifications";
+import { getIcon } from "../icons";
+import * as taskQueue from "../task-queue";
+
+const component: ClosureComponent = (): Component => {
+  return {
+    view: (vnode) => {
+      const device = vnode.attrs["device"];
+      const deviceId = device["DeviceID.ID"].value[0];
+      const files = store.fetch("uploads", [
+        "AND",
+        [">", ["PARAM", "_id"], `${deviceId}/`],
+        ["<", ["PARAM", "_id"], `${deviceId}/zzzz`],
+      ]);
+
+      const uploads: Record<string, any> = {};
+
+      for (const [k, v] of Object.entries(device)) {
+        if (k.startsWith("Uploads.")) {
+          const parts = k.split(".");
+          if (parts.length !== 3) continue;
+          uploads[parts[1]] = uploads[parts[1]] || {};
+          uploads[parts[1]][parts[2]] = v["value"][0];
+        }
+      }
+
+      const headers = ["Filename", "Type", "Timestamp", "Status"].map((l) =>
+        m("th", l)
+      );
+      const thead = m("thead", m("tr", headers));
+
+      let minTime = 999999;
+      const timestamp = store.getTimestamp();
+      const rows = [];
+      const uploadsSorted = Object.values(uploads).sort(
+        (a, b) => a["Upload"] - b["Upload"]
+      );
+
+      for (const u of Object.values(uploadsSorted)) {
+        if (!u["Upload"]) continue;
+        const ready = u["LastUpload"] >= u["Upload"];
+        if (!ready) {
+          minTime = Math.min(minTime, u["Upload"] - timestamp);
+          rows.push([
+            m("td", u["FileName"]),
+            m("td", u["FileType"]),
+            m("td", new Date(u["Upload"]).toLocaleString()),
+            m("td", "Waiting for upload"),
+          ]);
+        } else {
+          const filePath = `${deviceId}/${u["FileName"]}`;
+          if (!Array.isArray(files.value)) continue;
+          if (!files.value.some((f) => f["_id"] === filePath)) continue;
+
+          const deleteButton = m(
+            "button",
+            {
+              title: "Delete file",
+              onclick: (e) => {
+                e.redraw = false;
+                store
+                  .deleteResource("uploads", filePath)
+                  .then(() => {
+                    notifications.push("success", "File deleted");
+                    store.fulfill(Date.now(), Date.now());
+                    m.redraw();
+                  })
+                  .catch((err) => {
+                    notifications.push("error", err.message);
+                    store.fulfill(Date.now(), Date.now());
+                  });
+              },
+            },
+            getIcon("delete-instance")
+          );
+
+          rows.push([
+            m("td", u["FileName"]),
+            m("td", u["FileType"]),
+            m("td", new Date(u["LastUpload"]).toLocaleString()),
+            m(
+              "td",
+              m(
+                "a",
+                {
+                  href: `api/uploads/blob/${encodeURIComponent(filePath)}`,
+                },
+                "Ready"
+              )
+            ),
+            m("td", deleteButton),
+          ]);
+        }
+      }
+
+      if (minTime < 60000) {
+        setTimeout(() => {
+          store.fulfill(0, timestamp + 5000);
+        }, 10000);
+      }
+
+      const addButton = m(
+        "button",
+        {
+          title: "Fetch a new file",
+          onclick: () => {
+            taskQueue.stageUpload({
+              name: "upload",
+              devices: [deviceId],
+            });
+          },
+        },
+        getIcon("add-instance")
+      );
+
+      const empty = !rows.length;
+
+      rows.push([m("td", { colspan: headers.length }), m("td", addButton)]);
+
+      const tbody = m(
+        "tbody",
+        rows.map((r) => m("tr", r)),
+        empty
+          ? m("tr.empty", m("td", { colspan: headers.length }, "No uploads"))
+          : null
+      );
+
+      return m("table.table", thead, tbody);
+    },
+  };
+};
+
+export default component;

ui/css/app.css

@@ -106,7 +106,7 @@ body {
 
   & > .drawer-wrapper {
     position: fixed;
-    z-index: 10;
+    z-index: 101;
     width: 720px;
     margin-left: -360px;
     top: 0;

ui/drawer-component.ts

@@ -184,6 +184,122 @@ function renderStagingDownload(task: StageTask): Children {
   ];
 }
 
+const uploadTypesForDevice: WeakMap<
+  Record<string, unknown>,
+  Record<string, string>
+> = new WeakMap();
+function getUploadTypesForDevice(
+  device: Record<string, unknown>
+): Record<string, string> {
+  if (uploadTypesForDevice.has(device)) return uploadTypesForDevice.get(device);
+  const instances: Set<string> = new Set();
+  const filenames: Record<string, string> = {};
+  const versions: Record<string, string> = {};
+
+  if (device) {
+    for (const [k, v] of Object.entries(device)) {
+      if (/DeviceInfo\.VendorConfigFile\.[0-9]+(\.Name|\.Version)?$/.test(k)) {
+        const p = k.split(".");
+        if (p[p.length - 1] === "Name") {
+          const n = `3 Vendor Configuration File ${p[p.length - 2]}`;
+          instances.add(n);
+          if (v["value"]) filenames[n] = v["value"][0];
+        } else if (p[p.length - 1] === "Version") {
+          const n = `3 Vendor Configuration File ${p[p.length - 2]}`;
+          instances.add(n);
+          if (v["value"]) versions[n] = v["value"][0];
+        } else {
+          const n = `3 Vendor Configuration File ${p[p.length - 1]}`;
+          instances.add(n);
+        }
+      } else if (/DeviceInfo\.VendorLogFile\.[0-9]+(\.Name)?$/.test(k)) {
+        const p = k.split(".");
+        if (p[p.length - 1] === "Name") {
+          const n = `4 Vendor Log File ${p[p.length - 2]}`;
+          instances.add(n);
+          if (v["value"]) filenames[n] = v["value"][0];
+        } else {
+          const n = `4 Vendor Log File ${p[p.length - 1]}`;
+          instances.add(n);
+        }
+      }
+    }
+  }
+
+  if (!instances.size) {
+    instances.add("1 Vendor Configuration File");
+    instances.add("2 Vendor Log File");
+  }
+
+  const res: Record<string, string> = {};
+
+  for (const i of instances) {
+    let n = filenames[i];
+    const v = versions[i];
+    if (!n) n = i.replace(/\s+/g, "_") + ".txt";
+    if (v) n = `${n}-${v}`;
+    res[i] = n;
+  }
+
+  uploadTypesForDevice.set(device, res);
+  return res;
+}
+
+function renderStagingUpload(task: StageTask): Children {
+  if (!task.fileName || !task.fileType) invalid.add(task);
+  else invalid.delete(task);
+
+  const commonTypes: Record<string, number> = {};
+
+  const filenames: Record<string, string> = {};
+
+  for (const d of task.devices) {
+    const dev = store.fetch("devices", ["=", ["PARAM", "DeviceID.ID"], d])
+      .value[0];
+    if (dev) {
+      const u = getUploadTypesForDevice(dev);
+      Object.assign(filenames, u);
+      for (const k of Object.keys(u))
+        commonTypes[k] = (commonTypes[k] || 0) + 1;
+    }
+  }
+
+  const typesList = [""];
+  typesList.push(
+    ...Object.entries(commonTypes)
+      .filter(([, c]) => c === task.devices.length)
+      .map(([k]) => k)
+      .sort()
+  );
+
+  return [
+    "Fetch ",
+    m(
+      "select",
+      {
+        onchange: (e) => {
+          task.fileType = e.target.value;
+          task.fileName = filenames[task.fileType];
+        },
+      },
+      typesList.map((t) =>
+        m(
+          "option",
+          { disabled: !t, value: t, selected: (task.fileType || "") === t },
+          t
+        )
+      )
+    ),
+    " as ",
+    m("input", {
+      onchange: (e) => {
+        task.fileName = e.target.value;
+      },
+      value: task.fileName || "",
+    }),
+  ];
+}
+
 function renderStaging(staging: Set<StageTask>): Child[] {
   const elements: Child[] = [];
 
@@ -204,6 +320,7 @@ function renderStaging(staging: Set<StageTask>): Child[] {
     if (s.name === "setParameterValues")
       elms = renderStagingSpv(s, queueFunc, cancelFunc);
     else if (s.name === "download") elms = renderStagingDownload(s);
+    else if (s.name === "upload") elms = renderStagingUpload(s);
 
     const queue = m(
       "button.primary",
@@ -323,6 +440,14 @@ function renderQueue(queue: Set<QueueTask>): Child[] {
             m(".actions", actions)
           )
         );
+      } else if (t.name === "upload") {
+        details.push(
+          m(
+            `div.${t.status}`,
+            `Fetch file ${t.fileName} (${t.fileType})`,
+            m(".actions", actions)
+          )
+        );
       } else {
         details.push(m(`div.${t.status}`, t.name, m(".actions", actions)));
       }

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 {

ui/permissions-page.ts

@@ -47,6 +47,7 @@ const attributes = [
       "devices",
       "faults",
       "files",
+      "uploads",
       "permissions",
       "users",
       "presets",

ui/store.ts

@@ -58,11 +58,11 @@ const resources: Resources = {};
 for (const r of [
   "devices",
   "faults",
-  "files",
   "presets",
   "provisions",
   "virtualParameters",
   "files",
+  "uploads",
   "config",
   "users",
   "permissions",

ui/task-queue.ts

@@ -89,6 +89,14 @@ export function stageDownload(task: StageTask): void {
   staging.add(task);
 }
 
+export function stageUpload(task: StageTask): void {
+  if (queue.size + task.devices.length > MAX_QUEUE) {
+    notifications.push("error", "Too many tasks in queue");
+    return;
+  }
+  staging.add(task);
+}
+
 export function commit(
   tasks: QueueTask[],
   callback: (