RATIS-2634. Add TLS support for Netty RPC transport

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         |

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<TlsConf> 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<TlsConf> 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 {

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

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 @@
     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 @@
     private final NettyClient client = new NettyClient(peer.getAddress());
     private final Map<Long, CompletableFuture<RaftNettyServerReplyProto>> replies = new ConcurrentHashMap<>();
 
-    Connection(EventLoopGroup group) throws InterruptedException {
+    Connection(EventLoopGroup group, SslContext sslContext) throws InterruptedException {
       final ChannelInboundHandler inboundHandler
           = new SimpleChannelInboundHandler<RaftNettyServerReplyProto>() {
         @Override
@@ -191,6 +199,10 @@
         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 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);
   }
 

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<NettyRpcProxy> {
   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);
   }

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<NettyRpcProxy,
 
   public static final class Builder {
     private RaftServer server;
+    private Parameters parameters;
 
     private Builder() {}
 
@@ -71,8 +75,13 @@ public Builder setServer(RaftServer raftServer) {
       return this;
     }
 
+    public Builder setParameters(Parameters params) {
+      this.parameters = params;
+      return this;
+    }
+
     public NettyRpcService build() {
-      return new NettyRpcService(server);
+      return new NettyRpcService(server, parameters);
     }
   }
 
@@ -97,16 +106,22 @@ protected void channelRead0(ChannelHandlerContext ctx, RaftNettyServerRequestPro
   }
 
   /** Constructs a netty server with the given port. */
-  private NettyRpcService(RaftServer server) {
-    super(server::getId, id -> 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<SocketChannel> initializer
         = new ChannelInitializer<SocketChannel>() {
       @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());

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

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