Introduce Bolt Protocol 5.1 (#1065)

This version of the protocol introduces two new messages for manage credentials.

* LOGIN with a given auth token.
* LOGOFF of the current auth token.

The HELLO message is not more responsible for negotiating credentials in this version of the protocol.

Author: bigmontz

packages/bolt-connection/src/bolt/bolt-protocol-v1.js

@@ -24,15 +24,16 @@ import {
 // eslint-disable-next-line no-unused-vars
 import { Chunker } from '../channel'
 import { structure, v1 } from '../packstream'
-import RequestMessage from './request-message'
+import RequestMessage, { SIGNATURES } from './request-message'
 import {
   LoginObserver,
+  LogoffObserver,
   ResetObserver,
   ResultStreamObserver,
   // eslint-disable-next-line no-unused-vars
   StreamObserver
 } from './stream-observers'
-import { internal } from 'neo4j-driver-core'
+import { internal, newError } from 'neo4j-driver-core'
 import transformersFactories from './bolt-protocol-v1.transformer'
 import Transformer from './transformer'
 
@@ -99,6 +100,27 @@ export default class BoltProtocol {
     return BOLT_PROTOCOL_V1
   }
 
+  /**
+   * @property {boolean} supportsReAuth Either if the protocol version supports re-auth or not.
+   */
+  get supportsReAuth () {
+    return false
+  }
+
+  /**
+   * @property {boolean} initialized Either if the protocol was initialized or not
+   */
+  get initialized () {
+    return !!this._initialized
+  }
+
+  /**
+   * @property {object} authToken The token used in the last login
+   */
+  get authToken () {
+    return this._authToken
+  }
+
   /**
    * Get the packer.
    * @return {Packer} the protocol's packer.
@@ -162,6 +184,61 @@ export default class BoltProtocol {
     return observer
   }
 
+  /**
+   * Performs logoff of the underlying connection
+   *
+   * @param {Object} param
+   * @param {function(err: Error)} param.onError the callback to invoke on error.
+   * @param {function()} param.onComplete the callback to invoke on completion.
+   * @param {boolean} param.flush whether to flush the buffered messages.
+   *
+   * @returns {StreamObserver} the stream observer that monitors the corresponding server response.
+   */
+  logoff ({ onComplete, onError, flush } = {}) {
+    const observer = new LogoffObserver({
+      onCompleted: onComplete,
+      onError: onError
+    })
+
+    const error = newError(
+      'Driver is connected to a database that does not support logoff. ' +
+        'Please upgrade to Neo4j 5.5.0 or later in order to use this functionality.'
+    )
+
+    // unsupported API was used, consider this a fatal error for the current connection
+    this._onProtocolError(error.message)
+    observer.onError(error)
+    throw error
+  }
+
+  /**
+   * Performs login of the underlying connection
+   *
+   * @param {Object} args
+   * @param {Object} args.authToken the authentication token.
+   * @param {function(err: Error)} args.onError the callback to invoke on error.
+   * @param {function()} args.onComplete the callback to invoke on completion.
+   * @param {boolean} args.flush whether to flush the buffered messages.
+   *
+   * @returns {StreamObserver} the stream observer that monitors the corresponding server response.
+   */
+  logon ({ authToken, onComplete, onError, flush } = {}) {
+    const observer = new LoginObserver({
+      onCompleted: () => this._onLoginCompleted({}, authToken, onComplete),
+      onError: (error) => this._onLoginError(error, onError)
+    })
+
+    const error = newError(
+      'Driver is connected to a database that does not support logon. ' +
+        'Please upgrade to Neo4j 5.5.0 or later in order to use this functionality.'
+    )
+
+    // unsupported API was used, consider this a fatal error for the current connection
+    this._onProtocolError(error.message)
+    observer.onError(error)
+    throw error
+  }
+
   /**
    * Perform protocol related operations for closing this connection
    */
@@ -391,19 +468,19 @@ export default class BoltProtocol {
       this.packable(messageStruct)()
 
       this._chunker.messageBoundary()
-
       if (flush) {
         this._chunker.flush()
       }
     }
   }
 
-  isLastMessageLogin () {
-    return this._lastMessageSignature === 0x01
+  isLastMessageLogon () {
+    return this._lastMessageSignature === SIGNATURES.HELLO ||
+      this._lastMessageSignature === SIGNATURES.LOGON
   }
 
   isLastMessageReset () {
-    return this._lastMessageSignature === 0x0f
+    return this._lastMessageSignature === SIGNATURES.RESET
   }
 
   /**
@@ -472,7 +549,9 @@ export default class BoltProtocol {
     this._responseHandler._resetFailure()
   }
 
-  _onLoginCompleted (metadata, onCompleted) {
+  _onLoginCompleted (metadata, authToken, onCompleted) {
+    this._initialized = true
+    this._authToken = authToken
     if (metadata) {
       const serverVersion = metadata.server
       if (!this._server.version) {

packages/bolt-connection/src/bolt/bolt-protocol-v3.js

@@ -72,7 +72,7 @@ export default class BoltProtocol extends BoltProtocolV2 {
   initialize ({ userAgent, authToken, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
-      onCompleted: metadata => this._onLoginCompleted(metadata, onComplete)
+      onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
     this.write(RequestMessage.hello(userAgent, authToken), observer, true)

packages/bolt-connection/src/bolt/bolt-protocol-v4x1.js

@@ -75,7 +75,7 @@ export default class BoltProtocol extends BoltProtocolV4 {
   initialize ({ userAgent, authToken, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
-      onCompleted: metadata => this._onLoginCompleted(metadata, onComplete)
+      onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
     this.write(

packages/bolt-connection/src/bolt/bolt-protocol-v4x3.js

@@ -94,7 +94,7 @@ export default class BoltProtocol extends BoltProtocolV42 {
         if (metadata.patch_bolt !== undefined) {
           this._applyPatches(metadata.patch_bolt)
         }
-        return this._onLoginCompleted(metadata, onComplete)
+        return this._onLoginCompleted(metadata, authToken, onComplete)
       }
     })
 

packages/bolt-connection/src/bolt/bolt-protocol-v5x0.js

@@ -54,7 +54,7 @@ export default class BoltProtocol extends BoltProtocolV44 {
   initialize ({ userAgent, authToken, onError, onComplete } = {}) {
     const observer = new LoginObserver({
       onError: error => this._onLoginError(error, onError),
-      onCompleted: metadata => this._onLoginCompleted(metadata, onComplete)
+      onCompleted: metadata => this._onLoginCompleted(metadata, authToken, onComplete)
     })
 
     this.write(

packages/bolt-connection/src/bolt/bolt-protocol-v5x1.js

@@ -0,0 +1,132 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import BoltProtocolV5x0 from './bolt-protocol-v5x0'
+
+import transformersFactories from './bolt-protocol-v5x1.transformer'
+import Transformer from './transformer'
+import RequestMessage from './request-message'
+import { LoginObserver, LogoffObserver } from './stream-observers'
+
+import { internal } from 'neo4j-driver-core'
+
+const {
+  constants: { BOLT_PROTOCOL_V5_1 }
+} = internal
+
+export default class BoltProtocol extends BoltProtocolV5x0 {
+  get version () {
+    return BOLT_PROTOCOL_V5_1
+  }
+
+  get transformer () {
+    if (this._transformer === undefined) {
+      this._transformer = new Transformer(Object.values(transformersFactories).map(create => create(this._config, this._log)))
+    }
+    return this._transformer
+  }
+
+  get supportsReAuth () {
+    return true
+  }
+
+  /**
+   * Initialize a connection with the server
+   *
+   * @param {Object} param0 The params
+   * @param {string} param0.userAgent The user agent
+   * @param {any} param0.authToken The auth token
+   * @param {function(error)} param0.onError On error callback
+   * @param {function(onComplte)} param0.onComplete On complete callback
+   * @returns {LoginObserver} The Login observer
+   */
+  initialize ({ userAgent, authToken, onError, onComplete } = {}) {
+    const state = {}
+    const observer = new LoginObserver({
+      onError: error => this._onLoginError(error, onError),
+      onCompleted: metadata => {
+        state.metadata = metadata
+        return this._onLoginCompleted(metadata)
+      }
+    })
+
+    this.write(
+      RequestMessage.hello5x1(userAgent, this._serversideRouting),
+      observer,
+      false
+    )
+
+    return this.logon({
+      authToken,
+      onComplete: metadata => onComplete({ ...metadata, ...state.metadata }),
+      onError,
+      flush: true
+    })
+  }
+
+  /**
+   * Performs login of the underlying connection
+   *
+   * @param {Object} args
+   * @param {Object} args.authToken the authentication token.
+   * @param {function(err: Error)} args.onError the callback to invoke on error.
+   * @param {function()} args.onComplete the callback to invoke on completion.
+   * @param {boolean} args.flush whether to flush the buffered messages.
+   *
+   * @returns {StreamObserver} the stream observer that monitors the corresponding server response.
+   */
+  logon ({ authToken, onComplete, onError, flush } = {}) {
+    const observer = new LoginObserver({
+      onCompleted: () => this._onLoginCompleted(null, authToken, onComplete),
+      onError: (error) => this._onLoginError(error, onError)
+    })
+
+    this.write(
+      RequestMessage.logon(authToken),
+      observer,
+      flush
+    )
+
+    return observer
+  }
+
+  /**
+   * Performs logoff of the underlying connection
+   *
+   * @param {Object} param
+   * @param {function(err: Error)} param.onError the callback to invoke on error.
+   * @param {function()} param.onComplete the callback to invoke on completion.
+   * @param {boolean} param.flush whether to flush the buffered messages.
+   *
+   * @returns {StreamObserver} the stream observer that monitors the corresponding server response.
+  */
+  logoff ({ onComplete, onError, flush } = {}) {
+    const observer = new LogoffObserver({
+      onCompleted: onComplete,
+      onError: onError
+    })
+
+    this.write(
+      RequestMessage.logoff(),
+      observer,
+      flush
+    )
+
+    return observer
+  }
+}

packages/bolt-connection/src/bolt/bolt-protocol-v5x1.transformer.js

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import v5x0 from './bolt-protocol-v5x0.transformer'
+
+export default {
+  ...v5x0
+}

packages/bolt-connection/src/bolt/create.js

@@ -27,6 +27,7 @@ import BoltProtocolV4x2 from './bolt-protocol-v4x2'
 import BoltProtocolV4x3 from './bolt-protocol-v4x3'
 import BoltProtocolV4x4 from './bolt-protocol-v4x4'
 import BoltProtocolV5x0 from './bolt-protocol-v5x0'
+import BoltProtocolV5x1 from './bolt-protocol-v5x1'
 // eslint-disable-next-line no-unused-vars
 import { Chunker, Dechunker } from '../channel'
 import ResponseHandler from './response-handler'
@@ -191,6 +192,16 @@ function createProtocol (
         onProtocolError,
         serversideRouting
       )
+    case 5.1:
+      return new BoltProtocolV5x1(
+        server,
+        chunker,
+        packingConfig,
+        createResponseHandler,
+        log,
+        onProtocolError,
+        serversideRouting
+      )
     default:
       throw newError('Unknown Bolt protocol version: ' + version)
   }

packages/bolt-connection/src/bolt/handshake.js

@@ -76,7 +76,7 @@ function parseNegotiatedResponse (buffer) {
  */
 function newHandshakeBuffer () {
   return createHandshakeMessage([
-    version(5, 0),
+    [version(5, 1), version(5, 0)],
     [version(4, 4), version(4, 2)],
     version(4, 1),
     version(3, 0)