diff --git a/.eslintrc.json b/.eslintrc.json index 49d0614..c305923 100644 --- a/.eslintrc.json +++ b/.eslintrc.json @@ -32,6 +32,7 @@ { "allowExpressions": true } ], "@typescript-eslint/no-floating-promises": "error", + "@typescript-eslint/no-misused-promises": "error", "require-atomic-updates": 0, "no-prototype-builtins": 0 } diff --git a/bin/genieacs-cwmp.ts b/bin/genieacs-cwmp.ts index 1a5dd82..6910c16 100755 --- a/bin/genieacs-cwmp.ts +++ b/bin/genieacs-cwmp.ts @@ -95,7 +95,7 @@ if (!cluster.worker) { exception: err, pid: process.pid, }); - server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); + server.stop(false).then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); const initPromise = db @@ -111,13 +111,19 @@ if (!cluster.worker) { process.on("SIGINT", () => { initPromise.finally(() => { - server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); + server + .stop(false) + .then(exitWorkerGracefully) + .catch(exitWorkerUngracefully); }); }); process.on("SIGTERM", () => { initPromise.finally(() => { - server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); + server + .stop(false) + .then(exitWorkerGracefully) + .catch(exitWorkerUngracefully); }); }); } diff --git a/bin/genieacs-fs.ts b/bin/genieacs-fs.ts index ee360a9..c0d37f8 100755 --- a/bin/genieacs-fs.ts +++ b/bin/genieacs-fs.ts @@ -79,8 +79,6 @@ if (!cluster.worker) { timeout: 30000, }; - let stopping = false; - process.on("uncaughtException", (err) => { if ((err as NodeJS.ErrnoException).code === "ERR_IPC_DISCONNECTED") return; logger.error({ @@ -88,19 +86,13 @@ if (!cluster.worker) { exception: err, pid: process.pid, }); - stopping = true; server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); - const _listener = (req, res): void => { - if (stopping) res.setHeader("Connection", "close"); - void listener(req, res); - }; - const initPromise = db .connect() .then(() => { - server.start(options, _listener); + server.start(options, listener); }) .catch((err) => { setTimeout(() => { @@ -109,14 +101,12 @@ if (!cluster.worker) { }); process.on("SIGINT", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); }); process.on("SIGTERM", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); diff --git a/bin/genieacs-nbi.ts b/bin/genieacs-nbi.ts index 623dd50..ae7ede2 100755 --- a/bin/genieacs-nbi.ts +++ b/bin/genieacs-nbi.ts @@ -84,8 +84,6 @@ if (!cluster.worker) { timeout: 30000, }; - let stopping = false; - process.on("uncaughtException", (err) => { if ((err as NodeJS.ErrnoException).code === "ERR_IPC_DISCONNECTED") return; logger.error({ @@ -93,19 +91,13 @@ if (!cluster.worker) { exception: err, pid: process.pid, }); - stopping = true; server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); - const _listener = (req, res): void => { - if (stopping) res.setHeader("Connection", "close"); - listener(req, res); - }; - const initPromise = db .connect() .then(() => { - server.start(options, _listener); + server.start(options, listener); }) .catch((err) => { setTimeout(() => { @@ -114,14 +106,12 @@ if (!cluster.worker) { }); process.on("SIGINT", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); }); process.on("SIGTERM", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); diff --git a/bin/genieacs-ui.ts b/bin/genieacs-ui.ts index 2ab4f7d..4ef9714 100755 --- a/bin/genieacs-ui.ts +++ b/bin/genieacs-ui.ts @@ -84,8 +84,6 @@ if (!cluster.worker) { timeout: 30000, }; - let stopping = false; - process.on("uncaughtException", (err) => { if ((err as NodeJS.ErrnoException).code === "ERR_IPC_DISCONNECTED") return; logger.error({ @@ -93,19 +91,13 @@ if (!cluster.worker) { exception: err, pid: process.pid, }); - stopping = true; server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); - const _listener = (req, res): void => { - if (stopping) res.setHeader("Connection", "close"); - listener(req, res); - }; - const initPromise = db .connect() .then(() => { - server.start(options, _listener); + server.start(options, async (req, res) => listener(req, res)); }) .catch((err) => { setTimeout(() => { @@ -114,14 +106,12 @@ if (!cluster.worker) { }); process.on("SIGINT", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); }); process.on("SIGTERM", () => { - stopping = true; initPromise.finally(() => { server.stop().then(exitWorkerGracefully).catch(exitWorkerUngracefully); }); diff --git a/lib/cwmp.ts b/lib/cwmp.ts index 47ab7a5..f60a4ec 100644 --- a/lib/cwmp.ts +++ b/lib/cwmp.ts @@ -59,6 +59,7 @@ import { parseXmlDeclaration } from "./xml-parser"; import * as debug from "./debug"; import { getRequestOrigin } from "./forwarded"; import { getSocketEndpoints } from "./server"; +import { EventEmitter } from "events"; const gzipPromisified = promisify(zlib.gzip); const deflatePromisified = promisify(zlib.deflate); @@ -889,86 +890,98 @@ async function sendAcsRequest( return writeResponse(sessionContext, res); } -// When socket closes, store active sessions in cache -export function onConnection(socket: Socket): void { - socket.on("close", async () => { - const sessionContext = currentSessions.get(socket); - if (!sessionContext) return; - currentSessions.delete(socket); - if (sessionContext.authState !== 2) { - logger.accessError({ - message: "Authentication failure", - sessionContext: sessionContext, - }); - return; - } +function once( + emitter: EventEmitter, + event: string, + timeout: number +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`Event ${event} timed out after ${timeout} ms`)); + }, timeout); - const now = Date.now(); - - const lastActivity = sessionContext.lastActivity; - const timeoutMsg = { - sessionContext: sessionContext, - message: "Session timeout", - sessionTimestamp: sessionContext.timestamp, - }; - - const timeout = - sessionContext.lastActivity + sessionContext.timeout * 1000 - now; - - 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( - `session_${sessionContext.sessionId}` - ); - if (!sessionContextString) return; - const _sessionContext = await session.deserialize(sessionContextString); - if (_sessionContext.lastActivity === lastActivity) { - logger.accessError(timeoutMsg); - await endSession(sessionContext); - } - }, timeout + 1000).unref(); - - if (sessionContext.state === 0) return; - - const sessionContextString = await session.serialize(sessionContext); - await cache.set( - `session_${sessionContext.sessionId}`, - sessionContextString, - Math.ceil(timeout / 1000) + 3 - ); + emitter.once(event, (...args: unknown[]) => { + clearTimeout(timer); + resolve(args); + }); }); } -export function onClientError(err: Error, socket: Socket): void { - const remoteAddress = getSocketEndpoints(socket).remoteAddress; - localCache - .getCurrentSnapshot() - .then((cacheSnapshot) => { - const debugEnabled = !!localCache.getConfig( - cacheSnapshot, - "cwmp.debug", - { - remoteAddress: remoteAddress, - }, - Date.now(), - (e) => { - if (Array.isArray(e) && e[0] === "FUNC" && e[1] === "REMOTE_ADDRESS") - return remoteAddress; - return e; - } - ); +function setTimeoutPromise(delay: number, ref = true): Promise { + return new Promise((resolve) => { + const timerId = setTimeout(resolve, delay); + if (!ref) timerId.unref(); + }); +} - if (debugEnabled) debug.clientError(remoteAddress, err); - }) - .catch((err) => { - throw err; +// When socket closes, store active sessions in cache +export async function onConnection(socket: Socket): Promise { + await once(socket, "close", 300000); + const sessionContext = currentSessions.get(socket); + if (!sessionContext) return; + currentSessions.delete(socket); + if (sessionContext.authState !== 2) { + logger.accessError({ + message: "Authentication failure", + sessionContext: sessionContext, }); + return; + } + + const now = Date.now(); + + const lastActivity = sessionContext.lastActivity; + const timeoutMsg = { + sessionContext: sessionContext, + message: "Session timeout", + sessionTimestamp: sessionContext.timestamp, + }; + + const timeout = + sessionContext.lastActivity + sessionContext.timeout * 1000 - now; + + if (timeout <= 0) { + logger.accessError(timeoutMsg); + // TODO it's possible that lock would have already been expired + await endSession(sessionContext); + return; + } + + await cache.set( + `session_${sessionContext.sessionId}`, + await session.serialize(sessionContext), + Math.ceil(timeout / 1000) + 3 + ); + + await setTimeoutPromise(timeout + 1000, false); + const sessionStr = await cache.get(`session_${sessionContext.sessionId}`); + if (!sessionStr) return; + + const _sessionContext = await session.deserialize(sessionStr); + if (_sessionContext.lastActivity === lastActivity) { + logger.accessError(timeoutMsg); + await endSession(sessionContext); + } +} + +export async function onClientError(err: Error, socket: Socket): Promise { + const remoteAddress = getSocketEndpoints(socket).remoteAddress; + const cacheSnapshot = await localCache.getCurrentSnapshot(); + const debugEnabled = !!localCache.getConfig( + cacheSnapshot, + "cwmp.debug", + { + remoteAddress: remoteAddress, + }, + Date.now(), + (e) => { + if (Array.isArray(e) && e[0] === "FUNC" && e[1] === "REMOTE_ADDRESS") + return remoteAddress; + return e; + } + ); + + if (debugEnabled) debug.clientError(remoteAddress, err); } setInterval(() => { @@ -1341,29 +1354,19 @@ async function processRequest( } } -export function listener( +export async function listener( httpRequest: IncomingMessage, httpResponse: ServerResponse -): void { +): Promise { stats.concurrentRequests += 1; - listenerAsync(httpRequest, httpResponse) - .then(() => { - stats.concurrentRequests -= 1; - }) - .catch((err) => { - currentSessions.delete(httpRequest.socket); - stats.concurrentRequests -= 1; - setTimeout(() => { - throw err; - }); - try { - httpRequest.socket.unref(); - httpResponse.writeHead(500, { Connection: "close" }); - httpResponse.end(`${err.name}: ${err.message}`); - } catch (err) { - // Ignore - } - }); + try { + await listenerAsync(httpRequest, httpResponse); + } catch (err) { + currentSessions.delete(httpRequest.socket); + throw err; + } finally { + stats.concurrentRequests -= 1; + } } async function clientError( diff --git a/lib/db.ts b/lib/db.ts index 94d7959..6f2f67d 100644 --- a/lib/db.ts +++ b/lib/db.ts @@ -89,7 +89,7 @@ onConnect(async (db) => { }); export async function disconnect(): Promise { - if (clientPromise) await (await clientPromise).close(); + if (clientPromise != null) await (await clientPromise).close(); } // Optimize projection by removing overlaps diff --git a/lib/nbi.ts b/lib/nbi.ts index c4cfe37..a0ca92f 100644 --- a/lib/nbi.ts +++ b/lib/nbi.ts @@ -66,10 +66,10 @@ async function getBody(request: IncomingMessage): Promise { return Buffer.concat(chunks); } -export function listener( +export async function listener( request: IncomingMessage, response: ServerResponse -): void { +): Promise { response.setHeader("GenieACS-Version", VERSION); const origin = getRequestOrigin(request); @@ -78,32 +78,17 @@ export function listener( (origin.encrypted ? "https://" : "http://") + origin.host ); - let readableEnded = false; - // For Node 12.9+ we can just use stream.readableEnded - request.on("end", () => { - readableEnded = true; - }); + const body = await getBody(request).catch(() => null); + // Ignore incomplete requests + if (body == null) return; - getBody(request) - .then((body) => { - logger.accessInfo( - Object.assign({}, Object.fromEntries(url.searchParams), { - remoteAddress: origin.remoteAddress, - message: `${request.method} ${url.pathname}`, - }) - ); - return handler(request, response, url, body); + logger.accessInfo( + Object.assign({}, Object.fromEntries(url.searchParams), { + remoteAddress: origin.remoteAddress, + message: `${request.method} ${url.pathname}`, }) - .catch((err) => { - // Ignore incomplete requests - if (!readableEnded) return; - - if (!response.headersSent) { - response.writeHead(500, { Connection: "close" }); - response.end(`${err.name}: ${err.message}`); - } - throw err; - }); + ); + return handler(request, response, url, body); } async function handler( diff --git a/lib/sandbox.ts b/lib/sandbox.ts index 58f36db..311b652 100644 --- a/lib/sandbox.ts +++ b/lib/sandbox.ts @@ -50,7 +50,7 @@ function runExtension(sessionContext, key, extCall): Promise { } let prom = re.get(key); - if (!prom) { + if (prom == null) { re.set( key, (prom = new Promise((resolve, reject) => { diff --git a/lib/server.ts b/lib/server.ts index 47d1d2f..667a4bd 100644 --- a/lib/server.ts +++ b/lib/server.ts @@ -25,7 +25,8 @@ import * as path from "path"; import { ROOT_DIR } from "./config"; let server: http.Server | https.Server; -let listener: (...args) => void; +let listener: http.RequestListener; +let stopping = false; function closeServer(timeout, callback): void { if (!server) return void callback(); @@ -73,11 +74,29 @@ interface SocketEndpoint { // Save this info as they're not accessible after a socket has been closed const socketEndpoints: WeakMap = new WeakMap(); +type Promisify any> = ( + ...args: Parameters +) => Promise>; + export function start( options: ServerOptions, - _listener: http.RequestListener + _listener: Promisify ): void { - listener = _listener; + listener = (req, res) => { + if (stopping) res.setHeader("Connection", "close"); + _listener(req, res).catch((err) => { + try { + res.socket.unref(); + if (res.headersSent) { + res.writeHead(500, { Connection: "close" }); + res.end(`${err.name}: ${err.message}`); + } + } catch (err) { + // Ignore + } + throw err; + }); + }; if (options.ssl) { const opts = { @@ -124,7 +143,8 @@ export function start( server.listen({ port: options.port, host: options.host }); } -export function stop(): Promise { +export function stop(terminateConnections = true): Promise { + stopping = terminateConnections; return new Promise((resolve, reject) => { setTimeout(() => { reject(new Error("Could not close server in a timely manner")); diff --git a/lib/ui.ts b/lib/ui.ts index b71b040..588a0aa 100644 --- a/lib/ui.ts +++ b/lib/ui.ts @@ -59,7 +59,7 @@ const getAuthorizer = memoize( } ); -koa.on("error", async (err, ctx) => { +koa.on("error", (err, ctx) => { setTimeout(() => { // Ignored errors resulting from aborted requests if (ctx?.req.aborted) return;