elastic / elastic/beats

[Kafka output] Add SASL/OAUTHBEARER support with file-based JWT token provider

Open
#51,265 1 comment 2 reactions 0 assignees View on GitHub
enhancement Team:Elastic-Agent-Data-Plane
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.