genieacs/lib/cwmp.js

/**
 * Copyright 2013-2018  Zaid Abdulla
 *
 * 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/>.
 */
"use strict";

import zlib from "zlib";
import crypto from "crypto";
import fs from "fs";
import * as config from "./config";
import * as common from "./common";
import * as soap from "./soap";
import * as session from "./session";
import * as query from "./query";
import * as device from "./device";
import * as cache from "./cache";
import * as localCache from "./local-cache";
import * as db from "./db";
import * as logger from "./logger";
import * as scheduling from "./scheduling";

const MAX_CYCLES = 4;
const MAX_CONCURRENT_REQUESTS = config.get("MAX_CONCURRENT_REQUESTS");

const stats = {
  concurrentRequests: 0,
  totalRequests: 0,
  droppedRequests: 0,
  initiatedSessions: 0
};

function throwError(err, httpResponse) {
  if (httpResponse) {
    currentSessions.delete(httpResponse.connection);
    httpResponse.writeHead(500, { Connection: "close" });
    httpResponse.end(`${err.name}: ${err.message}`);
    stats.concurrentRequests -= 1;
  }
  throw err;
}

function writeResponse(sessionContext, res, close) {
  // Close connection after last request in session
  if (close) res.headers["Connection"] = "close";

  if (sessionContext.debug) {
    const dump =
      `# RESPONSE ${new Date(Date.now())}\n` +
      JSON.stringify(res.headers) +
      "\n" +
      res.data +
      "\n\n";
    fs.appendFile(`./debug/${sessionContext.deviceId}.dump`, dump, err => {
      if (err) throwError(err);
    });
  }

  function finish() {
    stats.concurrentRequests -= 1;
    if (sessionContext.httpRequest.connection.destroyed) {
      logger.accessError({
        sessionContext: sessionContext,
        message: "Connection dropped"
      });
    } else if (close) {
      endSession(sessionContext, (err, isNew) => {
        if (err) return void throwError(err);

        if (isNew) {
          logger.accessInfo({
            sessionContext: sessionContext,
            message: "New device registered"
          });
        }
      });
    } else {
      sessionContext.lastActivity = Date.now();
      currentSessions.set(
        sessionContext.httpRequest.connection,
        sessionContext
      );
    }
  }

  let compress;
  // Respond using the same content-encoding as the request
  if (
    sessionContext.httpRequest.headers["content-encoding"] &&
    res.data.length > 0
  ) {
    switch (sessionContext.httpRequest.headers["content-encoding"]) {
      case "gzip":
        res.headers["Content-Encoding"] = "gzip";
        compress = zlib.gzip;
        break;
      case "deflate":
        res.headers["Content-Encoding"] = "deflate";
        compress = zlib.deflate;
    }
  }

  if (compress) {
    compress(res.data, (err, data) => {
      if (err) return void throwError(err, sessionContext.httpResponse);

      res.headers["Content-Length"] = data.length;
      sessionContext.httpResponse.writeHead(res.code, res.headers);
      sessionContext.httpResponse.end(data);
      finish();
    });
  } else {
    res.headers["Content-Length"] = res.data.length;
    sessionContext.httpResponse.writeHead(res.code, res.headers);
    sessionContext.httpResponse.end(res.data);
    finish();
  }
}

function recordFault(sessionContext, fault, provisions, channels) {
  if (!provisions) {
    provisions = sessionContext.provisions;
    channels = sessionContext.channels;
  }

  const faults = sessionContext.faults;
  for (const channel of Object.keys(channels)) {
    const provs = sessionContext.faults[channel]
      ? sessionContext.faults[channel].provisions
      : [];
    faults[channel] = Object.assign({ provisions: provs }, fault);
    if (channel.startsWith("task_")) {
      const taskId = channel.slice(5);
      for (const t of sessionContext.tasks)
        if (t._id === taskId && t.expiry) faults[channel].expiry = t.expiry;
    }

    if (sessionContext.retries[channel] != null) {
      ++sessionContext.retries[channel];
    } else {
      sessionContext.retries[channel] = 0;
      if (Object.keys(channels).length !== 1) faults[channel].retryNow = true;
    }

    if (channels[channel] === 0) faults[channel].precondition = true;

    if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
    sessionContext.faultsTouched[channel] = true;

    logger.accessWarn({
      sessionContext: sessionContext,
      message: "Channel has faulted",
      fault: fault,
      channel: channel,
      retries: sessionContext.retries[channel]
    });
  }

  for (let i = 0; i < provisions.length; ++i) {
    for (const channel of Object.keys(channels)) {
      if ((channels[channel] >> i) & 1)
        faults[channel].provisions.push(provisions[i]);
    }
  }

  for (const channel of Object.keys(channels)) {
    const provs = faults[channel].provisions;
    faults[channel].provisions = [];
    appendProvisions(faults[channel].provisions, provs);
  }

  session.clearProvisions(sessionContext);
}

