Skip to content

[fix][broker] ServerCnx broken after recent cherry-picks #19521

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
michaeljmarshall merged 1 commit into from
Feb 15, 2023
Merged

[fix][broker] ServerCnx broken after recent cherry-picks #19521

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 25 additions & 26 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -634,16 +634,6 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) {

// complete the connect and sent newConnected command
private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) {
if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
if (!service.getAuthorizationService()
.isValidOriginalPrincipal(authRole, originalPrincipal, remoteAddress)) {
state = State.Failed;
service.getPulsarStats().recordConnectionCreateFail();
final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");
ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
return;
}
}
ctx.writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers));
state = State.Connected;
service.getPulsarStats().recordConnectionCreateSuccess();
Expand All @@ -660,16 +650,16 @@ private void completeConnect(int clientProtoVersion, String clientVersion, boole
}

// According to auth result, send newConnected or newAuthChallenge command.
private State doAuthentication(AuthData clientData,
private void doAuthentication(AuthData clientData,
boolean useOriginalAuthState,
int clientProtocolVersion,
String clientVersion) throws Exception {

// The original auth state can only be set on subsequent auth attempts (and only
// in presence of a proxy and if the proxy is forwarding the credentials).
// In this case, the re-validation needs to be done against the original client
// credentials.
boolean useOriginalAuthState = (originalAuthState != null);
AuthenticationState authState = useOriginalAuthState ? originalAuthState : this.authState;
AuthenticationState authState = useOriginalAuthState ? originalAuthState : this.authState;
String authRole = useOriginalAuthState ? originalPrincipal : this.authRole;
AuthData brokerData = authState.authenticate(clientData);

Expand Down Expand Up @@ -702,6 +692,16 @@ private State doAuthentication(AuthData clientData,

if (state != State.Connected) {
// First time authentication is done
if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
if (!service.getAuthorizationService()
.isValidOriginalPrincipal(this.authRole, originalPrincipal, remoteAddress)) {
state = State.Failed;
service.getPulsarStats().recordConnectionCreateFail();
final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");
ctx.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
return;
}
}
completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation);
} else {
// If the connection was already ready, it means we're doing a refresh
Expand All @@ -715,18 +715,16 @@ private State doAuthentication(AuthData clientData,
}
}
}
} else {

return State.Connected;
}

// auth not complete, continue auth with client side.
ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion));
if (log.isDebugEnabled()) {
log.debug("[{}] Authentication in progress client by method {}.",
remoteAddress, authMethod);
log.debug("[{}] connect state change to : [{}]", remoteAddress, State.Connecting.name());
// auth not complete, continue auth with client side.
ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, brokerData, clientProtocolVersion));
if (log.isDebugEnabled()) {
log.debug("[{}] Authentication in progress client by method {}.",
remoteAddress, authMethod);
log.debug("[{}] connect state change to : [{}]", remoteAddress, State.Connecting.name());
}
}
return State.Connecting;
}

public void refreshAuthenticationCredentials() {
Expand Down Expand Up @@ -824,6 +822,8 @@ protected void handleConnect(CommandConnect connect) {
return;
}

state = State.Connecting;

try {
byte[] authData = connect.hasAuthData() ? connect.getAuthData() : emptyArray;
AuthData clientData = AuthData.of(authData);
Expand Down Expand Up @@ -871,8 +871,6 @@ protected void handleConnect(CommandConnect connect) {
log.debug("[{}] Authenticate role : {}", remoteAddress, role);
}

state = doAuthentication(clientData, clientProtocolVersion, clientVersion);

// This will fail the check if:
// 1. client is coming through a proxy
// 2. we require to validate the original credentials
Expand Down Expand Up @@ -914,6 +912,7 @@ protected void handleConnect(CommandConnect connect) {
remoteAddress, originalPrincipal);
}
}
doAuthentication(clientData, false, clientProtocolVersion, clientVersion);
} catch (Exception e) {
service.getPulsarStats().recordConnectionCreateFail();
logAuthException(remoteAddress, "connect", getPrincipal(), Optional.empty(), e);
Expand All @@ -937,7 +936,7 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {

try {
AuthData clientData = AuthData.of(authResponse.getResponse().getAuthData());
doAuthentication(clientData, authResponse.getProtocolVersion(),
doAuthentication(clientData, originalAuthState != null, authResponse.getProtocolVersion(),
authResponse.hasClientVersion() ? authResponse.getClientVersion() : EMPTY);
} catch (AuthenticationException e) {
service.getPulsarStats().recordConnectionCreateFail();
Expand Down