neo4j
diff --git a/‎packages/bolt-connection/src/bolt/bolt-protocol-v1.js
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions b/‎packages/bolt-connection/src/bolt/bolt-protocol-v1.js
Copy file name to clipboard
+2Lines changed: 2 additions & 0 deletions
diff --git a/‎packages/bolt-connection/src/connection-provider/authentication-provider.js
Copy file name to clipboard
+66Lines changed: 66 additions & 0 deletions b/‎packages/bolt-connection/src/connection-provider/authentication-provider.js
Copy file name to clipboard
+66Lines changed: 66 additions & 0 deletions
diff --git a/‎packages/bolt-connection/src/connection-provider/connection-provider-direct.js
Copy file name to clipboard
+39-15Lines changed: 39 additions & 15 deletions b/‎packages/bolt-connection/src/connection-provider/connection-provider-direct.js
Copy file name to clipboard
+39-15Lines changed: 39 additions & 15 deletions
diff --git a/‎packages/bolt-connection/src/connection-provider/connection-provider-pooled.js
Copy file name to clipboard
+104-11Lines changed: 104 additions & 11 deletions b/‎packages/bolt-connection/src/connection-provider/connection-provider-pooled.js
Copy file name to clipboard
+104-11Lines changed: 104 additions & 11 deletions
@@ -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.'
 
@@ -0,0 +1,66 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [https://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
+ *
+ *     https://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)
+    }
+  }
+}
@@ -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 })
   }
 
@@ -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
+  }
 }