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.
  - move CertificateUtil to artemis-server
  - cache X509Certificate in NettyServerConnection
  - try to get certificate from channel if not already cached

Author: JeanLucGraphalo

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/AMQPBrokerConnection.java

@@ -49,7 +49,6 @@
 import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPMirrorBrokerConnectionElement;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.postoffice.QueueBinding;
-import org.apache.activemq.artemis.core.remoting.CertificateUtil;
 import org.apache.activemq.artemis.core.remoting.CloseListener;
 import org.apache.activemq.artemis.core.remoting.FailureListener;
 import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
@@ -99,6 +98,7 @@
 import org.apache.activemq.artemis.spi.core.remoting.ClientProtocolManager;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.security.scram.SCRAM;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.apache.activemq.artemis.utils.ConfigurationHelper;
 import org.apache.activemq.artemis.utils.UUIDGenerator;
 import org.apache.qpid.proton.amqp.Symbol;

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/ExternalServerSASLFactory.java

@@ -18,13 +18,13 @@
 
 import java.security.Principal;
 
-import org.apache.activemq.artemis.core.remoting.CertificateUtil;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.protocol.amqp.broker.AmqpInterceptor;
 import org.apache.activemq.artemis.protocol.amqp.proton.AMQPRoutingHandler;
 import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.lang.invoke.MethodHandles;

artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyServerConnection.java

@@ -16,11 +16,13 @@
  */
 package org.apache.activemq.artemis.core.remoting.impl.netty;
 