function inform(sessionContext, rpc) {
  session.inform(sessionContext, rpc.cpeRequest, (err, acsResponse) => {
    if (err) return void throwError(err, sessionContext.httpResponse);

    const res = soap.response({
      id: rpc.id,
      acsResponse: acsResponse,
      cwmpVersion: sessionContext.cwmpVersion
    });

    const cookiesPath = localCache.getConfig(
      sessionContext.cacheSnapshot,
      "cwmp.cookiesPath",
      sessionContext.configContext
    );

    if (cookiesPath) {
      res.headers["Set-Cookie"] = `session=${
        sessionContext.sessionId
      }; Path=${cookiesPath}`;
    } else {
      res.headers["Set-Cookie"] = `session=${sessionContext.sessionId}`;
    }

    writeResponse(sessionContext, res);
  });
}

function transferComplete(sessionContext, rpc) {
  session.transferComplete(
    sessionContext,
    rpc.cpeRequest,
    (err, acsResponse, operation, fault) => {
      if (err) return void throwError(err, sessionContext.httpResponse);

      if (!operation) {
        logger.accessWarn({
          sessionContext: sessionContext,
          message: "Unrecognized command key",
          rpc: rpc
        });
      }

      if (fault) {
        Object.assign(sessionContext.retries, operation.retries);
        recordFault(
          sessionContext,
          fault,
          operation.provisions,
          operation.channels
        );
      }

      const res = soap.response({
        id: rpc.id,
        acsResponse: acsResponse,
        cwmpVersion: sessionContext.cwmpVersion
      });

      writeResponse(sessionContext, res);
    }
  );
}

// Append providions and remove duplicates
function appendProvisions(original, toAppend) {
  let modified = false;
  const stringified = new WeakMap();

  for (const p of original) stringified.set(p, JSON.stringify(p));

  for (let i = toAppend.length - 1; i >= 0; --i) {
    let p = toAppend[i];
    const s = JSON.stringify(p);
    for (let j = original.length - 1; j >= 0; --j) {
      const ss = stringified.get(original[j]);
      if (s === ss) {
        if (!p || j >= original.length - (toAppend.length - i)) {
          p = null;
        } else {
          original.splice(j, 1);
          modified = true;
        }
      }
    }

    if (p) {
      original.splice(original.length - (toAppend.length - i) + 1, 0, p);
      stringified.set(p, s);
      modified = true;
    }
  }

  return modified;
}

