Do not reuse the same refresh token multiple times

- The "add a UAA token to HTTP request headers" flow works like this:
  - User makes a request using AbstractReactorOperations -> Operator
  - This adds an "Authorization" header, using Provider#getToken mono ;
    if a value is cached it uses that, otherwise it uses whatever flow
    is available to request a token.
  - If the response is unauthorized, it means the access token is
    expired, and the Operator calls Provider#invalidate ; and then
    retries the request, which will trigger another #getToken call.
- There was a race condition, when an access_token was cached and
  multiple request used it concurrently, they would all call
  AbstractUaaTokenProvider#invalidate, and all reuse the same refresh
  token. This is an issue when the UAA is configured with non-reusable
  refresh tokens (revocable + rotating or unique), only the first
  refresh token request succeeds, and all other refresh token requests
  fail.
- This PR addresses this by ensuring that the cached refresh token is
  removed from the cache right before being used. Any other call to
  #invalidate will be a no-op.
- This is NOT a perfect fix, and there are some smaller scale race
  conditions happening. For example, #invalidate calls
  refreshTokens.remove and accessTokens.put sequentially. It is possible
  that a concurrent request calls invalidate, finds the refreshTokens
  cache empty, and then will populate accessTokens through #getToken ;
  in that case there could be a race condition and two tokens fetched.
- Re-architecting the whole token logic is too big of a lift for the
  project, so we accept that this solution is not perfect - as long as
  the issues are recoverable.
- Fixes #1146

Signed-off-by: Daniel Garnier-Moiroux <[email protected]>

Author: Kehrlann

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/tokenprovider/AbstractUaaTokenProvider.java

@@ -80,7 +80,7 @@ public abstract class AbstractUaaTokenProvider implements TokenProvider {
     private final ConcurrentMap<ConnectionContext, RefreshToken> refreshTokenStreams =
             new ConcurrentHashMap<>(1);
 
-    private final ConcurrentMap<ConnectionContext, Mono<String>> refreshTokens =
+    private final ConcurrentMap<ConnectionContext, String> refreshTokens =
             new ConcurrentHashMap<>(1);
 
     /**
@@ -116,7 +116,10 @@ public final Mono<String> getToken(ConnectionContext connectionContext) {
 
     @Override
     public void invalidate(ConnectionContext connectionContext) {
-        this.accessTokens.put(connectionContext, token(connectionContext));
+        String refreshToken = this.refreshTokens.remove(connectionContext);
+        if (refreshToken != null) {
+            this.accessTokens.put(connectionContext, token(connectionContext, refreshToken));
+        }
     }
 
     /**
@@ -133,6 +136,30 @@ public void invalidate(ConnectionContext connectionContext) {
      */
     abstract void tokenRequestTransformer(HttpClientRequest request, HttpClientForm form);
 
+    private Mono<String> token(ConnectionContext connectionContext) {
+        Mono<String> token =
+                primaryToken(connectionContext)
+                        .doOnSubscribe(s -> LOGGER.debug("Negotiating using token provider"));
+
+        return cacheResult(connectionContext, token);
+    }
+
+    private Mono<String> token(ConnectionContext connectionContext, String refreshToken) {
+        Mono<String> token =
+                refreshToken(connectionContext, refreshToken)
+                        .doOnSubscribe(s -> LOGGER.debug("Negotiating using refresh token"))
+                        // fall back to primary token in case the refresh_token grant fails
+                        // (expired, revoked, ...)
+                        .switchIfEmpty(
+                                primaryToken(connectionContext)
+                                        .doOnSubscribe(
+                                                s ->
+                                                        LOGGER.debug(
+                                                                "Falling back to token provider")));
+
+        return cacheResult(connectionContext, token);
+    }
+
     private static String extractAccessToken(Map<String, String> payload) {
         String accessToken = payload.get(ACCESS_TOKEN);
 
@@ -227,8 +254,7 @@ private Consumer<Map<String, String>> extractRefreshToken(ConnectionContext conn
                                                         });
                                     }
 
-                                    this.refreshTokens.put(
-                                            connectionContext, Mono.just(refreshToken));
+                                    this.refreshTokens.put(connectionContext, refreshToken);
                                     getRefreshTokenStream(connectionContext)
                                             .sink
                                             .emitNext(refreshToken, FAIL_FAST);
@@ -297,30 +323,16 @@ private void setAuthorization(HttpHeaders headers) {
         headers.set(AUTHORIZATION, String.format("Basic %s", encoded));
     }
 
-    private Mono<String> token(ConnectionContext connectionContext) {
-        Mono<String> cached =
-                this.refreshTokens
-                        .getOrDefault(connectionContext, Mono.empty())
-                        .flatMap(
-                                refreshToken ->
-                                        refreshToken(connectionContext, refreshToken)
-                                                .doOnSubscribe(
-                                                        s ->
-                                                                LOGGER.debug(
-                                                                        "Negotiating using refresh"
-                                                                                + " token")))
-                        .switchIfEmpty(
-                                primaryToken(connectionContext)
-                                        .doOnSubscribe(
-                                                s ->
-                                                        LOGGER.debug(
-                                                                "Negotiating using token"
-                                                                        + " provider")));
-
+    /**
+     * Cache the given mono. If {@link ConnectionContext#getCacheDuration()} is not null, use that
+     * as the cache TTL. Otherwise, cache indefinitely.
+     */
+    private static Mono<String> cacheResult(
+            ConnectionContext connectionContext, Mono<String> token) {
         return connectionContext
                 .getCacheDuration()
-                .map(cached::cache)
-                .orElseGet(cached::cache)
+                .map(token::cache)
+                .orElseGet(token::cache)
                 .checkpoint();
     }