Skip to content

Commit 56b0cf0

Browse files
JeanLucGraphaloJeanLucGraphalo
committed
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 NettyConnection - try to get certificate from channel if not already cached (code is moved from CertificateUtil to NettyConnection)
1 parent 7c29b13 commit 56b0cf0

3 files changed

Lines changed: 109 additions & 55 deletions

File tree

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

Lines changed: 2 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -17,26 +17,17 @@
1717
package org.apache.activemq.artemis.core.remoting;
1818

1919
import javax.net.ssl.SSLPeerUnverifiedException;
20-
import java.io.ByteArrayInputStream;
2120
import java.security.Principal;
22-
import java.security.cert.Certificate;
23-
import java.security.cert.CertificateFactory;
2421
import java.security.cert.X509Certificate;
2522

26-
import io.netty.channel.Channel;
2723
import io.netty.channel.ChannelHandler;
2824
import io.netty.handler.ssl.SslHandler;
2925
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnection;
3026
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
3127
import org.apache.activemq.artemis.spi.core.remoting.Connection;
32-
import org.slf4j.Logger;
33-
import org.slf4j.LoggerFactory;
34-
import java.lang.invoke.MethodHandles;
3528

3629
public class CertificateUtil {
3730

38-
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
39-
4031
private static final String SSL_HANDLER_NAME = "ssl";
4132

4233
public static String getCertSubjectDN(RemotingConnection connection) {
@@ -50,11 +41,8 @@ public static String getCertSubjectDN(RemotingConnection connection) {
5041

5142
public static X509Certificate[] getCertsFromConnection(RemotingConnection remotingConnection) {
5243
X509Certificate[] certificates = null;
53-
if (remotingConnection != null) {
54-
Connection transportConnection = remotingConnection.getTransportConnection();
55-
if (transportConnection instanceof NettyConnection nettyConnection) {
56-
certificates = getCertsFromChannel(nettyConnection.getChannel());
57-
}
44+
if (remotingConnection != null && remotingConnection.getTransportConnection() instanceof NettyConnection nettyConnection) {
45+
certificates = nettyConnection.getCertificates();
5846
}
5947
return certificates;
6048
}
@@ -86,45 +74,4 @@ public static Principal getLocalPrincipalFromConnection(NettyConnection nettyCon
8674

8775
return result;
8876
}
89-
90-
private static X509Certificate[] getCertsFromChannel(Channel channel) {
91-
Certificate[] plainCerts = null;
92-
ChannelHandler channelHandler = channel.pipeline().get("ssl");
93-
if (channelHandler != null && channelHandler instanceof SslHandler sslHandler) {
94-
try {
95-
plainCerts = sslHandler.engine().getSession().getPeerCertificates();
96-
} catch (SSLPeerUnverifiedException e) {
97-
// ignore
98-
}
99-
}
100-
101-
/*
102-
* When using the OpenSSL provider on the broker the getPeerCertificates() method does *not* return a
103-
* X509Certificate[] so we need to convert the Certificate[] that is returned. This code is inspired by Tomcat's
104-
* org.apache.tomcat.util.net.jsse.JSSESupport class.
105-
*/
106-
X509Certificate[] x509Certs = null;
107-
if (plainCerts != null && plainCerts.length > 0) {
108-
x509Certs = new X509Certificate[plainCerts.length];
109-
for (int i = 0; i < plainCerts.length; i++) {
110-
if (plainCerts[i] instanceof X509Certificate x509Certificate) {
111-
x509Certs[i] = x509Certificate;
112-
} else {
113-
try {
114-
x509Certs[i] = (X509Certificate) CertificateFactory
115-
.getInstance("X.509").generateCertificate(new ByteArrayInputStream(plainCerts[i].getEncoded()));
116-
} catch (Exception ex) {
117-
logger.trace("Failed to convert SSL cert", ex);
118-
return null;
119-
}
120-
}
121-
122-
if (logger.isTraceEnabled()) {
123-
logger.trace("Cert #{} = {}", i, x509Certs[i]);
124-
}
125-
}
126-
}
127-
128-
return x509Certs;
129-
}
13077
}

‎artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,13 @@
1616
*/
1717
package org.apache.activemq.artemis.core.remoting.impl.netty;
1818

19+
import javax.net.ssl.SSLPeerUnverifiedException;
20+
import java.io.ByteArrayInputStream;
1921
import java.lang.invoke.MethodHandles;
2022
import java.net.SocketAddress;
23+
import java.security.cert.Certificate;
24+
import java.security.cert.CertificateFactory;
25+
import java.security.cert.X509Certificate;
2126
import java.util.ArrayList;
2227
import java.util.List;
2328
import java.util.Map;
@@ -29,9 +34,11 @@
2934
import io.netty.channel.Channel;
3035
import io.netty.channel.ChannelFuture;
3136
import io.netty.channel.ChannelFutureListener;
37+
import io.netty.channel.ChannelHandler;
3238
import io.netty.channel.ChannelOutboundBuffer;
3339
import io.netty.channel.ChannelPromise;
3440
import io.netty.channel.EventLoop;
41+
import io.netty.handler.ssl.SslHandler;
3542
import io.netty.util.concurrent.FastThreadLocal;
3643
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
3744
import org.apache.activemq.artemis.api.core.ActiveMQInterruptedException;
@@ -74,6 +81,8 @@ public class NettyConnection implements Connection {
7481

7582
private boolean ready = true;
7683

84+
private X509Certificate[] certificates;
85+
7786
public NettyConnection(final Map<String, Object> configuration,
7887
final Channel channel,
7988
final BaseConnectionLifeCycleListener<?> listener,
@@ -481,4 +490,48 @@ private void closeChannel(final Channel channel, boolean inEventLoop) {
481490
}
482491
}
483492

493+
public X509Certificate[] getCertificates() {
494+
return Objects.requireNonNullElse(certificates, getCertsFromChannel());
495+
}
496+
497+
private X509Certificate[] getCertsFromChannel() {
498+
Certificate[] plainCerts = null;
499+
ChannelHandler channelHandler = channel.pipeline().get("ssl");
500+
if (channelHandler instanceof SslHandler sslHandler) {
501+
try {
502+
plainCerts = sslHandler.engine().getSession().getPeerCertificates();
503+
} catch (SSLPeerUnverifiedException e) {
504+
// ignore
505+
}
506+
}
507+
508+
/*
509+
* When using the OpenSSL provider on the broker the getPeerCertificates() method does *not* return a
510+
* X509Certificate[] so we need to convert the Certificate[] that is returned. This code is inspired by Tomcat's
511+
* org.apache.tomcat.util.net.jsse.JSSESupport class.
512+
*/
513+
certificates = null;
514+
if (plainCerts != null && plainCerts.length > 0) {
515+
certificates = new X509Certificate[plainCerts.length];
516+
for (int i = 0; i < plainCerts.length; i++) {
517+
if (plainCerts[i] instanceof X509Certificate x509Certificate) {
518+
certificates[i] = x509Certificate;
519+
} else {
520+
try {
521+
certificates[i] = (X509Certificate) CertificateFactory
522+
.getInstance("X.509").generateCertificate(new ByteArrayInputStream(plainCerts[i].getEncoded()));
523+
} catch (Exception ex) {
524+
logger.trace("Failed to convert SSL cert", ex);
525+
return null;
526+
}
527+
}
528+
529+
if (logger.isTraceEnabled()) {
530+
logger.trace("Cert #{} = {}", i, certificates[i]);
531+
}
532+
}
533+
}
534+
return certificates;
535+
}
536+
484537
}

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

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,10 @@
3535
import org.apache.activemq.artemis.utils.RandomUtil;
3636
import org.apache.activemq.artemis.utils.Wait;
3737
import org.eclipse.paho.mqttv5.client.MqttClient;
38+
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
39+
import org.eclipse.paho.mqttv5.common.MqttException;
3840
import org.eclipse.paho.mqttv5.common.MqttMessage;
41+
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
3942
import org.junit.jupiter.api.TestTemplate;
4043
import org.junit.jupiter.api.Timeout;
4144
import org.junit.jupiter.api.extension.ExtendWith;
@@ -120,4 +123,55 @@ public void messageArrived(String topic, MqttMessage message) {
120123
producer.publish(topic, body, 1, false);
121124
assertTrue(latch.await(500, TimeUnit.MILLISECONDS));
122125
}
126+
127+
// Send will message using mutual TLS with certificate-based authentication
128+
@TestTemplate
129+
@Timeout(DEFAULT_TIMEOUT_SEC)
130+
void testSendWillMessage() throws Exception {
131+
final String willSenderId = RandomUtil.randomUUIDString();
132+
final String willTopic = RandomUtil.randomUUIDString();
133+
final byte[] willBody = RandomUtil.randomBytes(32);
134+
135+
CountDownLatch latch = new CountDownLatch(1);
136+
MqttClient willSender = createConnectedWillSender(willSenderId, willTopic, willBody);
137+
MqttClient willConsumer = createConnectedWillConsumer(latch, willTopic, willBody);
138+
139+
if (protocol.equals(WSS)) {
140+
willSender.disconnectForcibly(0, 0, false);
141+
} else {
142+
// for some reason disconnectForcibly doesn't work in this case so we trick the broker into sending the LWT
143+
getSessionStates().get(willSenderId).setFailed(true);
144+
getSessionStates().get(willSenderId).setAttached(false);
145+
}
146+
assertTrue(latch.await(3, TimeUnit.SECONDS));
147+
willConsumer.disconnect();
148+
}
149+
150+
private MqttClient createConnectedWillSender(String clientId, String topic, byte[] body) throws MqttException {
151+
MqttClient willSender = createPahoClient(protocol, clientId);
152+
MqttConnectionOptions options = getSslMqttConnectOptions();
153+
options.setSessionExpiryInterval(5L);
154+
options.setWill(topic, new MqttMessage(body));
155+
MqttProperties willMessageProperties = new MqttProperties();
156+
willMessageProperties.setWillDelayInterval(1L);
157+
options.setWillMessageProperties(willMessageProperties);
158+
willSender.connect(options);
159+
return willSender;
160+
}
161+
162+
private MqttClient createConnectedWillConsumer(CountDownLatch latch,
163+
String topic,
164+
byte[] body) throws MqttException {
165+
MqttClient willConsumer = createPahoClient(protocol, RandomUtil.randomUUIDString());
166+
willConsumer.connect(getSslMqttConnectOptions());
167+
willConsumer.setCallback(new DefaultMqttCallback() {
168+
@Override
169+
public void messageArrived(String topic, MqttMessage message) {
170+
assertEqualsByteArrays(body, message.getPayload());
171+
latch.countDown();
172+
}
173+
});
174+
willConsumer.subscribe(topic, AT_LEAST_ONCE);
175+
return willConsumer;
176+
}
123177
}

0 commit comments

Comments
 (0)