function applyPresets(sessionContext) {
  const deviceData = sessionContext.deviceData;
  const presets = localCache.getPresets(sessionContext.cacheSnapshot);

  // Filter presets based on existing faults
  const blackList = {};
  let whiteList = null;
  let whiteListProvisions = null;
  const RETRY_DELAY = localCache.getConfig(
    sessionContext.cacheSnapshot,
    "cwmp.retryDelay",
    sessionContext.configContext
  );

  if (sessionContext.faults) {
    for (const [channel, fault] of Object.entries(sessionContext.faults)) {
      let retryTimestamp = 0;
      if (!fault.retryNow) {
        retryTimestamp =
          fault.timestamp +
          RETRY_DELAY * Math.pow(2, sessionContext.retries[channel]) * 1000;
      }

      if (retryTimestamp <= sessionContext.timestamp) {
        whiteList = channel;
        whiteListProvisions = fault.provisions;
        break;
      }

      blackList[channel] = fault.precondition ? 1 : 2;
    }
  }

  deviceData.timestamps.revision = 1;
  deviceData.attributes.revision = 1;

  const deviceEvents = {};
  for (const p of deviceData.paths.find(["Events", "*"], false, true)) {
    const attrs = deviceData.attributes.get(p);
    if (attrs && attrs.value && attrs.value[1][0] >= sessionContext.timestamp)
      deviceEvents[p[1]] = true;
  }

  const parameters = {};
  const filteredPresets = [];

  for (const preset of presets) {
    if (whiteList != null) {
      if (preset.channel !== whiteList) continue;
    } else if (blackList[preset.channel] === 1) {
      continue;
    }

    let eventsMatch = true;
    for (const [k, v] of Object.entries(preset.events)) {
      if (!v !== !deviceEvents[k.replace(/\s+/g, "_")]) {
        eventsMatch = false;
        break;
      }
    }

    if (!eventsMatch) continue;

    if (preset.schedule && preset.schedule.schedule) {
      const r = scheduling.cron(
        sessionContext.timestamp,
        preset.schedule.schedule
      );
      if (!(r[0] + preset.schedule.duration > sessionContext.timestamp))
        continue;
    }

    filteredPresets.push(preset);
    for (const k of Object.keys(preset.precondition)) {
      sessionContext.channels[preset.channel] = 0;
      const p = k.split(/([^a-zA-Z0-9\-_.].*)/, 1)[0];
      parameters[p] = common.parsePath(p);
    }
  }

  const declarations = Object.values(parameters).map(v => ({
    path: v,
    pathGet: 1,
    pathSet: null,
    attrGet: { value: 1 },
    attrSet: null,
    defer: true
  }));

  session.rpcRequest(
    sessionContext,
    declarations,
    (err, flt, reqId, acsReq) => {
      if (err) return void throwError(err, sessionContext.httpResponse);

      if (flt) {
        recordFault(sessionContext, flt);
        session.clearProvisions(sessionContext);
        return void applyPresets(sessionContext);
      }

      if (acsReq) return void sendAcsRequest(sessionContext, reqId, acsReq);

      session.clearProvisions(sessionContext);

      const parameterValues = {};
      for (const [k, v] of Object.entries(parameters)) {
        const unpacked = device.unpack(deviceData, v);
        if (!unpacked[0]) continue;
        const attrs = deviceData.attributes.get(unpacked[0]);
        if (attrs && attrs.value && attrs.value[1])
          parameterValues[k] = attrs.value[1][0];
      }

      if (whiteList != null)
        session.addProvisions(sessionContext, whiteList, whiteListProvisions);

      const appendProvisionsToFaults = {};
      for (const p of filteredPresets) {
        if (query.testFilter(parameterValues, p.precondition)) {
          if (blackList[p.channel] === 2) {
            appendProvisionsToFaults[p.channel] = (
              appendProvisionsToFaults[p.channel] || []
            ).concat(p.provisions);
          } else {
            session.addProvisions(sessionContext, p.channel, p.provisions);
          }
        }
      }

      for (const [channel, provisions] of Object.entries(
        appendProvisionsToFaults
      )) {
        if (
          appendProvisions(
            sessionContext.faults[channel].provisions,
            provisions
          )
        ) {
          if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
          sessionContext.faultsTouched[channel] = true;
        }
      }

      // Don't increment when processing a single channel (e.g. after fault)
      if (whiteList == null)
        sessionContext.presetCycles = (sessionContext.presetCycles || 0) + 1;

      if (sessionContext.presetCycles > MAX_CYCLES) {
        const fault = {
          code: "preset_loop",
          message: "The presets are stuck in an endless configuration loop",
          timestamp: sessionContext.timestamp
        };
        recordFault(sessionContext, fault);
        // No need to save retryNow
        for (const f of Object.values(sessionContext.faults)) delete f.retryNow;
        session.clearProvisions(sessionContext);
        return void sendAcsRequest(sessionContext);
      }

      deviceData.timestamps.dirty = 0;
      deviceData.attributes.dirty = 0;
      session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) => {
        if (err) return void throwError(err, sessionContext.httpResponse);

        if (fault) {
          recordFault(sessionContext, fault);
          session.clearProvisions(sessionContext);
          return void applyPresets(sessionContext);
        }

        if (!acsRequest) {
          for (const channel of Object.keys(sessionContext.channels)) {
            if (sessionContext.faults[channel]) {
              delete sessionContext.faults[channel];
              if (!sessionContext.faultsTouched)
                sessionContext.faultsTouched = {};
              sessionContext.faultsTouched[channel] = true;
            }
          }

          if (whiteList != null) return void applyPresets(sessionContext);

          if (
            sessionContext.deviceData.timestamps.dirty > 1 ||
            sessionContext.deviceData.attributes.dirty > 1
          )
            return void applyPresets(sessionContext);
        }

        sendAcsRequest(sessionContext, id, acsRequest);
      });
    }
  );
}

