Re-authentication (#1050)

**⚠️ This API is released as preview.**

This changes introduce two ways of changing the connection credentials in a driver instance, each of them solving a different use case. 

### Token Expiration / Change Credentials for the whole driver instance

This use case is related to the issue #993 in the repository. For solving this, the driver is now able to receive a `AuthTokenManager` in the driver creation. This interface enables the user code provide new auth tokens to the driver and be notified by token expiration failures. 

For simplifying the usage, the driver also provides a default implementation of `AuthTokenManager` which can be created with `neo4j.temporalAuthDataManager` and receives a function for renewing the auth token as parameters. 

Example:

```typescript
import neo4j, { AuthToken } from 'neo4j-driver'

/**
 * Method called whenever the driver needs to refresh the token.
 *
 * The refresh will happen if the driver is notified by the server
 * about a token expiration or if the `Date.now() > tokenData.expiry`
 *
 * Important, the driver will block all the connections creation until
 * this function resolves the new auth token.
 */
async function fetchAuthTokenFromMyProvider () {
   const bearer: string = await myProvider.getBearerToken()
   const token: AuthToken = neo4j.auth.bearer(bearer)
   const expiration: Date = myProvider.getExpiryDate()  
   return {
      token,
      // if expiration is not provided, 
      // the driver will only fetch a new token when a failure happens
      expiration 
   }
}

const driver = neo4j.driver(
    'neo4j://localhost:7687', 
    neo4j.expirationBasedAuthTokenManager({ 
        getAuthData: fetchAuthTokenFromMyProvider 
    })
)
```

### User Switching

In this scenario, different credentials can be configured in a session providing a way for change the user context for the session. For using this feature, it needed to check if your server supports session auth by calling `driver.supportsSessionAuth()`.

Example:

```typescript
import neo4j from 'neo4j-driver'


const driver =  neo4j.driver(
    'neo4j://localhost:7687', 
    neo4j.auth.basic('neo4j', 'password')
)


const sessionWithUserB = driver.session({
  database: 'neo4j',
  auth: neo4j.auth.basic('userB', 'userBpassword')
})


try {
  // run some queries as userB
  const result = await sessionWithUserB.executeRead(tx => tx.run('RETURN 1'))
} finally {
  // close the session as usual
  await sessionWithUserB.close()
}
```

**⚠️ This API is released as preview.**

Author: bigmontz

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

@@ -205,6 +205,7 @@ export default class BoltProtocol {
       onError: onError
     })
 
+    // TODO: Verify the Neo4j version in the message
     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.'
@@ -233,6 +234,7 @@ export default class BoltProtocol {
       onError: (error) => this._onLoginError(error, onError)
     })
 
+    // TODO: Verify the Neo4j version in the message
     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.'

packages/bolt-connection/src/connection-provider/authentication-provider.js

@@ -0,0 +1,66 @@
+/**
+ * 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 { expirationBasedAuthTokenManager } from 'neo4j-driver-core'
+import { object } from '../lang'
+
+/**
+ * Class which provides Authorization for {@link Connection}
+ */
+export default class AuthenticationProvider {
+  constructor ({ authTokenManager, userAgent }) {
+    this._authTokenManager = authTokenManager || expirationBasedAuthTokenManager({
+      tokenProvider: () => {}
+    })
+    this._userAgent = userAgent
+  }
+
+  async authenticate ({ connection, auth, skipReAuth, waitReAuth, forceReAuth }) {
+    if (auth != null) {
+      const shouldReAuth = connection.supportsReAuth === true && (
+        (!object.equals(connection.authToken, auth) && skipReAuth !== true) ||
+        forceReAuth === true
+      )
+      if (connection.authToken == null || shouldReAuth) {
+        return await connection.connect(this._userAgent, auth, waitReAuth || false)
+      }
+      return connection
+    }
+
+    const authToken = await this._authTokenManager.getToken()
+
+    if (!object.equals(authToken, connection.authToken)) {
+      return await connection.connect(this._userAgent, authToken)
+    }
+
+    return connection
+  }
+
+  handleError ({ connection, code }) {
+    if (
+      connection &&
+      [
+        'Neo.ClientError.Security.Unauthorized',
+        'Neo.ClientError.Security.TokenExpired'
+      ].includes(code)
+    ) {
+      this._authTokenManager.onTokenExpired(connection.authToken)
+    }
+  }
+}

packages/bolt-connection/src/connection-provider/connection-provider-direct.js