+import java.security.cert.X509Certificate;
 import java.util.Map;
 
 import io.netty.channel.Channel;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.apache.activemq.artemis.spi.core.remoting.ServerConnectionLifeCycleListener;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.apache.activemq.artemis.utils.ProxyProtocolUtil;
 
 public class NettyServerConnection extends NettyConnection {
@@ -29,6 +31,8 @@ public class NettyServerConnection extends NettyConnection {
 
    private final String router;
 
+   private X509Certificate[] certificates;
+
    public NettyServerConnection(Map<String, Object> configuration,
                                 Channel channel,
                                 ServerConnectionLifeCycleListener listener,
@@ -54,6 +58,13 @@ public String getRouter() {
       return router;
    }
 
+   public X509Certificate[] getPeerCertificates() {
+      if (certificates == null) {
+         certificates = CertificateUtil.getCertsFromChannel(channel);
+      }
+      return certificates;
+   }
+
    /**
     * {@return a string representation of the remote address of this connection; if this connection was made via the
     * proxy protocol then this will be the original address, not the proxy address}

artemis-server/src/main/java/org/apache/activemq/artemis/core/security/impl/SecurityStoreImpl.java

@@ -36,7 +36,6 @@
 import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
 import org.apache.activemq.artemis.core.management.impl.ManagementRemotingConnection;
-import org.apache.activemq.artemis.core.remoting.CertificateUtil;
 import org.apache.activemq.artemis.core.security.CheckType;
 import org.apache.activemq.artemis.core.security.Role;
 import org.apache.activemq.artemis.core.security.SecurityAuth;
@@ -58,6 +57,7 @@
 import org.apache.activemq.artemis.spi.core.security.jaas.NoCacheLoginException;
 import org.apache.activemq.artemis.spi.core.security.jaas.UserPrincipal;
 import org.apache.activemq.artemis.utils.ByteUtil;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;

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

@@ -67,7 +67,6 @@
 import org.apache.activemq.artemis.core.postoffice.QueueBinding;
 import org.apache.activemq.artemis.core.postoffice.RoutingStatus;
 import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
-import org.apache.activemq.artemis.core.remoting.CertificateUtil;
 import org.apache.activemq.artemis.core.remoting.CloseListener;
 import org.apache.activemq.artemis.core.remoting.FailureListener;
 import org.apache.activemq.artemis.core.security.CheckType;
@@ -104,6 +103,7 @@
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
 import org.apache.activemq.artemis.utils.ByteUtil;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.apache.activemq.artemis.utils.CompositeAddress;
 import org.apache.activemq.artemis.utils.JsonLoader;
 import org.apache.activemq.artemis.utils.PrefixUtil;

artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/impl/NotificationActiveMQServerPlugin.java

@@ -23,7 +23,6 @@
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
 import org.apache.activemq.artemis.api.core.management.ManagementHelper;
-import org.apache.activemq.artemis.core.remoting.CertificateUtil;
 import org.apache.activemq.artemis.core.server.ActiveMQServer;
 import org.apache.activemq.artemis.core.server.MessageReference;
 import org.apache.activemq.artemis.core.server.ServerConsumer;
@@ -32,6 +31,7 @@
 import org.apache.activemq.artemis.core.server.management.Notification;
 import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerPlugin;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
+import org.apache.activemq.artemis.utils.CertificateUtil;
 import org.apache.activemq.artemis.utils.collections.TypedProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/ActiveMQJAASSecurityManager.java

@@ -35,7 +35,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getCertsFromConnection;
+import static org.apache.activemq.artemis.utils.CertificateUtil.getCertsFromConnection;
 
 /**
  * This implementation delegates to the JAAS security interfaces.

artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/JaasCallbackHandler.java

@@ -16,8 +16,8 @@
  */
 package org.apache.activemq.artemis.spi.core.security.jaas;
 
-import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getCertsFromConnection;
-import static org.apache.activemq.artemis.core.remoting.CertificateUtil.getPeerPrincipalFromConnection;
+import static org.apache.activemq.artemis.utils.CertificateUtil.getCertsFromConnection;
+import static org.apache.activemq.artemis.utils.CertificateUtil.getPeerPrincipalFromConnection;
 
 import java.io.IOException;
 import java.security.Principal;

…temis/core/remoting/CertificateUtil.java …ivemq/artemis/utils/CertificateUtil.javaartemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/CertificateUtil.java renamed to artemis-server/src/main/java/org/apache/activemq/artemis/utils/CertificateUtil.java artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/CertificateUtil.java renamed to artemis-server/src/main/java/org/apache/activemq/artemis/utils/CertificateUtil.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.activemq.artemis.core.remoting;
+package org.apache.activemq.artemis.utils;
 
 import javax.net.ssl.SSLPeerUnverifiedException;
 import java.io.ByteArrayInputStream;
@@ -27,6 +27,7 @@
 import io.netty.channel.ChannelHandler;
 import io.netty.handler.ssl.SslHandler;
 import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
+import org.apache.activemq.artemis.core.remoting.impl.netty.NettyServerConnection;
 import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 import org.apache.activemq.artemis.spi.core.remoting.Connection;
 import org.slf4j.Logger;
@@ -52,7 +53,9 @@ public static X509Certificate[] getCertsFromConnection(RemotingConnection remoti
       X509Certificate[] certificates = null;
       if (remotingConnection != null) {
          Connection transportConnection = remotingConnection.getTransportConnection();
-         if (transportConnection instanceof NettyConnection nettyConnection) {
+         if (transportConnection instanceof NettyServerConnection nettyServerConnection) {
+            certificates = nettyServerConnection.getPeerCertificates();
+         } else if (transportConnection instanceof NettyConnection nettyConnection) {
             certificates = getCertsFromChannel(nettyConnection.getChannel());
          }
       }
@@ -87,7 +90,7 @@ public static Principal getLocalPrincipalFromConnection(NettyConnection nettyCon
       return result;
    }
 
-   private static X509Certificate[] getCertsFromChannel(Channel channel) {
+   public static X509Certificate[] getCertsFromChannel(Channel channel) {
       Certificate[] plainCerts = null;
       ChannelHandler channelHandler = channel.pipeline().get("ssl");
       if (channelHandler != null && channelHandler instanceof SslHandler sslHandler) {

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;
+   }
 }