This is an automated email from the ASF dual-hosted git repository.
omkreddy pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9c7fad33973 Require expected issuer and audience for the
SASL/OAUTHBEARER broker validator
9c7fad33973 is described below
commit 9c7fad339731199da395d90690fc62efae69615b
Author: Evan Zhou <[email protected]>
AuthorDate: Wed Jun 17 23:51:55 2026 -0500
Require expected issuer and audience for the SASL/OAUTHBEARER broker
validator
When a JWKS endpoint is configured, the broker's SASL/OAUTHBEARER validator
now requires
both sasl.oauthbearer.expected.issuer and
sasl.oauthbearer.expected.audience to be set, and
the broker fails fast at startup otherwise, so the issuer and audience of
incoming tokens are
verified by default.
Operators that intentionally do not restrict the issuer or audience can opt
out via
sasl.oauthbearer.allow.unverified.issuer=true and/or
sasl.oauthbearer.allow.unverified.audience=true;
a startup warning is logged when an opt-out is enabled.
- Add sasl.oauthbearer.allow.unverified.issuer and
sasl.oauthbearer.allow.unverified.audience
(boolean, default false) to both the client and broker config definitions.
- Raise the importance of sasl.oauthbearer.expected.issuer and
sasl.oauthbearer.expected.audience
to HIGH.
- BrokerJwtValidator fails fast when an expected value is missing without
the matching opt-out,
and skips jose4j's default audience validation when the audience opt-out
is enabled.
Reviewers: Manikumar Reddy <[email protected]>
---
checkstyle/import-control.xml | 1 +
.../apache/kafka/common/config/SaslConfigs.java | 34 +++-
.../config/internals/BrokerSecurityConfigs.java | 7 +-
.../security/oauthbearer/BrokerJwtValidator.java | 60 +++++-
.../oauthbearer/BrokerJwtValidatorTest.java | 212 ++++++++++++++++++++-
.../oauthbearer/DefaultJwtValidatorTest.java | 66 ++++++-
.../OAuthBearerValidatorCallbackHandlerTest.java | 36 +++-
.../kafka/api/ClientOAuthIntegrationTest.scala | 1 +
docs/getting-started/upgrade.md | 1 +
9 files changed, 398 insertions(+), 20 deletions(-)
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 49c57b8dffa..c76980b572e 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -135,6 +135,7 @@
</subpackage>
<subpackage name="oauthbearer">
<allow pkg="com.fasterxml.jackson.databind" />
+ <allow class="org.apache.logging.log4j.Level" />
<allow pkg="org.jose4j" />
<allow pkg="javax.crypto"/>
<allow pkg="org.testcontainers" />
diff --git
a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
index 01f7ad1f927..02704a2300d 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
@@ -352,15 +352,31 @@ public class SaslConfigs {
public static final String SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS_DOC = "The
(optional) value in seconds to allow for differences between the time of the
OAuth/OIDC identity provider and"
+ " the broker.";
+ public static final String SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE =
"sasl.oauthbearer.allow.unverified.audience";
+ public static final String SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE_DOC
= "The (optional) setting that allows the broker to validate JWTs without
verifying the audience. By default the broker"
+ + " requires sasl.oauthbearer.expected.audience to be set whenever
" + SASL_OAUTHBEARER_JWKS_ENDPOINT_URL + " is configured, and rejects any JWT
whose \"aud\" claim does"
+ + " not match one of the expected audiences. When this is set to
true that requirement is lifted: the \"aud\" claim is no longer checked and a
JWT bearing any (or no)"
+ + " audience is accepted. Leaving the audience unverified is
insecure and strongly discouraged; enable it only for deployments that
intentionally accept tokens regardless"
+ + " of their intended audience. The default value is 'false'.";
+
public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE =
"sasl.oauthbearer.expected.audience";
- public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC = "The
(optional) comma-delimited setting for the broker to use to verify that the JWT
was issued for one of the"
- + " expected audiences. The JWT will be inspected for the standard
OAuth \"aud\" claim and if this value is set, the broker will match the value
from JWT's \"aud\" claim "
- + " to see if there is an exact match. If there is no match, the
broker will reject the JWT and authentication will fail.";
+ public static final String SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC = "The
comma-delimited setting for the broker to use to verify that the JWT was issued
for one of the"
+ + " expected audiences. The JWT will be inspected for the standard
OAuth \"aud\" claim and the broker will match the value(s) from the JWT's
\"aud\" claim"
+ + " to see if there is an exact match. If there is no match, the
broker will reject the JWT and authentication will fail. Required if " +
SASL_OAUTHBEARER_JWKS_ENDPOINT_URL + " is set," +
+ " with the ability to opt-out using " +
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE + ".";
+
+ public static final String SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER =
"sasl.oauthbearer.allow.unverified.issuer";
+ public static final String SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER_DOC =
"The (optional) setting that allows the broker to validate JWTs without
verifying the issuer. By default the broker"
+ + " requires sasl.oauthbearer.expected.issuer to be set whenever "
+ SASL_OAUTHBEARER_JWKS_ENDPOINT_URL + " is configured, and rejects any JWT
whose \"iss\" claim does not"
+ + " match the expected issuer. When this is set to true that
requirement is lifted: the \"iss\" claim is no longer checked and a JWT bearing
any (or no) issuer is accepted."
+ + " Leaving the issuer unverified is insecure and strongly
discouraged; enable it only for deployments that intentionally accept tokens
regardless of which issuer minted"
+ + " them. The default value is 'false'.";
public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER =
"sasl.oauthbearer.expected.issuer";
- public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC = "The
(optional) setting for the broker to use to verify that the JWT was created by
the expected issuer. The JWT will"
- + " be inspected for the standard OAuth \"iss\" claim and if this
value is set, the broker will match it exactly against what is in the JWT's
\"iss\" claim. If there is no"
- + " match, the broker will reject the JWT and authentication will
fail.";
+ public static final String SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC = "The
setting for the broker to use to verify that the JWT was created by the
expected issuer. The JWT will"
+ + " be inspected for the standard OAuth \"iss\" claim and the
broker will match it exactly against what is in the JWT's \"iss\" claim. If
there is no"
+ + " match, the broker will reject the JWT and authentication will
fail. Required if " + SASL_OAUTHBEARER_JWKS_ENDPOINT_URL + " is set, with the
ability to " +
+ "opt-out using " + SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER + ".";
public static final String SASL_OAUTHBEARER_HEADER_URLENCODE =
"sasl.oauthbearer.header.urlencode";
public static final boolean DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE =
false;
@@ -409,8 +425,10 @@ public class SaslConfigs {
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS,
ConfigDef.Type.LONG,
DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS,
ConfigDef.Importance.LOW,
SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS,
ConfigDef.Type.LONG, DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS,
ConfigDef.Type.INT, DEFAULT_SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
ConfigDef.Type.LIST, List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true,
false), ConfigDef.Importance.LOW, SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW,
SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC)
+
.define(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE,
ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM,
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
ConfigDef.Type.LIST, List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true,
false), ConfigDef.Importance.HIGH, SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER,
ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM,
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
ConfigDef.Type.STRING, null, ConfigDef.Importance.HIGH,
SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_HEADER_URLENCODE,
ConfigDef.Type.BOOLEAN, DEFAULT_SASL_OAUTHBEARER_HEADER_URLENCODE,
ConfigDef.Importance.LOW, SASL_OAUTHBEARER_HEADER_URLENCODE_DOC);
}
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
index ad714803e04..a7c7c1b2073 100644
---
a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
+++
b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
@@ -27,6 +27,7 @@ import org.apache.kafka.common.utils.Utils;
import java.util.List;
+import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
import static org.apache.kafka.common.config.ConfigDef.Type.BOOLEAN;
@@ -217,6 +218,8 @@ public class BrokerSecurityConfigs {
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS, LONG,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS, LOW,
SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS, LONG,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS, LOW,
SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_RETRY_BACKOFF_MAX_MS_DOC)
.define(SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, INT,
SaslConfigs.DEFAULT_SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS, LOW,
SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE, LIST,
List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true, false), LOW,
SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
- .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, STRING,
null, LOW, SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC);
+ .define(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE,
BOOLEAN, false, MEDIUM,
SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE, LIST,
List.of(), ConfigDef.ValidList.anyNonDuplicateValues(true, false), HIGH,
SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER,
BOOLEAN, false, MEDIUM,
SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER_DOC)
+ .define(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, STRING,
null, HIGH, SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER_DOC);
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
index 60b065a5e45..9a3c6810640 100644
---
a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
+++
b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.security.oauthbearer;
+import org.apache.kafka.common.config.ConfigException;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.BasicOAuthBearerToken;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.ClaimValidationUtils;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.CloseableVerificationKeyResolver;
@@ -43,6 +44,8 @@ import java.util.Set;
import javax.security.auth.login.AppConfigurationEntry;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER;
import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_CLOCK_SKEW_SECONDS;
import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE;
import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
@@ -74,6 +77,20 @@ import static
org.jose4j.jwa.AlgorithmConstraints.DISALLOW_NONE;
* Signature matching validation against the <code>kid</code> and
those provided by
* the OAuth/OIDC provider's JWKS
* </li>
+ * <li>
+ * Validation of the <code>aud</code> (audience) claim. The broker
must be configured with
+ * {@code sasl.oauthbearer.expected.audience}; the token's
<code>aud</code> claim is matched
+ * against those values. If no expected audience is configured, the
audience cannot be
+ * verified, so configuration fails fast (the broker refuses to start)
unless
+ * {@code sasl.oauthbearer.allow.unverified.audience} is set to {@code
true}.
+ * </li>
+ * <li>
+ * Validation of the <code>iss</code> (issuer) claim. The broker must
be configured with
+ * {@code sasl.oauthbearer.expected.issuer}; the token's
<code>iss</code> claim is matched
+ * exactly against that value. If the expected issuer is not
configured, the issuer cannot be
+ * verified, so configuration fails fast (the broker refuses to start)
unless
+ * {@code sasl.oauthbearer.allow.unverified.issuer} is set to {@code
true}.
+ * </li>
* </ol>
*/
public class BrokerJwtValidator implements JwtValidator {
@@ -110,6 +127,30 @@ public class BrokerJwtValidator implements JwtValidator {
String expectedIssuer =
cu.validateString(SASL_OAUTHBEARER_EXPECTED_ISSUER, false);
String scopeClaimName =
cu.validateString(SASL_OAUTHBEARER_SCOPE_CLAIM_NAME);
String subClaimName =
cu.validateString(SASL_OAUTHBEARER_SUB_CLAIM_NAME);
+ boolean allowUnverifiedAudience =
Boolean.TRUE.equals(cu.validateBoolean(SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE,
false));
+ boolean allowUnverifiedIssuer =
Boolean.TRUE.equals(cu.validateBoolean(SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER,
false));
+
+ // The broker must be configured with at least one expected audience
so it can verify that the token was minted
+ // for this cluster. Without it, a token issued for any other service
by the same provider would be accepted.
+ // Fail fast at configuration time unless the operator has explicitly
opted out.
+ if (expectedAudiences.isEmpty() && !allowUnverifiedAudience) {
+ throw new ConfigException(String.format(
+ "The OAuth validator for the broker requires \"%s\" to be
configured so that the token's audience" +
+ " (\"aud\") claim can be verified, but it was not set. Set
\"%s\" to the audience(s) that identify this" +
+ " cluster, or set \"%s\" to true to accept tokens regardless
of their audience.",
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE));
+ }
+
+ // The broker must be configured with an expected issuer. Without it,
jose4j accepts a token bearing any (or no)
+ // "iss" claim, so the issuer cannot be verified. Fail fast at
configuration time unless the operator has
+ // explicitly opted out.
+ if (expectedIssuer == null && !allowUnverifiedIssuer) {
+ throw new ConfigException(String.format(
+ "The OAuth validator for the broker requires \"%s\" to be
configured so that the token's issuer" +
+ " (\"iss\") claim can be verified, but it was not set. Set
\"%s\" to the issuer URL of the OAuth/OIDC" +
+ " provider that signs the tokens, or set \"%s\" to true to
accept tokens regardless of their issuer.",
+ SASL_OAUTHBEARER_EXPECTED_ISSUER,
SASL_OAUTHBEARER_EXPECTED_ISSUER, SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER));
+ }
CloseableVerificationKeyResolver verificationKeyResolver =
verificationKeyResolverOpt.orElseGet(
() -> VerificationKeyResolverFactory.get(configs, saslMechanism,
jaasConfigEntries)
@@ -120,11 +161,26 @@ public class BrokerJwtValidator implements JwtValidator {
if (clockSkew != null)
jwtConsumerBuilder.setAllowedClockSkewInSeconds(clockSkew);
- if (!expectedAudiences.isEmpty())
+ if (!expectedAudiences.isEmpty()) {
jwtConsumerBuilder.setExpectedAudience(expectedAudiences.toArray(new
String[0]));
+ } else {
+ // Reached only when the audience opt-out is enabled (otherwise
configuration fails fast above). jose4j
+ // otherwise rejects any token that carries an "aud" claim when no
expected audience is set, so skip its
+ // default audience validation entirely.
+ log.warn("The OAuth broker validator is configured with a JWKS
endpoint but without \"{}\", and \"{}\" is" +
+ " set to true, so it will accept a JWT regardless of its
\"aud\" (audience) claim.",
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE);
+ jwtConsumerBuilder.setSkipDefaultAudienceValidation();
+ }
- if (expectedIssuer != null)
+ if (expectedIssuer != null) {
jwtConsumerBuilder.setExpectedIssuer(expectedIssuer);
+ } else {
+ // Reached only when the issuer opt-out is enabled (otherwise
configuration fails fast above).
+ log.warn("The OAuth broker validator is configured with a JWKS
endpoint but without \"{}\", and \"{}\" is" +
+ " set to true, so it will accept a JWT bearing any (or no)
\"iss\" (issuer) claim.",
+ SASL_OAUTHBEARER_EXPECTED_ISSUER,
SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER);
+ }
this.jwtConsumer = jwtConsumerBuilder
.setJwsAlgorithmConstraints(DISALLOW_NONE)
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
index 5f76f508513..6bc992673c2 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidatorTest.java
@@ -17,22 +17,37 @@
package org.apache.kafka.common.security.oauthbearer;
+import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SaslConfigs;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.CloseableVerificationKeyResolver;
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.logging.log4j.Level;
import org.jose4j.jwk.PublicJsonWebKey;
import org.jose4j.jws.AlgorithmIdentifiers;
import org.jose4j.lang.InvalidAlgorithmException;
import org.junit.jupiter.api.Test;
+import java.util.List;
import java.util.Map;
+import static
org.apache.kafka.common.config.SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER;
import static
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule.OAUTHBEARER_MECHANISM;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class BrokerJwtValidatorTest extends JwtValidatorTest {
+ private static final String ATTACKER_ISSUER = "https://evil.example/";;
+
+ private static final String EXPECTED_ISSUER = "https://legit.example/";;
+
+ private static final String ATTACKER_AUDIENCE =
"https://evil.example/other-service";;
+
+ private static final String EXPECTED_AUDIENCE =
"https://legit.example/other-service";;
+
@Override
protected JwtValidator createJwtValidator(AccessTokenBuilder builder) {
CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
builder.jwk().getKey();
@@ -69,10 +84,15 @@ public class BrokerJwtValidatorTest extends
JwtValidatorTest {
.jwk(jwk)
.alg(AlgorithmIdentifiers.RSA_USING_SHA256)
.addCustomClaim(subClaimName, subject)
+ .addCustomClaim("iss", EXPECTED_ISSUER)
+ .audience(EXPECTED_AUDIENCE)
.subjectClaimName(subClaimName)
.subject(null);
JwtValidator validator = createJwtValidator(tokenBuilder);
- Map<String, ?> saslConfigs =
getSaslConfigs(SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME, subClaimName);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_SUB_CLAIM_NAME, subClaimName,
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
// Validation should succeed (e.g. signature verification) even if sub
claim is missing
@@ -81,10 +101,196 @@ public class BrokerJwtValidatorTest extends
JwtValidatorTest {
assertEquals(subject, token.principalName());
}
+ @Test
+ public void testRejectMissingIssuerClaimWhenExpectedIssuerSet() throws
Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .audience(EXPECTED_AUDIENCE); // valid audience, but no iss
claim is set
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ // With an expected issuer configured, jose4j requires the "iss" claim
to be present; a token without it
+ // must be rejected at validation time rather than silently accepted.
+ JwtValidatorException e = assertThrows(JwtValidatorException.class,
+ () -> validator.validate(accessToken));
+ assertErrorMessageContains(e.getMessage(), EXPECTED_ISSUER);
+ }
+
+ @Test
+ public void testConfigureFailsWhenExpectedIssuerUnset() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ assertThrowsWithMessage(ConfigException.class,
+ () -> validator.configure(
+
getSaslConfigs(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)),
+ OAUTHBEARER_MECHANISM, getJaasConfigEntries()),
+ SASL_OAUTHBEARER_EXPECTED_ISSUER);
+ }
+
+ @Test
+ public void testExpectedIssuerSet() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", ATTACKER_ISSUER)
+ .audience(EXPECTED_AUDIENCE);
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ JwtValidatorException e = assertThrows(JwtValidatorException.class,
+ () -> validator.validate(accessToken));
+ // jose4j reports the issuer mismatch; the message includes the
offending issuer value
+ assertErrorMessageContains(e.getMessage(), ATTACKER_ISSUER);
+ }
+
+ @Test
+ public void testConfigureFailsWhenExpectedAudienceUnset() throws Exception
{
+ PublicJsonWebKey jwk = createRsaJwk();
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ assertThrowsWithMessage(ConfigException.class,
+ () -> validator.configure(
+ getSaslConfigs(SASL_OAUTHBEARER_EXPECTED_ISSUER,
EXPECTED_ISSUER),
+ OAUTHBEARER_MECHANISM, getJaasConfigEntries()),
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ }
+
+ @Test
+ public void testRejectWrongAudienceWhenExpectedAudienceSet() throws
Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", EXPECTED_ISSUER)
+ .audience(ATTACKER_AUDIENCE);
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ JwtValidatorException e = assertThrows(JwtValidatorException.class,
+ () -> validator.validate(accessToken));
+ // jose4j reports the audience mismatch; the message includes the
offending audience value
+ assertErrorMessageContains(e.getMessage(), ATTACKER_AUDIENCE);
+ }
+
+ @Test
+ public void testAllowUnverifiedAudience() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", EXPECTED_ISSUER)
+ .audience(ATTACKER_AUDIENCE); // an audience that matches no
expected audience
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE, true));
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ OAuthBearerToken token = validator.validate(accessToken);
+ assertEquals(builder.subject(), token.principalName());
+ }
+
+ @Test
+ public void testAllowUnverifiedIssuer() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", ATTACKER_ISSUER) // an issuer that
matches no expected issuer
+ .audience(EXPECTED_AUDIENCE);
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE),
+ SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER, true));
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ OAuthBearerToken token = validator.validate(accessToken);
+ assertEquals(builder.subject(), token.principalName());
+ }
+
+ @Test
+ public void testAllowUnverifiedIssuerAndAudience() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", ATTACKER_ISSUER) // neither the
issuer nor the audience matches anything expected
+ .audience(ATTACKER_AUDIENCE);
+ String accessToken = builder.build();
+
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ // Neither expected.issuer nor expected.audience is set; both opt-outs
are enabled so configuration succeeds.
+ Map<String, ?> saslConfigs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER, true,
+ SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE, true));
+
+ try (LogCaptureAppender appender =
LogCaptureAppender.createAndRegister(BrokerJwtValidator.class)) {
+ appender.setClassLogger(BrokerJwtValidator.class, Level.WARN);
+ validator.configure(saslConfigs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+
+ // Each opt-out path must log a WARN naming the config that
disabled the check, so the weakened
+ // posture is visible in the broker logs.
+ List<String> warnings = appender.getMessages("WARN");
+ assertTrue(warnings.stream().anyMatch(m ->
m.contains(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER)),
+ "Expected a WARN mentioning " +
SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_ISSUER + ", got " + warnings);
+ assertTrue(warnings.stream().anyMatch(m ->
m.contains(SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE)),
+ "Expected a WARN mentioning " +
SaslConfigs.SASL_OAUTHBEARER_ALLOW_UNVERIFIED_AUDIENCE + ", got " + warnings);
+ }
+
+ // With both opt-outs in effect, a token whose issuer and audience
match nothing configured is still accepted.
+ OAuthBearerToken token = validator.validate(accessToken);
+ assertEquals(builder.subject(), token.principalName());
+ }
+
+ @Test
+ public void testConfigureFailsWhenNeitherExpectedSet() throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ CloseableVerificationKeyResolver resolver = (jws, nestingContext) ->
jwk.getKey();
+ BrokerJwtValidator validator = new BrokerJwtValidator(resolver);
+ // With neither expected value set and no opt-out, configuration must
fail fast. The audience guard runs
+ // first, so its message is the one that surfaces.
+ assertThrowsWithMessage(ConfigException.class,
+ () -> validator.configure(getSaslConfigs(),
OAUTHBEARER_MECHANISM, getJaasConfigEntries()),
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ }
+
private void testEncryptionAlgorithm(PublicJsonWebKey jwk, String alg)
throws Exception {
- AccessTokenBuilder builder = new
AccessTokenBuilder().jwk(jwk).alg(alg);
+ AccessTokenBuilder builder = new AccessTokenBuilder().jwk(jwk).alg(alg)
+ .addCustomClaim("iss",
EXPECTED_ISSUER).audience(EXPECTED_AUDIENCE);
JwtValidator validator = createJwtValidator(builder);
- validator.configure(getSaslConfigs(), OAUTHBEARER_MECHANISM,
getJaasConfigEntries());
+ validator.configure(getSaslConfigs(Map.of(
+ SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE))),
+ OAUTHBEARER_MECHANISM, getJaasConfigEntries());
String accessToken = builder.build();
OAuthBearerToken token = validator.validate(accessToken);
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
index cf3754a77ac..a8e95ecfb4d 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidatorTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.security.oauthbearer;
+import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import
org.apache.kafka.common.security.oauthbearer.internals.secured.AccessTokenBuilder;
@@ -29,7 +30,10 @@ import org.jose4j.jwk.PublicJsonWebKey;
import org.jose4j.jws.AlgorithmIdentifiers;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static
org.apache.kafka.common.config.internals.BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG;
@@ -37,9 +41,16 @@ import static
org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModul
import static org.apache.kafka.test.TestUtils.tempFile;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.fail;
public class DefaultJwtValidatorTest extends OAuthBearerTest {
+ private static final String ATTACKER_ISSUER =
"https://evil.example/attacker";;
+
+ private static final String EXPECTED_ISSUER = "https://idp.legit.example/";;
+
+ private static final String EXPECTED_AUDIENCE = "kafka-cluster";
+
@AfterEach
public void tearDown() {
System.clearProperty(BrokerSecurityConfigs.ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
@@ -50,7 +61,9 @@ public class DefaultJwtValidatorTest extends OAuthBearerTest {
AccessTokenBuilder builder = new AccessTokenBuilder()
.alg(AlgorithmIdentifiers.RSA_USING_SHA256);
CloseableVerificationKeyResolver verificationKeyResolver =
createVerificationKeyResolver(builder);
- Map<String, ?> configs = getSaslConfigs();
+ Map<String, ?> configs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
DefaultJwtValidator jwtValidator = new
DefaultJwtValidator(verificationKeyResolver);
assertDoesNotThrow(() -> jwtValidator.configure(configs,
OAUTHBEARER_MECHANISM, getJaasConfigEntries()));
assertInstanceOf(BrokerJwtValidator.class, jwtValidator.delegate());
@@ -69,14 +82,20 @@ public class DefaultJwtValidatorTest extends
OAuthBearerTest {
PublicJsonWebKey jwk = createRsaJwk();
AccessTokenBuilder builder = new AccessTokenBuilder()
.jwk(jwk)
- .alg(AlgorithmIdentifiers.RSA_USING_SHA256);
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", EXPECTED_ISSUER)
+ .audience(EXPECTED_AUDIENCE);
String accessToken = builder.build();
JsonWebKeySet jwks = new JsonWebKeySet(jwk);
String jwksJson =
jwks.toJson(JsonWebKey.OutputControlLevel.PUBLIC_ONLY);
String fileUrl = tempFile(jwksJson).toURI().toString();
System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, fileUrl);
- Map<String, ?> configs =
getSaslConfigs(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL, fileUrl);
+ Map<String, Object> rawConfigs = new HashMap<>();
+ rawConfigs.put(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
fileUrl);
+ rawConfigs.put(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER,
EXPECTED_ISSUER);
+ rawConfigs.put(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE));
+ Map<String, ?> configs = getSaslConfigs(rawConfigs);
DefaultJwtValidator jwtValidator = new DefaultJwtValidator();
assertDoesNotThrow(() -> jwtValidator.configure(configs,
OAUTHBEARER_MECHANISM, getJaasConfigEntries()));
@@ -84,7 +103,48 @@ public class DefaultJwtValidatorTest extends
OAuthBearerTest {
assertDoesNotThrow(() -> jwtValidator.validate(accessToken));
}
+ @Test
+ public void testRejectsAttackerIssuerViaDefaultJwtValidatorWithJwksUrl()
throws Exception {
+ PublicJsonWebKey jwk = createRsaJwk();
+ AccessTokenBuilder builder = new AccessTokenBuilder()
+ .jwk(jwk)
+ .alg(AlgorithmIdentifiers.RSA_USING_SHA256)
+ .addCustomClaim("iss", ATTACKER_ISSUER);
+ String accessToken = builder.build();
+
+ JsonWebKeySet jwks = new JsonWebKeySet(jwk);
+ String jwksJson =
jwks.toJson(JsonWebKey.OutputControlLevel.PUBLIC_ONLY);
+ String fileUrl = tempFile(jwksJson).toURI().toString();
+ System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, fileUrl);
+
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL, fileUrl);
+ // expected.issuer intentionally omitted; expected.audience is set so
the issuer check is the one that fails
+ configs.put(SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE));
+
+ assertSecurelyRejected(() -> {
+ DefaultJwtValidator validator = new DefaultJwtValidator();
+ validator.configure(getSaslConfigs(configs),
OAUTHBEARER_MECHANISM, getJaasConfigEntries());
+ validator.validate(accessToken);
+ }, "an attacker-issuer token via DefaultJwtValidator with
jwks.endpoint.url set (expected.issuer unset)");
+ }
+
private CloseableVerificationKeyResolver
createVerificationKeyResolver(AccessTokenBuilder builder) {
return (jws, nestingContext) -> builder.jwk().getPublicKey();
}
+
+ private void assertSecurelyRejected(Executable flow, String what) {
+ Throwable thrown = null;
+ try {
+ flow.execute();
+ } catch (Throwable t) {
+ thrown = t;
+ }
+ if (thrown == null) {
+ fail("INSECURE: " + what + " was accepted");
+ } else if (!(thrown instanceof JwtValidatorException || thrown
instanceof ConfigException)) {
+ fail("Unexpected failure validating " + what + ": " + thrown,
thrown);
+ }
+ // else: ConfigException or JwtValidatorException -> securely rejected
-> pass
+ }
}
diff --git
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java
index 96ef6aecc6e..0e17df1ee1d 100644
---
a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandlerTest.java
@@ -59,6 +59,8 @@ public class OAuthBearerValidatorCallbackHandlerTest extends
OAuthBearerTest {
private static final String EXPECTED_ISSUER = "https://idp.legit.example/";;
+ private static final String EXPECTED_AUDIENCE = "kafka-cluster";
+
@AfterEach
public void tearDown() {
System.clearProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG);
@@ -209,7 +211,8 @@ public class OAuthBearerValidatorCallbackHandlerTest
extends OAuthBearerTest {
Map<String, ?> configs = getSaslConfigs(Map.of(
SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL, fileUrl,
- SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER));
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE, List.of(EXPECTED_AUDIENCE)));
OAuthBearerValidatorCallbackHandler handler = new
OAuthBearerValidatorCallbackHandler();
assertDoesNotThrow(() -> handler.configure(configs,
OAUTHBEARER_MECHANISM, getJaasConfigEntries()));
@@ -227,6 +230,32 @@ public class OAuthBearerValidatorCallbackHandlerTest
extends OAuthBearerTest {
}
}
+ @Test
+ public void testFailsToStartWhenExpectedAudienceMissingWithJwksUrl() {
+ // A broker validator handler configured with a JWKS endpoint URL and
expected.issuer but no expected.audience
+ // must fail to start, rather than come up and accept tokens whose
audience cannot be verified.
+ Map<String, ?> configs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
"https://example.com/jwks";,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER));
+ OAuthBearerValidatorCallbackHandler handler = new
OAuthBearerValidatorCallbackHandler();
+ assertThrowsWithMessage(ConfigException.class,
+ () -> handler.configure(configs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries()),
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE);
+ }
+
+ @Test
+ public void testFailsToStartWhenExpectedIssuerMissingWithJwksUrl() {
+ // A broker validator handler configured with a JWKS endpoint URL and
expected.audience but no expected.issuer
+ // must fail to start, rather than come up and accept tokens whose
issuer cannot be verified.
+ Map<String, ?> configs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL,
"https://example.com/jwks";,
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE,
List.of(EXPECTED_AUDIENCE)));
+ OAuthBearerValidatorCallbackHandler handler = new
OAuthBearerValidatorCallbackHandler();
+ assertThrowsWithMessage(ConfigException.class,
+ () -> handler.configure(configs, OAUTHBEARER_MECHANISM,
getJaasConfigEntries()),
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER);
+ }
+
@Test
public void testConfigureAcceptsCustomValidatorClass() {
// A custom (non-default) validator class must be accepted; the
startup check only applies
@@ -244,7 +273,10 @@ public class OAuthBearerValidatorCallbackHandlerTest
extends OAuthBearerTest {
private void assertInvalidAccessTokenFails(String accessToken, String
expectedMessageSubstring) throws Exception {
AccessTokenBuilder builder = new AccessTokenBuilder()
.alg(AlgorithmIdentifiers.RSA_USING_SHA256);
- Map<String, ?> configs = getSaslConfigs();
+ // The injected resolver routes to BrokerJwtValidator, which requires
expected.issuer and expected.audience.
+ Map<String, ?> configs = getSaslConfigs(Map.of(
+ SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER, EXPECTED_ISSUER,
+ SASL_OAUTHBEARER_EXPECTED_AUDIENCE, List.of(EXPECTED_AUDIENCE)));
CloseableVerificationKeyResolver verificationKeyResolver =
createVerificationKeyResolver(builder);
JwtValidator jwtValidator =
createJwtValidator(verificationKeyResolver);
diff --git
a/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
index d1b2161f7cb..c7e26c1be7d 100644
--- a/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ClientOAuthIntegrationTest.scala
@@ -95,6 +95,7 @@ class ClientOAuthIntegrationTest extends
IntegrationTestHarness with SaslSetup {
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_JAAS_CONFIG}",
s"${classOf[OAuthBearerLoginModule].getName} required ;")
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_EXPECTED_AUDIENCE}",
issuerId)
+
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_EXPECTED_ISSUER}",
mockOAuthServer.issuerUrl(issuerId).toString)
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL}",
jwksUrl)
serverConfig.setProperty(s"$listenerNamePrefix.oauthbearer.${BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS_CONFIG}",
classOf[OAuthBearerValidatorCallbackHandler].getName)
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index c6d3052059c..04f79cdb95b 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -40,6 +40,7 @@ type: docs
`kafka.server:type=group-coordinator-metrics,name=offset-count` instead of
`kafka.coordinator.group:type=GroupMetadataManager,name=NumOffsets`, and
`kafka.server:type=group-coordinator-metrics,name=classic-group-count,state={PreparingRebalance|CompletingRebalance|Stable|Dead|Empty}`
instead of the
`kafka.coordinator.group:type=GroupMetadataManager,name=NumGroups{PreparingRebalance|CompletingRebalance|Stable|Dead|Empty}`
metrics.
For further details, please refer to
[KIP-1301](https://cwiki.apache.org/confluence/x/Z5U8G).
+ * The broker-side OAUTHBEARER JWT validator now fails fast at startup when a
JWKS endpoint (`sasl.oauthbearer.jwks.endpoint.url`) is configured but
`sasl.oauthbearer.expected.audience` or `sasl.oauthbearer.expected.issuer` is
not set. Brokers that previously started without these settings will now fail
to start until they are configured. To intentionally accept tokens regardless
of their audience or issuer, set the new
`sasl.oauthbearer.allow.unverified.audience` or `sasl.oauthbearer.a [...]
## Upgrading to 4.3.0
