This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-23995 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 91e86474240b2df9f7e3a87d04c5094505208e5e Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 14 19:45:17 2026 +0200 CAMEL-23995: Fix saslAuthType overriding explicit securityProtocol and generating non-working configs Fix three bugs in the saslAuthType convenience layer: 1. Remove tautology (saslAuthType != NONE always true inside isSasl() branch) that forced SSL on all SASL types, ignoring explicit securityProtocol=SASL_PLAINTEXT. The configurer already defaults to SSL; only downgrade when user explicitly requests SASL_PLAINTEXT. 2. Make saslAuthType=KERBEROS not throw at startup by returning null from buildJaasConfig() when no principal is set. This lets Kerberos users rely on external JAAS config while saslAuthType only sets the protocol and mechanism. 3. Inject the required callback handler classes for OAUTH (KIP-768 OAuthBearerLoginCallbackHandler) and AWS_MSK_IAM (IAMClientCallbackHandler). Move OAuth token endpoint URL from JAAS string to the top-level sasl.oauthbearer.token.endpoint.url config property. Fix OAuth scope JAAS key from oauth.scope to scope. Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/kafka/KafkaConfiguration.java | 18 ++--- .../kafka/security/KafkaSecurityConfigurer.java | 41 ++++++++++- .../camel/component/kafka/KafkaComponentTest.java | 62 ++++++++++++++++ .../security/KafkaSecurityConfigurerTest.java | 83 +++++++++++++++++++++- 4 files changed, 192 insertions(+), 12 deletions(-) diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java index 57372b08af62..1e0ec5d2d957 100755 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java @@ -449,15 +449,17 @@ public class KafkaConfiguration implements Cloneable, HeaderFilterStrategyAware // The saslAuthType just helps set the mechanism and protocol } - // Determine if we should use SSL (check if any SSL properties are set) - boolean hasSslConfig = ObjectHelper.isNotEmpty(sslTruststoreLocation) - || ObjectHelper.isNotEmpty(sslKeystoreLocation) - || sslContextParameters != null; - - // For SASL types, default to SSL unless explicitly using SASL_PLAINTEXT + // For SASL types, the configurer defaults to useSsl=true (SASL_SSL). + // Only downgrade to SASL_PLAINTEXT if the user explicitly requested it. if (saslAuthType.isSasl()) { - boolean useSsl = !securityProtocol.equals("SASL_PLAINTEXT") && !securityProtocol.equals("PLAINTEXT"); - configurer.withSsl(useSsl || hasSslConfig || saslAuthType != KafkaAuthType.NONE); + if (securityProtocol.equals("SASL_PLAINTEXT")) { + configurer.withSsl(false); + } + } else if (saslAuthType == KafkaAuthType.NONE) { + boolean hasSslConfig = ObjectHelper.isNotEmpty(sslTruststoreLocation) + || ObjectHelper.isNotEmpty(sslKeystoreLocation) + || sslContextParameters != null; + configurer.withSsl(hasSslConfig || securityProtocol.equals("SSL")); } // Apply the configuration diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java index 7538bd3d4fc3..2675667736a8 100644 --- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java +++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurer.java @@ -336,9 +336,8 @@ public class KafkaSecurityConfigurer { validateOAuth(); sb.append(" clientId=\"").append(escapeJaasValue(oauthClientId)).append("\""); sb.append(" clientSecret=\"").append(escapeJaasValue(oauthClientSecret)).append("\""); - sb.append(" oauth.token.endpoint.uri=\"").append(oauthTokenEndpointUri).append("\""); if (ObjectHelper.isNotEmpty(oauthScope)) { - sb.append(" oauth.scope=\"").append(oauthScope).append("\""); + sb.append(" scope=\"").append(escapeJaasValue(oauthScope)).append("\""); } break; @@ -347,6 +346,9 @@ public class KafkaSecurityConfigurer { break; case KERBEROS: + if (ObjectHelper.isEmpty(kerberosPrincipal)) { + return null; + } validateKerberos(); sb.append(" useKeyTab=").append(kerberosUseKeyTab); sb.append(" storeKey=").append(kerberosStoreKey); @@ -405,6 +407,26 @@ public class KafkaSecurityConfigurer { if (jaasConfig != null) { configuration.setSaslJaasConfig(jaasConfig); } + applySaslProperties(configuration.getAdditionalProperties()); + } + } + + private void applySaslProperties(java.util.Map<String, Object> additionalProperties) { + switch (authType) { + case OAUTH: + additionalProperties.putIfAbsent(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, + "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler"); + if (ObjectHelper.isNotEmpty(oauthTokenEndpointUri)) { + additionalProperties.putIfAbsent(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, + oauthTokenEndpointUri); + } + break; + case AWS_MSK_IAM: + additionalProperties.putIfAbsent(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, + "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); + break; + default: + break; } } @@ -426,6 +448,21 @@ public class KafkaSecurityConfigurer { if (jaasConfig != null) { props.setProperty(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig); } + switch (authType) { + case OAUTH: + props.setProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, + "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler"); + if (ObjectHelper.isNotEmpty(oauthTokenEndpointUri)) { + props.setProperty(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL, oauthTokenEndpointUri); + } + break; + case AWS_MSK_IAM: + props.setProperty(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, + "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); + break; + default: + break; + } } return props; diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java index f2622e32c552..15154285dbac 100644 --- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaComponentTest.java @@ -36,6 +36,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -420,6 +421,8 @@ public class KafkaComponentTest { assertEquals("SASL_SSL", props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); assertEquals("AWS_MSK_IAM", props.getProperty(SaslConfigs.SASL_MECHANISM)); assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("IAMLoginModule")); + assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler", + props.getProperty(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS)); } @Test @@ -446,6 +449,65 @@ public class KafkaComponentTest { assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG)); } + @Test + public void testSaslAuthTypePlainWithExplicitSaslPlaintext() { + String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=PLAIN" + + "&saslUsername=user&saslPassword=pass&securityProtocol=SASL_PLAINTEXT"; + + KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class); + + Properties props = endpoint.getConfiguration().createProducerProperties(); + assertEquals("SASL_PLAINTEXT", props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); + assertEquals("PLAIN", props.getProperty(SaslConfigs.SASL_MECHANISM)); + } + + @Test + public void testSaslAuthTypeKerberosDoesNotThrow() { + String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=KERBEROS"; + + KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class); + assertEquals(KafkaAuthType.KERBEROS, endpoint.getConfiguration().getSaslAuthType()); + + Properties props = endpoint.getConfiguration().createProducerProperties(); + assertEquals("SASL_SSL", props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); + assertEquals("GSSAPI", props.getProperty(SaslConfigs.SASL_MECHANISM)); + assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG)); + } + + @Test + public void testSaslAuthTypeOAuth() { + String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=OAUTH" + + "&oauthClientId=myClient&oauthClientSecret=mySecret" + + "&oauthTokenEndpointUri=https://auth.example.com/token&oauthScope=kafka";; + + KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class); + assertEquals(KafkaAuthType.OAUTH, endpoint.getConfiguration().getSaslAuthType()); + + Properties props = endpoint.getConfiguration().createProducerProperties(); + assertEquals("SASL_SSL", props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); + assertEquals("OAUTHBEARER", props.getProperty(SaslConfigs.SASL_MECHANISM)); + assertNotNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG)); + assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("OAuthBearerLoginModule")); + assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("clientId=\"myClient\"")); + assertTrue(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("scope=\"kafka\"")); + assertFalse(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG).contains("oauth.token.endpoint.uri")); + assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler", + props.getProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS)); + assertEquals("https://auth.example.com/token";, + props.getProperty(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL)); + } + + @Test + public void testSaslAuthTypeNoneWithExplicitSsl() { + String uri = "kafka:mytopic?brokers=broker1:12345&saslAuthType=NONE&securityProtocol=SSL"; + + KafkaEndpoint endpoint = context.getEndpoint(uri, KafkaEndpoint.class); + assertEquals(KafkaAuthType.NONE, endpoint.getConfiguration().getSaslAuthType()); + + Properties props = endpoint.getConfiguration().createProducerProperties(); + assertEquals("SSL", props.getProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); + } + // ======================================================================== // Backward compatibility tests - existing security options should still work // ======================================================================== diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java index b70ad3449bc4..f2de4a435fe6 100644 --- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java +++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/security/KafkaSecurityConfigurerTest.java @@ -231,8 +231,9 @@ public class KafkaSecurityConfigurerTest { assertTrue(jaasConfig.contains("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required")); assertTrue(jaasConfig.contains("clientId=\"my-client\"")); assertTrue(jaasConfig.contains("clientSecret=\"my-secret\"")); - assertTrue(jaasConfig.contains("oauth.token.endpoint.uri=\"https://auth.example.com/token\"";)); - assertTrue(jaasConfig.contains("oauth.scope=\"kafka\"")); + assertTrue(jaasConfig.contains("scope=\"kafka\"")); + assertFalse(jaasConfig.contains("oauth.token.endpoint.uri")); + assertFalse(jaasConfig.contains("oauth.scope")); } @Test @@ -466,6 +467,84 @@ public class KafkaSecurityConfigurerTest { assertNull(props.getProperty(SaslConfigs.SASL_JAAS_CONFIG)); } + // ======================================================================== + // Kerberos without principal (external JAAS) tests + // ======================================================================== + + @Test + public void testBuildJaasConfigKerberosNoPrincipal() { + KafkaSecurityConfigurer configurer = KafkaSecurityConfigurer.forAuthType(KafkaAuthType.KERBEROS); + assertNull(configurer.buildJaasConfig()); + } + + @Test + public void testKerberosNoPrincipalSetsProtocolAndMechanism() { + KafkaConfiguration config = new KafkaConfiguration(); + KafkaSecurityConfigurer.forAuthType(KafkaAuthType.KERBEROS) + .configure(config); + + assertEquals("SASL_SSL", config.getSecurityProtocol()); + assertEquals("GSSAPI", config.getSaslMechanism()); + assertNull(config.getSaslJaasConfig()); + } + + // ======================================================================== + // Callback handler and additional SASL properties tests + // ======================================================================== + + @Test + public void testToPropertiesOAuthIncludesCallbackHandler() { + Properties props = KafkaSecurityConfigurer.oauth("client", "secret", "https://auth.example.com/token";) + .toProperties(); + + assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler", + props.getProperty(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS)); + assertEquals("https://auth.example.com/token";, + props.getProperty(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL)); + } + + @Test + public void testToPropertiesAwsMskIamIncludesCallbackHandler() { + Properties props = KafkaSecurityConfigurer.awsMskIam().toProperties(); + + assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler", + props.getProperty(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS)); + } + + @Test + public void testConfigureOAuthSetsCallbackHandler() { + KafkaConfiguration config = new KafkaConfiguration(); + KafkaSecurityConfigurer.oauth("client", "secret", "https://auth.example.com/token";) + .configure(config); + + assertEquals("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler", + config.getAdditionalProperties().get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS)); + assertEquals("https://auth.example.com/token";, + config.getAdditionalProperties().get(SaslConfigs.SASL_OAUTHBEARER_TOKEN_ENDPOINT_URL)); + } + + @Test + public void testConfigureAwsMskIamSetsCallbackHandler() { + KafkaConfiguration config = new KafkaConfiguration(); + KafkaSecurityConfigurer.awsMskIam() + .configure(config); + + assertEquals("software.amazon.msk.auth.iam.IAMClientCallbackHandler", + config.getAdditionalProperties().get(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS)); + } + + @Test + public void testConfigureOAuthDoesNotOverrideUserCallbackHandler() { + KafkaConfiguration config = new KafkaConfiguration(); + config.getAdditionalProperties().put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS, "com.custom.MyHandler"); + + KafkaSecurityConfigurer.oauth("client", "secret", "https://auth.example.com/token";) + .configure(config); + + assertEquals("com.custom.MyHandler", + config.getAdditionalProperties().get(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS)); + } + // ======================================================================== // toString() test // ========================================================================
