Adds support for RabbitMQ dynamic credentials from Vault

.github/native-tests.json

@@ -63,7 +63,7 @@
         {
             "category": "Messaging2",
             "timeout": 70,
-            "test-modules": "reactive-messaging-amqp, reactive-messaging-http, reactive-messaging-rabbitmq",
+            "test-modules": "reactive-messaging-amqp, reactive-messaging-http, reactive-messaging-rabbitmq, reactive-messaging-rabbitmq-dyn",
             "os-name": "ubuntu-latest"
         },
         {

docs/src/main/asciidoc/credentials-provider.adoc

@@ -35,6 +35,7 @@ by the following credentials consumer extensions:
 * `reactive-pg-client`
 * `oidc`
 * `oidc-client`
+* `smallrye-reactive-messaging-rabbitmq`
 
 All extensions that rely on username/password authentication also allow setting configuration
 properties in the `application.properties` as an alternative. But the `Credentials Provider` is the only option

docs/src/main/asciidoc/rabbitmq-reference.adoc

@@ -393,6 +393,67 @@ Note that a message processing failures nacks the message, which is then handled
 It's the responsibility of the `failure-strategy` to report the failure and influence the outcome of the checks.
 The `fail` failure strategy reports the failure, and so the check will report the fault.
 
+[[dynamic-credentials]]
+== Dynamic Credentials
+
+Quarkus and the RabbitMQ connector support https://www.vaultproject.io/docs/secrets/rabbitmq[Vault's RabbitMQ secrets engine]
+for generating short-lived dynamic credentials. This allows Vault to create and retire RabbitMQ credentials on a regular basis.
+
+First we need to enable Vault's `rabbitmq` secret engine, configure it with RabbitMQ's connection and authentication
+information, and create a Vault role `my-role` (replace `10.0.0.3` by the actual host that is running the
+RabbitMQ container):
+[source,bash, subs=attributes+]
+----
+vault secrets enable rabbitmq
+
+vault write rabbitmq/config/connection \
+    connection_uri=http://10.0.0.3:15672 \
+    username=guest \
+    password=guest
+
+vault write rabbitmq/roles/my-role \
+    vhosts='{"/":{"write": ".*", "read": ".*"}}'
+----
+
+[NOTE]
+====
+For this use case, user `guest` configured above needs to be a RabbitMQ admin user with the capability to create
+credentials.
+====
+
+Then we need to give a read capability to the Quarkus application on path `rabbitmq/creds/my-role`.
+[source,bash]
+----
+cat <<EOF | vault policy write vault-rabbitmq-policy -
+path "secret/data/myapps/vault-rabbitmq-test/*" {
+  capabilities = ["read"]
+}
+path "rabbitmq/creds/my-role" {
+  capabilities = [ "read" ]
+}
+EOF
+----
+
+Now that Vault knows how to create users in RabbitMQ, we need to configure Quarkus to use a credentials-provider
+for RabbitMQ.
+
+First we tell Quarkus to request dynamic credentials using a credentials-provider named `rabbitmq`.
+----
+quarkus.rabbitmq.credentials-provider = rabbitmq
+----
+
+Next we configure the `rabbitmq` credentials provider. The `credentials-role` option must be set to the name of the
+role we created in Vault, in our case `my-role`. The `credentials-mount` option must be set to `rabbitmq`.
+[source, properties]
+----
+quarkus.vault.credentials-provider.rabbitmq.credentials-role = my-role
+quarkus.vault.credentials-provider.rabbitmq.credentials-mount = rabbitmq
+----
+
+NOTE: The `credentials-mount` is used directly as the mount of the secret engine in Vault. Here we are using
+the default mount path of `rabbitmq`. If the RabbitMQ secret engine was mounted at a custom path, the
+`credentials-mount` option must be set to that path instead.
+
 [[configuration-reference]]
 == RabbitMQ Connector Configuration Reference
 

extensions/smallrye-reactive-messaging-rabbitmq/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/rabbitmq/deployment/RabbitMQBuildTimeConfig.java

@@ -1,5 +1,7 @@
 package io.quarkus.smallrye.reactivemessaging.rabbitmq.deployment;
 
+import java.util.Optional;
+
 import io.quarkus.runtime.annotations.ConfigItem;
 import io.quarkus.runtime.annotations.ConfigPhase;
 import io.quarkus.runtime.annotations.ConfigRoot;
@@ -12,4 +14,21 @@ public class RabbitMQBuildTimeConfig {
      */
     @ConfigItem
     public RabbitMQDevServicesBuildTimeConfig devservices;
+
+    /**
+     * The credentials provider name.
+     */
+    @ConfigItem
+    public Optional<String> credentialsProvider = Optional.empty();
+
+    /**
+     * The credentials provider bean name.
+     * <p>
+     * It is the {@code &#64;Named} value of the credentials provider bean. It is used to discriminate if multiple
+     * CredentialsProvider beans are available.
+     * <p>
+     * For Vault it is: vault-credentials-provider. Not necessary if there is only one credentials provider available.
+     */
+    @ConfigItem
+    public Optional<String> credentialsProviderName = Optional.empty();
 }

extensions/smallrye-reactive-messaging-rabbitmq/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/rabbitmq/deployment/SmallRyeReactiveMessagingRabbitMQProcessor.java

@@ -1,8 +1,23 @@
 package io.quarkus.smallrye.reactivemessaging.rabbitmq.deployment;
 
+import javax.enterprise.context.ApplicationScoped;
+
+import com.rabbitmq.client.impl.CredentialsProvider;
+
+import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.arc.deployment.SyntheticBeanBuildItem.ExtendedBeanConfigurator;
 import io.quarkus.deployment.Feature;
+import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.annotations.ExecutionTime;
+import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
+import io.quarkus.deployment.builditem.RunTimeConfigurationDefaultBuildItem;
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.smallrye.reactivemessaging.rabbitmq.runtime.CredentialsProviderLink;
+import io.quarkus.smallrye.reactivemessaging.rabbitmq.runtime.RabbitMQRecorder;
+import io.smallrye.common.annotation.Identifier;
 
 public class SmallRyeReactiveMessagingRabbitMQProcessor {
 
@@ -11,4 +26,38 @@ FeatureBuildItem feature() {
         return new FeatureBuildItem(Feature.SMALLRYE_REACTIVE_MESSAGING_RABBITMQ);
     }
 
+    @BuildStep
+    @Record(ExecutionTime.RUNTIME_INIT)
+    public void dynamicCredentials(RabbitMQRecorder recorder,
+            RabbitMQBuildTimeConfig rabbitMQBuildTimeConfig,
+            BuildProducer<SyntheticBeanBuildItem> syntheticBeans,
+            BuildProducer<AdditionalBeanBuildItem> additionalBeans,
+            BuildProducer<RunTimeConfigurationDefaultBuildItem> configDefaults) {
+
+        additionalBeans.produce(AdditionalBeanBuildItem.builder().addBeanClass(Identifier.class).build());
+
+        if (rabbitMQBuildTimeConfig.credentialsProvider.isPresent()) {
+            String credentialsProvider = rabbitMQBuildTimeConfig.credentialsProvider.get();
+
+            RuntimeValue<CredentialsProviderLink> credentialsProviderLink = recorder.configureOptions(
+                    credentialsProvider,
+                    rabbitMQBuildTimeConfig.credentialsProviderName);
+
+            String identifier = "credentials-provider-link-" + credentialsProvider;
+
+            ExtendedBeanConfigurator rabbitMQOptionsConfigurator = SyntheticBeanBuildItem
+                    .configure(CredentialsProviderLink.class)
+                    .defaultBean()
+                    .addType(CredentialsProvider.class)
+                    .addQualifier().annotation(Identifier.class).addValue("value", identifier).done()
+                    .scope(ApplicationScoped.class)
+                    .runtimeValue(credentialsProviderLink)
+                    .unremovable()
+                    .setRuntimeInit();
+
+            configDefaults.produce(new RunTimeConfigurationDefaultBuildItem("rabbitmq-credentials-provider-name", identifier));
+            syntheticBeans.produce(rabbitMQOptionsConfigurator.done());
+        }
+    }
+
 }

extensions/smallrye-reactive-messaging-rabbitmq/runtime/pom.xml

@@ -30,6 +30,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-vertx</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-credentials</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-jackson</artifactId>

extensions/smallrye-reactive-messaging-rabbitmq/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/rabbitmq/runtime/CredentialsProviderLink.java

@@ -0,0 +1,62 @@
+package io.quarkus.smallrye.reactivemessaging.rabbitmq.runtime;
+
+import static io.quarkus.credentials.CredentialsProvider.EXPIRATION_TIMESTAMP_PROPERTY_NAME;
+import static io.quarkus.credentials.CredentialsProvider.PASSWORD_PROPERTY_NAME;
+import static io.quarkus.credentials.CredentialsProvider.USER_PROPERTY_NAME;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+
+import io.quarkus.credentials.CredentialsProvider;
+
+public class CredentialsProviderLink implements com.rabbitmq.client.impl.CredentialsProvider {
+
+    private final CredentialsProvider credentialsProvider;
+    private final String credentialsProviderName;
+    private String username;
+    private String password;
+    private Instant expiresAt;
+
+    public CredentialsProviderLink(CredentialsProvider credentialsProvider, String credentialsProviderName) {
+        this.credentialsProvider = credentialsProvider;
+        this.credentialsProviderName = credentialsProviderName;
+        this.expiresAt = Instant.MIN;
+    }
+
+    private void refreshIfExpired() {
+        if (expiresAt.isAfter(Instant.now())) {
+            return;
+        }
+        refresh();
+    }
+
+    @Override
+    public String getUsername() {
+        refreshIfExpired();
+        return username;
+    }
+
+    @Override
+    public String getPassword() {
+        refreshIfExpired();
+        return password;
+    }
+
+    @Override
+    public Duration getTimeBeforeExpiration() {
+        return Duration.between(Instant.now(), expiresAt);
+    }
+
+    @Override
+    public void refresh() {
+        Map<String, String> credentials = credentialsProvider.getCredentials(credentialsProviderName);
+        username = credentials.get(USER_PROPERTY_NAME);
+        password = credentials.get(PASSWORD_PROPERTY_NAME);
+        expiresAt = Instant.parse(credentials.getOrDefault(EXPIRATION_TIMESTAMP_PROPERTY_NAME, getDefaultExpiresAt()));
+    }
+
+    private String getDefaultExpiresAt() {
+        return Instant.now().plusSeconds(10).toString();
+    }
+}

extensions/smallrye-reactive-messaging-rabbitmq/runtime/src/main/java/io/quarkus/smallrye/reactivemessaging/rabbitmq/runtime/RabbitMQRecorder.java

@@ -0,0 +1,20 @@
+package io.quarkus.smallrye.reactivemessaging.rabbitmq.runtime;
+
+import java.util.Optional;
+
+import io.quarkus.credentials.CredentialsProvider;
+import io.quarkus.credentials.runtime.CredentialsProviderFinder;
+import io.quarkus.runtime.RuntimeValue;
+import io.quarkus.runtime.annotations.Recorder;
+
+@Recorder
+public class RabbitMQRecorder {
+
+    public RuntimeValue<CredentialsProviderLink> configureOptions(String credentialsProviderName,
+            Optional<String> credentialsProviderBeanName) {
+
+        CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(credentialsProviderBeanName.orElse(null));
+
+        return new RuntimeValue<>(new CredentialsProviderLink(credentialsProvider, credentialsProviderName));
+    }
+}

integration-tests/pom.xml

@@ -263,6 +263,7 @@
                 <module>reactive-messaging-kafka</module>
                 <module>reactive-messaging-http</module>
                 <module>reactive-messaging-rabbitmq</module>
+                <module>reactive-messaging-rabbitmq-dyn</module>
                 <module>rest-client</module>
                 <module>resteasy-reactive-kotlin</module>
                 <module>rest-client-reactive</module>