Support OAuth2 authentication

This commit introduces OAuth2 settings (token endpoint URI, client
secret, etc) in the stream environment builder. When OAuth2 is in use,
the library fetches a token from the HTTP(S) endpoint and use it to
authenticate against the broker. The library also takes care of
refreshing tokens: when a token is about to expire, a new one is
fetched, and connections re-authenticate.

Fixes #781

Author: acogoluegnes

ci/start-broker.sh

@@ -20,7 +20,7 @@ cp -R "${PWD}"/tls-gen/basic/result/* rabbitmq-configuration/tls
 chmod o+r rabbitmq-configuration/tls/*
 chmod g+r rabbitmq-configuration/tls/*
 
-echo "[rabbitmq_stream,rabbitmq_mqtt,rabbitmq_stomp,rabbitmq_amqp1_0,rabbitmq_auth_mechanism_ssl]." >> rabbitmq-configuration/enabled_plugins
+echo "[rabbitmq_stream,rabbitmq_mqtt,rabbitmq_stomp,rabbitmq_amqp1_0,rabbitmq_auth_mechanism_ssl,rabbitmq_auth_backend_oauth2]." >> rabbitmq-configuration/enabled_plugins
 
 echo "loopback_users = none
 
@@ -37,8 +37,25 @@ auth_mechanisms.1 = PLAIN
 auth_mechanisms.2 = ANONYMOUS
 auth_mechanisms.3 = EXTERNAL
 
+auth_backends.1 = internal
+auth_backends.2 = rabbit_auth_backend_oauth2
+
 stream.listeners.ssl.1 = 5551" >> rabbitmq-configuration/rabbitmq.conf
 
+echo "[
+  {rabbitmq_auth_backend_oauth2, [{key_config,
+         [{signing_keys,
+              #{<<\"token-key\">> =>
+                    {map,
+                        #{<<\"alg\">> => <<\"HS256\">>,
+                          <<\"k\">> => <<\"abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGH\">>,
+                          <<\"kid\">> => <<\"token-key\">>,
+                          <<\"kty\">> => <<\"oct\">>,
+                          <<\"use\">> => <<\"sig\">>,
+                          <<\"value\">> => <<\"token-key\">>}}}}]},
+     {resource_server_id,<<\"rabbitmq\">>}]}
+]." >> rabbitmq-configuration/advanced.config
+
 echo "Running RabbitMQ ${RABBITMQ_IMAGE}"
 
 docker rm -f rabbitmq 2>/dev/null || echo "rabbitmq was not running"

pom.xml

@@ -54,6 +54,7 @@
     <proton-j.version>0.34.1</proton-j.version>
     <metrics.version>4.2.33</metrics.version>
     <micrometer.version>1.15.3</micrometer.version>
+    <gson.version>2.13.1</gson.version>
     <swiftmq-client.version>13.1.2</swiftmq-client.version>
     <picocli.version>4.7.5</picocli.version>
     <commons-compress.version>1.28.0</commons-compress.version>
@@ -66,10 +67,10 @@
     <amqp-client.version>5.26.0</amqp-client.version>
     <commons-lang3.version>3.18.0</commons-lang3.version>
     <commons-codec.version>1.19.0</commons-codec.version>
-    <gson.version>2.13.1</gson.version>
     <vavr.version>0.10.7</vavr.version>
     <paho.version>1.2.5</paho.version>
     <micrometer-tracing-test.version>1.5.3</micrometer-tracing-test.version>
+    <bouncycastle.version>1.81</bouncycastle.version>
     <micrometer-docs-generator.version>1.0.4</micrometer-docs-generator.version>
     <netty-tcnative.version>2.0.72.Final</netty-tcnative.version>
     <maven.compiler.plugin.version>3.14.0</maven.compiler.plugin.version>
@@ -105,6 +106,7 @@
     <central-publishing-maven-plugin.version>0.8.0</central-publishing-maven-plugin.version>
     <maven.javadoc.skip>true</maven.javadoc.skip>
     <gpg.skip>true</gpg.skip>
+    <jose4j.version>0.9.6</jose4j.version>
   </properties>
 
   <dependencies>
@@ -185,6 +187,13 @@
       <optional>true</optional>
     </dependency>
 
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+      <version>${gson.version}</version>
+      <optional>true</optional>
+    </dependency>
+
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-engine</artifactId>
@@ -285,13 +294,6 @@
       <scope>test</scope>
     </dependency>
 
-    <dependency>
-      <groupId>com.google.code.gson</groupId>
-      <artifactId>gson</artifactId>
-      <version>${gson.version}</version>
-      <scope>test</scope>
-    </dependency>
-
     <dependency>
       <groupId>io.vavr</groupId>
       <artifactId>vavr</artifactId>
@@ -313,6 +315,20 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk18on</artifactId>
+      <version>${bouncycastle.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.bitbucket.b_c</groupId>
+      <artifactId>jose4j</artifactId>
+      <version>${jose4j.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.openjdk.jmh</groupId>
       <artifactId>jmh-core</artifactId>

src/main/java/com/rabbitmq/stream/EnvironmentBuilder.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
+import javax.net.ssl.SSLContext;
 
 /**
  * API to configure and create an {@link Environment}.
@@ -517,4 +518,90 @@ interface NettyConfiguration {
      */
     EnvironmentBuilder environmentBuilder();
   }
