Skip to content

Commit 48cab90

Browse files
nodecenodece
committed
[fix][proxy] Fix refresh client auth
Signed-off-by: Zixuan Liu <nodeces@gmail.com>
1 parent 08df28a commit 48cab90

File tree

4 files changed

+338
-74
lines changed

4 files changed

+338
-74
lines changed

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

+59-41
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import java.util.concurrent.Semaphore;
4545
import java.util.concurrent.TimeUnit;
4646
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
47+
import javax.naming.AuthenticationException;
4748
import lombok.AccessLevel;
4849
import lombok.Getter;
4950
import org.apache.commons.lang3.exception.ExceptionUtils;
@@ -253,8 +254,25 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
253254
} else {
254255
log.info("{} Connected through proxy to target broker at {}", ctx.channel(), proxyToTargetBrokerAddress);
255256
}
256-
// Send CONNECT command
257-
ctx.writeAndFlush(newConnectCommand())
257+
completeActive();
258+
}
259+
260+
protected void completeActive() throws Exception {
261+
sendConnectCommand(null, null, null);
262+
}
263+
264+
protected final void sendConnectCommand(String originalPrincipal, AuthData originalAuthData,
265+
String originalAuthMethod) throws Exception {
266+
// mutual authentication is to auth between `remoteHostName` and this client for this channel.
267+
// each channel will have a mutual client/server pair, mutual client evaluateChallenge with init data,
268+
// and return authData to server.
269+
authenticationDataProvider = authentication.getAuthData(remoteHostName);
270+
AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA);
271+
272+
ByteBuf byteBuf = Commands.newConnect(authentication.getAuthMethodName(), authData, this.protocolVersion,
273+
PulsarVersion.getVersion(), proxyToTargetBrokerAddress, originalPrincipal, originalAuthData,
274+
originalAuthMethod);
275+
ctx.writeAndFlush(byteBuf)
258276
.addListener(future -> {
259277
if (future.isSuccess()) {
260278
if (log.isDebugEnabled()) {
@@ -268,16 +286,6 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
268286
});
269287
}
270288

271-
protected ByteBuf newConnectCommand() throws Exception {
272-
// mutual authentication is to auth between `remoteHostName` and this client for this channel.
273-
// each channel will have a mutual client/server pair, mutual client evaluateChallenge with init data,
274-
// and return authData to server.
275-
authenticationDataProvider = authentication.getAuthData(remoteHostName);
276-
AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA);
277-
return Commands.newConnect(authentication.getAuthMethodName(), authData, this.protocolVersion,
278-
PulsarVersion.getVersion(), proxyToTargetBrokerAddress, null, null, null);
279-
}
280-
281289
@Override
282290
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
283291
super.channelInactive(ctx);
@@ -361,51 +369,54 @@ protected void handleConnected(CommandConnected connected) {
361369
state = State.Ready;
362370
}
363371