function nextRpc(sessionContext) {
  session.rpcRequest(sessionContext, null, (err, fault, id, acsRequest) => {
    if (err) return void throwError(err, sessionContext.httpResponse);

    if (fault) {
      recordFault(sessionContext, fault);
      session.clearProvisions(sessionContext);
      return void nextRpc(sessionContext);
    }

    if (acsRequest) return void sendAcsRequest(sessionContext, id, acsRequest);

    for (const [channel, flags] of Object.entries(sessionContext.channels)) {
      if (flags && sessionContext.faults[channel]) {
        delete sessionContext.faults[channel];
        if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};

        sessionContext.faultsTouched[channel] = true;
      }
      if (channel.startsWith("task_")) {
        const taskId = channel.slice(5);
        if (!sessionContext.doneTasks) sessionContext.doneTasks = [];
        sessionContext.doneTasks.push(taskId);

        for (let j = 0; j < sessionContext.tasks.length; ++j) {
          if (sessionContext.tasks[j]._id === taskId) {
            sessionContext.tasks.splice(j, 1);
            break;
          }
        }
      }
    }

    session.clearProvisions(sessionContext);

    // Clear expired tasks
    sessionContext.tasks = sessionContext.tasks.filter(task => {
      if (!(task.expiry <= sessionContext.timestamp)) return true;

      logger.accessInfo({
        sessionContext: sessionContext,
        message: "Task expired",
        task: task
      });

      if (!sessionContext.doneTasks) sessionContext.doneTasks = [];
      sessionContext.doneTasks.push(task._id);

      const channel = `task_${task._id}`;
      if (sessionContext.faults[channel]) {
        delete sessionContext.faults[channel];
        if (!sessionContext.faultsTouched) sessionContext.faultsTouched = {};
        sessionContext.faultsTouched[channel] = true;
      }

      return false;
    });

    const task = sessionContext.tasks.find(
      t => !sessionContext.faults[`task_${t._id}`]
    );

    if (!task) return void applyPresets(sessionContext);

    let alias;

    switch (task.name) {
      case "getParameterValues":
        // Set channel in case params array is empty
        sessionContext.channels[`task_${task._id}`] = 0;
        for (const p of task.parameterNames) {
          session.addProvisions(sessionContext, `task_${task._id}`, [
            ["refresh", p]
          ]);
        }

        break;
      case "setParameterValues":
        // Set channel in case params array is empty
        sessionContext.channels[`task_${task._id}`] = 0;
        for (const p of task.parameterValues) {
          session.addProvisions(sessionContext, `task_${task._id}`, [
            ["value", p[0], p[1]]
          ]);
        }

        break;
      case "refreshObject":
        session.addProvisions(sessionContext, `task_${task._id}`, [
          ["refresh", task.objectName]
        ]);
        break;
      case "reboot":
        session.addProvisions(sessionContext, `task_${task._id}`, [["reboot"]]);
        break;
      case "factoryReset":
        session.addProvisions(sessionContext, `task_${task._id}`, [["reset"]]);
        break;
      case "download":
        session.addProvisions(sessionContext, `task_${task._id}`, [
          ["download", task.fileType, task.fileName, task.targetFileName]
        ]);
        break;
      case "addObject":
        alias = (task.parameterValues || [])
          .map(p => `${p[0]}:${JSON.stringify(p[1])}`)
          .join(",");
        session.addProvisions(sessionContext, `task_${task._id}`, [
          ["instances", `${task.objectName}.[${alias}]`, "+1"]
        ]);
        break;
      case "deleteObject":
        session.addProvisions(sessionContext, `task_${task._id}`, [
          ["instances", task.objectName, 0]
        ]);
        break;
      default:
        return void throwError(
          new Error("Task name not recognized"),
          sessionContext.httpResponse
        );
    }
    nextRpc(sessionContext);
  });
}