+
+  /**
+   * OAuth 2 settings.
+   *
+   * @return OAuth 2 settings
+   * @see OAuth2Configuration
+   * @since 1.3.0
+   */
+  OAuth2Configuration oauth2();
+
+  /**
+   * Configuration to retrieve a token using the <a
+   * href="https://tools.ietf.org/html/rfc6749#section-4.4">OAuth 2 Client Credentials flow</a>.
+   *
+   * @since 1.3.0
+   */
+  interface OAuth2Configuration {
+
+    /**
+     * Set the URI to access to get the token.
+     *
+     * <p>TLS is supported by providing a <code>HTTPS</code> URI and setting a {@link
+     * javax.net.ssl.SSLContext}. See {@link #tls()} for more information. <em>Applications in
+     * production should always use HTTPS to retrieve tokens.</em>
+     *
+     * @param uri access URI
+     * @return OAuth 2 configuration
+     * @see #sslContext(javax.net.ssl.SSLContext)
+     */
+    OAuth2Configuration tokenEndpointUri(String uri);
+
+    /**
+     * Set the OAuth 2 client ID
+     *
+     * <p>The client ID usually identifies the application that requests a token.
+     *
+     * @param clientId client ID
+     * @return OAuth 2 configuration
+     */
+    OAuth2Configuration clientId(String clientId);
+
+    /**
+     * Set the secret (password) to use to get a token.
+     *
+     * @param clientSecret client secret
+     * @return OAuth 2 configuration
+     */
+    OAuth2Configuration clientSecret(String clientSecret);
+
+    /**
+     * Set the grant type to use when requesting the token.
+     *
+     * <p>The default is <code>client_credentials</code>, but some OAuth 2 servers can use
+     * non-standard grant types to request tokens with extra-information.
+     *
+     * @param grantType grant type
+     * @return OAuth 2 configuration
+     */
+    OAuth2Configuration grantType(String grantType);
+
+    /**
+     * Set a parameter to pass in the request.
+     *
+     * <p>The OAuth 2 server may require extra parameters to narrow down the identity of the user.
+     *
+     * @param name name of the parameter
+     * @param value value of the parameter
+     * @return OAuth 2 configuration
+     */
+    OAuth2Configuration parameter(String name, String value);
+
+    /**
+     * {@link javax.net.ssl.SSLContext} for HTTPS requests.
+     *
+     * @param sslContext the SSL context
+     * @return OAuth 2 configuration
+     */
+    OAuth2Configuration sslContext(SSLContext sslContext);
+
+    /**
+     * Go back to the environment builder
+     *
+     * @return the environment builder
+     */
+    EnvironmentBuilder environmentBuilder();
+  }
 }

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -48,6 +48,8 @@
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandlerInfo;
 import com.rabbitmq.stream.metrics.MetricsCollector;
 import com.rabbitmq.stream.metrics.NoOpMetricsCollector;
