Skip to content

[feat][broker] PIP 97: Implement for ServerCnx #19409

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
lhotari merged 9 commits into from
Feb 4, 2023
Merged

[feat][broker] PIP 97: Implement for ServerCnx #19409

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
165ee96
[feat][broker] PIP 97: Implement for ServerCnx
michaeljmarshall Feb 2, 2023
d5341dc
Handle exception to maintain current behavior
michaeljmarshall Feb 3, 2023
f34b2d7
Use correct AuthState to get newAuthRole
michaeljmarshall Feb 3, 2023
b29b74a
Fix test that relied on hard coded message for auth failure
michaeljmarshall Feb 3, 2023
9865d89
Merge remote-tracking branch 'origin/master' into pip-97-server-cnx
michaeljmarshall Feb 3, 2023
6a66c61
Move ServerCnxAuthorizationTest to ServerCnxTest
michaeljmarshall Feb 3, 2023
539ad03
Only authenticate original auth data when non-null
michaeljmarshall Feb 3, 2023
cdbcd87
Fix another hard coded error message assertion
michaeljmarshall Feb 3, 2023
c03a39e
Add test for missed edge case fixed two commits ago
michaeljmarshall Feb 3, 2023
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
222 changes: 132 additions & 90 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 @@ -194,6 +194,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
// it will hold the credentials of the original client
private AuthenticationState originalAuthState;
private AuthenticationDataSource originalAuthData;
// Keep temporarily in order to verify after verifying proxy's authData
private AuthData originalAuthDataCopy;
private boolean pendingAuthChallengeResponse = false;

// Max number of pending requests per connections. If multiple producers are sharing the same connection the flow
Expand Down Expand Up @@ -690,8 +692,8 @@ ByteBuf createConsumerStatsResponse(Consumer consumer, long requestId) {
}

// complete the connect and sent newConnected command
private void completeConnect(int clientProtoVersion, String clientVersion, boolean supportsTopicWatchers) {
writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, supportsTopicWatchers));
private void completeConnect(int clientProtoVersion, String clientVersion) {
writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, enableSubscriptionPatternEvaluation));
state = State.Connected;
service.getPulsarStats().recordConnectionCreateSuccess();
if (log.isDebugEnabled()) {
Expand All @@ -706,74 +708,135 @@ private void completeConnect(int clientProtoVersion, String clientVersion, boole
}
}

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

// According to auth result, send Connected, AuthChallenge, or Error command.
private void doAuthentication(AuthData clientData,
boolean useOriginalAuthState,
int clientProtocolVersion,
final String clientVersion) {
// 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);

if (log.isDebugEnabled()) {
log.debug("Authenticate using original auth state : {}, role = {}", useOriginalAuthState, authRole);
}
authState
.authenticateAsync(clientData)
.whenCompleteAsync((authChallenge, throwable) -> {
if (throwable == null) {
authChallengeSuccessCallback(authChallenge, useOriginalAuthState, authRole,
clientProtocolVersion, clientVersion);
} else {
authenticationFailed(throwable);
}
}, ctx.executor());
}