@@ -26,14 +26,19 @@ import {
 import { internal, error } from 'neo4j-driver-core'
 
 const {
-  constants: { BOLT_PROTOCOL_V3, BOLT_PROTOCOL_V4_0, BOLT_PROTOCOL_V4_4 }
+  constants: {
+    BOLT_PROTOCOL_V3,
+    BOLT_PROTOCOL_V4_0,
+    BOLT_PROTOCOL_V4_4,
+    BOLT_PROTOCOL_V5_1
+  }
 } = internal
 
 const { SERVICE_UNAVAILABLE } = error
 
 export default class DirectConnectionProvider extends PooledConnectionProvider {
-  constructor ({ id, config, log, address, userAgent, authToken }) {
-    super({ id, config, log, userAgent, authToken })
+  constructor ({ id, config, log, address, userAgent, authTokenManager, newPool }) {
+    super({ id, config, log, userAgent, authTokenManager, newPool })
 
     this._address = address
   }
@@ -42,27 +47,33 @@ export default class DirectConnectionProvider extends PooledConnectionProvider {
    * See {@link ConnectionProvider} for more information about this method and
    * its arguments.
    */
-  acquireConnection ({ accessMode, database, bookmarks } = {}) {
+  async acquireConnection ({ accessMode, database, bookmarks, auth, forceReAuth } = {}) {
     const databaseSpecificErrorHandler = ConnectionErrorHandler.create({
       errorCode: SERVICE_UNAVAILABLE,
-      handleAuthorizationExpired: (error, address) =>
-        this._handleAuthorizationExpired(error, address, database)
+      handleAuthorizationExpired: (error, address, conn) =>
+        this._handleAuthorizationExpired(error, address, conn, database)
     })
 
-    return this._connectionPool
-      .acquire(this._address)
-      .then(
-        connection =>
-          new DelegateConnection(connection, databaseSpecificErrorHandler)
-      )
+    const connection = await this._connectionPool.acquire({ auth, forceReAuth }, this._address)
+
+    if (auth) {
+      await this._verifyStickyConnection({
+        auth,
+        connection,
+        address: this._address
+      })
+      return connection
+    }
+
+    return new DelegateConnection(connection, databaseSpecificErrorHandler)
   }
 
-  _handleAuthorizationExpired (error, address, database) {
+  _handleAuthorizationExpired (error, address, connection, database) {
     this._log.warn(
       `Direct driver ${this._id} will close connection to ${address} for database '${database}' because of an error ${error.code} '${error.message}'`
     )
-    this._connectionPool.purge(address).catch(() => {})
-    return error
+
+    return super._handleAuthorizationExpired(error, address, connection)
   }
 
   async _hasProtocolVersion (versionPredicate) {
@@ -111,6 +122,19 @@ export default class DirectConnectionProvider extends PooledConnectionProvider {
     )
   }
 
+  async supportsSessionAuth () {
+    return await this._hasProtocolVersion(
+      version => version >= BOLT_PROTOCOL_V5_1
+    )
+  }
+
+  async verifyAuthentication ({ auth }) {
+    return this._verifyAuthentication({
+      auth,
+      getAddress: () => this._address
+    })
+  }
+
   async verifyConnectivityAndGetServerInfo () {
     return await this._verifyConnectivityAndGetServerVersion({ address: this._address })
   }

packages/bolt-connection/src/connection-provider/connection-provider-pooled.js

@@ -19,21 +19,30 @@
 
 import { createChannelConnection, ConnectionErrorHandler } from '../connection'
 import Pool, { PoolConfig } from '../pool'
-import { error, ConnectionProvider, ServerInfo } from 'neo4j-driver-core'
+import { error, ConnectionProvider, ServerInfo, newError, isStaticAuthTokenManger } from 'neo4j-driver-core'
+import AuthenticationProvider from './authentication-provider'
+import { object } from '../lang'
 
 const { SERVICE_UNAVAILABLE } = error
+const AUTHENTICATION_ERRORS = [
+  'Neo.ClientError.Security.CredentialsExpired',
+  'Neo.ClientError.Security.Forbidden',
+  'Neo.ClientError.Security.TokenExpired',
+  'Neo.ClientError.Security.Unauthorized'
+]
+
 export default class PooledConnectionProvider extends ConnectionProvider {
   constructor (
-    { id, config, log, userAgent, authToken },
+    { id, config, log, userAgent, authTokenManager, newPool = (...args) => new Pool(...args) },
     createChannelConnectionHook = null
   ) {
     super()
 
     this._id = id
     this._config = config
     this._log = log
-    this._userAgent = userAgent
-    this._authToken = authToken
+    this._authTokenManager = authTokenManager
+    this._authenticationProvider = new AuthenticationProvider({ authTokenManager, userAgent })
     this._createChannelConnection =
       createChannelConnectionHook ||
       (address => {
@@ -44,10 +53,11 @@ export default class PooledConnectionProvider extends ConnectionProvider {
           this._log
         )
       })
-    this._connectionPool = new Pool({
+    this._connectionPool = newPool({
       create: this._createConnection.bind(this),
       destroy: this._destroyConnection.bind(this),
-      validate: this._validateConnection.bind(this),
+      validateOnAcquire: this._validateConnectionOnAcquire.bind(this),
+      validateOnRelease: this._validateConnectionOnRelease.bind(this),
       installIdleObserver: PooledConnectionProvider._installIdleObserverOnConnection.bind(
         this
       ),
@@ -57,6 +67,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
       config: PoolConfig.fromDriverConfig(config),
       log: this._log
     })
+    this._userAgent = userAgent
     this._openConnections = {}
   }
 
@@ -69,14 +80,13 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    * @return {Promise<Connection>} promise resolved with a new connection or rejected when failed to connect.
    * @access private
    */
-  _createConnection (address, release) {
+  _createConnection ({ auth }, address, release) {
     return this._createChannelConnection(address).then(connection => {
       connection._release = () => {
         return release(address, connection)
       }
       this._openConnections[connection.id] = connection
-      return connection
-        .connect(this._userAgent, this._authToken)
+      return this._authenticationProvider.authenticate({ connection, auth })
         .catch(error => {
           // let's destroy this connection
           this._destroyConnection(connection)
@@ -86,6 +96,26 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     })
   }
 
+  async _validateConnectionOnAcquire ({ auth, skipReAuth }, conn) {
+    if (!this._validateConnection(conn)) {
+      return false
+    }
+
+    try {
+      await this._authenticationProvider.authenticate({ connection: conn, auth, skipReAuth })
+      return true
+    } catch (error) {
+      this._log.debug(
+        `The connection ${conn.id} is not valid because of an error ${error.code} '${error.message}'`
+      )
+      return false
+    }
+  }
+
+  _validateConnectionOnRelease (conn) {
+    return conn._sticky !== true && this._validateConnection(conn)
+  }
+
   /**
    * Check that a connection is usable
    * @return {boolean} true if the connection is open
@@ -98,7 +128,11 @@ export default class PooledConnectionProvider extends ConnectionProvider {
 
     const maxConnectionLifetime = this._config.maxConnectionLifetime
     const lifetime = Date.now() - conn.creationTimestamp
-    return lifetime <= maxConnectionLifetime
+    if (lifetime > maxConnectionLifetime) {
+      return false
+    }
+
+    return true
   }
 
   /**
@@ -118,7 +152,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    * @return {Promise<ServerInfo>} the server info
    */
   async _verifyConnectivityAndGetServerVersion ({ address }) {
-    const connection = await this._connectionPool.acquire(address)
+    const connection = await this._connectionPool.acquire({}, address)
     const serverInfo = new ServerInfo(connection.server, connection.protocol().version)
     try {
       if (!connection.protocol().isLastMessageLogon()) {
@@ -130,6 +164,47 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     return serverInfo
   }
 
+  async _verifyAuthentication ({ getAddress, auth }) {
+    const connectionsToRelease = []
+    try {
+      const address = await getAddress()
+      const connection = await this._connectionPool.acquire({ auth, skipReAuth: true }, address)
+      connectionsToRelease.push(connection)
+
+      const lastMessageIsNotLogin = !connection.protocol().isLastMessageLogon()
+
+      if (!connection.supportsReAuth) {
+        throw newError('Driver is connected to a database that does not support user switch.')
+      }
+      if (lastMessageIsNotLogin && connection.supportsReAuth) {
+        await this._authenticationProvider.authenticate({ connection, auth, waitReAuth: true, forceReAuth: true })
+      } else if (lastMessageIsNotLogin && !connection.supportsReAuth) {
+        const stickyConnection = await this._connectionPool.acquire({ auth }, address, { requireNew: true })
+        stickyConnection._sticky = true
+        connectionsToRelease.push(stickyConnection)
+      }
+      return true
+    } catch (error) {
+      if (AUTHENTICATION_ERRORS.includes(error.code)) {
+        return false
+      }
+      throw error
+    } finally {
+      await Promise.all(connectionsToRelease.map(conn => conn._release()))
+    }
+  }
+
+  async _verifyStickyConnection ({ auth, connection, address }) {
+    const connectionWithSameCredentials = object.equals(auth, connection.authToken)
+    const shouldCreateStickyConnection = !connectionWithSameCredentials
+    connection._sticky = connectionWithSameCredentials && !connection.supportsReAuth
+
+    if (shouldCreateStickyConnection || connection._sticky) {
+      await connection._release()
+      throw newError('Driver is connected to a database that does not support user switch.')
+    }
+  }
+
   async close () {
     // purge all idle connections in the connection pool
     await this._connectionPool.close()
@@ -146,4 +221,22 @@ export default class PooledConnectionProvider extends ConnectionProvider {
   static _removeIdleObserverOnConnection (conn) {
     conn._updateCurrentObserver()
   }
+
+  _handleAuthorizationExpired (error, address, connection) {
+    this._authenticationProvider.handleError({ connection, code: error.code })
+
+    if (error.code === 'Neo.ClientError.Security.AuthorizationExpired') {
+      this._connectionPool.apply(address, (conn) => { conn.authToken = null })
+    }
+
+    if (connection) {
+      connection.close().catch(() => undefined)
+    }
+
+    if (error.code === 'Neo.ClientError.Security.TokenExpired' && !isStaticAuthTokenManger(this._authTokenManager)) {
+      error.retriable = true
+    }
+
+    return error
+  }
 }