Add test for update secret

References rabbitmq/rabbitmq-server#9187

Author: acogoluegnes

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

@@ -35,6 +35,8 @@ public final class Constants {
   public static final short RESPONSE_CODE_PRECONDITION_FAILED = 17;
   public static final short RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST = 18;
   public static final short RESPONSE_CODE_NO_OFFSET = 19;
+  public static final short RESPONSE_CODE_SASL_CANNOT_CHANGE_MECHANISM = 20;
+  public static final short RESPONSE_CODE_SASL_CANNOT_CHANGE_USERNAME = 21;
 
   public static final short CODE_MESSAGE_ENQUEUEING_FAILED = 10_001;
   public static final short CODE_PRODUCER_NOT_AVAILABLE = 10_002;

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

@@ -190,6 +190,7 @@ public long applyAsLong(Object value) {
   private final Map<String, String> serverProperties;
   private final Map<String, String> connectionProperties;
   private final Duration rpcTimeout;
+  private final List<String> saslMechanisms;
   private volatile ShutdownReason shutdownReason = null;
   private final Runnable exchangeCommandVersionsCheck;
   private final boolean filteringSupported;
@@ -367,7 +368,8 @@ public void initChannel(SocketChannel ch) {
               parameters.requestedMaxFrameSize, (int) parameters.requestedHeartbeat.getSeconds());
       this.clientProperties = clientProperties(parameters.clientProperties);
       this.serverProperties = peerProperties();
-      authenticate();
+      this.saslMechanisms = getSaslMechanisms();
+      authenticate(this.credentialsProvider);
       this.tuneState.await(Duration.ofSeconds(10));
       this.maxFrameSize = this.tuneState.getMaxFrameSize();
       this.frameSizeCopped = this.maxFrameSize() > 0;
@@ -483,14 +485,13 @@ private Map<String, String> peerProperties() {
     }
   }
 
-  private void authenticate() {
-    List<String> saslMechanisms = getSaslMechanisms();
-    SaslMechanism saslMechanism = this.saslConfiguration.getSaslMechanism(saslMechanisms);
+  void authenticate(CredentialsProvider credentialsProvider) {
+    SaslMechanism saslMechanism = this.saslConfiguration.getSaslMechanism(this.saslMechanisms);
 
     byte[] challenge = null;
     boolean authDone = false;
     while (!authDone) {
-      byte[] saslResponse = saslMechanism.handleChallenge(challenge, this.credentialsProvider);
+      byte[] saslResponse = saslMechanism.handleChallenge(challenge, credentialsProvider);
       SaslAuthenticateResponse saslAuthenticateResponse =
           sendSaslAuthenticate(saslMechanism, saslResponse);
       if (saslAuthenticateResponse.isOk()) {

src/test/java/com/rabbitmq/stream/Host.java

@@ -147,6 +147,34 @@ public static Process killStreamLeaderProcess(String stream) throws IOException
             + "\">>) of {ok, Pid} -> exit(Pid, kill); Pid -> exit(Pid, kill) end.'");
   }
 
+  public static void addUser(String username, String password) throws IOException {
+    rabbitmqctl(format("add_user %s %s", username, password));
+  }
+
+  public static void setPermissions(String username, String vhost, String permission)
+      throws IOException {
+    rabbitmqctl(
+        format(
+            "set_permissions --vhost %s %s '%s' '%s' '%s'",
+            vhost, username, permission, permission, permission));
+  }
+
+  public static void changePassword(String username, String newPassword) throws IOException {
+    rabbitmqctl(format("change_password %s %s", username, newPassword));
+  }
+
+  public static void deleteUser(String username) throws IOException {
+    rabbitmqctl(format("delete_user %s", username));
+  }
+
+  public static void addVhost(String vhost) throws IOException {
+    rabbitmqctl("add_vhost " + vhost);
+  }
+
+  public static void deleteVhost(String vhost) throws Exception {
+    rabbitmqctl("delete_vhost " + vhost);
+  }
+
   public static void setEnv(String parameter, String value) throws IOException {
     rabbitmqctl(format("eval 'application:set_env(rabbitmq_stream, %s, %s).'", parameter, value));
   }

src/test/java/com/rabbitmq/stream/impl/AuthenticationTest.java

@@ -13,7 +13,9 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.Host.*;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.rabbitmq.stream.AuthenticationFailureException;
 import com.rabbitmq.stream.Constants;
@@ -49,8 +51,8 @@ void authenticateShouldFailWhenUsingBadCredentials() {
     try {
       cf.get(new Client.ClientParameters().username("bad").password("bad"));
     } catch (AuthenticationFailureException e) {
-      assertThat(
-          e.getMessage().contains(String.valueOf(Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE)));
+      assertThat(e.getMessage())
+          .contains(String.valueOf(Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE));
     }
   }
 
@@ -74,9 +76,8 @@ public byte[] handleChallenge(
                         }
                       }));
     } catch (StreamException e) {
-      assertThat(
-          e.getMessage()
-              .contains(String.valueOf(Constants.RESPONSE_CODE_SASL_MECHANISM_NOT_SUPPORTED)));
+      assertThat(e.getMessage())
+          .contains(String.valueOf(Constants.RESPONSE_CODE_SASL_MECHANISM_NOT_SUPPORTED));
     }
   }
 
