Skip to content

Commit

Permalink
HBASE-27279 Make SslHandler work with SaslWrapHandler/SaslUnwrapHandler
Browse files Browse the repository at this point in the history
  • Loading branch information
Apache9 committed Aug 14, 2022
1 parent 73759be commit 2e66d19
Show file tree
Hide file tree
Showing 15 changed files with 829 additions and 718 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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 {
Expand Down
Loading

0 comments on commit 2e66d19

Please sign in to comment.