Introduce AuthToken rotation and session auth support

This update brings support for `AuthToken` rotation (refresh) by replacing existing token with a new token during driver's lifetime.

The main objective of this feature is to allow token rotation for the same identity. Therefore, the rotation support is not intended for a change of identity.

A new type called `AuthTokenManager` has the following 2 primary responsibilities:
- supplying a valid token, which may be one of the following:
  - the current token
  - a new token, which instructs the driver to use the new token
- handling a token expired failure that originates from the server if it determines the current token to be expired (a timely rotation should generally reduce the likelihood of this happening)

The driver does not make judgements on whether `AuthToken` should be updated. Instead, it calls `AuthTokenManager` to check if the provided token is the same as the previously used token and takes action if not. The driver reserves the right to call the token manager as often as it deems necessary. The manager implementation must be thread-safe and non-blocking for for caller threads. For instance, IO operations must not be done on the calling thread.

The `GraphDatabase` class has been updated to include a set of new methods that accept `AuthTokenManager` instance. A sample of the driver instantiation is shown below:
```java
var manager = /* the manager implementation */
var driver = GraphDatabase.driver(uri, manager);
```

The token rotation benefits from [Bolt 5.1](#bolt-51-support), but works on previous Bolt versions at the expence of replacing existing connections with new connections.

A temporal `AuthTokenManager` implementation is available in a new `AuthTokenManagers` factory. It manages `AuthToken` instances that come with a UTC expiration timestamp and calls a new token supplier, which is provided by the user, when a new token is required.

The temporal manager instantiation example:
```java
import org.neo4j.driver.AuthTokenManagers.TemporalAuthData;

var manager = AuthTokenManagers.temporal(() -> {
  var token = // get new token logic
  return TemporalAuthData.of(token, timestamp);
});
```

This update includes Bolt 5.1 support. The new `LOGOFF` and `LOGON` messages allow for token management on active Bolt connections.

In addition to the token rotation support, this update also includes support for setting a static `AuthToken` instance on the driver session level.

Unlike the rotation feature, this feature may be used for identity change. As such, it might be referred to as user switching.

It requires a minimum Bolt 5.1 version.

The `Driver` interface has 2 new `session` methods that accept an `AuthToken` instance.

Sample usage:
```java
var token = AuthTokens.bearer("token");
var session = driver.session(Session.class, token);
```

The `Driver` includes a new method that checks session auth is supported.

Sample usage:
```java
var supports = driver.supportsSessionAuth();
```

The `Driver` includes a new method that verifies a given `AuthToken` instance by communicating with the server.

It requires a minimum Bolt 5.1 version.

Sample usage:
```java
var token = AuthTokens.bearer("token");
var successful = driver.verifyAuthentication(token);
```

There are 3 new exceptions:
- `AuthTokenManagerExecutionException` - Indicates that `AuthTokenManager` execution has lead to an unexpected result. This includes invalid results and errors.
- `UnsupportedFeatureException` - Indicates that a given feature is not supported in a particular setup. For instance, session auth feature is not supported on Bolt versions below 5.1.
- `TokenExpiredRetryableException` - Indicates that the token supplied by the `AuthTokenManager` has been deemed as expired by the server. This is a retryable variant of the `TokenExpiredException` used when the driver has an explicit `AuthTokenManager` that might supply a new token following this failure. If driver is instantiated with the static `AuthToken`, the `TokenExpiredException` will be used instead.

Author: injectives

driver/clirr-ignored-differences.xml

@@ -433,4 +433,28 @@
         <method>org.neo4j.driver.BookmarkManager queryTaskBookmarkManager()</method>
     </difference>
 
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.BaseSession session(java.lang.Class, org.neo4j.driver.AuthToken)</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>org.neo4j.driver.BaseSession session(java.lang.Class, org.neo4j.driver.SessionConfig, org.neo4j.driver.AuthToken)</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>boolean verifyAuthentication(org.neo4j.driver.AuthToken)</method>
+    </difference>
+
+    <difference>
+        <className>org/neo4j/driver/Driver</className>
+        <differenceType>7012</differenceType>
+        <method>boolean supportsSessionAuth()</method>
+    </difference>
+
 </differences>

driver/src/main/java/org/neo4j/driver/AuthTokenManager.java

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.neo4j.driver;
+
+import java.util.concurrent.CompletionStage;
+
+/**
+ * A manager of {@link AuthToken} instances used by the driver.
+ * <p>
+ * The manager must manage tokens for the same identity. Therefore, it is not intended for a change of identity.
+ * <p>
+ * Implementations should supply the same token unless it needs to be updated since a change of token might result in
+ * extra processing by the driver.
+ * <p>
+ * Driver initializes new connections with a token supplied by the manager. If token changes, driver action depends on
+ * connection's Bolt protocol version:
+ * <ul>
+ *     <li>Bolt 5.1 or above - {@code LOGOFF} and {@code LOGON} messages are dispatched to update the token on next interaction</li>
+ *     <li>Bolt 5.0 or below - connection is closed an a new one is initialized with the new token</li>
+ * </ul>
+ * <p>
+ * All implementations of this interface must be thread-safe and non-blocking for caller threads. For instance, IO operations must not
+ * be done on the calling thread.
+ * @since 5.7
+ */
+public interface AuthTokenManager {
+    /**
+     * Returns a {@link CompletionStage} for a valid {@link AuthToken}.
+     * <p>
+     * Driver invokes this method often to check if token has changed.
+     * <p>
+     * Failures will surface via the driver API, like {@link Session#beginTransaction()} method and others.
+     * @return a stage for a valid token, must not be {@code null} or complete with {@code null}
+     * @see org.neo4j.driver.exceptions.AuthTokenManagerExecutionException
+     */
+    CompletionStage<AuthToken> getToken();
+
+    /**
+     * Handles an error notification emitted by the server if the token is expired.
+     * <p>
+     * This will be called when driver emits the {@link org.neo4j.driver.exceptions.TokenExpiredRetryableException}.
+     *
+     * @param authToken the expired token
+     */
+    void onExpired(AuthToken authToken);
+}

driver/src/main/java/org/neo4j/driver/AuthTokenManagers.java

@@ -0,0 +1,113 @@
+/*
+ * 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.
+ */
+package org.neo4j.driver;
+
+import java.time.Clock;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ForkJoinPool;
+import java.util.function.Supplier;
+import org.neo4j.driver.internal.security.TemporalAuthTokenManager;
+
+/**
+ * Implementations of {@link AuthTokenManager}.
+ * @since 5.7
+ */
+public final class AuthTokenManagers {
+    private AuthTokenManagers() {}
+
+    /**
+     * Returns an {@link AuthTokenManager} that manages {@link AuthToken} instances with UTC expiration timestamp.
+     * <p>
+     * The implementation will only use the token supplier when it needs a new token instance. This includes the
+     * following conditions:
+     * <ol>
+     *     <li>token's UTC timestamp is expired</li>
+     *     <li>server rejects the current token (see {@link AuthTokenManager#onExpired(AuthToken)})</li>
+     * </ol>
+     * <p>
+     * The supplier will be called by a task running in the {@link ForkJoinPool#commonPool()} as documented in the
+     * {@link CompletableFuture#supplyAsync(Supplier)}.
+     *
+     * @param newTokenSupplier a new token supplier
+     * @return a new token manager
+     */
+    public static AuthTokenManager temporal(Supplier<TemporalAuthData> newTokenSupplier) {
+        return new TemporalAuthTokenManager(() -> CompletableFuture.supplyAsync(newTokenSupplier), Clock.systemUTC());
+    }
+
+    /**
+     * Returns an {@link AuthTokenManager} that manages {@link AuthToken} instances with UTC expiration timestamp.
+     * <p>
+     * The implementation will only use the token supplier when it needs a new token instance. This includes the
+     * following conditions:
+     * <ol>
+     *     <li>token's UTC timestamp is expired</li>
+     *     <li>server rejects the current token (see {@link AuthTokenManager#onExpired(AuthToken)})</li>
+     * </ol>
+     * <p>
+     * The provided supplier and its completion stages must be non-blocking as documented in the {@link AuthTokenManager}.
+     *
+     * @param newTokenStageSupplier a new token stage supplier
+     * @return a new token manager
+     */
+    public static AuthTokenManager temporalAsync(Supplier<CompletionStage<TemporalAuthData>> newTokenStageSupplier) {
+        return new TemporalAuthTokenManager(newTokenStageSupplier, Clock.systemUTC());
+    }
+
+    /**
+     * A container used by the temporal {@link AuthTokenManager} implementation provided by the driver, it contains an
+     * {@link AuthToken} and its UTC expiration timestamp.
+     * @since 5.7
+     */
+    public interface TemporalAuthData {
+        /**
+         * Returns a new instance with the provided token and {@link Long#MAX_VALUE} expiration timestamp.
+         * @param authToken the token
+         * @return a new instance
+         */
+        static TemporalAuthData of(AuthToken authToken) {
+            return of(authToken, Long.MAX_VALUE);
+        }
+
+        /**
+         * Returns a new instance with the provided token and its expiration timestamp.
+         * @param authToken the token
+         * @param expirationTimestamp the expiration timestamp
+         * @return a new instance
+         */
+        static TemporalAuthData of(AuthToken authToken, long expirationTimestamp) {
+            return new TemporalAuthTokenManager.InternalTemporalAuthData(authToken, expirationTimestamp);
+        }
+
+        /**
+         * Returns the {@link AuthToken}.
+         *
+         * @return the token
+         */
+        AuthToken authToken();
+
+        /**
+         * Returns the token's UTC expiration timestamp.
+         *
+         * @return the token's UTC expiration timestamp
+         */
+        long expirationTimestamp();
+    }
+}

driver/src/main/java/org/neo4j/driver/Driver.java

@@ -144,6 +144,42 @@ default <T extends BaseSession> T session(Class<T> sessionClass) {
         return session(sessionClass, SessionConfig.defaultConfig());
     }
 
+    /**
+     * Instantiate a new session of a supported type with the supplied {@link AuthToken}.
+     * <p>
+     * This method allows creating a session with a different {@link AuthToken} to the one used on the driver level.
+     * The minimum Bolt protocol version is 5.1. An {@link IllegalStateException} will be emitted on session interaction
+     * for previous Bolt versions.
+     * <p>
+     * Supported types are:
+     * <ul>
+     *     <li>{@link org.neo4j.driver.Session} - synchronous session</li>
+     *     <li>{@link org.neo4j.driver.async.AsyncSession} - asynchronous session</li>
+     *     <li>{@link org.neo4j.driver.reactive.ReactiveSession} - reactive session using Flow API</li>
+     *     <li>{@link org.neo4j.driver.reactivestreams.ReactiveSession} - reactive session using Reactive Streams
+     * API</li>
+     *     <li>{@link org.neo4j.driver.reactive.RxSession} - deprecated reactive session using Reactive Streams
+     * API, superseded by {@link org.neo4j.driver.reactivestreams.ReactiveSession}</li>
+     * </ul>
+     * <p>
+     * Sample usage:
+     * <pre>
+     * {@code
+     * var session = driver.session(AsyncSession.class);
+     * }
+     * </pre>
+     *
+     * @param sessionClass session type class, must not be null
+     * @param sessionAuthToken a token, null will result in driver-level configuration being used
+     * @return session instance
+     * @param <T> session type
+     * @throws IllegalArgumentException for unsupported session types
+     * @since 5.7
+     */
+    default <T extends BaseSession> T session(Class<T> sessionClass, AuthToken sessionAuthToken) {
+        return session(sessionClass, SessionConfig.defaultConfig(), sessionAuthToken);
+    }
+
     /**
      * Create a new session of supported type with a specified {@link SessionConfig session configuration}.
      * <p>
@@ -172,7 +208,45 @@ default <T extends BaseSession> T session(Class<T> sessionClass) {
      * @throws IllegalArgumentException for unsupported session types
      * @since 5.2
      */
-    <T extends BaseSession> T session(Class<T> sessionClass, SessionConfig sessionConfig);
+    default <T extends BaseSession> T session(Class<T> sessionClass, SessionConfig sessionConfig) {
+        return session(sessionClass, sessionConfig, null);
+    }
+
+    /**
+     * Instantiate a new session of a supported type with the supplied {@link SessionConfig session configuration} and
+     * {@link AuthToken}.
+     * <p>
+     * This method allows creating a session with a different {@link AuthToken} to the one used on the driver level.
+     * The minimum Bolt protocol version is 5.1. An {@link IllegalStateException} will be emitted on session interaction
+     * for previous Bolt versions.
+     * <p>
+     * Supported types are:
+     * <ul>
+     *     <li>{@link org.neo4j.driver.Session} - synchronous session</li>
+     *     <li>{@link org.neo4j.driver.async.AsyncSession} - asynchronous session</li>
+     *     <li>{@link org.neo4j.driver.reactive.ReactiveSession} - reactive session using Flow API</li>
+     *     <li>{@link org.neo4j.driver.reactivestreams.ReactiveSession} - reactive session using Reactive Streams
+     * API</li>
+     *     <li>{@link org.neo4j.driver.reactive.RxSession} - deprecated reactive session using Reactive Streams
+     * API, superseded by {@link org.neo4j.driver.reactivestreams.ReactiveSession}</li>
+     * </ul>
+     * <p>
+     * Sample usage:
+     * <pre>
+     * {@code
+     * var session = driver.session(AsyncSession.class);
+     * }
+     * </pre>
+     *
+     * @param sessionClass session type class, must not be null
+     * @param sessionConfig session config, must not be null
+     * @param sessionAuthToken a token, null will result in driver-level configuration being used
+     * @return session instance
+     * @param <T> session type
+     * @throws IllegalArgumentException for unsupported session types
+     * @since 5.7
+     */
+    <T extends BaseSession> T session(Class<T> sessionClass, SessionConfig sessionConfig, AuthToken sessionAuthToken);
 
     /**
      * Create a new general purpose {@link RxSession} with default {@link SessionConfig session configuration}. The {@link RxSession} provides a reactive way to
@@ -325,6 +399,24 @@ default AsyncSession asyncSession(SessionConfig sessionConfig) {
      */
     CompletionStage<Void> verifyConnectivityAsync();
 
+    /**
+     * Verifies if the given {@link AuthToken} is valid.
+     * <p>
+     * This check works on Bolt 5.1 version or above only.
+     * @param authToken the token
+     * @return the verification outcome
+     * @since 5.7
+     */
+    boolean verifyAuthentication(AuthToken authToken);
+
+    /**
+     * Checks if session auth is supported.
+     * @return the check outcome
+     * @since 5.7
+     * @see Driver#session(Class, SessionConfig, AuthToken)
+     */
+    boolean supportsSessionAuth();
+
     /**
      * Returns true if the server or cluster the driver connects to supports multi-databases, otherwise false.
      *