You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2024/05/10 13:16:01 UTC
(rocketmq-mqtt) 03/05: feat: support WebSocket over TLS transport
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
commit 0643cd3a7ee95bbd32cb9f7320c08cbe5429360f
Author: Li Zhanhui <li...@gmail.com>
AuthorDate: Tue Apr 30 11:40:50 2024 +0800
feat: support WebSocket over TLS transport
Signed-off-by: Li Zhanhui <li...@gmail.com>
---
.../mqtt/cs/channel/AdaptiveTlsHandler.java | 93 ++++++++++++++++++++++
.../rocketmq/mqtt/cs/protocol/ssl/SslFactory.java | 4 +
.../rocketmq/mqtt/cs/starter/MqttServer.java | 20 +++++
3 files changed, 117 insertions(+)
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java
new file mode 100644
index 0000000..d2271ac
--- /dev/null
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/AdaptiveTlsHandler.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.mqtt.cs.channel;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.ssl.SslHandler;
+import java.util.NoSuchElementException;
+import org.apache.rocketmq.mqtt.cs.protocol.ssl.SslFactory;
+import org.apache.rocketmq.remoting.common.TlsMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ChannelHandler.Sharable
+public class AdaptiveTlsHandler extends SimpleChannelInboundHandler<ByteBuf> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(AdaptiveTlsHandler.class);
+
+ private final TlsMode tlsMode;
+
+ private static final byte HANDSHAKE_MAGIC_CODE = 0x16;
+
+ private final SslFactory sslFactory;
+
+ public AdaptiveTlsHandler(TlsMode tlsMode, SslFactory sslFactory) {
+ this.tlsMode = tlsMode;
+ this.sslFactory = sslFactory;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+
+ // Peek the current read index byte to determine if the content is starting with TLS handshake
+ byte b = msg.getByte(msg.readerIndex());
+
+ if (b == HANDSHAKE_MAGIC_CODE) {
+ switch (tlsMode) {
+ case DISABLED:
+ ctx.close();
+ LOGGER.warn("Clients intend to establish an SSL connection while this server is running in SSL disabled mode");
+ throw new UnsupportedOperationException("The NettyRemotingServer in SSL disabled mode doesn't support ssl client");
+ case PERMISSIVE:
+ case ENFORCING:
+ if (null != sslFactory.getSslContext()) {
+ ctx.pipeline()
+ .addAfter(AdaptiveTlsHandler.class.getSimpleName(), SslHandler.class.getSimpleName(),
+ sslFactory.getSslContext().newHandler(ctx.alloc()));
+ LOGGER.info("Handlers prepended to channel pipeline to establish SSL connection");
+ } else {
+ ctx.close();
+ LOGGER.error("Trying to establish an SSL connection but sslContext is null");
+ }
+ break;
+
+ default:
+ LOGGER.warn("Unknown TLS mode");
+ break;
+ }
+ } else if (tlsMode == TlsMode.ENFORCING) {
+ ctx.close();
+ LOGGER.warn("Clients intend to establish an insecure connection while this server is running in SSL enforcing mode");
+ }
+
+ try {
+ // Remove this handler
+ ctx.pipeline().remove(this);
+ } catch (NoSuchElementException e) {
+ LOGGER.error("Error while removing TlsModeHandler", e);
+ }
+
+ // Hand over this message to the next handler.
+ ctx.fireChannelRead(msg.retain());
+ }
+
+
+}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
index b9cc770..08d776e 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/protocol/ssl/SslFactory.java
@@ -53,6 +53,7 @@ public class SslFactory {
@PostConstruct
private void initSslContext() {
if (!connectConf.isEnableTlsSever()) {
+ LOG.info("SSL/TLS is disabled");
return;
}
@@ -139,5 +140,8 @@ public class SslFactory {
return sslEngine;
}
+ public SslContext getSslContext() {
+ return sslContext;
+ }
}
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
index f4d0183..94d67ff 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/starter/MqttServer.java
@@ -32,12 +32,14 @@ import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.rocketmq.mqtt.cs.channel.AdaptiveTlsHandler;
import org.apache.rocketmq.mqtt.cs.channel.ConnectHandler;
import org.apache.rocketmq.mqtt.cs.config.ConnectConf;
import org.apache.rocketmq.mqtt.cs.protocol.mqtt.MqttPacketDispatcher;
import org.apache.rocketmq.mqtt.cs.protocol.ssl.SslFactory;
import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketServerHandler;
import org.apache.rocketmq.mqtt.cs.protocol.ws.WebSocketEncoder;
+import org.apache.rocketmq.remoting.common.TlsMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
@@ -69,8 +71,12 @@ public class MqttServer {
@Resource
private SslFactory sslFactory;
+ private AdaptiveTlsHandler adaptiveTlsHandler;
+
@PostConstruct
public void init() throws Exception {
+ adaptiveTlsHandler = new AdaptiveTlsHandler(TlsMode.PERMISSIVE, sslFactory);
+
start();
startWs();
startTls();
@@ -129,6 +135,19 @@ public class MqttServer {
logger.warn("start mqtt tls server , port:{}", tlsPort);
}
+ /**
+ * Support Web Socket Transport.
+ * </p>
+ *
+ * Both WebSocket and WebSocket-Over-TLS/SSL are supported.
+ * </p>
+ *
+ * Clients are supposed to use one of the following Server URIs to connect:
+ * <ul>
+ * <li>ws://host:port/mqtt</li>
+ * <li>wss://host:port/mqtt</li>
+ * </ul>
+ */
private void startWs() {
int port = connectConf.getMqttWsPort();
wsServerBootstrap
@@ -144,6 +163,7 @@ public class MqttServer {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
+ pipeline.addLast(AdaptiveTlsHandler.class.getSimpleName(), adaptiveTlsHandler);
pipeline.addLast("connectHandler", connectHandler);
pipeline.addLast("http-codec", new HttpServerCodec(1024, 32 * 1024, connectConf.getMaxPacketSizeInByte() * 2, true));
pipeline.addLast("aggregator", new HttpObjectAggregator(connectConf.getMaxPacketSizeInByte() * 2));