function endSession(sessionContext, callback) {
  let saveCache = sessionContext.cacheUntil != null;
  let counter = 3;

  counter += 2;
  db.saveDevice(
    sessionContext.deviceId,
    sessionContext.deviceData,
    sessionContext.new,
    sessionContext.timestamp,
    err => {
      if (err) {
        if (counter & 1) callback(err);
        return void (counter = 0);
      }
      if ((counter -= 2) === 1) callback(null, sessionContext.new);
    }
  );

  if (sessionContext.operationsTouched) {
    for (const k of Object.keys(sessionContext.operationsTouched)) {
      counter += 2;
      saveCache = true;
      if (sessionContext.operations[k]) {
        db.saveOperation(
          sessionContext.deviceId,
          k,
          sessionContext.operations[k],
          err => {
            if (err) {
              if (counter & 1) callback(err);
              return void (counter = 0);
            }
            if ((counter -= 2) === 1) callback(null, sessionContext.new);
          }
        );
      } else {
        db.deleteOperation(sessionContext.deviceId, k, err => {
          if (err) {
            if (counter & 1) callback(err);
            return void (counter = 0);
          }
          if ((counter -= 2) === 1) callback(null, sessionContext.new);
        });
      }
    }
  }

  if (sessionContext.doneTasks && sessionContext.doneTasks.length) {
    counter += 2;
    saveCache = true;
    db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, err => {
      if (err) {
        if (counter & 1) callback(err);
        return void (counter = 0);
      }
      if ((counter -= 2) === 1) callback(null, sessionContext.new);
    });
  }

  if (sessionContext.faultsTouched) {
    for (const k of Object.keys(sessionContext.faultsTouched)) {
      counter += 2;
      saveCache = true;
      if (sessionContext.faults[k]) {
        sessionContext.faults[k].retries = sessionContext.retries[k];
        db.saveFault(
          sessionContext.deviceId,
          k,
          sessionContext.faults[k],
          err => {
            if (err) {
              if (counter & 1) callback(err);
              return void (counter = 0);
            }
            if ((counter -= 2) === 1) callback(null, sessionContext.new);
          }
        );
      } else {
        db.deleteFault(sessionContext.deviceId, k, err => {
          if (err) {
            if (counter & 1) callback(err);
            return void (counter = 0);
          }
          if ((counter -= 2) === 1) callback(null, sessionContext.new);
        });
      }
    }
  }

  if (saveCache) {
    counter += 2;
    cacheDueTasksAndFaultsAndOperations(
      sessionContext.deviceId,
      sessionContext.tasks,
      sessionContext.faults,
      sessionContext.operations,
      sessionContext.cacheUntil,
      err => {
        if (err) {
          if (counter & 1) callback(err);
          return void (counter = 0);
        }
        if ((counter -= 2) === 1) callback(null, sessionContext.new);
      }
    );
  }
  if ((counter -= 2) === 1) callback(null, sessionContext.new);
}

function sendAcsRequest(sessionContext, id, acsRequest) {
  if (!acsRequest)
    return void writeResponse(sessionContext, soap.response(null), true);

  if (acsRequest.name === "Download") {
    acsRequest.fileSize = 0;
    if (!acsRequest.url) {
      const FS_PORT = config.get("FS_PORT");
      const FS_HOSTNAME = config.get("FS_HOSTNAME");
      const FS_SSL = config.get("FS_SSL");
      acsRequest.url = FS_SSL ? "https://" : "http://";
      acsRequest.url += FS_HOSTNAME;
      if (FS_PORT !== 80) acsRequest.url += ":" + FS_PORT;
      acsRequest.url += "/" + encodeURI(acsRequest.fileName);

      const files = localCache.getFiles(sessionContext.cacheSnapshot);
      if (files[acsRequest.fileName])
        acsRequest.fileSize = files[acsRequest.fileName].length;
    }
  }

  const rpc = {
    id: id,
    acsRequest: acsRequest,
    cwmpVersion: sessionContext.cwmpVersion
  };

  logger.accessInfo({
    sessionContext: sessionContext,
    message: "ACS request",
    rpc: rpc
  });

  const res = soap.response(rpc);
  writeResponse(sessionContext, res);
}

function getSession(connection, sessionId, callback) {
  const sessionContext = currentSessions.get(connection);
  if (sessionContext) {
    currentSessions.delete(connection);
    return void callback(null, sessionContext);
  }

  if (!sessionId) return void callback();

  setTimeout(() => {
    cache.pop(`session_${sessionId}`, (err, sessionContextString) => {
      if (err || !sessionContextString) return void callback(err);
      session.deserialize(sessionContextString, callback);
    });
  }, 100);
}

const currentSessions = new WeakMap();