+import com.rabbitmq.stream.oauth2.CredentialsManager;
+import com.rabbitmq.stream.oauth2.CredentialsManager.Registration;
 import com.rabbitmq.stream.sasl.CredentialsProvider;
 import com.rabbitmq.stream.sasl.DefaultSaslConfiguration;
 import com.rabbitmq.stream.sasl.DefaultUsernamePasswordCredentialsProvider;
@@ -115,6 +117,7 @@
  */
 public class Client implements AutoCloseable {
 
+  private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);
   private static final Charset CHARSET = StandardCharsets.UTF_8;
   public static final int DEFAULT_PORT = 5552;
   public static final int DEFAULT_TLS_PORT = 5551;
@@ -170,7 +173,6 @@ public long applyAsLong(Object value) {
       };
   private final AtomicInteger correlationSequence = new AtomicInteger(0);
   private final SaslConfiguration saslConfiguration;
-  private final CredentialsProvider credentialsProvider;
   private final Runnable nettyClosing;
   private final int maxFrameSize;
   private final boolean frameSizeCapped;
@@ -190,6 +192,7 @@ public long applyAsLong(Object value) {
   private final Runnable streamStatsCommandVersionsCheck;
   private final boolean filteringSupported;
   private final Runnable superStreamManagementCommandVersionsCheck;
+  private final Registration credentialsRegistration;
 
   @SuppressFBWarnings("CT_CONSTRUCTOR_THROW")
   public Client() {
@@ -206,7 +209,6 @@ public Client(ClientParameters parameters) {
     this.creditNotification = parameters.creditNotification;
     this.codec = parameters.codec == null ? Codecs.DEFAULT : parameters.codec;
     this.saslConfiguration = parameters.saslConfiguration;
-    this.credentialsProvider = parameters.credentialsProvider;
     this.chunkChecksum = parameters.chunkChecksum;
     this.metricsCollector = parameters.metricsCollector;
     this.metadataListener = parameters.metadataListener;
@@ -381,8 +383,36 @@ public void initChannel(SocketChannel ch) {
       debug(() -> "starting SASL handshake");
       this.saslMechanisms = getSaslMechanisms();
       debug(() -> "SASL mechanisms supported by server ({})", this.saslMechanisms);
+
+      CredentialsProvider credentialsProvider = parameters.credentialsProvider;
+      CredentialsManager credentialsManager = parameters.credentialsManager;
+      CredentialsManager.AuthenticationCallback authCallback, renewCallback;
+      String regName =
+          clientConnectionName.isBlank()
+              ? String.valueOf(ID_SEQUENCE.getAndIncrement())
+              : clientConnectionName + "-" + ID_SEQUENCE.getAndIncrement();
+      if (credentialsManager == null) {
+        this.credentialsRegistration = CredentialsManagerFactory.get();
+        authCallback = (u, p) -> this.authenticate(credentialsProvider);
+      } else {
+        renewCallback =
+            authCallback =
+                (u, p) -> {
+                  if (u == null && p == null) {
+                    // no username/password provided by the credentials manager
+                    // using the credentials manager
+                    this.authenticate(credentialsProvider);
+                  } else {
+                    // the credentials manager provides username/password (e.g. after token
+                    // retrieval)
+                    // we use them with a one-time credentials provider
+                    this.authenticate(new DefaultUsernamePasswordCredentialsProvider(u, p));
+                  }
+                };
+        this.credentialsRegistration = credentialsManager.register(regName, renewCallback);
+      }
       debug(() -> "starting authentication");
-      authenticate(this.credentialsProvider);
+      this.credentialsRegistration.connect(authCallback);
       debug(() -> "authenticated");
       this.tuneState.await(Duration.ofSeconds(10));
       this.maxFrameSize = this.tuneState.getMaxFrameSize();
@@ -1454,6 +1484,9 @@ void closingSequence(ShutdownContext.ShutdownReason reason) {
     if (reason != null) {
       this.shutdownListenerCallback.accept(reason);
     }
+    if (this.credentialsRegistration != null) {
+      this.credentialsRegistration.close();
+    }
     this.nettyClosing.run();
     this.failOutstandingRequests();
     if (this.closeDispatchingExecutorService != null) {
@@ -2378,6 +2411,10 @@ String label() {
     }
   }
 
+  static ClientParameters cp() {
+    return new ClientParameters();
+  }
+
   public static class ClientParameters {
 
     private final Map<String, String> clientProperties = new ConcurrentHashMap<>();
@@ -2410,6 +2447,7 @@ public static class ClientParameters {
     private SaslConfiguration saslConfiguration = DefaultSaslConfiguration.PLAIN;
     private CredentialsProvider credentialsProvider =
         new DefaultUsernamePasswordCredentialsProvider(DEFAULT_USERNAME, "guest");
+    private CredentialsManager credentialsManager;
     private ChunkChecksum chunkChecksum = JdkChunkChecksum.CRC32_SINGLETON;
     private MetricsCollector metricsCollector = NoOpMetricsCollector.SINGLETON;
     private SslContext sslContext;
@@ -2492,6 +2530,11 @@ public ClientParameters credentialsProvider(CredentialsProvider credentialsProvi
       return this;
     }
 
+    public ClientParameters credentialsManager(CredentialsManager credentialsManager) {
+      this.credentialsManager = credentialsManager;
+      return this;
+    }
+
     public ClientParameters username(String username) {
       if (this.credentialsProvider instanceof UsernamePasswordCredentialsProvider) {
         this.credentialsProvider =

src/main/java/com/rabbitmq/stream/impl/ConsumersCoordinator.java

@@ -506,6 +506,18 @@ SubscriptionState state() {
       return this.state.get();
     }
 
+    private void markConsuming() {
+      if (this.consumer != null) {
+        this.consumer.consuming();
+      }
+    }
+
+    private void markNotConsuming() {
+      if (this.consumer != null) {
+        this.consumer.notConsuming();
+      }
+    }
+
     String label() {
       return String.format(
           "[id=%d, stream=%s, name=%s, consumer=%d]",
@@ -700,6 +712,7 @@ private ClientSubscriptionsManager(
                   "Subscription connection has {} consumer(s) over {} stream(s) to recover",
                   this.subscriptionTrackers.stream().filter(Objects::nonNull).count(),
                   this.streamToStreamSubscriptions.size());
+              iterate(this.subscriptionTrackers, SubscriptionTracker::markNotConsuming);
               environment
                   .scheduledExecutorService()
                   .execute(
@@ -774,6 +787,7 @@ private ClientSubscriptionsManager(
             }
 
             if (affectedSubscriptions != null && !affectedSubscriptions.isEmpty()) {
+              iterate(affectedSubscriptions, SubscriptionTracker::markNotConsuming);
               environment
                   .scheduledExecutorService()
                   .execute(
@@ -1132,6 +1146,7 @@ void add(
           throw e;
         }
         subscriptionTracker.state(SubscriptionState.ACTIVE);
+        subscriptionTracker.markConsuming();
         LOGGER.debug("Subscribed to '{}'", subscriptionTracker.stream);
       } finally {
         this.subscriptionManagerLock.unlock();
@@ -1397,4 +1412,13 @@ static Broker pickBroker(
       Function<List<Broker>, Broker> picker, Collection<BrokerWrapper> candidates) {
     return picker.apply(keepReplicasIfPossible(candidates));
   }
+
+  private static void iterate(
+      Collection<SubscriptionTracker> l, java.util.function.Consumer<SubscriptionTracker> c) {
+    for (SubscriptionTracker tracker : l) {
+      if (tracker != null) {
+        c.accept(tracker);
+      }
+    }
+  }
 }