Merge branch 'master' into breaking-changes

Author: mikuso

README.md

@@ -195,6 +195,7 @@ await cli.connect();
   * [Event: 'error'](#event-error)
   * [Event: 'close'](#event-close)
   * [Event: 'closing'](#event-closing)
+  * [Event: 'upgradeAborted'](#event-upgradeaborted)
   * [server.auth(callback)](#serverauthcallback)
   * [server.handleUpgrade(request)](#serverhandleupgraderequest-socket-head)
   * [server.reconfigure(options)](#serverreconfigureoptions)
@@ -276,6 +277,16 @@ Emitted when the server has fully closed and all clients have been disconnected.
 
 Emitted when the server has begun closing. Beyond this point, no more clients will be accepted and the `'client'` event will no longer fire.
 
+#### Event: 'upgradeAborted'
+
+* `event` {Object}
+  * `error` {Error} - The cause of the abort.
+  * `socket` {net.Socket} - Network socket between the server and client
+  * `request` {http.IncomingMessage} - The full HTTP request received by the underlying webserver.
+  * `identity` {String} - The identity portion of the connection URL, decoded.
+
+Emitted when a websocket upgrade has been aborted. This could be caused by an authentication rejection, socket error or websocket handshake error.
+
 #### server.auth(callback)
 
 * `callback` {Function}
@@ -319,7 +330,7 @@ rpcServer.auth((accept, reject, handshake) => {
 #### server.handleUpgrade(request, socket, head)
 
 * `request` {http.IncomingMessage}
-* `socket` {stream.Duplex} - Network socket between the server and client
+* `socket` {net.Socket} - Network socket between the server and client
 * `head` {Buffer} - The first packet of the upgraded stream (may be empty)
 
 Converts an HTTP upgrade request into a WebSocket client to be handled by this RPCServer. This method is bound to the server instance, so it is suitable to pass directly as an `http.Server`'s `'upgrade'` event handler.

lib/errors.js

@@ -63,7 +63,15 @@ class RPCFrameworkError extends RPCError {
     rpcErrorCode = 'RpcFrameworkError';
 };
 
+class WebsocketUpgradeError extends Error {
+    constructor(code, message) {
+        super(message);
+        this.code = code;
+    }
+}
+
 module.exports = {
+    WebsocketUpgradeError,
     TimeoutError,
     UnexpectedHttpResponse,
     RPCError,

lib/server.js

@@ -5,6 +5,7 @@ const RPCServerClient = require('./server-client');
 const { abortHandshake, parseSubprotocols } = require('./ws-util');
 const standardValidators = require('./standard-validators');
 const { getPackageIdent } = require('./util');
+const { WebsocketUpgradeError } = require('./errors');
 
 class RPCServer extends EventEmitter {
     constructor(options) {
@@ -78,24 +79,50 @@ class RPCServer extends EventEmitter {
 
     get handleUpgrade() {
         return async (request, socket, head) => {
-            try {
-                if (this._state !== OPEN) {
-                    throw Error("Server not open");
+
+            let resolved = false;
+            let emittedAbort = false;
+
+            const url = new URL(request.url, 'http://localhost');
+            const pathParts = url.pathname.split('/');
+            const identity = decodeURIComponent(pathParts.pop());
+
+            const abortUpgrade = (error) => {
+                resolved = true;
+
+                if (error && error instanceof WebsocketUpgradeError) {
+                    abortHandshake(socket, error.code, error.message);
+                } else {
+                    abortHandshake(socket, 500);
                 }
 
-                if (request.headers.upgrade.toLowerCase() !== 'websocket') {
-                    throw Error("Can only upgrade websocket upgrade requests");
+                if (!emittedAbort) {
+                    emittedAbort = true;
+                    this.emit('upgradeAborted', {
+                        error,
+                        socket,
+                        request,
+                        identity,
+                    });
                 }
+            };
 
-                const remoteAddress = request.socket.remoteAddress;
+            try {
+                if (this._state !== OPEN) {
+                    throw new WebsocketUpgradeError(500, "Server not open");
+                }
+                
                 const headers = request.headers;
+                
+                if (headers.upgrade.toLowerCase() !== 'websocket') {
+                    throw new WebsocketUpgradeError(400, "Can only upgrade websocket upgrade requests");
+                }
+                
+                const endpoint = pathParts.join('/');
+                const remoteAddress = request.socket.remoteAddress;
                 const protocols = ('sec-websocket-protocol' in request.headers)
                     ? parseSubprotocols(request.headers['sec-websocket-protocol'])
                     : new Set();
-                const url = new URL(request.url, 'http://localhost');
-                const pathParts = url.pathname.split('/');
-                const identity = decodeURIComponent(pathParts.pop());
-                const endpoint = pathParts.join('/');
 
                 let password;
                 if (headers.authorization) {
@@ -130,8 +157,6 @@ class RPCServer extends EventEmitter {
                     password,
                 };
 
-                let resolved = false;
-
                 const accept = (session, protocol) => {
                     if (resolved) return;
                     resolved = true;
@@ -140,7 +165,8 @@ class RPCServer extends EventEmitter {
                         // pick first subprotocol (preferred by server) that is also supported by the client
                         protocol = (this._options.protocols ?? []).find(p => protocols.has(p));
                     } else if (protocol !== false && !protocols.has(protocol)) {
-                        return abortHandshake(socket, 400, `Client doesn't support expected subprotocol`);
+                        abortUpgrade(new WebsocketUpgradeError(400, `Client doesn't support expected subprotocol`));
+                        return;
                     }
 
                     // cache auth results for connection creation
@@ -155,14 +181,18 @@ class RPCServer extends EventEmitter {
                     });
                 };
 
-                const reject = (code = 404, message) => {
+                const reject = (code = 404, message = 'Not found') => {
                     if (resolved) return;
                     resolved = true;
-                    abortHandshake(socket, code, message);
+                    abortUpgrade(new WebsocketUpgradeError(code, message));
                 };
 
-                socket.on('error', () => {
-                    reject();
+                socket.once('close', () => {
+                    reject(400, `Client connection closed before upgrade complete`);
+                });
+
+                socket.on('error', (err) => {
+                    abortUpgrade(err);
                 });
 
                 if (this.authCallback) {
@@ -176,7 +206,7 @@ class RPCServer extends EventEmitter {
                 }
 
             } catch (err) {
-                abortHandshake(socket, 500);
+                abortUpgrade(err);
             }
         };
     }

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "ocpp-rpc",
-  "version": "1.7.4",
+  "version": "1.8.0",
   "description": "A client & server implementation of the WAMP-like RPC-over-websocket system defined in the OCPP protcols (e.g. OCPP1.6-J and OCPP2.0.1).",
   "main": "index.js",
   "scripts": {

package.json

@@ -2,7 +2,7 @@ const assert = require('assert/strict');
 const http = require('http');
 const { once } = require('events');
 const RPCClient = require("../lib/client");
-const { TimeoutError, UnexpectedHttpResponse } = require('../lib/errors');
+const { TimeoutError, UnexpectedHttpResponse, WebsocketUpgradeError } = require('../lib/errors');
 const RPCServer = require("../lib/server");
 const { setTimeout } = require('timers/promises');
 const { createValidator } = require('../lib/validator');
@@ -823,8 +823,14 @@ describe('RPCServer', function(){
 
         it("should abort handshake if server not open", async () => {
 
-            let authed = false;
             const server = new RPCServer();
+            
+            let abortEvent;
+            server.on('upgradeAborted', event => {
+                abortEvent = event;
+            });
+            
+            let authed = false;
             server.auth((accept) => {
                 // shouldn't get this far
                 authed = true;
@@ -853,6 +859,7 @@ describe('RPCServer', function(){
             await server.close();
             assert.doesNotReject(server.handleUpgrade(...upgrade));
             assert.equal(authed, false);
+            assert.equal(abortEvent.error.code, 500);
             httpServer.close();
 
         });
@@ -862,6 +869,11 @@ describe('RPCServer', function(){
 
             const {endpoint, close, server} = await createServer();
 
+            let abortEvent;
+            server.on('upgradeAborted', event => {
+                abortEvent = event;
+            });
+
             let authed = false;
             server.auth((accept) => {
                 // shouldn't get this far
@@ -875,14 +887,49 @@ describe('RPCServer', function(){
                     headers: {
                         connection: 'Upgrade',
                         upgrade: '_UNKNOWN_',
+                        'user-agent': 'test/0',
                     }
                 });
                 req.end();
 
                 const [res] = await once(req, 'response');
 
-                assert.equal(res.statusCode, 500);
+                assert.equal(res.statusCode, 400);
                 assert.equal(authed, false);
+                assert.ok(abortEvent.error instanceof WebsocketUpgradeError);
+                assert.equal(abortEvent.request.headers['user-agent'], 'test/0');
+                
+            } finally {
+                close();
+            }
+
+        });
+        
+
+        it("should emit upgradeAborted event on auth reject", async () => {
+
+            const {endpoint, close, server} = await createServer();
+
+            let abortEvent;
+            server.on('upgradeAborted', event => {
+                abortEvent = event;
+            });
+
+            server.auth((accept, reject) => {
+                reject(499);
+            });
+
+            try {
+
+                const cli = new RPCClient({
+                    endpoint,
+                    identity: 'X'
+                });
+
+                await assert.rejects(cli.connect());
+                
+                assert.ok(abortEvent.error instanceof WebsocketUpgradeError);
+                assert.equal(abortEvent.error.code, 499);
 
             } finally {
                 close();