## Summary

A distinct alternate trigger was confirmed on the vulnerable Apache Kafka 4.1.0 broker-side SASL/OAUTHBEARER path, but no bypass of the Kafka 4.1.2 fix was found. Instead of replaying an expired JWT, this variant matrix used non-expired JWTs with (1) an unexpected issuer, (2) an unexpected audience, and (3) a signature produced by an untrusted key while retaining the trusted `kid`. Kafka 4.1.0 accepted all three invalid tokens over the real network SASL/OAUTHBEARER broker listener because `DefaultJwtValidator` delegated to `ClientJwtValidator`; Kafka 4.1.2 rejected the same inputs because the fixed `DefaultJwtValidator` selected `BrokerJwtValidator` when `sasl.oauthbearer.jwks.endpoint.url` was configured. Therefore this is an alternate trigger on the vulnerable version, not a fixed-version bypass.

## Fix Coverage / Assumptions

The fix relies on the invariant that a broker configured for secured OAuth/OIDC validation has `sasl.oauthbearer.jwks.endpoint.url` in the effective listener-prefixed configuration, and that this should cause the default validator to instantiate `BrokerJwtValidator` instead of `ClientJwtValidator`.

The relevant source-level change between Kafka 4.1.0 and 4.1.2 is in:

- `clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidator.java`
  - Kafka 4.1.0: if no injected `CloseableVerificationKeyResolver` is present, `DefaultJwtValidator.configure()` always uses `new ClientJwtValidator()`.
  - Kafka 4.1.2: if no injected resolver is present, `DefaultJwtValidator.configure()` constructs `ConfigurationUtils` and checks `SaslConfigs.SASL_OAUTHBEARER_JWKS_ENDPOINT_URL`; when present it uses `new BrokerJwtValidator()`, otherwise it falls back to `new ClientJwtValidator()`.

The runtime proof also captures the shipped bytecode comparison in:

- `bundle/logs/vuln_variant/javap_defaultjwtvalidator_4.1.0.txt`
- `bundle/logs/vuln_variant/javap_defaultjwtvalidator_4.1.2.txt`

The fix explicitly covers the production broker path using `OAuthBearerValidatorCallbackHandler` plus a configured JWKS endpoint. It does not attempt to remove all uses of `ClientJwtValidator`, because that class remains appropriate for client-side token parsing and for configurations that intentionally do not provide JWKS. The tested broker path was configured with:

- `listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler`
- `listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.jwks.endpoint.url=<mock JWKS URL>`
- `sasl.oauthbearer.expected.issuer=https://mock-idp.example.com`
- `sasl.oauthbearer.expected.audience=kafka-broker`

The Apache Kafka security model treats broker listener authentication as in scope when security is enabled. It also states that security is off by default and that trusted operators, classpath/JAR loading, and intentionally unsecured/development tooling are out of scope. This variant stays inside the in-scope boundary: an untrusted network peer sends a JWT over a broker SASL listener configured for OAUTHBEARER validation.

## Variant / Alternate Trigger

The parent reproduction demonstrated an expired JWT replay. The variant matrix tested three materially different token-validation failures that reach the same sink from the same network trust boundary:

1. `wrong_issuer`: JWT has a valid future `exp`, trusted signature, and correct audience, but `iss=https://attacker-idp.example.net` instead of the configured expected issuer.
2. `wrong_audience`: JWT has a valid future `exp`, trusted signature, and correct issuer, but `aud=other-service` instead of `kafka-broker`.
3. `tampered_signature`: JWT has valid future `exp`, correct issuer, and correct audience, but is signed with an untrusted RSA key while using the trusted key id.

Exact entry point:

- Network protocol: TCP connection to Kafka broker listener.
- Kafka API messages: `SaslHandshake(OAUTHBEARER)` followed by `SaslAuthenticate` containing the RFC 7628 initial client response `auth=Bearer <JWT>`.
- Target callback path: `OAuthBearerValidatorCallbackHandler.handle()` -> `handleValidatorCallback()` -> `DefaultJwtValidator.validate()`.

Specific code paths:

