Skip to content

KAFKA-7462: Make token optional for OAuthBearerLoginModule #5733

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
rajinisivaram merged 3 commits into from
Oct 8, 2018
Merged

KAFKA-7462: Make token optional for OAuthBearerLoginModule #5733

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
23fd758
KAFKA-7462: Make token optional for OAuthBearerLoginModule
rajinisivaram Oct 3, 2018
90cf558
Address review comments
rajinisivaram Oct 4, 2018
f4f073d
Address review comments
rajinisivaram Oct 8, 2018
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
107 changes: 74 additions & 33 deletions ...ts/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,21 @@
* @see SaslConfigs#SASL_LOGIN_REFRESH_BUFFER_SECONDS_DOC
*/
public class OAuthBearerLoginModule implements LoginModule {

/**
* Login state transitions:
* Initial state: NOT_LOGGED_IN
* login() : NOT_LOGGED_IN => LOGGED_IN_NOT_COMMITTED
* commit() : LOGGED_IN_NOT_COMMITTED => COMMITTED
* abort() : LOGGED_IN_NOT_COMMITTED => NOT_LOGGED_IN
* logout() : Any state => NOT_LOGGED_IN
*/
private enum LoginState {
NOT_LOGGED_IN,

@rondagostino rondagostino Oct 3, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would be good to comment that this implies logged-out or aborted since the state is null at the beginning (when we aren't logged in, either, but we don't set the state to NOT_LOGGED_IN). Either comment or rename to LOGGED_OUT_OR_ABORTED?

@rajinisivaram rajinisivaram Oct 4, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added comment.

LOGGED_IN_NOT_COMMITTED,
COMMITTED
}

/**
* The SASL Mechanism name for OAuth 2: {@code OAUTHBEARER}
*/
Expand All @@ -248,6 +263,7 @@ public class OAuthBearerLoginModule implements LoginModule {
private OAuthBearerToken myCommittedToken = null;
private SaslExtensions extensionsRequiringCommit = null;
private SaslExtensions myCommittedExtensions = null;
private LoginState loginState;

static {
OAuthBearerSaslClientProvider.initialize(); // not part of public API
Expand All @@ -266,17 +282,29 @@ public void initialize(Subject subject, CallbackHandler callbackHandler, Map<Str

@Override
public boolean login() throws LoginException {
if (tokenRequiringCommit != null)
throw new IllegalStateException(String.format(
if (loginState == LoginState.LOGGED_IN_NOT_COMMITTED) {
if (tokenRequiringCommit != null)
throw new IllegalStateException(String.format(
"Already have an uncommitted token with private credential token count=%d", committedTokenCount()));
if (myCommittedToken != null)
throw new IllegalStateException(String.format(
else
throw new IllegalStateException("Already logged in without a token");
}
if (loginState == LoginState.COMMITTED) {
if (myCommittedToken != null)
throw new IllegalStateException(String.format(
"Already have a committed token with private credential token count=%d; must login on another login context or logout here first before reusing the same login context",
committedTokenCount()));
else
throw new IllegalStateException("Login has already been committed without a token");
}

identifyToken();
identifyExtensions();
if (tokenRequiringCommit != null)
identifyExtensions();
else
log.debug("Logged in without a token, this login cannot be used to establish client connections");

loginState = LoginState.LOGGED_IN_NOT_COMMITTED;
log.info("Login succeeded; invoke commit() to commit it; current committed token count={}",
committedTokenCount());
return true;
Expand All @@ -292,7 +320,7 @@ private void identifyToken() throws LoginException {
}

tokenRequiringCommit = tokenCallback.token();
if (tokenRequiringCommit == null) {
if (tokenCallback.errorCode() != null) {
log.info("Login failed: {} : {} (URI={})", tokenCallback.errorCode(), tokenCallback.errorDescription(),
tokenCallback.errorUri());
throw new LoginException(tokenCallback.errorDescription());
Expand Down Expand Up @@ -322,64 +350,77 @@ private void identifyExtensions() throws LoginException {

@Override
public boolean logout() {
if (tokenRequiringCommit != null)
if (loginState == LoginState.LOGGED_IN_NOT_COMMITTED)
throw new IllegalStateException(
"Cannot call logout() immediately after login(); need to first invoke commit() or abort()");
if (myCommittedToken == null) {
if (loginState != LoginState.COMMITTED) {
if (log.isDebugEnabled())
log.debug("Nothing here to log out");
return false;
}
log.info("Logging out my token; current committed token count = {}", committedTokenCount());
for (Iterator<Object> iterator = subject.getPrivateCredentials().iterator(); iterator.hasNext();) {
Object privateCredential = iterator.next();
if (privateCredential == myCommittedToken) {
iterator.remove();
myCommittedToken = null;
break;
if (myCommittedToken != null) {
log.info("Logging out my token; current committed token count = {}", committedTokenCount());
for (Iterator<Object> iterator = subject.getPrivateCredentials().iterator(); iterator.hasNext(); ) {
Object privateCredential = iterator.next();
if (privateCredential == myCommittedToken) {
iterator.remove();
myCommittedToken = null;
break;
}
}
}
log.info("Done logging out my token; committed token count is now {}", committedTokenCount());
log.info("Done logging out my token; committed token count is now {}", committedTokenCount());
} else
log.debug("No tokens to logout for this login");

log.info("Logging out my extensions");
if (subject.getPublicCredentials().removeIf(e -> myCommittedExtensions == e))
myCommittedExtensions = null;
log.info("Done logging out my extensions");
if (myCommittedExtensions != null) {
log.info("Logging out my extensions");
if (subject.getPublicCredentials().removeIf(e -> myCommittedExtensions == e))
myCommittedExtensions = null;
log.info("Done logging out my extensions");
} else
log.debug("No extensions to logout for this login");

loginState = LoginState.NOT_LOGGED_IN;
return true;
}

@Override
public boolean commit() {
if (tokenRequiringCommit == null) {
if (loginState != LoginState.LOGGED_IN_NOT_COMMITTED) {
if (log.isDebugEnabled())
log.debug("Nothing here to commit");
return false;
}

log.info("Committing my token; current committed token count = {}", committedTokenCount());
subject.getPrivateCredentials().add(tokenRequiringCommit);
myCommittedToken = tokenRequiringCommit;
tokenRequiringCommit = null;
log.info("Done committing my token; committed token count is now {}", committedTokenCount());
if (tokenRequiringCommit != null) {
log.info("Committing my token; current committed token count = {}", committedTokenCount());
subject.getPrivateCredentials().add(tokenRequiringCommit);
myCommittedToken = tokenRequiringCommit;
tokenRequiringCommit = null;
log.info("Done committing my token; committed token count is now {}", committedTokenCount());
} else
log.debug("No tokens to commit, this login cannot be used to establish client connections");

subject.getPublicCredentials().add(extensionsRequiringCommit);
myCommittedExtensions = extensionsRequiringCommit;
extensionsRequiringCommit = null;
if (extensionsRequiringCommit != null) {
subject.getPublicCredentials().add(extensionsRequiringCommit);
myCommittedExtensions = extensionsRequiringCommit;
extensionsRequiringCommit = null;
}

loginState = LoginState.COMMITTED;
return true;
}

@Override
public boolean abort() {
if (tokenRequiringCommit != null) {
if (loginState == LoginState.LOGGED_IN_NOT_COMMITTED) {
log.info("Login aborted");
tokenRequiringCommit = null;
extensionsRequiringCommit = null;
loginState = LoginState.NOT_LOGGED_IN;
return true;
}
if (log.isDebugEnabled())
log.debug("Nothing here to abort");
log.debug("Nothing here to abort");
return false;
}

Expand Down
4 changes: 2 additions & 2 deletions .../src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerTokenCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,10 +90,10 @@ public String errorUri() {
* Set the token. All error-related values are cleared.
*
* @param token
* the mandatory token to set
* the optional token to set
*/
public void token(OAuthBearerToken token) {
this.token = Objects.requireNonNull(token);
this.token = token;
this.errorCode = null;
this.errorDescription = null;
this.errorUri = null;
Expand Down
8 changes: 8 additions & 0 deletions ...on/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,14 @@ public void close() {
private void handleTokenCallback(OAuthBearerTokenCallback callback) {
if (callback.token() != null)
throw new IllegalArgumentException("Callback had a token already");
if (moduleOptions.isEmpty()) {
log.debug("Token not provided, this login cannot be used to establish client connections");
callback.token(null);
return;
}
if (moduleOptions.keySet().stream().noneMatch(name -> !name.startsWith(EXTENSION_PREFIX))) {
throw new OAuthBearerConfigException("Extensions provided in login context without a token");
}
String principalClaimNameValue = optionValue(PRINCIPAL_CLAIM_NAME_OPTION);
String principalClaimName = principalClaimNameValue != null && !principalClaimNameValue.trim().isEmpty()
? principalClaimNameValue.trim()
Expand Down
35 changes: 35 additions & 0 deletions ...s/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@
import org.apache.kafka.common.security.plain.internals.PlainServerCallbackHandler;

import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
Expand Down Expand Up @@ -1207,6 +1208,40 @@ public void testValidSaslOauthBearerMechanism() throws Exception {
createAndCheckClientConnection(securityProtocol, node);
}

/**
* Tests OAUTHBEARER client channels without tokens for the server.
*/
@Test
public void testValidSaslOauthBearerMechanismWithoutServerTokens() throws Exception {
String node = "0";
SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("OAUTHBEARER"));
saslClientConfigs.put(SaslConfigs.SASL_JAAS_CONFIG,
TestJaasConfig.jaasConfigProperty("OAUTHBEARER", Collections.singletonMap("unsecuredLoginStringClaim_sub", TestJaasConfig.USERNAME)));
saslServerConfigs.put("listener.name.sasl_ssl.oauthbearer." + SaslConfigs.SASL_JAAS_CONFIG,
TestJaasConfig.jaasConfigProperty("OAUTHBEARER", Collections.emptyMap()));

// Server without a token should start up successfully and authenticate clients.
server = createEchoServer(securityProtocol);
createAndCheckClientConnection(securityProtocol, node);

// Client without a token should fail to connect
saslClientConfigs.put(SaslConfigs.SASL_JAAS_CONFIG,
TestJaasConfig.jaasConfigProperty("OAUTHBEARER", Collections.emptyMap()));
createAndCheckClientConnectionFailure(securityProtocol, node);

// Server with extensions, but without a token should fail to start up since it could indicate a configuration error
saslServerConfigs.put("listener.name.sasl_ssl.oauthbearer." + SaslConfigs.SASL_JAAS_CONFIG,
TestJaasConfig.jaasConfigProperty("OAUTHBEARER", Collections.singletonMap("unsecuredLoginExtension_test", "something")));
try {
createEchoServer(securityProtocol);
fail("Server created with invalid login config containing extensions without a token");
} catch (Throwable e) {
assertTrue("Unexpected exception " + Utils.stackTrace(e), e.getCause() instanceof LoginException);
}
}

/**
* Tests OAUTHBEARER fails the connection when the client presents a token with
* insufficient scope .
Expand Down