364-
@Override
365-
protected void handleAuthChallenge(CommandAuthChallenge authChallenge) {
366-
checkArgument(authChallenge.hasChallenge());
367-
checkArgument(authChallenge.getChallenge().hasAuthData());
372+
protected final void sendMutualAuthCommand(String authMethod, AuthData authData) {
373+
if (log.isDebugEnabled()) {
374+
log.debug("{} Mutual auth {}", ctx.channel(), authMethod);
375+
}
368376

377+
ByteBuf request = Commands.newAuthResponse(authMethod,
378+
authData,
379+
this.protocolVersion,
380+
PulsarVersion.getVersion());
381+
382+
ctx.writeAndFlush(request).addListener(writeFuture -> {
383+
if (!writeFuture.isSuccess()) {
384+
log.warn("{} Failed to send request for mutual auth to broker: {}", ctx.channel(),
385+
writeFuture.cause().getMessage());
386+
close(writeFuture.cause());
387+
}
388+
});
389+
}
390+
391+
protected void prepareMutualAuth(CommandAuthChallenge authChallenge) throws AuthenticationException {
369392
if (Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData())) {
370393
try {
371394
authenticationDataProvider = authentication.getAuthData(remoteHostName);
372395
} catch (PulsarClientException e) {
373396
log.error("{} Error when refreshing authentication data provider: {}", ctx.channel(), e);
374-
connectionFuture.completeExceptionally(e);
397+
close(e);
375398
return;
376399
}
377400
}
378-
379401
// mutual authn. If auth not complete, continue auth; if auth complete, complete connectionFuture.
380-
try {
381-
AuthData authData = authenticationDataProvider
382-
.authenticate(AuthData.of(authChallenge.getChallenge().getAuthData()));
383-
384-
checkState(!authData.isComplete());
385-
386-
ByteBuf request = Commands.newAuthResponse(authentication.getAuthMethodName(),
387-
authData,
388-
this.protocolVersion,
389-
PulsarVersion.getVersion());
390-
391-
if (log.isDebugEnabled()) {
392-
log.debug("{} Mutual auth {}", ctx.channel(), authentication.getAuthMethodName());
393-
}
394-
395-
ctx.writeAndFlush(request).addListener(writeFuture -> {
396-
if (!writeFuture.isSuccess()) {
397-
log.warn("{} Failed to send request for mutual auth to broker: {}", ctx.channel(),
398-
writeFuture.cause().getMessage());
399-
connectionFuture.completeExceptionally(writeFuture.cause());
400-
}
401-
});
402+
AuthData authData =
403+
authenticationDataProvider.authenticate(AuthData.of(authChallenge.getChallenge().getAuthData()));
404+
checkState(!authData.isComplete());
405+
sendMutualAuthCommand(authentication.getAuthMethodName(), authData);
406+
}
402407

408+
@Override
409+
protected void handleAuthChallenge(CommandAuthChallenge authChallenge) {
410+
checkArgument(authChallenge.hasChallenge());
411+
checkArgument(authChallenge.getChallenge().hasAuthData());
412+
try {
413+
prepareMutualAuth(authChallenge);
403414
if (state == State.SentConnectFrame) {
404415
state = State.Connecting;
405416
}
406417
} catch (Exception e) {
407418
log.error("{} Error mutual verify: {}", ctx.channel(), e);
408-
connectionFuture.completeExceptionally(e);
419+
close(e);
409420
return;
410421
}
411422
}
@@ -1243,6 +1254,13 @@ public void close() {
12431254
}
12441255
}
12451256

