Create RefreshProtectedCredentialsProvider class

To protect from several actual token retrievals to happen at the same
time. This class is used in the OAuth 2 client credentials grant
provider.

Add also a builder to make the OAuth 2 provider easier to configure, add
TLS settings and a test.

(cherry picked from commit 5752e55)

Conflicts:
	src/test/java/com/rabbitmq/client/test/ClientTests.java

Author: acogoluegnes

pom.xml

@@ -64,6 +64,7 @@
     <mockito.version>3.0.0</mockito.version>
     <assertj.version>3.12.2</assertj.version>
     <jetty.version>9.4.19.v20190610</jetty.version>
+    <bouncycastle.version>1.61</bouncycastle.version>
 
     <maven.javadoc.plugin.version>3.0.1</maven.javadoc.plugin.version>
     <maven.release.plugin.version>2.5.3</maven.release.plugin.version>
@@ -751,6 +752,12 @@
       <version>${jetty.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk15on</artifactId>
+      <version>${bouncycastle.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

src/main/java/com/rabbitmq/client/impl/OAuth2ClientCredentialsGrantCredentialsProvider.java

@@ -16,51 +16,60 @@
 package com.rabbitmq.client.impl;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.HostnameVerifier;
+import javax.net.ssl.HttpsURLConnection;
+import javax.net.ssl.SSLSocketFactory;
 import java.io.*;
 import java.net.HttpURLConnection;
 import java.net.URL;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
-
-public class OAuth2ClientCredentialsGrantCredentialsProvider implements CredentialsProvider {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(OAuth2ClientCredentialsGrantCredentialsProvider.class);
+/**
+ *
+ * @see RefreshProtectedCredentialsProvider
+ */
+public class OAuth2ClientCredentialsGrantCredentialsProvider extends RefreshProtectedCredentialsProvider<OAuth2ClientCredentialsGrantCredentialsProvider.Token> {
 
     private static final String UTF_8_CHARSET = "UTF-8";
-    private final String serverUri; // should be renamed to tokenEndpointUri?
+    private final String tokenEndpointUri;
     private final String clientId;
     private final String clientSecret;
     private final String grantType;
-    // UAA specific, to distinguish between different users
-    private final String username, password;
+
+    private final Map<String, String> parameters;
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
     private final String id;
 
-    private final AtomicReference<Token> token = new AtomicReference<>();
+    private final HostnameVerifier hostnameVerifier;
+    private final SSLSocketFactory sslSocketFactory;
+
+    public OAuth2ClientCredentialsGrantCredentialsProvider(String tokenEndpointUri, String clientId, String clientSecret, String grantType) {
+        this(tokenEndpointUri, clientId, clientSecret, grantType, new HashMap<>());
+    }
+
+    public OAuth2ClientCredentialsGrantCredentialsProvider(String tokenEndpointUri, String clientId, String clientSecret, String grantType,
+                                                           HostnameVerifier hostnameVerifier, SSLSocketFactory sslSocketFactory) {
+        this(tokenEndpointUri, clientId, clientSecret, grantType, new HashMap<>(), hostnameVerifier, sslSocketFactory);
+    }
 
-    private final Lock refreshLock = new ReentrantLock();
-    private final AtomicReference<CountDownLatch> latch = new AtomicReference<>();
-    private AtomicBoolean refreshInProcess = new AtomicBoolean(false);
+    public OAuth2ClientCredentialsGrantCredentialsProvider(String tokenEndpointUri, String clientId, String clientSecret, String grantType, Map<String, String> parameters) {
+        this(tokenEndpointUri, clientId, clientSecret, grantType, parameters, null, null);
+    }
 
-    public OAuth2ClientCredentialsGrantCredentialsProvider(String serverUri, String clientId, String clientSecret, String grantType, String username, String password) {
-        this.serverUri = serverUri;
+    public OAuth2ClientCredentialsGrantCredentialsProvider(String tokenEndpointUri, String clientId, String clientSecret, String grantType, Map<String, String> parameters,
+                                                           HostnameVerifier hostnameVerifier, SSLSocketFactory sslSocketFactory) {
+        this.tokenEndpointUri = tokenEndpointUri;
         this.clientId = clientId;
         this.clientSecret = clientSecret;
         this.grantType = grantType;
-        this.username = username;
-        this.password = password;
+        this.parameters = Collections.unmodifiableMap(new HashMap<>(parameters));
+        this.hostnameVerifier = hostnameVerifier;
+        this.sslSocketFactory = sslSocketFactory;
         this.id = UUID.randomUUID().toString();
     }
 
@@ -90,19 +99,8 @@ public String getUsername() {
     }
 
     @Override
-    public String getPassword() {
-        if (token.get() == null) {
-            refresh();
-        }
-        return token.get().getAccess();
-    }
-
-    @Override
-    public Date getExpiration() {
-        if (token.get() == null) {
-            refresh();
-        }
-        return token.get().getExpiration();
+    protected String usernameFromToken(Token token) {
+        return "";
     }
 
     protected Token parseToken(String response) {
@@ -118,47 +116,21 @@ protected Token parseToken(String response) {
     }
 
     @Override
-    public void refresh() {
-        // refresh should happen at once. Other calls wait for the refresh to finish and move on.
-        if (refreshLock.tryLock()) {
-            LOGGER.debug("Refreshing token");
-            try {
-                latch.set(new CountDownLatch(1));
-                refreshInProcess.set(true);
-                token.set(retrieveToken());
-                LOGGER.debug("Token refreshed");
-            } finally {
-                latch.get().countDown();
-                refreshInProcess.set(false);
-                refreshLock.unlock();
-            }
-        } else {
-            try {
-                LOGGER.debug("Waiting for token refresh to be finished");
-                while (!refreshInProcess.get()) {
-                    Thread.sleep(10);
-                }
-                latch.get().await();
-                LOGGER.debug("Done waiting for token refresh");
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-            }
-        }
-    }
-
     protected Token retrieveToken() {
-        // FIXME handle TLS specific settings
         try {
             StringBuilder urlParameters = new StringBuilder();
             encode(urlParameters, "grant_type", grantType);
-            encode(urlParameters, "username", username);
-            encode(urlParameters, "password", password);
+            for (Map.Entry<String, String> parameter : parameters.entrySet()) {
+                encode(urlParameters, parameter.getKey(), parameter.getValue());
+            }
             byte[] postData = urlParameters.toString().getBytes(StandardCharsets.UTF_8);
             int postDataLength = postData.length;
-            URL url = new URL(serverUri);
+            URL url = new URL(tokenEndpointUri);
+
             // FIXME close connection?
             // FIXME set timeout on request
             HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+
             conn.setDoOutput(true);
             conn.setInstanceFollowRedirects(false);
             conn.setRequestMethod("POST");
@@ -168,6 +140,9 @@ protected Token retrieveToken() {
             conn.setRequestProperty("accept", "application/json");
             conn.setRequestProperty("content-length", Integer.toString(postDataLength));
             conn.setUseCaches(false);
+
+            configureHttpConnection(conn);
+
             try (DataOutputStream wr = new DataOutputStream(conn.getOutputStream())) {
                 wr.write(postData);
             }
@@ -196,6 +171,28 @@ protected Token retrieveToken() {
         }
     }
 
+    @Override
+    protected String passwordFromToken(Token token) {
+        return token.getAccess();
+    }
+
+    @Override
+    protected Date expirationFromToken(Token token) {
+        return token.getExpiration();
+    }
+
+    protected void configureHttpConnection(HttpURLConnection connection) {
+        if (connection instanceof HttpsURLConnection) {
+            HttpsURLConnection securedConnection = (HttpsURLConnection) connection;
+            if (this.hostnameVerifier != null) {
+                securedConnection.setHostnameVerifier(this.hostnameVerifier);
+            }
+            if (this.sslSocketFactory != null) {
+                securedConnection.setSSLSocketFactory(this.sslSocketFactory);
+            }
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -230,4 +227,59 @@ public String getAccess() {
             return access;
         }
     }
+
+    public static class OAuth2ClientCredentialsGrantCredentialsProviderBuilder {
+
+        private final Map<String, String> parameters = new HashMap<>();
+        private String tokenEndpointUri;
+        private String clientId;
+        private String clientSecret;
+        private String grantType = "client_credentials";
+        private HostnameVerifier hostnameVerifier;
+
+        private SSLSocketFactory sslSocketFactory;
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder tokenEndpointUri(String tokenEndpointUri) {
+            this.tokenEndpointUri = tokenEndpointUri;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder clientId(String clientId) {
+            this.clientId = clientId;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder clientSecret(String clientSecret) {
+            this.clientSecret = clientSecret;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder grantType(String grantType) {
+            this.grantType = grantType;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder parameter(String name, String value) {
+            this.parameters.put(name, value);
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder setHostnameVerifier(HostnameVerifier hostnameVerifier) {
+            this.hostnameVerifier = hostnameVerifier;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProviderBuilder setSslSocketFactory(SSLSocketFactory sslSocketFactory) {
+            this.sslSocketFactory = sslSocketFactory;
+            return this;
+        }
+
+        public OAuth2ClientCredentialsGrantCredentialsProvider build() {
+            return new OAuth2ClientCredentialsGrantCredentialsProvider(
+                    tokenEndpointUri, clientId, clientSecret, grantType, parameters,
+                    hostnameVerifier, sslSocketFactory
+            );
+        }
+
+    }
 }

src/main/java/com/rabbitmq/client/impl/RefreshProtectedCredentialsProvider.java

@@ -0,0 +1,113 @@
+// Copyright (c) 2019 Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+
+package com.rabbitmq.client.impl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * An abstract {@link CredentialsProvider} that does not let token refresh happen concurrently.
+ * <p>
+ * A token is usually long-lived (several minutes or more), can be re-used inside the same application,
+ * and refreshing it is a costly operation. This base class lets a first call to {@link #refresh()}
+ * pass and block concurrent calls until the first call is over. Concurrent calls are then unblocked and
+ * can benefit from the refresh. This avoids unnecessary refresh operations to happen if a token
+ * is already being renewed.
+ * <p>
+ * Subclasses need to provide the actual token retrieval (whether is a first retrieval or a renewal is
+ * a implementation detail) and how to extract information (username, password, expiration date) from the retrieved
+ * token.
+ *
+ * @param <T> the type of token (usually specified by the subclass)
+ */
+public abstract class RefreshProtectedCredentialsProvider<T> implements CredentialsProvider {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RefreshProtectedCredentialsProvider.class);
+
+    private final AtomicReference<T> token = new AtomicReference<>();
+
+    private final Lock refreshLock = new ReentrantLock();
+    private final AtomicReference<CountDownLatch> latch = new AtomicReference<>();
+    private AtomicBoolean refreshInProcess = new AtomicBoolean(false);
+
+    @Override
+    public String getUsername() {
+        if (token.get() == null) {
+            refresh();
+        }
+        return usernameFromToken(token.get());
+    }
+
+    @Override
+    public String getPassword() {
+        if (token.get() == null) {
+            refresh();
+        }
+        return passwordFromToken(token.get());
+    }
+
+    @Override
+    public Date getExpiration() {
+        if (token.get() == null) {
+            refresh();
+        }
+        return expirationFromToken(token.get());
+    }
+
+    @Override
+    public void refresh() {
+        // refresh should happen at once. Other calls wait for the refresh to finish and move on.
+        if (refreshLock.tryLock()) {
+            LOGGER.debug("Refreshing token");
+            try {
+                latch.set(new CountDownLatch(1));
+                refreshInProcess.set(true);
+                token.set(retrieveToken());
+                LOGGER.debug("Token refreshed");
+            } finally {
+                latch.get().countDown();
+                refreshInProcess.set(false);
+                refreshLock.unlock();
+            }
+        } else {
+            try {
+                LOGGER.debug("Waiting for token refresh to be finished");
+                while (!refreshInProcess.get()) {
+                    Thread.sleep(10);
+                }
+                latch.get().await();
+                LOGGER.debug("Done waiting for token refresh");
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+    }
+
+    protected abstract T retrieveToken();
+
+    protected abstract String usernameFromToken(T token);
+
+    protected abstract String passwordFromToken(T token);
+
+    protected abstract Date expirationFromToken(T token);
+}