diff --git a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginModule.java b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginModule.java index 1dcd1991aedde..e3a78103560a9 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginModule.java +++ b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerLoginModule.java @@ -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, + LOGGED_IN_NOT_COMMITTED, + COMMITTED + } + /** * The SASL Mechanism name for OAuth 2: {@code OAUTHBEARER} */ @@ -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 @@ -266,17 +282,29 @@ public void initialize(Subject subject, CallbackHandler callbackHandler, Map 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 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; } diff --git a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerTokenCallback.java b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerTokenCallback.java index 62ce492eb9140..3f4f269606ab2 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerTokenCallback.java +++ b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerTokenCallback.java @@ -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; diff --git a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java index 8d259e30895b1..e7a4f2cc798d1 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java +++ b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java @@ -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() diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java index dfefabbd5ecca..297cba5d2a5e9 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java @@ -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; @@ -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 .