From 4fba3ad0d49762e78fe9cbc528a82f8666e4b966 Mon Sep 17 00:00:00 2001 From: Abhishek Pal Date: Sun, 2 Aug 2026 16:29:16 +0530 Subject: [PATCH] RATIS-2634. Add TLS support for Netty RPC transport --- ratis-docs/src/site/markdown/security.md | 2 + .../apache/ratis/netty/NettyConfigKeys.java | 20 ++++++ .../org/apache/ratis/netty/NettyFactory.java | 10 ++- .../org/apache/ratis/netty/NettyRpcProxy.java | 29 ++++++-- .../ratis/netty/client/NettyClientRpc.java | 5 +- .../ratis/netty/server/NettyRpcService.java | 21 +++++- .../ratis/netty/MiniRaftClusterWithNetty.java | 11 ++- .../apache/ratis/netty/TestNettyRpcTls.java | 69 +++++++++++++++++++ 8 files changed, 152 insertions(+), 15 deletions(-) create mode 100644 ratis-test/src/test/java/org/apache/ratis/netty/TestNettyRpcTls.java diff --git a/ratis-docs/src/site/markdown/security.md b/ratis-docs/src/site/markdown/security.md index eb233bf932..6ad80f5875 100644 --- a/ratis-docs/src/site/markdown/security.md +++ b/ratis-docs/src/site/markdown/security.md @@ -44,3 +44,5 @@ Applications may use them to build `RaftServer`/`RaftClient` objects for establi | `raft.grpc.admin.tls.conf` | gRPC admin TLS conf | | `raft.netty.dataStream.server.tls.conf` | Netty data stream server TLS conf | | `raft.netty.dataStream.client.tls.conf` | Netty data stream client TLS conf | +| `raft.netty.server.tls.conf` | Netty RPC server TLS conf | +| `raft.netty.client.tls.conf` | Netty RPC client TLS conf | diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java index fd0906ebfe..2a18ab91bc 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyConfigKeys.java @@ -74,6 +74,16 @@ static boolean useEpoll(RaftProperties properties) { static void setUseEpoll(RaftProperties properties, boolean enable) { setBoolean(properties::setBoolean, USE_EPOLL_KEY, enable); } + + String TLS_CONF_PARAMETER = PREFIX + ".tls.conf"; + Class TLS_CONF_CLASS = TlsConf.class; + static TlsConf tlsConf(Parameters parameters) { + return parameters == null ? null + : getTlsConf(key -> parameters.get(key, TLS_CONF_CLASS), TLS_CONF_PARAMETER, getDefaultLog()); + } + static void setTlsConf(Parameters parameters, TlsConf conf) { + ConfUtils.setTlsConf((key, value) -> parameters.put(key, value, TLS_CONF_CLASS), TLS_CONF_PARAMETER, conf); + } } interface Client { @@ -92,6 +102,16 @@ static boolean useEpoll(RaftProperties properties) { static void setUseEpoll(RaftProperties properties, boolean enable) { setBoolean(properties::setBoolean, USE_EPOLL_KEY, enable); } + + String TLS_CONF_PARAMETER = PREFIX + ".tls.conf"; + Class TLS_CONF_CLASS = TlsConf.class; + static TlsConf tlsConf(Parameters parameters) { + return parameters == null ? null + : getTlsConf(key -> parameters.get(key, TLS_CONF_CLASS), TLS_CONF_PARAMETER, getDefaultLog()); + } + static void setTlsConf(Parameters parameters, TlsConf conf) { + ConfUtils.setTlsConf((key, value) -> parameters.put(key, value, TLS_CONF_CLASS), TLS_CONF_PARAMETER, conf); + } } interface DataStream { diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyFactory.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyFactory.java index 24f0142268..8d10a9195f 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyFactory.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyFactory.java @@ -28,7 +28,11 @@ import org.apache.ratis.server.ServerFactory; public class NettyFactory implements ServerFactory, ClientFactory { - public NettyFactory(Parameters parameters) {} + private final Parameters parameters; + + public NettyFactory(Parameters parameters) { + this.parameters = parameters; + } @Override public SupportedRpcType getRpcType() { @@ -37,11 +41,11 @@ public SupportedRpcType getRpcType() { @Override public NettyRpcService newRaftServerRpc(RaftServer server) { - return NettyRpcService.newBuilder().setServer(server).build(); + return NettyRpcService.newBuilder().setServer(server).setParameters(parameters).build(); } @Override public NettyClientRpc newRaftClientRpc(ClientId clientId, RaftProperties properties) { - return new NettyClientRpc(clientId, properties); + return new NettyClientRpc(clientId, properties, parameters); } } diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java index d2eb38859a..3964624d8d 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java @@ -18,11 +18,13 @@ package org.apache.ratis.netty; import org.apache.ratis.client.RaftClientConfigKeys; +import org.apache.ratis.conf.Parameters; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.protocol.exceptions.TimeoutIOException; import org.apache.ratis.thirdparty.io.netty.channel.*; import org.apache.ratis.thirdparty.io.netty.channel.socket.SocketChannel; +import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContext; import org.apache.ratis.thirdparty.io.netty.handler.codec.protobuf.ProtobufDecoder; import org.apache.ratis.thirdparty.io.netty.handler.codec.protobuf.ProtobufEncoder; import org.apache.ratis.thirdparty.io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder; @@ -45,6 +47,7 @@ import java.io.Closeable; import java.io.IOException; +import java.net.InetSocketAddress; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; @@ -60,14 +63,19 @@ public static class PeerMap extends PeerProxyMap { private final EventLoopGroup group; public PeerMap(String name, RaftProperties properties) { - this(name, properties, NettyUtils.newEventLoopGroup(name, 0, - NettyConfigKeys.Client.useEpoll(properties))); + this(name, properties, (Parameters) null); } - private PeerMap(String name, RaftProperties properties, EventLoopGroup group) { + public PeerMap(String name, RaftProperties properties, Parameters parameters) { + this(name, properties, + NettyUtils.buildSslContextForClient(NettyConfigKeys.Client.tlsConf(parameters)), + NettyUtils.newEventLoopGroup(name, 0, NettyConfigKeys.Client.useEpoll(properties))); + } + + private PeerMap(String name, RaftProperties properties, SslContext sslContext, EventLoopGroup group) { super(name, peer -> { try { - return new NettyRpcProxy(peer, properties, group); + return new NettyRpcProxy(peer, properties, group, sslContext); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw IOUtils.toInterruptedIOException("Failed connecting to " + peer, e); @@ -150,7 +158,7 @@ class Connection implements Closeable { private final NettyClient client = new NettyClient(peer.getAddress()); private final Map> replies = new ConcurrentHashMap<>(); - Connection(EventLoopGroup group) throws InterruptedException { + Connection(EventLoopGroup group, SslContext sslContext) throws InterruptedException { final ChannelInboundHandler inboundHandler = new SimpleChannelInboundHandler() { @Override @@ -191,6 +199,10 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { protected void initChannel(SocketChannel ch) { final ChannelPipeline p = ch.pipeline(); + if (sslContext != null) { + final InetSocketAddress address = peer.getAddress(); + p.addLast("ssl", sslContext.newHandler(ch.alloc(), address.getHostName(), address.getPort())); + } // LoggingHandler emits all events at the chosen level; use DEBUG to reduce noise by default. p.addLast(new LoggingHandler(LogLevel.DEBUG)); p.addLast(new ProtobufVarint32FrameDecoder()); @@ -273,8 +285,13 @@ private synchronized void failOutstandingRequests(Throwable cause) { private final TimeDuration requestTimeoutDuration; public NettyRpcProxy(RaftPeer peer, RaftProperties properties, EventLoopGroup group) throws InterruptedException { + this(peer, properties, group, null); + } + + public NettyRpcProxy(RaftPeer peer, RaftProperties properties, EventLoopGroup group, SslContext sslContext) + throws InterruptedException { this.peer = peer; - this.connection = new Connection(group); + this.connection = new Connection(group, sslContext); this.requestTimeoutDuration = RaftClientConfigKeys.Rpc.requestTimeout(properties); } diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientRpc.java b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientRpc.java index ef34caf17d..7afb9397f1 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientRpc.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientRpc.java @@ -20,6 +20,7 @@ import org.apache.ratis.client.RaftClientConfigKeys; import org.apache.ratis.client.impl.ClientProtoUtils; import org.apache.ratis.client.impl.RaftClientRpcWithProxy; +import org.apache.ratis.conf.Parameters; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.netty.NettyRpcProxy; import org.apache.ratis.protocol.*; @@ -47,8 +48,8 @@ public class NettyClientRpc extends RaftClientRpcWithProxy { private final TimeDuration requestTimeout; private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance(); - public NettyClientRpc(ClientId clientId, RaftProperties properties) { - super(new NettyRpcProxy.PeerMap(clientId.toString(), properties)); + public NettyClientRpc(ClientId clientId, RaftProperties properties, Parameters parameters) { + super(new NettyRpcProxy.PeerMap(clientId.toString(), properties, parameters)); this.clientId = clientId; this.requestTimeout = RaftClientConfigKeys.Rpc.requestTimeout(properties); } diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java index f7d2805e8a..cf452363c5 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyRpcService.java @@ -18,8 +18,10 @@ package org.apache.ratis.netty.server; import org.apache.ratis.client.impl.ClientProtoUtils; +import org.apache.ratis.conf.Parameters; import org.apache.ratis.netty.NettyConfigKeys; import org.apache.ratis.netty.NettyRpcProxy; +import org.apache.ratis.security.TlsConf; import org.apache.ratis.util.NettyUtils; import org.apache.ratis.protocol.GroupInfoReply; import org.apache.ratis.protocol.GroupListReply; @@ -37,6 +39,7 @@ import org.apache.ratis.thirdparty.io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender; import org.apache.ratis.thirdparty.io.netty.handler.logging.LogLevel; import org.apache.ratis.thirdparty.io.netty.handler.logging.LoggingHandler; +import org.apache.ratis.thirdparty.io.netty.handler.ssl.SslContext; import org.apache.ratis.proto.RaftProtos.*; import org.apache.ratis.proto.netty.NettyProtos.RaftNettyExceptionReplyProto; import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto; @@ -63,6 +66,7 @@ public final class NettyRpcService extends RaftServerRpcWithProxy new NettyRpcProxy.PeerMap(id.toString(), server.getProperties())); + private NettyRpcService(RaftServer server, Parameters parameters) { + super(server::getId, id -> new NettyRpcProxy.PeerMap(id.toString(), server.getProperties(), parameters)); this.server = server; + final TlsConf tlsConf = NettyConfigKeys.Server.tlsConf(parameters); + final SslContext sslContext = NettyUtils.buildSslContextForServer(tlsConf); + final ChannelInitializer initializer = new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) { final ChannelPipeline p = ch.pipeline(); + if (sslContext != null) { + p.addLast("ssl", sslContext.newHandler(ch.alloc())); + } p.addLast(new ProtobufVarint32FrameDecoder()); p.addLast(new ProtobufDecoder(RaftNettyServerRequestProto.getDefaultInstance())); p.addLast(new ProtobufVarint32LengthFieldPrepender()); diff --git a/ratis-netty/src/test/java/org/apache/ratis/netty/MiniRaftClusterWithNetty.java b/ratis-netty/src/test/java/org/apache/ratis/netty/MiniRaftClusterWithNetty.java index b912e14927..97baee563f 100644 --- a/ratis-netty/src/test/java/org/apache/ratis/netty/MiniRaftClusterWithNetty.java +++ b/ratis-netty/src/test/java/org/apache/ratis/netty/MiniRaftClusterWithNetty.java @@ -55,10 +55,19 @@ protected MiniRaftClusterWithNetty(String[] ids, String[] listenerIds, RaftPrope super(ids, listenerIds, properties, null); } + public MiniRaftClusterWithNetty(String[] ids, RaftProperties properties, Parameters parameters) { + this(ids, new String[0], properties, parameters); + } + + public MiniRaftClusterWithNetty(String[] ids, String[] listenerIds, RaftProperties properties, + Parameters parameters) { + super(ids, listenerIds, properties, parameters); + } + @Override protected Parameters setPropertiesAndInitParameters(RaftPeerId id, RaftGroup group, RaftProperties properties) { NettyConfigKeys.Server.setPort(properties, getPort(id, group)); - return null; + return parameters; } @Override diff --git a/ratis-test/src/test/java/org/apache/ratis/netty/TestNettyRpcTls.java b/ratis-test/src/test/java/org/apache/ratis/netty/TestNettyRpcTls.java new file mode 100644 index 0000000000..b4e4a015ec --- /dev/null +++ b/ratis-test/src/test/java/org/apache/ratis/netty/TestNettyRpcTls.java @@ -0,0 +1,69 @@ +/* + * 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.ratis.netty; + +import org.apache.ratis.BaseTest; +import org.apache.ratis.RaftConfigKeys; +import org.apache.ratis.RaftTestUtil; +import org.apache.ratis.RaftTestUtil.SimpleMessage; +import org.apache.ratis.client.RaftClient; +import org.apache.ratis.conf.Parameters; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.protocol.RaftClientReply; +import org.apache.ratis.rpc.SupportedRpcType; +import org.apache.ratis.security.SecurityTestUtils; +import org.apache.ratis.server.RaftServer; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** + * Test TLS on the Netty RPC transport, covering both server-to-server RPC (election/replication) + * and client-to-server RPC. + */ +public class TestNettyRpcTls extends BaseTest { + @Test + public void testRpcWithMutualTls() throws Exception { + final RaftProperties properties = new RaftProperties(); + RaftConfigKeys.Rpc.setType(properties, SupportedRpcType.NETTY); + + // A single Parameters object carries both the server and client TLS configuration. It is applied + // to every server (its inbound endpoint and its outbound peer connections) and to the client + // created by the cluster, so all Netty RPC traffic is mutually authenticated over TLS. + final Parameters parameters = new Parameters(); + NettyConfigKeys.Server.setTlsConf(parameters, SecurityTestUtils.newServerTlsConfig(true)); + NettyConfigKeys.Client.setTlsConf(parameters, SecurityTestUtils.newClientTlsConfig(true)); + + final MiniRaftClusterWithNetty cluster = new MiniRaftClusterWithNetty( + new String[]{"s0", "s1", "s2"}, properties, parameters); + try { + cluster.start(); + + // A successful election proves server-to-server RPC (RequestVote/AppendEntries) works over TLS. + final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster); + Assertions.assertNotNull(leader, "A leader should be elected over the TLS-secured Netty RPC"); + + // A successful client write proves client-to-server RPC works over TLS. + try (RaftClient client = cluster.createClient(leader.getId())) { + final RaftClientReply reply = client.io().send(new SimpleMessage("hello-tls")); + Assertions.assertTrue(reply.isSuccess()); + } + } finally { + cluster.shutdown(); + } + } +}