1257+
public void close(Throwable e) {
1258+
if (ctx != null) {
1259+
ctx.close();
1260+
connectionFuture.completeExceptionally(e);
1261+
}
1262+
}
1263+
12461264
private void checkRequestTimeout() {
12471265
while (!requestTimeoutQueue.isEmpty()) {
12481266
RequestTime request = requestTimeoutQueue.peek();

‎pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyClientCnx.java

+45-25
Original file line numberDiff line numberDiff line change
@@ -18,46 +18,66 @@
1818
*/
1919
package org.apache.pulsar.proxy.server;
2020

21-
import io.netty.buffer.ByteBuf;
2221
import io.netty.channel.EventLoopGroup;
23-
import org.apache.pulsar.PulsarVersion;
22+
import java.util.Arrays;
23+
import java.util.concurrent.CompletableFuture;
24+
import java.util.function.Function;
25+
import javax.naming.AuthenticationException;
26+
import lombok.extern.slf4j.Slf4j;
2427
import org.apache.pulsar.client.impl.ClientCnx;
2528
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
2629
import org.apache.pulsar.common.api.AuthData;
27-
import org.apache.pulsar.common.protocol.Commands;
28-
import org.slf4j.Logger;
29-
import org.slf4j.LoggerFactory;
30+
import org.apache.pulsar.common.api.proto.CommandAuthChallenge;
3031

32+
@Slf4j
3133
public class ProxyClientCnx extends ClientCnx {
32-
33-
String clientAuthRole;
34-
AuthData clientAuthData;
35-
String clientAuthMethod;
36-
int protocolVersion;
34+
private final boolean forwardClientAuth;
35+
private final String clientAuthMethod;
36+
private final String clientAuthRole;
37+
private final Function<Boolean, CompletableFuture<AuthData>> clientAuthDataSupplier;
3738

3839
public ProxyClientCnx(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, String clientAuthRole,
39-
AuthData clientAuthData, String clientAuthMethod, int protocolVersion) {
40-
super(conf, eventLoopGroup);
40+
Function<Boolean, CompletableFuture<AuthData>> clientAuthDataSupplier,
41+
String clientAuthMethod,
42+
int protocolVersion, boolean forwardClientAuth) {
43+
super(conf, eventLoopGroup, protocolVersion);
4144
this.clientAuthRole = clientAuthRole;
42-
this.clientAuthData = clientAuthData;
45+
this.clientAuthDataSupplier = clientAuthDataSupplier;
4346
this.clientAuthMethod = clientAuthMethod;
44-
this.protocolVersion = protocolVersion;
47+
this.forwardClientAuth = forwardClientAuth;
4548
}
4649

4750
@Override
48-
protected ByteBuf newConnectCommand() throws Exception {
49-
if (log.isDebugEnabled()) {
50-
log.debug("New Connection opened via ProxyClientCnx with params clientAuthRole = {},"
51-
+ " clientAuthData = {}, clientAuthMethod = {}",
52-
clientAuthRole, clientAuthData, clientAuthMethod);
51+
protected void completeActive() throws Exception {
52+
if (!forwardClientAuth) {
53+
super.completeActive();
54+
return;
5355
}
5456

55-
authenticationDataProvider = authentication.getAuthData(remoteHostName);
56-
AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA);
57-
return Commands.newConnect(authentication.getAuthMethodName(), authData, this.protocolVersion,
58-
PulsarVersion.getVersion(), proxyToTargetBrokerAddress, clientAuthRole, clientAuthData,
59-
clientAuthMethod);
57+
clientAuthDataSupplier.apply(false).thenAccept(clientAuthData -> {
58+
try {
59+
sendConnectCommand(clientAuthRole, clientAuthData, clientAuthMethod);
60+
} catch (Exception e) {
61+
log.error("{} Error during handshake", ctx.channel(), e);
62+
close(e);
63+
}
64+
});
6065
}
6166

62-
private static final Logger log = LoggerFactory.getLogger(ProxyClientCnx.class);
67+
@Override
68+
protected void prepareMutualAuth(CommandAuthChallenge authChallenge) throws AuthenticationException {
69+
boolean isRefresh = Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData());
70+
if (!forwardClientAuth || !isRefresh) {
71+
super.prepareMutualAuth(authChallenge);
72+
return;
73+
}
74+
75+
clientAuthDataSupplier.apply(true).thenAccept(originalClientAuthData -> {
76+
sendMutualAuthCommand(clientAuthMethod, originalClientAuthData);
77+
}).exceptionally(e -> {
78+
log.error("{} Error mutual verify", ctx.channel(), e);
79+
close(e);
80+
return null;
81+
});
82+
}
6383
}

‎pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java

+59-8
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import java.util.List;
3838
import java.util.Map;
3939
import java.util.Optional;
40+
import java.util.concurrent.CompletableFuture;
4041
import java.util.concurrent.RejectedExecutionException;
4142
import java.util.concurrent.ThreadLocalRandom;
4243
import java.util.concurrent.TimeUnit;
@@ -45,6 +46,7 @@
4546
import javax.naming.AuthenticationException;
4647
import javax.net.ssl.SSLSession;
4748
import lombok.Getter;
49+
import lombok.extern.slf4j.Slf4j;
4850
import org.apache.pulsar.broker.PulsarServerException;
4951
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
5052
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
@@ -78,6 +80,7 @@
7880
* Handles incoming discovery request from client and sends appropriate response back to client.
7981
*
8082
*/
83+
@Slf4j
8184
public class ProxyConnection extends PulsarHandler {
8285
private static final Logger LOG = LoggerFactory.getLogger(ProxyConnection.class);
8386
// ConnectionPool is used by the proxy to issue lookup requests
@@ -97,6 +100,7 @@ public class ProxyConnection extends PulsarHandler {
97100
String clientAuthRole;
98101
AuthData clientAuthData;
99102
String clientAuthMethod;
103+
private CompletableFuture<AuthData> refreshAuthFuture;
100104

101105
private String authMethod = "none";
102106
AuthenticationProvider authenticationProvider;
@@ -316,12 +320,12 @@ private synchronized void completeConnect(AuthData clientData) throws PulsarClie
316320
this.clientAuthData = clientData;
317321
this.clientAuthMethod = authMethod;
318322
}
319-
clientCnxSupplier =
320-
() -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole, clientAuthData,
321-
clientAuthMethod, protocolVersionToAdvertise);
323+
clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole,
324+
this::getOrRefreshOriginalClientAuthData,
325+
clientAuthMethod, protocolVersionToAdvertise,
326+
service.getConfiguration().isForwardAuthorizationCredentials());
322327
} else {
323-
clientCnxSupplier =
324-
() -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise);
328+
clientCnxSupplier = () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise);
325329
}
326330