@@ -100,7 +101,7 @@ public byte[] handleChallenge(
                         }
                       }));
     } catch (StreamException e) {
-      assertThat(e.getMessage().contains(String.valueOf(Constants.RESPONSE_CODE_SASL_ERROR)));
+      assertThat(e.getMessage()).contains(String.valueOf(Constants.RESPONSE_CODE_SASL_ERROR));
     }
   }
 
@@ -109,9 +110,40 @@ void accessToNonExistingVirtualHostShouldFail() {
     try {
       cf.get(new Client.ClientParameters().virtualHost(UUID.randomUUID().toString()));
     } catch (StreamException e) {
-      assertThat(
-          e.getMessage()
-              .contains(String.valueOf(Constants.RESPONSE_CODE_VIRTUAL_HOST_ACCESS_FAILURE)));
+      assertThat(e.getMessage())
+          .contains(String.valueOf(Constants.RESPONSE_CODE_VIRTUAL_HOST_ACCESS_FAILURE));
     }
   }
+
+  @Test
+  @TestUtils.BrokerVersionAtLeast(TestUtils.BrokerVersion.RABBITMQ_3_13_0)
+  void updateSecret() throws Exception {
+    String username = "stream";
+    String password = "stream";
+    String newPassword = "new-password";
+    try {
+      addUser(username, password);
+      setPermissions(username, "/", "^stream.*$");
+      Client client = cf.get(new Client.ClientParameters().username("stream").password(username));
+      changePassword(username, newPassword);
+      // OK
+      client.authenticate(credentialsProvider(username, newPassword));
+      // wrong password
+      assertThatThrownBy(() -> client.authenticate(credentialsProvider(username, "dummy")))
+          .isInstanceOf(AuthenticationFailureException.class)
+          .hasMessageContaining(String.valueOf(Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE));
+      // cannot change username
+      assertThatThrownBy(() -> client.authenticate(credentialsProvider("guest", "guest")))
+          .isInstanceOf(StreamException.class)
+          .hasMessageContaining(
+              String.valueOf(Constants.RESPONSE_CODE_SASL_CANNOT_CHANGE_USERNAME));
+      client.close();
+    } finally {
+      deleteUser(username);
+    }
+  }
+
+  private static CredentialsProvider credentialsProvider(String username, String password) {
+    return new DefaultUsernamePasswordCredentialsProvider(username, password);
+  }
 }

src/test/java/com/rabbitmq/stream/impl/AuthorisationTest.java

@@ -13,13 +13,13 @@
 // [email protected].
 package com.rabbitmq.stream.impl;
 
+import static com.rabbitmq.stream.Host.*;
 import static com.rabbitmq.stream.impl.TestUtils.b;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.rabbitmq.stream.Constants;
-import com.rabbitmq.stream.Host;
 import com.rabbitmq.stream.OffsetSpecification;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
@@ -43,21 +43,16 @@ public class AuthorisationTest {
 
   @BeforeAll
   static void init() throws Exception {
-    Host.rabbitmqctl("add_vhost " + VH);
-    Host.rabbitmqctl("add_user " + USERNAME + " " + PASSWORD);
-    Host.rabbitmqctl(
-        "set_permissions --vhost "
-            + VH
-            + " "
-            + USERNAME
-            + " '^stream.*$' '^stream.*$' '^stream.*$'");
-    Host.rabbitmqctl("set_permissions --vhost " + VH + " guest '.*' '.*' '.*'");
+    addVhost(VH);
+    addUser(USERNAME, PASSWORD);
+    setPermissions(USERNAME, VH, "^stream.*$");
+    setPermissions("guest", VH, ".*");
   }
 
   @AfterAll
   static void tearDown() throws Exception {
-    Host.rabbitmqctl("delete_user stream");
-    Host.rabbitmqctl("delete_vhost test_stream");
+    deleteUser(USERNAME);
+    deleteVhost(VH);
   }
 
   static boolean await(CountDownLatch latch) {