if (authState.isComplete()) {
// Authentication has completed. It was either:
// 1. the 1st time the authentication process was done, in which case we'll send
// a `CommandConnected` response
// 2. an authentication refresh, in which case we need to refresh authenticationData

String newAuthRole = authState.getAuthRole();

// Refresh the auth data.
this.authenticationData = authState.getAuthDataSource();
if (log.isDebugEnabled()) {
log.debug("[{}] Auth data refreshed for role={}", remoteAddress, this.authRole);
}
public void authChallengeSuccessCallback(AuthData authChallenge,
boolean useOriginalAuthState,
String authRole,
int clientProtocolVersion,
String clientVersion) {
try {
if (authChallenge == null) {
// Authentication has completed. It was either:
// 1. the 1st time the authentication process was done, in which case we'll send
// a `CommandConnected` response
// 2. an authentication refresh, in which case we need to refresh authenticationData
AuthenticationState authState = useOriginalAuthState ? originalAuthState : this.authState;
String newAuthRole = authState.getAuthRole();

// Refresh the auth data.
this.authenticationData = authState.getAuthDataSource();
if (log.isDebugEnabled()) {
log.debug("[{}] Auth data refreshed for role={}", remoteAddress, this.authRole);
}

if (!useOriginalAuthState) {
this.authRole = newAuthRole;
}
if (!useOriginalAuthState) {
this.authRole = newAuthRole;
}

if (log.isDebugEnabled()) {
log.debug("[{}] Client successfully authenticated with {} role {} and originalPrincipal {}",
remoteAddress, authMethod, this.authRole, originalPrincipal);
}
if (log.isDebugEnabled()) {
log.debug("[{}] Client successfully authenticated with {} role {} and originalPrincipal {}",
remoteAddress, authMethod, this.authRole, originalPrincipal);
}

if (state != State.Connected) {
// First time authentication is done
completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation);
} else {
// If the connection was already ready, it means we're doing a refresh
if (!StringUtils.isEmpty(authRole)) {
if (!authRole.equals(newAuthRole)) {
log.warn("[{}] Principal cannot change during an authentication refresh expected={} got={}",
remoteAddress, authRole, newAuthRole);
ctx.close();
if (state != State.Connected) {
// First time authentication is done
if (originalAuthState != null) {
// We only set originalAuthState when we are going to use it.
authenticateOriginalData(clientProtocolVersion, clientVersion);
} else {
log.info("[{}] Refreshed authentication credentials for role {}", remoteAddress, authRole);
completeConnect(clientProtocolVersion, clientVersion);
}
} else {
// If the connection was already ready, it means we're doing a refresh
if (!StringUtils.isEmpty(authRole)) {
if (!authRole.equals(newAuthRole)) {
log.warn("[{}] Principal cannot change during an authentication refresh expected={} got={}",
remoteAddress, authRole, newAuthRole);
ctx.close();
} else {
log.info("[{}] Refreshed authentication credentials for role {}", remoteAddress, authRole);
}
}
}
} else {
// auth not complete, continue auth with client side.
ctx.writeAndFlush(Commands.newAuthChallenge(authMethod, authChallenge, clientProtocolVersion));
if (log.isDebugEnabled()) {
log.debug("[{}] Authentication in progress client by method {}.", remoteAddress, authMethod);
}
}

return State.Connected;
} catch (Exception e) {
authenticationFailed(e);
}
}

// auth not complete, continue auth with client side.
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());
private void authenticateOriginalData(int clientProtoVersion, String clientVersion) {
originalAuthState
.authenticateAsync(originalAuthDataCopy)
.whenCompleteAsync((authChallenge, throwable) -> {
if (throwable != null) {
authenticationFailed(throwable);
} else if (authChallenge != null) {
// The protocol does not yet handle an auth challenge here.
// See https://github.com/apache/pulsar/issues/19291.
authenticationFailed(new AuthenticationException("Failed to authenticate original auth data "
+ "due to unsupported authChallenge."));
} else {
try {
// No need to retain these bytes anymore
originalAuthDataCopy = null;
originalAuthData = originalAuthState.getAuthDataSource();
originalPrincipal = originalAuthState.getAuthRole();
if (log.isDebugEnabled()) {
log.debug("[{}] Authenticated original role (forwarded from proxy): {}",
remoteAddress, originalPrincipal);
}
completeConnect(clientProtoVersion, clientVersion);
} catch (Exception e) {
authenticationFailed(e);
}
}
}, ctx.executor());
}

// Handle authentication and authentication refresh failures. Must be called from event loop.
private void authenticationFailed(Throwable t) {
String operation;
if (state == State.Connecting) {
service.getPulsarStats().recordConnectionCreateFail();
operation = "connect";
} else {
operation = "authentication-refresh";
}
return State.Connecting;
state = State.Failed;
logAuthException(remoteAddress, operation, getPrincipal(), Optional.empty(), t);
final ByteBuf msg = Commands.newError(-1, ServerError.AuthenticationError, "Failed to authenticate");
NettyChannelUtil.writeAndFlushWithClosePromise(ctx, msg);
}

public void refreshAuthenticationCredentials() {
Expand Down Expand Up @@ -871,10 +934,13 @@ protected void handleConnect(CommandConnect connect) {
}

if (!service.isAuthenticationEnabled()) {
completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation);
completeConnect(clientProtocolVersion, clientVersion);
return;
}

// Go to Connecting state now because auth can be async.
state = State.Connecting;

try {
byte[] authData = connect.hasAuthData() ? connect.getAuthData() : emptyArray;
AuthData clientData = AuthData.of(authData);
Expand All @@ -899,10 +965,9 @@ protected void handleConnect(CommandConnect connect) {
authRole = getBrokerService().getAuthenticationService().getAnonymousUserRole()
.orElseThrow(() ->
new AuthenticationException("No anonymous role, and no authentication provider configured"));
completeConnect(clientProtocolVersion, clientVersion, enableSubscriptionPatternEvaluation);
completeConnect(clientProtocolVersion, clientVersion);
return;
}

// init authState and other var
ChannelHandler sslHandler = ctx.channel().pipeline().get(PulsarChannelInitializer.TLS_HANDLER);
SSLSession sslSession = null;
Expand All @@ -922,14 +987,11 @@ 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
// 3. no credentials were passed
if (connect.hasOriginalPrincipal() && service.getPulsar().getConfig().isAuthenticateOriginalAuthData()) {
// init authentication
// Flow:
// 1. Initialize original authentication.
// 2. Authenticate the proxy's authentication data.
// 3. Authenticate the original authentication data.
String originalAuthMethod;
if (connect.hasOriginalAuthMethod()) {
originalAuthMethod = connect.getOriginalAuthMethod();
Expand All @@ -947,32 +1009,23 @@ protected void handleConnect(CommandConnect connect) {
+ " using auth method [%s] is not available", originalAuthMethod));
}

AuthData originalAuthDataCopy = AuthData.of(connect.getOriginalAuthData().getBytes());
originalAuthDataCopy = AuthData.of(connect.getOriginalAuthData().getBytes());
originalAuthState = originalAuthenticationProvider.newAuthState(
originalAuthDataCopy,
remoteAddress,
sslSession);
originalAuthState.authenticate(originalAuthDataCopy);
originalAuthData = originalAuthState.getAuthDataSource();
originalPrincipal = originalAuthState.getAuthRole();
} else if (connect.hasOriginalPrincipal()) {
originalPrincipal = connect.getOriginalPrincipal();

if (log.isDebugEnabled()) {
log.debug("[{}] Authenticate original role : {}", remoteAddress, originalPrincipal);
}
} else {
originalPrincipal = connect.hasOriginalPrincipal() ? connect.getOriginalPrincipal() : null;

if (log.isDebugEnabled()) {
log.debug("[{}] Authenticate original role (forwarded from proxy): {}",
log.debug("[{}] Setting original role (forwarded from proxy): {}",
remoteAddress, originalPrincipal);
}
}

doAuthentication(clientData, false, clientProtocolVersion, clientVersion);
} catch (Exception e) {
service.getPulsarStats().recordConnectionCreateFail();
state = State.Failed;
logAuthException(remoteAddress, "connect", getPrincipal(), Optional.empty(), e);
ByteBuf msg = Commands.newError(-1, ServerError.AuthenticationError, "Unable to authenticate");
NettyChannelUtil.writeAndFlushWithClosePromise(ctx, msg);
authenticationFailed(e);
}
}

Expand All @@ -990,21 +1043,10 @@ 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();
state = State.Failed;
log.warn("[{}] Authentication failed: {} ", remoteAddress, e.getMessage());
ByteBuf msg = Commands.newError(-1, ServerError.AuthenticationError, "Unable to authenticate");
NettyChannelUtil.writeAndFlushWithClosePromise(ctx, msg);
} catch (Exception e) {
service.getPulsarStats().recordConnectionCreateFail();
state = State.Failed;
String msg = "Unable to handleAuthResponse";
log.warn("[{}] {} ", remoteAddress, msg, e);
ByteBuf command = Commands.newError(-1, ServerError.UnknownError, msg);
NettyChannelUtil.writeAndFlushWithClosePromise(ctx, command);
authenticationFailed(e);
}
}

Expand Down
Loading