Refactor HTTP connection requests

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();
-        reject(new Error("Device is offline"));
+        req.destroy();
+        reject(err);
         if (_debug) debug.outgoingHttpRequestError(req, deviceId, options, err);
       })
-      .on("socket", (socket) => {
-        socket.setTimeout(timeout);
-        socket.on("timeout", () => {
-          req.abort();
-        });
+      .on("timeout", () => {
+        req.destroy();
       });
   });
 }
@@ -105,6 +101,7 @@ export async function httpConnectionRequest(
   options.agent = new http.Agent({
     maxSockets: 1,
     keepAlive: true,
+    timeout: timeout,
   });
 
   let authHeader: Record<string, string>;
@@ -146,24 +143,42 @@ export async function httpConnectionRequest(
       }
     }
 
-    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 (authHeader) {
+        try {
+          res = await httpGet(opts, _debug, deviceId);
+        } catch (err) {
+          return `Connection request error: ${err.message}`;
+        }
+      }
+
+      if (err["code"] === "ECONNRESET" || err["code"] === "ECONNREFUSED")
+        return "Device is offline";
+
+      return `Connection request error: ${err.message}`;
+    }
 
-    // Workaround for some devices unexpectedly closing the connection
-    if (res.statusCode === 0 && authHeader)
-      res = await httpGet(opts, timeout, _debug, deviceId);
-    if (res.statusCode === 0) return "Device is offline";
     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(
         res.headers["www-authenticate"]
       );
       [username, password, authExp] = await extractAuth(authExp, false);
     } else {
-      return `Unexpected response code from device: ${res.statusCode}`;
+      return `Connection request error: Unexpected status code ${res.statusCode}`;
     }
   }
-  return "Incorrect connection request credentials";
+
+  return "Connection request error: Incorrect connection request credentials";
 }
 
 export async function udpConnectionRequest(

lib/nbi.ts

@@ -394,12 +394,19 @@ export function listener(
             .insertTasks(task)
             .then(async () => {
               const lastInform = Date.now();
+
+              const socketTimeout: number = request.socket["timeout"];
+
+              // Disable socket timeout while waiting for session
+              if (socketTimeout) request.socket.setTimeout(0);
+
               const notInSession = await apiFunctions.awaitSessionEnd(
                 deviceId,
                 30000
               );
               await cache.del(`${deviceId}_tasks_faults_operations`);
               if (urlParts.query.connection_request == null) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
                 response.writeHead(202, {
                   "Content-Type": "application/json",
                 });
@@ -408,6 +415,7 @@ export function listener(
               }
 
               if (!notInSession) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
                 response.writeHead(202, "Task queued but not processed", {
                   "Content-Type": "application/json",
                 });
@@ -418,6 +426,7 @@ export function listener(
               const status = await apiFunctions.connectionRequest(deviceId);
 
               if (status) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
                 response.writeHead(202, status, {
                   "Content-Type": "application/json",
                 });
@@ -436,6 +445,7 @@ export function listener(
                 onlineThreshold
               );
               if (!sessionStarted) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
                 response.writeHead(202, "Task queued but not processed", {
                   "Content-Type": "application/json",
                 });
@@ -448,6 +458,7 @@ export function listener(
                 120000
               );
               if (!sessionEnded) {
+                if (socketTimeout) request.socket.setTimeout(socketTimeout);
                 response.writeHead(202, "Task queued but not processed", {
                   "Content-Type": "application/json",
                 });
@@ -467,6 +478,10 @@ export function listener(
               );
 
               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",

lib/ui/api.ts

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