327331
if (this.connectionPool == null) {
@@ -428,11 +432,17 @@ private void doAuthentication(AuthData clientData) throws Exception {
428432
// authentication has completed, will send newConnected command.
429433
if (authState.isComplete()) {
430434
clientAuthRole = authState.getAuthRole();
435+
clientAuthData = clientData;
431436
if (LOG.isDebugEnabled()) {
432437
LOG.debug("[{}] Client successfully authenticated with {} role {}",
433-
remoteAddress, authMethod, clientAuthRole);
438+
remoteAddress, authMethod, clientAuthRole);
439+
}
440+
if (refreshAuthFuture != null) {
441+
refreshAuthFuture.complete(clientData);
442+
refreshAuthFuture = null;
443+
} else {
444+
completeConnect(clientData);
434445
}
435-
completeConnect(clientData);
436446
return;
437447
}
438448

@@ -523,7 +533,9 @@ remoteAddress, protocolVersionToAdvertise, getRemoteEndpointProtocolVersion(),
523533

524534
@Override
525535
protected void handleAuthResponse(CommandAuthResponse authResponse) {
526-
checkArgument(state == State.Connecting);
536+
if (refreshAuthFuture == null) {
537+
checkArgument(state == State.Connecting);
538+
}
527539
checkArgument(authResponse.hasResponse());
528540
checkArgument(authResponse.getResponse().hasAuthData() && authResponse.getResponse().hasAuthMethodName());
529541

@@ -543,6 +555,45 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {
543555
}
544556
}
545557

558+
private CompletableFuture<AuthData> getOrRefreshOriginalClientAuthData(boolean isRefresh) {
559+
if (!isRefresh) {
560+
return CompletableFuture.completedFuture(clientAuthData);
561+
}
562+
563+
if (refreshAuthFuture != null && !refreshAuthFuture.isDone()) {
564+
log.error("{} Mutual auth timeout", ctx.channel());
565+
ctx.close();
566+
return CompletableFuture.failedFuture(new AuthenticationException("Mutual auth timeout"));
567+
}
568+
569+
refreshAuthFuture = new CompletableFuture<>();
570+
try {
571+
AuthData refreshAuthData = authState.refreshAuthentication();
572+
ctx.writeAndFlush(Commands.newAuthChallenge(clientAuthMethod, refreshAuthData, protocolVersionToAdvertise))
573+
.addListener(writeFuture -> {
574+
if (writeFuture.isSuccess()) {
575+
if (LOG.isDebugEnabled()) {
576+
LOG.debug("{} Sent auth challenge to client to refresh credentials with method: {}",
577+
ctx.channel(), clientAuthMethod);
578+
}
579+
} else {
580+
LOG.error("{} Failed to send request for mutual auth to client", ctx.channel(),
581+
writeFuture.cause());
582+
refreshAuthFuture.completeExceptionally(writeFuture.cause());
583+
ctx.close();
584+
}
585+
});
586+
} catch (AuthenticationException e) {
587+
log.error("{} Failed to refresh authentication", ctx.channel(), e);
588+
ctx.writeAndFlush(
589+
Commands.newError(-1, ServerError.AuthenticationError, "Failed to refresh authentication"))
590+
.addListener(ChannelFutureListener.CLOSE);
591+
refreshAuthFuture.completeExceptionally(e);
592+
}
593+
594+
return refreshAuthFuture;
595+
}
596+
546597
@Override
547598
protected void handlePartitionMetadataRequest(CommandPartitionedTopicMetadata partitionMetadata) {
548599
checkArgument(state == State.ProxyLookupRequests);

0 commit comments

Comments
 (0)