## Summary

Apache Kafka 4.1.0/4.1.1 broker-side SASL/OAUTHBEARER validation can accept attacker-supplied JWTs through the network authentication path when the default broker validator is used. The current reproduction focuses on the judge-requested observable vulnerable-vs-fixed divergence for an expired JWT replayed directly over Kafka's TCP SASL boundary: Kafka 4.1.0 returns a successful `SaslAuthenticate` response for the expired replayed JWT, while Kafka 4.1.2 rejects the same token with `invalid_token` and broker-side `BrokerJwtValidator` expiration evidence. This avoids Kafka client-side expiry enforcement by using a raw Kafka protocol client only for the SASL frames.

## Impact

- Package/component affected: Apache Kafka broker SASL/OAUTHBEARER authentication, specifically `org.apache.kafka.common.security.oauthbearer.DefaultJwtValidator` used by `OAuthBearerValidatorCallbackHandler`.
- Affected versions: Apache Kafka 4.1.0 through 4.1.1 according to the Apache advisory for CVE-2026-33557.
- Fixed versions: Apache Kafka 4.1.2 and 4.2.0+.
- Risk level and consequences: Important/high authentication validation weakness. The advisory impact is that a broker using the default validator may accept JWTs without validating signature, issuer, or audience. In this run, the expired-token replay path demonstrates that the vulnerable broker accepts an expired JWT at the SASL authentication step, while the fixed broker rejects it.

## Impact Parity

- Disclosed/claimed maximum impact: `authz_bypass` / authentication validation bypass for Kafka SASL/OAUTHBEARER.
- Reproduced impact from this run: network-protocol authentication divergence on the real Kafka broker TCP listener. The vulnerable broker returns `SaslAuthenticate(error_code=0)` for an expired JWT replayed directly in the SASL initial client response; the fixed broker returns an invalid-token challenge and logs expiration validation failure.
- Parity: `full` for the broker-side token validation bypass at the SASL/OAUTHBEARER TCP authentication boundary.
- Not demonstrated: The proof does not require a Kafka CLI client to use the expired credential because the standard client enforces expiry before replay. The primary proof intentionally bypasses that client-side behavior with raw Kafka SASL frames to test the broker-side validator directly.

## Root Cause

Kafka 4.1.0's `DefaultJwtValidator.configure()` always delegates to `ClientJwtValidator` when no explicit verification key resolver is passed. `ClientJwtValidator` performs only structural and claim-presence checks for `scope`, `exp`, `sub`, and `iat`; it does not verify the JWT signature and does not reject an `exp` timestamp that is already in the past. Therefore, the broker-side `OAuthBearerValidatorCallbackHandler` can receive a replayed JWT over TCP and treat it as valid.

Kafka 4.1.2 changes `DefaultJwtValidator.configure()` so that when `sasl.oauthbearer.jwks.endpoint.url` is present, the default delegate becomes `BrokerJwtValidator`, which performs signature and claims validation, including expiration. The fix commit referenced by the advisory is `01d8e7db8d08dbd538892b409457ea6bfcc2a422`.

## Reproduction Steps

1. Run `bundle/repro/reproduction_steps.sh`.
2. The script installs required Python dependencies, reuses or downloads Kafka 4.1.0 and Kafka 4.1.2 binaries, starts a real mock JWKS/OAuth issuer, starts real Kafka KRaft brokers with SASL/OAUTHBEARER enabled, and sends raw Kafka `SaslHandshake` and `SaslAuthenticate` TCP frames containing the same expired attacker-controlled JWT.
3. Expected evidence of reproduction:
   - Kafka 4.1.0: expired JWT replay receives `SaslAuthenticate` success (`error_code=0`, no invalid-token auth bytes).
   - Kafka 4.1.2: the same expired JWT receives an invalid-token response and broker logs show `BrokerJwtValidator` rejecting the `exp` claim as expired.
   - Valid JWT controls are also run so the fixed broker's rejection is not caused by a broken test setup.

## Evidence

Primary evidence is written under `bundle/logs/`:

- `bundle/logs/evidence.log`: summary, decoded JWT claims, version comparison, and key protocol outcomes.
- `bundle/logs/raw_vulnerable_expired.json`: parsed raw Kafka SASL response showing vulnerable acceptance of the expired JWT.
- `bundle/logs/raw_fixed_expired.json`: parsed raw Kafka SASL response showing fixed rejection of the same expired JWT.
- `bundle/logs/broker_vulnerable.log`: real Kafka 4.1.0 broker runtime log.
- `bundle/logs/broker_fixed.log`: real Kafka 4.1.2 broker runtime log with expiration rejection details.
- `bundle/logs/mock_oauth_server.log`: generated expired/valid JWTs and JWKS requests.

Key expected excerpts:

- Vulnerable parsed response: `"sasl_authenticate_error_code": 0`, `"accepted_by_broker": true`, and an expired JWT claim where `exp < replay_time`.
- Fixed parsed response: `"rejected_by_broker": true` with `"invalid_token"` in the auth bytes/error payload.
- Fixed broker log: `The JWT is no longer valid ... Expiration Time ...` from `BrokerJwtValidator`.

Environment details captured include Java version, Kafka binary versions, mock issuer/JWKS URL, decoded token timestamps, and the raw TCP proof JSON files.

## Recommendations / Next Steps

- Upgrade affected Kafka deployments to 4.1.2, 4.2.0, or later.
- If upgrade is not immediately possible, explicitly configure `sasl.oauthbearer.jwt.validator.class=org.apache.kafka.common.security.oauthbearer.BrokerJwtValidator` for brokers using SASL/OAUTHBEARER and configure a trusted `sasl.oauthbearer.jwks.endpoint.url`.
- Add integration tests that replay expired, wrong-issuer, wrong-audience, and unsigned/tampered JWTs directly against the broker-side SASL/OAUTHBEARER listener, not only through Kafka clients.

## Additional Notes

The reproduction script is designed to be idempotent: it creates isolated per-run KRaft data directories, uses unique ports per attempt where practical, kills broker/mock processes on exit, and writes fresh runtime evidence on every run. The proof intentionally uses a raw Kafka TCP client after broker startup because the standard Kafka CLI client refuses to replay expired credentials locally; using it would mask the broker-side validator path that this ticket asks to verify.