Mark all connections for closure on LDAP Auth Credential expiration (#731)

When an authenticated connection is made by the drivers, the server takes the raw credentials passed in the HELLO message and uses them to generate a token which is then cached and used to authorize all subsequent requests.

In some implementations of auth (principally LDAP), this cached token can expire and the original raw credentials are required to generate a new token. Since these raw credentials are not stored on the server and are only provided during connection initialization, the only way currently to refresh these tokens is by terminating the existing connection and create new connections.

When the driver receives an error with status `Status.Security.AuthorizationExpired` this should have the effect of marking all current connections as stale/invalid, forcing the driver to establish new connections and therefore refreshing the credentials cached on the server. Existing connections which are currently in use should not be interrupted.

Author: bigmontz

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

@@ -18,13 +18,19 @@
  */
 
 import PooledConnectionProvider from './connection-provider-pooled'
-import { createChannelConnection, DelegateConnection } from '../connection'
-import { internal } from 'neo4j-driver-core'
+import {
+  createChannelConnection,
+  DelegateConnection,
+  ConnectionErrorHandler
+} from '../connection'
+import { internal, error } from 'neo4j-driver-core'
 
 const {
   constants: { BOLT_PROTOCOL_V4_0, BOLT_PROTOCOL_V3 }
 } = internal
 
+const { SERVICE_UNAVAILABLE, newError } = error
+
 export default class DirectConnectionProvider extends PooledConnectionProvider {
   constructor ({ id, config, log, address, userAgent, authToken }) {
     super({ id, config, log, userAgent, authToken })
@@ -37,9 +43,26 @@ export default class DirectConnectionProvider extends PooledConnectionProvider {
    * its arguments.
    */
   acquireConnection ({ accessMode, database, bookmarks } = {}) {
+    const databaseSpecificErrorHandler = ConnectionErrorHandler.create({
+      errorCode: SERVICE_UNAVAILABLE,
+      handleAuthorizationExpired: (error, address) =>
+        this._handleAuthorizationExpired(error, address, database)
+    })
+
     return this._connectionPool
       .acquire(this._address)
-      .then(connection => new DelegateConnection(connection, null))
+      .then(
+        connection =>
+          new DelegateConnection(connection, databaseSpecificErrorHandler)
+      )
+  }
+
+  _handleAuthorizationExpired (error, address, 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
   }
 
   async _hasProtocolVersion (versionPredicate) {

bolt-connection/src/connection-provider/connection-provider-routing.js

@@ -100,6 +100,14 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     return error
   }
 
+  _handleAuthorizationExpired (error, address, database) {
+    this._log.warn(
+      `Routing driver ${this._id} will close connections to ${address} for database '${database}' because of an error ${error.code} '${error.message}'`
+    )
+    this._connectionPool.purge(address).catch(() => {})
+    return error
+  }
+
   _handleWriteFailure (error, address, database) {
     this._log.warn(
       `Routing driver ${this._id} will forget writer ${address} for database '${database}' because of an error ${error.code} '${error.message}'`
@@ -122,7 +130,9 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     const databaseSpecificErrorHandler = new ConnectionErrorHandler(
       SESSION_EXPIRED,
       (error, address) => this._handleUnavailability(error, address, database),
-      (error, address) => this._handleWriteFailure(error, address, database)
+      (error, address) => this._handleWriteFailure(error, address, database),
+      (error, address) =>
+        this._handleAuthorizationExpired(error, address, database)
     )
 
     const routingTable = await this._freshRoutingTable({

bolt-connection/src/connection/connection-error-handler.js

@@ -22,10 +22,30 @@ import { error } from 'neo4j-driver-core'
 const { SERVICE_UNAVAILABLE, SESSION_EXPIRED } = error
 
 export default class ConnectionErrorHandler {
-  constructor (errorCode, handleUnavailability, handleWriteFailure) {
+  constructor (
+    errorCode,
+    handleUnavailability,
+    handleWriteFailure,
+    handleAuthorizationExpired
+  ) {
     this._errorCode = errorCode
     this._handleUnavailability = handleUnavailability || noOpHandler
     this._handleWriteFailure = handleWriteFailure || noOpHandler
+    this._handleAuthorizationExpired = handleAuthorizationExpired || noOpHandler
+  }
+
+  static create ({
+    errorCode,
+    handleUnavailability,
+    handleWriteFailure,
+    handleAuthorizationExpired
+  }) {
+    return new ConnectionErrorHandler(
+      errorCode,
+      handleUnavailability,
+      handleWriteFailure,
+      handleAuthorizationExpired
+    )
   }
 
   /**
@@ -43,6 +63,9 @@ export default class ConnectionErrorHandler {
    * @return {Neo4jError} new error that should be propagated to the user.
    */
   handleAndTransformError (error, address) {
+    if (isAutorizationExpiredError(error)) {
+      return this._handleAuthorizationExpired(error, address)
+    }
     if (isAvailabilityError(error)) {
       return this._handleUnavailability(error, address)
     }
@@ -53,6 +76,10 @@ export default class ConnectionErrorHandler {
   }
 }
 
+function isAutorizationExpiredError (error) {
+  return error && error.code === 'Neo.ClientError.Security.AuthorizationExpired'
+}
+
 function isAvailabilityError (error) {
   if (error) {
     return (

…ernal/connection-provider-direct.test.js …vider/connection-provider-direct.test.jstest/internal/connection-provider-direct.test.js renamed to bolt-connection/test/connection-provider/connection-provider-direct.test.js test/internal/connection-provider-direct.test.js renamed to bolt-connection/test/connection-provider/connection-provider-direct.test.js

@@ -17,12 +17,10 @@
  * limitations under the License.
  */
 
-import { READ } from '../../src/driver'
-import DirectConnectionProvider from '../../bolt-connection/lib/connection-provider/connection-provider-direct'
-import Pool from '../../bolt-connection/lib/pool/pool'
-import Connection from '../../bolt-connection/lib/connection/connection'
-import DelegateConnection from '../../bolt-connection/lib/connection/connection-delegate'
-import { internal } from 'neo4j-driver-core'
+import DirectConnectionProvider from '../../src/connection-provider/connection-provider-direct'
+import { Pool } from '../../src/pool'
+import { Connection, DelegateConnection } from '../../src/connection'
+import { internal, newError } from 'neo4j-driver-core'
 
 const {
   serverAddress: { ServerAddress },
@@ -36,7 +34,7 @@ describe('#unit DirectConnectionProvider', () => {
     const connectionProvider = newDirectConnectionProvider(address, pool)
 
     connectionProvider
-      .acquireConnection({ accessMode: READ, database: '' })
+      .acquireConnection({ accessMode: 'READ', database: '' })
       .then(connection => {
         expect(connection).toBeDefined()
         expect(connection.address).toEqual(address)
@@ -52,18 +50,59 @@ describe('#unit DirectConnectionProvider', () => {
     const connectionProvider = newDirectConnectionProvider(address, pool)
 
     const conn = await connectionProvider.acquireConnection({
-      accessMode: READ,
+      accessMode: 'READ',
       database: ''
     })
     expect(conn instanceof DelegateConnection).toBeTruthy()
   })
+
+  it('should purge connections for address when AuthorizationExpired happens', async () => {
+    const address = ServerAddress.fromUrl('localhost:123')
+    const pool = newPool()
+    jest.spyOn(pool, 'purge')
+    const connectionProvider = newDirectConnectionProvider(address, pool)
+
+    const conn = await connectionProvider.acquireConnection({
+      accessMode: 'READ',
+      database: ''
+    })
+
+    const error = newError(
+      'Message',
+      'Neo.ClientError.Security.AuthorizationExpired'
+    )
+
+    conn.handleAndTransformError(error, address)
+
+    expect(pool.purge).toHaveBeenCalledWith(address)
+  })
+
+  it('should purge not change error when AuthorizationExpired happens', async () => {
+    const address = ServerAddress.fromUrl('localhost:123')
+    const pool = newPool()
+    const connectionProvider = newDirectConnectionProvider(address, pool)
+
+    const conn = await connectionProvider.acquireConnection({
+      accessMode: 'READ',
+      database: ''
+    })
+
+    const expectedError = newError(
+      'Message',
+      'Neo.ClientError.Security.AuthorizationExpired'
+    )
+
+    const error = conn.handleAndTransformError(expectedError, address)
+
+    expect(error).toBe(expectedError)
+  })
 })
 
 function newDirectConnectionProvider (address, pool) {
   const connectionProvider = new DirectConnectionProvider({
     id: 0,
     config: {},
-    logger: Logger.noOp(),
+    log: Logger.noOp(),
     address: address
   })
   connectionProvider._connectionPool = pool