// When socket closes, store active sessions in cache
export function onConnection(socket) {
  // The property remoteAddress may be undefined after the connection is
  // closed, unless we read it at least once (caching?)
  socket.remoteAddress;

  socket.on("close", () => {
    const sessionContext = currentSessions.get(socket);
    if (!sessionContext) return;
    currentSessions.delete(socket);
    const now = Date.now();

    const lastActivity = sessionContext.lastActivity;
    const timeoutMsg = logger.flatten({
      sessionContext: sessionContext,
      message: "Session timeout",
      sessionTimestamp: sessionContext.timestamp
    });

    const timeout =
      sessionContext.lastActivity + sessionContext.timeout * 1000 - now;
    if (timeout <= 0) return void logger.accessError(timeoutMsg);

    setTimeout(() => {
      cache.get(
        `session_${sessionContext.sessionId}`,
        (err, sessionContextString) => {
          if (err) return void throwError(err);

          if (!sessionContextString) return;

          session.deserialize(sessionContextString, (err, _sessionContext) => {
            if (err) return void throwError(err);
            if (_sessionContext.lastActivity === lastActivity)
              logger.accessError(timeoutMsg);
          });
        }
      );
    }, timeout + 1000).unref();

    session.serialize(sessionContext, (err, sessionContextString) => {
      if (err) return void throwError(err);
      cache.set(
        `session_${sessionContext.sessionId}`,
        sessionContextString,
        Math.ceil(timeout / 1000) + 3,
        err => {
          if (err) return void throwError(err);
        }
      );
    });
  });
}

setInterval(() => {
  if (stats.droppedRequests) {
    logger.warn({
      message: "Worker overloaded",
      droppedRequests: stats.droppedRequests,
      totalRequests: stats.totalRequests,
      initiatedSessions: stats.initiatedSessions,
      pid: process.pid
    });
  }

  stats.totalRequests = 0;
  stats.droppedRequests = 0;
  stats.initiatedSessions = 0;
}, 10000).unref();

function getDueTasksAndFaultsAndOperations(deviceId, timestamp, callback) {
  cache.get(`${deviceId}_tasks_faults_operations`, (err, res) => {
    if (err) return void callback(err);

    if (res) {
      res = JSON.parse(res);
      return void callback(
        null,
        res.tasks || [],
        res.faults || {},
        res.operations || {}
      );
    }

    let faults, tasks, operations, cacheUntil;

    db.getFaults(deviceId, (err, _faults) => {
      if (err) {
        if (callback) callback(err);
        return void (callback = null);
      }
      faults = _faults;
      if (tasks && operations)
        callback(null, tasks, faults, operations, cacheUntil);
    });

    db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) => {
      if (err) {
        if (callback) callback(err);
        return void (callback = null);
      }
      tasks = dueTasks;
      cacheUntil = nextTimestamp || 0;
      if (faults && operations)
        callback(null, tasks, faults, operations, cacheUntil);
    });

    db.getOperations(deviceId, (err, _operations) => {
      if (err) {
        if (callback) callback(err);
        return void (callback = null);
      }
      operations = _operations;
      if (faults && tasks)
        callback(null, tasks, faults, operations, cacheUntil);
    });

    if (faults && tasks && operations)
      callback(null, tasks, faults, operations, cacheUntil);
  });
}

function cacheDueTasksAndFaultsAndOperations(
  deviceId,
  tasks,
  faults,
  operations,
  cacheUntil,
  callback
) {
  const v = {};
  if (tasks.length) v.tasks = tasks;
  if (Object.keys(faults).length) v.faults = faults;
  if (Object.keys(operations).length) v.operations = operations;

  let ttl;
  if (cacheUntil) ttl = Math.trunc((Date.now() - cacheUntil) / 1000);
  else ttl = config.get("MAX_CACHE_TTL", deviceId);

  cache.set(
    `${deviceId}_tasks_faults_operations`,
    JSON.stringify(v),
    ttl,
    callback
  );
}