- `clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java`: selects the broker SASL callback handler. The default OAUTHBEARER server handler is the unsecured test handler unless the operator configures `sasl.server.callback.handler.class`; this test configures the secured `OAuthBearerValidatorCallbackHandler`.
- `clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerValidatorCallbackHandler.java`: receives the attacker-supplied token from the SASL exchange and delegates to the configured `JwtValidator`.
- `clients/src/main/java/org/apache/kafka/common/security/oauthbearer/DefaultJwtValidator.java`: vulnerable-vs-fixed delegate selection.
- `clients/src/main/java/org/apache/kafka/common/security/oauthbearer/ClientJwtValidator.java`: parses claims and requires claim presence but does not validate issuer, audience, signature trust, or current-time expiry.
- `clients/src/main/java/org/apache/kafka/common/security/oauthbearer/BrokerJwtValidator.java`: fixed broker validator using jose4j with signature verification, expected issuer/audience checks, required `exp`/`iat`, and expiration enforcement.

## Impact

- Package/component affected: Apache Kafka broker SASL/OAUTHBEARER authentication, specifically the default broker-side JWT validator selected behind `OAuthBearerValidatorCallbackHandler`.
- Affected versions as tested: Apache Kafka 4.1.0 binary distribution, release tag commit `13f70256db3c994c590e5d262a7cc50b9e973204`.
- Fixed version as tested: Apache Kafka 4.1.2 binary distribution, release tag commit `c82fd9b934b4c1e6fa799e3f1dcc8f08d997740c`.
- Consequences: On the vulnerable version, a network peer can authenticate to the broker with a token that should be rejected for issuer, audience, or signature trust reasons. This is an authentication validation bypass at the broker listener boundary.

## Impact Parity

- Disclosed/claimed maximum impact for the parent: `authz_bypass` / broker-side SASL/OAUTHBEARER authentication validation bypass.
- Reproduced impact from this variant run: Kafka 4.1.0 accepted invalid but non-expired JWTs over the production broker TCP SASL path; Kafka 4.1.2 rejected the same invalid inputs.
- Parity: `full` for the vulnerable-version broker-side JWT validation bypass, because the invalid tokens were accepted by the broker authentication handshake.
- Not demonstrated: No fixed-version bypass was demonstrated. The proof does not demonstrate post-auth topic read/write operations; it proves SASL authentication success/failure at the broker boundary.

## Root Cause

The same underlying bug is reached because Kafka 4.1.0's `DefaultJwtValidator.configure()` used `ClientJwtValidator` whenever no explicit verification-key resolver was injected. In the secured broker callback handler path, this meant a broker with `sasl.oauthbearer.jwks.endpoint.url`, expected issuer, and expected audience configured still ended up using a validator intended for client-side token parsing. `ClientJwtValidator` extracts `scope`, `exp`, `sub`, and `iat` and constructs a `BasicOAuthBearerToken`, but it does not verify the token signature, compare `iss` to the expected issuer, compare `aud` to the expected audience, or enforce current-time expiration.

Kafka 4.1.2 closes the tested path by changing `DefaultJwtValidator.configure()` to instantiate `BrokerJwtValidator` when `sasl.oauthbearer.jwks.endpoint.url` is present. `BrokerJwtValidator.configure()` builds a jose4j `JwtConsumer` with `DISALLOW_NONE`, `setRequireExpirationTime()`, `setRequireIssuedAt()`, `setVerificationKeyResolver(...)`, and optional expected issuer/audience. The fixed broker logs show `BrokerJwtValidator` rejection evidence for the tested invalid issuer/audience/signature tokens.

Fix reference from the parent reproduction: `01d8e7db8d08dbd538892b409457ea6bfcc2a422` was identified as the advisory fix commit, while the tested fixed release tag resolves to `c82fd9b934b4c1e6fa799e3f1dcc8f08d997740c`.

## Reproduction Steps

