HBASE-27279 Make SslHandler work with SaslWrapHandler/SaslUnwrapHandler (#4705)

Signed-off-by: Bryan Beaudreault <[email protected]>
Reviewed-by: Andor Molnár <[email protected]>

Author: Apache9

hbase-asyncfs/src/test/java/org/apache/hadoop/hbase/security/HBaseKerberosUtils.java

@@ -188,4 +188,13 @@ public static UserGroupInformation loginAndReturnUGI(Configuration conf, String
       UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keyTabFileLocation);
     return ugi;
   }
+
+  public static UserGroupInformation loginKerberosPrincipal(String krbKeytab, String krbPrincipal)
+    throws Exception {
+    Configuration conf = new Configuration();
+    conf.set(CommonConfigurationKeys.HADOOP_SECURITY_AUTHENTICATION, "kerberos");
+    UserGroupInformation.setConfiguration(conf);
+    UserGroupInformation.loginUserFromKeytab(krbPrincipal, krbKeytab);
+    return UserGroupInformation.getLoginUser();
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java

@@ -221,7 +221,8 @@ private void saslNegotiate(final Channel ch) {
       return;
     }
     ch.pipeline().addBefore(BufferCallBeforeInitHandler.NAME, null, new SaslChallengeDecoder())
-      .addBefore(BufferCallBeforeInitHandler.NAME, null, saslHandler);
+      .addBefore(BufferCallBeforeInitHandler.NAME, NettyHBaseSaslRpcClientHandler.HANDLER_NAME,
+        saslHandler);
     NettyFutureUtils.addListener(saslPromise, new FutureListener<Boolean>() {
 
       @Override

hbase-client/src/main/java/org/apache/hadoop/hbase/security/NettyHBaseSaslRpcClient.java

@@ -45,16 +45,16 @@ public NettyHBaseSaslRpcClient(Configuration conf, SaslClientAuthenticationProvi
     super(conf, provider, token, serverAddr, securityInfo, fallbackAllowed, rpcProtection);
   }
 
-  public void setupSaslHandler(ChannelPipeline p) {
+  public void setupSaslHandler(ChannelPipeline p, String addAfter) {
     String qop = (String) saslClient.getNegotiatedProperty(Sasl.QOP);
     LOG.trace("SASL client context established. Negotiated QoP {}", qop);
     if (qop == null || "auth".equalsIgnoreCase(qop)) {
       return;
     }
     // add wrap and unwrap handlers to pipeline.
-    p.addFirst(new SaslWrapHandler(saslClient::wrap),
-      new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
-      new SaslUnwrapHandler(saslClient::unwrap));
+    p.addAfter(addAfter, null, new SaslUnwrapHandler(saslClient::unwrap))
+      .addAfter(addAfter, null, new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4))
+      .addAfter(addAfter, null, new SaslWrapHandler(saslClient::wrap));
   }
 
   public String getSaslQOP() {

hbase-client/src/main/java/org/apache/hadoop/hbase/security/NettyHBaseSaslRpcClientHandler.java

@@ -47,6 +47,8 @@ public class NettyHBaseSaslRpcClientHandler extends SimpleChannelInboundHandler<
 
   private static final Logger LOG = LoggerFactory.getLogger(NettyHBaseSaslRpcClientHandler.class);
 
+  public static final String HANDLER_NAME = "SaslRpcClientHandler";
+
   private final Promise<Boolean> saslPromise;
 
   private final UserGroupInformation ugi;
@@ -93,7 +95,7 @@ private void tryComplete(ChannelHandlerContext ctx) {
       LOG.trace("SASL negotiation for {} is complete", provider.getSaslAuthMethod().getName());
     }
     ChannelPipeline p = ctx.pipeline();
-    saslRpcClient.setupSaslHandler(p);
+    saslRpcClient.setupSaslHandler(p, HANDLER_NAME);
     p.remove(SaslChallengeDecoder.class);
     p.remove(this);
 

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyHBaseSaslRpcServerHandler.java

@@ -89,11 +89,11 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Excep
       boolean useWrap = qop != null && !"auth".equalsIgnoreCase(qop);
       ChannelPipeline p = ctx.pipeline();
       if (useWrap) {
-        p.addFirst(new SaslWrapHandler(saslServer::wrap));
-        p.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
+        p.addBefore(DECODER_NAME, null, new SaslWrapHandler(saslServer::wrap)).addLast(
+          new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
           new SaslUnwrapHandler(saslServer::unwrap));
       }
-      conn.setupDecoder();
+      conn.setupHandler();
       p.remove(this);
       p.remove(DECODER_NAME);
     }

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServer.java

@@ -117,9 +117,8 @@ protected void initChannel(Channel ch) throws Exception {
           if (conf.getBoolean(HBASE_SERVER_NETTY_TLS_ENABLED, false)) {
             initSSL(pipeline, conf.getBoolean(HBASE_SERVER_NETTY_TLS_SUPPORTPLAINTEXT, true));
           }
-          pipeline.addLast(NettyRpcServerPreambleHandler.DECODER_NAME, preambleDecoder);
-          pipeline.addLast(createNettyRpcServerPreambleHandler(),
-            new NettyRpcServerResponseEncoder(metrics));
+          pipeline.addLast(NettyRpcServerPreambleHandler.DECODER_NAME, preambleDecoder)
+            .addLast(createNettyRpcServerPreambleHandler());
         }
       });
     try {

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcServerPreambleHandler.java

@@ -61,7 +61,7 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Excep
       p.addLast(NettyHBaseSaslRpcServerHandler.DECODER_NAME, decoder);
       p.addLast(new NettyHBaseSaslRpcServerHandler(rpcServer, conn));
     } else {
-      conn.setupDecoder();
+      conn.setupHandler();
     }
     // add first and then remove, so the single decode decoder will pass the remaining bytes to the
     // handler above.

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/NettyServerRpcConnection.java

@@ -33,7 +33,6 @@
 import org.apache.hbase.thirdparty.com.google.protobuf.Message;
 import org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf;
 import org.apache.hbase.thirdparty.io.netty.channel.Channel;
-import org.apache.hbase.thirdparty.io.netty.channel.ChannelPipeline;
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;
 
@@ -70,10 +69,11 @@ class NettyServerRpcConnection extends ServerRpcConnection {
     this.remotePort = inetSocketAddress.getPort();
   }
 
-  void setupDecoder() {
-    ChannelPipeline p = channel.pipeline();
-    p.addLast("frameDecoder", new NettyRpcFrameDecoder(rpcServer.maxRequestSize, this));
-    p.addLast("decoder", new NettyRpcServerRequestDecoder(rpcServer.metrics, this));
+  void setupHandler() {
+    channel.pipeline()
+      .addLast("frameDecoder", new NettyRpcFrameDecoder(rpcServer.maxRequestSize, this))
+      .addLast("decoder", new NettyRpcServerRequestDecoder(rpcServer.metrics, this))
+      .addLast("encoder", new NettyRpcServerResponseEncoder(rpcServer.metrics));
   }
 
   void process(ByteBuf buf) throws IOException, InterruptedException {