function processRequest(sessionContext, rpc) {
  if (rpc.cpeRequest) {
    if (rpc.cpeRequest.name === "Inform") {
      logger.accessInfo({
        sessionContext: sessionContext,
        message: "Inform",
        rpc: rpc
      });
      inform(sessionContext, rpc);
    } else if (rpc.cpeRequest.name === "TransferComplete") {
      logger.accessInfo({
        sessionContext: sessionContext,
        message: "CPE request",
        rpc: rpc
      });
      transferComplete(sessionContext, rpc);
    } else if (rpc.cpeRequest.name === "GetRPCMethods") {
      logger.accessInfo({
        sessionContext: sessionContext,
        message: "CPE request",
        rpc: rpc
      });
      const res = soap.response({
        id: rpc.id,
        acsResponse: {
          name: "GetRPCMethodsResponse",
          methodList: ["Inform", "GetRPCMethods", "TransferComplete"]
        },
        cwmpVersion: sessionContext.cwmpVersion
      });
      writeResponse(sessionContext, res);
    } else {
      return void throwError(
        new Error("ACS method not supported"),
        sessionContext.httpResponse
      );
    }
  } else if (rpc.cpeResponse) {
    session.rpcResponse(sessionContext, rpc.id, rpc.cpeResponse, err => {
      if (err) return void throwError(err, sessionContext.httpResponse);
      nextRpc(sessionContext);
    });
  } else if (rpc.cpeFault) {
    logger.accessWarn({
      sessionContext: sessionContext,
      message: "CPE fault",
      rpc: rpc
    });

    session.rpcFault(sessionContext, rpc.id, rpc.cpeFault, (err, fault) => {
      if (err) return void throwError(err, sessionContext.httpResponse);
      if (fault) {
        recordFault(sessionContext, fault);
        session.clearProvisions(sessionContext);
      }
      nextRpc(sessionContext);
    });
  } else {
    // CPE sent empty response
    session.timeoutOperations(sessionContext, (err, faults, operations) => {
      if (err) return void throwError(err, sessionContext.httpResponse);

      for (const i of faults) {
        for (const [k, v] of Object.entries(operations[i].retries))
          sessionContext.retries[k] = v;

        recordFault(
          sessionContext,
          faults[i],
          operations[i].provisions,
          operations[i].channels
        );
      }

      nextRpc(sessionContext);
    });
  }
}