1. Run `bundle/vuln_variant/reproduction_steps.sh`.
2. The script downloads or reuses Kafka 4.1.0 and Kafka 4.1.2 binary distributions, starts a mock OAuth/JWKS issuer, starts isolated real Kafka KRaft brokers configured with SASL/OAUTHBEARER and `OAuthBearerValidatorCallbackHandler`, and sends raw Kafka `SaslHandshake`/`SaslAuthenticate` frames with valid and invalid JWTs.
3. Expected evidence:
   - Kafka 4.1.0 accepts all three variant tokens: wrong issuer, wrong audience, and tampered signature.
   - Kafka 4.1.2 accepts the valid control token but rejects all three variant tokens with `invalid_token` and broker log evidence from `BrokerJwtValidator`.
   - The script exits `1` by design because no fixed-version bypass is reproduced; this is still a successful negative-bypass / alternate-trigger run.

The script was executed twice successfully and produced the same verdict both times: alternate trigger confirmed on Kafka 4.1.0, no bypass on Kafka 4.1.2.

## Evidence

Primary evidence locations:

- `bundle/logs/vuln_variant/reproduction_steps.log`: full latest execution log.
- `bundle/logs/vuln_variant/variant_evidence.log`: summarized variant matrix and raw response excerpts.
- `bundle/logs/vuln_variant/raw_vulnerable_wrong_issuer.json`
- `bundle/logs/vuln_variant/raw_vulnerable_wrong_audience.json`
- `bundle/logs/vuln_variant/raw_vulnerable_tampered_signature.json`
- `bundle/logs/vuln_variant/raw_fixed_wrong_issuer.json`
- `bundle/logs/vuln_variant/raw_fixed_wrong_audience.json`
- `bundle/logs/vuln_variant/raw_fixed_tampered_signature.json`
- `bundle/logs/vuln_variant/broker_vulnerable.log`
- `bundle/logs/vuln_variant/broker_fixed.log`
- `bundle/logs/vuln_variant/fixed_version.txt`: exact tested release/tag identity.

Key evidence from the latest run:

```json
{
  "alternate_trigger_confirmed_on_vulnerable": true,
  "bypass_confirmed_on_fixed": false,
  "valid_controls_ok": true,
  "vulnerable_variant_accepts": {
    "wrong_issuer": true,
    "wrong_audience": true,
    "tampered_signature": true
  },
  "fixed_variant_rejects": {
    "wrong_issuer": true,
    "wrong_audience": true,
    "tampered_signature": true
  },
  "fixed_variant_accepts": {
    "wrong_issuer": false,
    "wrong_audience": false,
    "tampered_signature": false
  }
}
```

The fixed broker returned `invalid_token` for all three variant inputs. For example, the wrong-issuer raw response has `accepted_by_broker=false`, `rejected_by_broker=true`, and SASL auth bytes `{"status":"invalid_token"}`. The fixed broker log includes rejection from `BrokerJwtValidator` with: `Issuer (iss) claim value (https://attacker-idp.example.net) doesn't match expected value of https://mock-idp.example.com`.

Environment details captured by the script include Java version, Kafka binary paths, release tag commits, mock issuer/JWKS URL, token claims, raw Kafka response JSON, and broker runtime logs.

## Recommendations / Next Steps

- Treat the Kafka 4.1.2 fix as covering the tested secured broker callback path. No additional patch is required for the wrong-issuer, wrong-audience, or tampered-signature data paths when `sasl.oauthbearer.jwks.endpoint.url` is configured.
- Keep regression tests for more than expiration. The fix should remain covered by integration tests that send raw SASL/OAUTHBEARER frames for expired tokens, wrong issuer, wrong audience, unsigned/`none` algorithm tokens, and tokens signed by untrusted keys.
- Document and lint broker configurations so production deployments using OAUTHBEARER do not accidentally use `OAuthBearerUnsecuredValidatorCallbackHandler` or omit the JWKS endpoint while expecting production-grade JWT validation.
- Consider an operator-facing warning when a broker listener uses OAUTHBEARER without a secured server callback handler or without JWKS configuration, because Kafka's security model permits insecure configurations but operators may misinterpret them.

## Additional Notes

The reproduction script is idempotent: it creates per-run working directories under `bundle/vuln_variant/artifacts/`, chooses fresh local ports, and kills mock/broker processes on exit. Both required verification runs completed without crashing; both returned exit code `1` because the fixed version rejected every variant input, which is the expected negative-bypass outcome. The `Killed` messages in the log correspond to deliberate broker shutdown with `kill -9` during cleanup after successful probes, not an infrastructure failure.