[Kafka output] Add SASL/OAUTHBEARER support with file-based JWT token provider
- Dominant language
- Go
- Stars
- 12.7k
- Forks
- 5k
- Avg merge
- 2d 15m
- Merged PRs (30d)
- 385
Description
## Summary
Add SASL/OAUTHBEARER support to the Kafka output, using a file-based token provider that reads a JWT from disk on each authentication handshake.
## Use case
SASL/OAUTHBEARER ([KIP-255](https://cwiki.apache.org/confluence/display/KAFKA/KIP-255%3A+OAuth+Authentication+via+SASL%2FOAUTHBEARER), standardized in Apache Kafka 2.0) is the mechanism used for token-based authentication across modern Kafka deployments:
- **Confluent Cloud** — [OAuth/OIDC workload identity](https://docs.confly/authenticate/workload-identities/identity-providers/oauth/overview.html)
- **AWS MSK** — IAM-based OAUTHBEARER authentication
- **Azure Event Hubs for Kafka** — OAuth 2.0 token access
- **Strimzi / Red Hat OpenShift Streams** — OIDC-based client auth
- **Self-hosted Apache Kafka** — [SASL/OAUTHBEARER](https://kafka.apache.org/43/security/authentication-using-sasl/) since 2.0, pluggable token validation
On Kubernetes, short-lived JWTs are commonly issued by the platform's credential system (SPIFFE/SPIRE, AWS IRSA, GCP Workload Identity, Azure Workload Identity) and delivered to the workload as **a file on disk**, rotated automatically by the platform. The consumer re-reads the file on each authentication challenge to pick up rotations without restarting.
The Kafka output currently supports `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, and `Kerberos/GSSAPI`. None of these work with platform-issued short-lived credentials. Without `OAUTHBEARER` support, Beats cannot participate in zero-long-lived-secret Kubernetes deployments with any of the Kafka platforms listed above.
## Proposed configuration
AWS MSK example (no extensions needed in this case):
```yaml
sasl.mechanism: OAUTHBEARER
sasl.credentials_path: /var/run/secrets/eks.amazonaws.com/serviceaccount/token
```
**Behavior:**
- On each SASL handshake, read the file at `credentials_path` and return its contents (whitespace-trimmed) as the bearer token.
- No caching — re-reading on every call ensures rotated tokens are picked up automatically, without restarting the beat.
- `credentials_path` is required when `mechanism: OAUTHBEARER`; `username` and `password` are not used.
- `extensions` is a free-form `map[string]string` that populates the `ext` field defined in [RFC 7628 §3.1](https://www.rfc-editor.org/rfc/rfc7628#section-3.1). Different platforms use different extension keys; the generic map keeps the API provider-agnostic.
## Implementation sketch
The [elastic/sarama](https://github.com/IBM/sarama) library already provides the `AccessTokenProvider` interface and handles the SASL/OAUTHBEARER protocol. The change is purely in Beats, wiring existing sarama support to a file-based provider. The diff touches four files in `libbeat`:
**New `libbeat/common/kafka/oauthbearer.go`** — a `fileTokenProvider` implementing `sarama.AccessTokenProvider`:
```go
type fileTokenProvider struct {
credentialsPath string
extensions map[string]string
}
func (p *fileTokenProvider) Token() (*sarama.AccessToken, error) {
raw, err := os.ReadFile(p.credentialsPath)
if err != nil {
return nil, fmt.Errorf("reading OAUTHBEARER credentials file %q: %w", p.credentialsPath, err)
}
return &sarama.AccessToken{
Token: strings.TrimSpace(string(raw)),
Extensions: p.extensions,
}, nil
}
```
**`libbeat/common/kafka/sasl.go`** — two new fields on `SaslConfig`, one new case in `ConfigureSarama`:
```go
type SaslConfig struct {
SaslMechanism string `config:"mechanism"`
CredentialsPath string `config:"credentials_path"` // new
Extensions map[string]string `config:"extensions"` // new
// in ConfigureSarama():
case saslTypeOAuthBearer:
cfg.Net.SASL.Enable = true
cfg.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeOAuth)
cfg.Net.SASL.TokenProvider = &fileTokenProvider{
credentialsPath: c.CredentialsPath,
extensions: c.Extensions,
}
```
**`libbeat/common/kafka/sasl_fips.go` / `sasl_nofips.go`** — add `OAUTHBEARER` to each file's `Validate()` allowlist.
**`libbeat/outputs/kafka/config.go`** — add a standalone `OAUTHBEARER` case in `newSaramaConfig` that does not require `username`/`password`, alongside the existing `username != ""` case for PLAIN/SCRAM.
## References
- Apache Kafka: [Authentication using SASL](https://kafka.apache.org/43/security/authentication-using-sasl/)
- Confluent Cloud: [OAuth/OIDC workload identity overview](https://docs.confluent.io/cloud/current/security/authenticate/workload-identities/identity-providers/oauth/overview.html)
- [KIP-255: OAuth Authentication via SASL/OAUTHBEARER](https://cwiki.apache.org/confluence/display/KAFKA/KIP-255%3A+OAuth+Authentication+via+SASL%2FOAUTHBEARER)
- [RFC 7628](https://www.rfc-editor.org/rfc/rfc7628) — SASL Mechanisms for OAuth
- [IBM/sarama](https://github.com/IBM/sarama) — `AccessTokenProvider` interface this change wires into is already implemented in the library Beats uses
- Related: #42248 (MSK IAM auth, stalled)
Contributor guide
Research direction
Start by reading libbeat/common/kafka/sasl.go, sasl_fips.go, sasl_nofips.go, and libbeat/outputs/kafka/config.go to trace SASL validation and Sarama configuration. Then add the file-based provider in libbeat/common/kafka/oauthbearer.go and verify that OAUTHBEARER reads and trims the token file on each handshake, accepts extensions, and does not require username or password.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- backend, distributed-systems, security
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100