ARTEMIS-5163 Artemis fails to send mqtt will message using mutual TLS

The client's certificate is cached in order to successfully authenticate when sending LWT using mutual TLS.
  - cache X509Certificate in RemotingConnection
  - try to get certificate from channel if not already cached (CertificateUtil)

Author: JeanLucGraphalo

artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/CertificateUtil.java

@@ -51,9 +51,11 @@ public static String getCertSubjectDN(RemotingConnection connection) {
    public static X509Certificate[] getCertsFromConnection(RemotingConnection remotingConnection) {
       X509Certificate[] certificates = null;
       if (remotingConnection != null) {
-         Connection transportConnection = remotingConnection.getTransportConnection();
-         if (transportConnection instanceof NettyConnection nettyConnection) {
+         if (remotingConnection.getCertificates() != null) {
+            certificates = remotingConnection.getCertificates();
+         } else if (remotingConnection.getTransportConnection() instanceof NettyConnection nettyConnection) {
             certificates = getCertsFromChannel(nettyConnection.getChannel());
+            remotingConnection.setCertificates(certificates);
          }
       }
       return certificates;

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java

@@ -16,6 +16,7 @@
  */
 package org.apache.activemq.artemis.spi.core.protocol;
 
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -52,6 +53,7 @@ public abstract class AbstractRemotingConnection implements RemotingConnection {
    protected volatile boolean dataReceived;
    private String clientId;
    private Subject subject;
+   private X509Certificate[] certificates;
 
    public AbstractRemotingConnection(final Connection transportConnection, final Executor executor) {
       this.transportConnection = transportConnection;
@@ -282,6 +284,16 @@ public Subject getSubject() {
       return subject;
    }
 
+   @Override
+   public void setCertificates(X509Certificate[] certificates) {
+      this.certificates = certificates;
+   }
+
+   @Override
+   public X509Certificate[] getCertificates() {
+      return certificates;
+   }
+
    @Override
    public void setClientID(String clientId) {
       this.clientId = clientId;

artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java

@@ -17,6 +17,7 @@
 package org.apache.activemq.artemis.spi.core.protocol;
 
 import javax.security.auth.Subject;
+import java.security.cert.X509Certificate;
 import java.util.List;
 import java.util.concurrent.Future;
 
@@ -219,6 +220,16 @@ default void disconnect(DisconnectReason reason, String targetNodeID, TransportC
     */
    Subject getSubject();
 
+   /**
+    * sets the associated certificates for this connection
+    */
+   void setCertificates(X509Certificate[] certificates);
+
+   /**
+    * the possibly null certificates associated with this connection
+    */
+   X509Certificate[] getCertificates();
+
    /**
     * {@return the name of the protocol for this Remoting Connection}
     */

artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/ManagementRemotingConnection.java

@@ -18,6 +18,7 @@
 package org.apache.activemq.artemis.core.management.impl;
 
 import javax.security.auth.Subject;
+import java.security.cert.X509Certificate;
 import java.util.List;
 import java.util.concurrent.Future;
 
@@ -180,6 +181,15 @@ public Subject getSubject() {
       return subject;
    }
 
+   @Override
+   public void setCertificates(X509Certificate[] certificates) {
+   }
+
+   @Override
+   public X509Certificate[] getCertificates() {
+      return null;
+   }
+
    @Override
    public String getProtocolName() {
       return null;

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt5/ssl/CertificateAuthenticationSslTests.java

@@ -35,7 +35,10 @@
 import org.apache.activemq.artemis.utils.RandomUtil;
 import org.apache.activemq.artemis.utils.Wait;
 import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttException;
 import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -120,4 +123,55 @@ public void messageArrived(String topic, MqttMessage message) {
       producer.publish(topic, body, 1, false);
       assertTrue(latch.await(500, TimeUnit.MILLISECONDS));
    }
+
+   // Send will message using mutual TLS with certificate-based authentication
+   @TestTemplate
+   @Timeout(DEFAULT_TIMEOUT_SEC)
+   void testSendWillMessage() throws Exception {
+      final String willSenderId = RandomUtil.randomUUIDString();
+      final String willTopic = RandomUtil.randomUUIDString();
+      final byte[] willBody = RandomUtil.randomBytes(32);
+
+      CountDownLatch latch = new CountDownLatch(1);
+      MqttClient willSender = createConnectedWillSender(willSenderId, willTopic, willBody);
+      MqttClient willConsumer = createConnectedWillConsumer(latch, willTopic, willBody);
+
+      if (protocol.equals(WSS)) {
+         willSender.disconnectForcibly(0, 0, false);
+      } else {
+         // for some reason disconnectForcibly doesn't work in this case so we trick the broker into sending the LWT
+         getSessionStates().get(willSenderId).setFailed(true);
+         getSessionStates().get(willSenderId).setAttached(false);
+      }
+      assertTrue(latch.await(3, TimeUnit.SECONDS));
+      willConsumer.disconnect();
+   }
+
+   private MqttClient createConnectedWillSender(String clientId, String topic, byte[] body) throws MqttException {
+      MqttClient willSender = createPahoClient(protocol, clientId);
+      MqttConnectionOptions options = getSslMqttConnectOptions();
+      options.setSessionExpiryInterval(5L);
+      options.setWill(topic, new MqttMessage(body));
+      MqttProperties willMessageProperties = new MqttProperties();
+      willMessageProperties.setWillDelayInterval(1L);
+      options.setWillMessageProperties(willMessageProperties);
+      willSender.connect(options);
+      return willSender;
+   }
+
+   private MqttClient createConnectedWillConsumer(CountDownLatch latch,
+                                                  String topic,
+                                                  byte[] body) throws MqttException {
+      MqttClient willConsumer = createPahoClient(protocol, RandomUtil.randomUUIDString());
+      willConsumer.connect(getSslMqttConnectOptions());
+      willConsumer.setCallback(new DefaultMqttCallback() {
+         @Override
+         public void messageArrived(String topic, MqttMessage message) {
+            assertEqualsByteArrays(body, message.getPayload());
+            latch.countDown();
+         }
+      });
+      willConsumer.subscribe(topic, AT_LEAST_ONCE);
+      return willConsumer;
+   }
 }