export function listener(httpRequest, httpResponse) {
  stats.totalRequests += 1;

  if (httpRequest.method !== "POST") {
    httpResponse.writeHead(405, {
      Allow: "POST",
      Connection: "close"
    });
    httpResponse.end("405 Method Not Allowed");
    return;
  }

  let sessionId;
  // Separation by comma is important as some devices don't comform to standard
  const COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g;
  let match;
  while ((match = COOKIE_REGEX.exec(httpRequest.headers.cookie)))
    if (match[1] === "session") sessionId = match[2];

  // If overloaded, ask CPE to retry in 60 seconds
  if (!sessionId && stats.concurrentRequests > MAX_CONCURRENT_REQUESTS) {
    httpResponse.writeHead(503, {
      "Retry-after": 60,
      Connection: "close"
    });
    httpResponse.end("503 Service Unavailable");
    stats.droppedRequests += 1;
    return;
  }

  let stream = httpRequest;
  if (httpRequest.headers["content-encoding"]) {
    switch (httpRequest.headers["content-encoding"]) {
      case "gzip":
        stream = httpRequest.pipe(zlib.createGunzip());
        break;
      case "deflate":
        stream = httpRequest.pipe(zlib.createInflate());
        break;
      default:
        httpResponse.writeHead(415, { Connection: "close" });
        httpResponse.end("415 Unsupported Media Type");
        return;
    }
  }

  stats.concurrentRequests += 1;
  httpRequest.on("aborted", () => {
    stats.concurrentRequests -= 1;
    // In some cases event end can be emitted after aborted event
    httpRequest.removeAllListeners("end");
  });

  const chunks = [];
  let bytes = 0;
  stream.on("data", chunk => {
    chunks.push(chunk);
    bytes += chunk.length;
  });

  stream.on("end", () => {
    const body = Buffer.allocUnsafe(bytes);
    let offset = 0;
    for (const chunk of chunks) {
      chunk.copy(body, offset, 0, chunk.length);
      offset += chunk.length;
    }

    function parsedRpc(sessionContext, rpc, parseWarnings) {
      for (const w of parseWarnings) {
        w.sessionContext = sessionContext;
        w.rpc = rpc;
        logger.accessWarn(w);
      }

      if (sessionContext.debug) {
        const dump =
          `# REQUEST ${new Date(Date.now())}\n` +
          JSON.stringify(httpRequest.headers) +
          "\n" +
          body +
          "\n\n";
        fs.appendFile(`./debug/${sessionContext.deviceId}.dump`, dump, err => {
          if (err) return void throwError(err);
        });
      }

      processRequest(sessionContext, rpc);
    }

    getSession(httpRequest.connection, sessionId, (err, sessionContext) => {
      if (err) return void throwError(err, httpResponse);

      if (sessionContext) {
        sessionContext.httpRequest = httpRequest;
        sessionContext.httpResponse = httpResponse;
        if (
          sessionContext.sessionId !== sessionId ||
          sessionContext.lastActivity + sessionContext.timeout * 1000 <
            Date.now()
        ) {
          logger.accessError({
            message: "Invalid session",
            sessionContext: sessionContext
          });

          httpResponse.writeHead(400, { Connection: "close" });
          httpResponse.end("Invalid session");
          stats.concurrentRequests -= 1;
          return;
        }
      } else if (stats.concurrentRequests > MAX_CONCURRENT_REQUESTS) {
        // Check again just in case device included old session ID
        // from the previous session
        httpResponse.writeHead(503, { "Retry-after": 60, Connection: "close" });
        httpResponse.end("503 Service Unavailable");
        stats.droppedRequests += 1;
        stats.concurrentRequests -= 1;
        return;
      }

      const parseWarnings = [];
      let rpc;
      try {
        rpc = soap.request(
          body,
          sessionContext ? sessionContext.cwmpVersion : null,
          parseWarnings
        );
      } catch (error) {
        logger.accessError({
          message: "XML parse error",
          parseError: error.message.trim(),
          sessionContext: sessionContext || {
            httpRequest: httpRequest,
            httpResponse: httpResponse
          }
        });
        httpResponse.writeHead(400, { Connection: "close" });
        httpResponse.end(error.message);
        stats.concurrentRequests -= 1;
        return;
      }

      if (sessionContext) {
        if (
          (rpc.cpeRequest && rpc.cpeRequest.name === "Inform") ||
          !sessionContext.rpcRequest ^ !(rpc.cpeResponse || rpc.cpeFault)
        ) {
          logger.accessError({
            message: "Bad session state",
            sessionContext: sessionContext
          });
          httpResponse.writeHead(400, { Connection: "close" });
          httpResponse.end("Bad session state");
          stats.concurrentRequests -= 1;
          return;
        }
        return void parsedRpc(sessionContext, rpc, parseWarnings);
      }

      if (!(rpc.cpeRequest && rpc.cpeRequest.name === "Inform")) {
        logger.accessError({
          message: "Invalid session",
          sessionContext: sessionContext || {
            httpRequest: httpRequest,
            httpResponse: httpResponse
          }
        });
        httpResponse.writeHead(400, { Connection: "close" });
        httpResponse.end("Invalid session");
        stats.concurrentRequests -= 1;
        return;
      }

      stats.initiatedSessions += 1;
      const deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId);

      localCache.getCurrentSnapshot((err, cacheSnapshot) => {
        if (err) return void throwError(err, httpResponse);

        const configContext = {
          id: deviceId,
          serialNumber: rpc.cpeRequest.deviceId["SerialNumber"],
          productClass: rpc.cpeRequest.deviceId["ProductClass"],
          oui: rpc.cpeRequest.deviceId["OUI"]
        };

        const sessionTimeout =
          rpc.sessionTimeout ||
          localCache.getConfig(
            cacheSnapshot,
            "cwmp.sessionTimeout",
            configContext
          );

        session.init(
          deviceId,
          rpc.cwmpVersion,
          sessionTimeout,
          (err, _sessionContext) => {
            if (err) return void throwError(err, httpResponse);

            _sessionContext.cacheSnapshot = cacheSnapshot;
            _sessionContext.configContext = configContext;
            _sessionContext.debug = localCache.getConfig(
              cacheSnapshot,
              "cwmp.debug",
              configContext
            );

            _sessionContext.httpRequest = httpRequest;
            _sessionContext.httpResponse = httpResponse;
            _sessionContext.sessionId = crypto.randomBytes(8).toString("hex");
            httpRequest.connection.setTimeout(_sessionContext.timeout * 1000);

            getDueTasksAndFaultsAndOperations(
              deviceId,
              _sessionContext.timestamp,
              (err, dueTasks, faults, operations, cacheUntil) => {
                if (err) return void throwError(err, httpResponse);

                _sessionContext.tasks = dueTasks;
                _sessionContext.operations = operations;
                _sessionContext.cacheUntil = cacheUntil;
                _sessionContext.faults = faults;
                _sessionContext.retries = {};
                for (const [k, v] of Object.entries(_sessionContext.faults)) {
                  if (v.expiry >= _sessionContext.timestamp) {
                    // Delete expired faults
                    delete _sessionContext.faults[k];
                    if (!_sessionContext.faultsTouched)
                      _sessionContext.faultsTouched = {};
                    _sessionContext.faultsTouched[k] = true;
                  } else {
                    _sessionContext.retries[k] = v.retries;
                  }
                }
                parsedRpc(_sessionContext, rpc, parseWarnings);
              }
            );
          }
        );
      });
    });
  });
}