Propagate authentication errors

Authentication errors happen during connection initialization when INIT
message is send. Server validates provided credentials and then responds
with either SUCCESS if credentials are fine or with FAILURE and closes
the connection when credentials are wrong. Previously auth errors were
not propagated to external observers (including those defined by user's
Result promise). This means auth errors were only propagated to
`Driver.onError` callback but all other places received `ServiceUnavailable`
or `SessionExpired` because of closed connection.

This commit makes driver treat INIT error as fatal and corresponding
connection as broken. Initialization error is memorized and propagated
to all message observers. So they see specific failure reason and not
generic `ServiceUnavailable` \ `SessionExpired`.

Author: lutovich

src/v1/internal/connector.js

@@ -288,6 +288,20 @@ class Connection {
     }
   }
 
+  /**
+   * Mark this connection as failed because processing of INIT message failed and server will close the connection.
+   * Initialization failure is a fatal error for the connection.
+   * @param {Neo4jError} error the initialization error.
+   * @param {StreamObserver} initObserver the initialization observer that noticed the failure.
+   * @protected
+   */
+  _initializationFailed(error, initObserver) {
+    if (this._currentObserver === initObserver) {
+      this._currentObserver = null; // init observer detected the failure and should not be notified again
+    }
+    this._handleFatalError(error);
+  }
+
   _handleMessage( msg ) {
     const payload = msg.fields[0];
 
@@ -351,7 +365,8 @@ class Connection {
   /** Queue an INIT-message to be sent to the database */
   initialize( clientName, token, observer ) {
     log("C", "INIT", clientName, token);
-    this._queueObserver(observer);
+    const initObserver = new InitObserver(this, observer);
+    this._queueObserver(initObserver);
     this._packer.packStruct( INIT, [this._packable(clientName), this._packable(token)],
       (err) => this._handleFatalError(err) );
     this._chunker.messageBoundary();
@@ -489,6 +504,40 @@ function connect(url, config = {}, connectionErrorCode = null) {
   return new Connection( new Ch(channelConfig), completeUrl);
 }
 
+/**
+ * Observer that wraps user-defined observer for INIT message and handles initialization failures. Connection is
+ * closed by the server if processing of INIT message fails so this observer will handle initialization failure
+ * as a fatal error.
+ */
+class InitObserver {
+
+  /**
+   * @constructor
+   * @param {Connection} connection the connection used to send INIT message.
+   * @param {StreamObserver} originalObserver the observer to wrap and delegate calls to.
+   */
+  constructor(connection, originalObserver) {
+    this._connection = connection;
+    this._originalObserver = originalObserver || NO_OP_OBSERVER;
+  }
+
+  onNext(record) {
+    this._originalObserver.onNext(record);
+  }
+
+  onError(error) {
+    try {
+      this._originalObserver.onError(error);
+    } finally {
+      this._connection._initializationFailed(error, this);
+    }
+  }
+
+  onCompleted(metaData) {
+    this._originalObserver.onCompleted(metaData);
+  }
+}
+
 export {
     connect,
     parseScheme,

src/v1/internal/get-servers-util.js

@@ -23,6 +23,7 @@ import Integer, {int} from '../integer';
 
 const PROCEDURE_CALL = 'CALL dbms.cluster.routing.getServers';
 const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';
+const UNAUTHORIZED_CODE = 'Neo.ClientError.Security.Unauthorized';
 
 export default class GetServersUtil {
 
@@ -35,10 +36,14 @@ export default class GetServersUtil {
         // throw when getServers procedure not found because this is clearly a configuration issue
         throw newError('Server ' + routerAddress + ' could not perform routing. ' +
           'Make sure you are connecting to a causal cluster', SERVICE_UNAVAILABLE);
+      } else if (error.code === UNAUTHORIZED_CODE) {
+        // auth error is a sign of a configuration issue, rediscovery should not proceed
+        throw error;
+      } else {
+        // return nothing when failed to connect because code higher in the callstack is still able to retry with a
+        // different session towards a different router
+        return null;
       }
-      // return nothing when failed to connect because code higher in the callstack is still able to retry with a
-      // different session towards a different router
-      return null;
     });
   }
 

test/internal/connector.test.js

@@ -117,6 +117,30 @@ describe('connector', () => {
     channel.onmessage(packedFailureMessage(errorCode, errorMessage));
   });
 
+  it('should notify provided observer when connection initialization completes', done => {
+    const connection = connect('bolt://localhost');
+
+    connection.initialize('mydriver/0.0.0', basicAuthToken(), {
+      onCompleted: metaData => {
+        expect(connection.isOpen()).toBeTruthy();
+        expect(metaData).toBeDefined();
+        done();
+      },
+    });
+  });
+
+  it('should notify provided observer when connection initialization fails', done => {
+    const connection = connect('bolt://localhost:7474'); // wrong port
+
+    connection.initialize('mydriver/0.0.0', basicAuthToken(), {
+      onError: error => {
+        expect(connection.isOpen()).toBeFalsy();
+        expect(error).toBeDefined();
+        done();
+      },
+    });
+  });
+
   function packedHandshakeMessage() {
     const result = alloc(4);
     result.putInt32(0, 1);

test/resources/boltkit/failed_auth.script

@@ -0,0 +1,6 @@
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: INIT "neo4j-javascript/0.0.0-dev" {"credentials": "neo4j", "scheme": "basic", "principal": "neo4j"}
+S: FAILURE {"code": "Neo.ClientError.Security.Unauthorized", "message": "Some server auth error message"}
+S: <EXIT>

test/v1/driver.test.js

@@ -80,7 +80,7 @@ describe('driver', () => {
 
   it('should fail early on wrong credentials', done => {
     // Given
-    driver = neo4j.driver("bolt://localhost", neo4j.auth.basic("neo4j", "who would use such a password"));
+    driver = neo4j.driver("bolt://localhost", wrongCredentials());
 
     // Expect
     driver.onError = err => {
@@ -93,6 +93,16 @@ describe('driver', () => {
     startNewTransaction(driver);
   });
 
+  it('should fail queries on wrong credentials', done => {
+    driver = neo4j.driver("bolt://localhost", wrongCredentials());
+
+    const session = driver.session();
+    session.run('RETURN 1').catch(error => {
+      expect(error.code).toEqual('Neo.ClientError.Security.Unauthorized');
+      done();
+    });
+  });
+
   it('should indicate success early on correct credentials', done => {
     // Given
     driver = neo4j.driver("bolt://localhost", sharedNeo4j.authToken);
@@ -207,4 +217,8 @@ describe('driver', () => {
     expect(session.beginTransaction()).toBeDefined();
   }
 
+  function wrongCredentials() {
+    return neo4j.auth.basic('neo4j', 'who would use such a password');
+  }
+
 });

test/v1/routing.driver.boltkit.it.js

@@ -1549,6 +1549,33 @@ describe('routing driver', () => {
     });
   });
 
+  it('should fail rediscovery on auth error', done => {
+    if (!boltkit.BoltKitSupport) {
+      done();
+      return;
+    }
+
+    const kit = new boltkit.BoltKit();
+    const router = kit.start('./test/resources/boltkit/failed_auth.script', 9010);
+
+    kit.run(() => {
+      const driver = newDriver('bolt+routing://127.0.0.1:9010');
+      const session = driver.session();
+      session.run('RETURN 1').catch(error => {
+        expect(error.code).toEqual('Neo.ClientError.Security.Unauthorized');
+        expect(error.message).toEqual('Some server auth error message');
+
+        session.close(() => {
+          driver.close();
+          router.exit(code => {
+            expect(code).toEqual(0);
+            done();
+          });
+        });
+      });
+    });
+  });
+
   function moveNextDateNow30SecondsForward() {
     const currentTime = Date.now();
     hijackNextDateNowCall(currentTime + 30 * 1000 + 1);