[Feature] Support SASL/OAUTHBEARER authentication for Fluss RPC

[litiliu]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Fluss currently supports pluggable authentication and built-in SASL/PLAIN authentication. In many production environments, service identity is managed through OAuth2/OIDC providers and clients authenticate with bearer tokens, usually JWTs.

It would be useful for Fluss to support the standard SASL/OAUTHBEARER mechanism in the native Fluss RPC authentication layer. This would allow clients to authenticate with OAuth bearer tokens, and allow Fluss servers to validate those tokens and map validated claims to FlussPrincipal for the existing ACL authorization flow.

This feature can be designed to work together with FIP-29: (m)TLS Support. FIP-29 keeps transport security and authentication orthogonal: TLS protects the transport, while client.security.protocol and security.protocol.map select the authentication protocol. Following that model, OAuth bearer token authentication should be introduced as a SASL mechanism.

Solution

Add support for OAUTHBEARER as a SASL mechanism:

client.security.protocol=SASL
client.security.sasl.mechanism=OAUTHBEARER

Server-side example:

security.protocol.map: CLIENT:SASL
security.sasl.enabled.mechanisms: OAUTHBEARER
security.sasl.oauthbearer.jwks.endpoint.url: https://issuer.example.com/.well-known/jwks.json
security.sasl.oauthbearer.expected.issuer: https://issuer.example.com
security.sasl.oauthbearer.expected.audience: fluss
security.sasl.oauthbearer.principal.claim.name: sub

Proposed implementation direction:

• Refactor the current SASL implementation so SASL mechanisms are extensible instead of hard-coding only PLAIN.
• Preserve existing SASL/PLAIN behavior and compatibility.
• Add an OAUTHBEARER mechanism for the native Fluss RPC authentication layer.
• Add client-side token provider support.
• Add server-side JWT/JWKS token validation.
• Validate token signature, issuer, audience, expiration, not-before, and configurable clock skew.
• Map a configurable token claim to FlussPrincipal.
• Reuse the existing ACL authorization flow after authentication succeeds.
• Add tests for successful authentication, invalid token, expired token, wrong issuer, wrong audience, and principal mapping.

When TLS support from FIP-29 is available, users can run SASL/OAUTHBEARER over TLS by enabling TLS separately. Since bearer tokens are sensitive credentials, the implementation should warn or optionally fail fast when OAUTHBEARER is used without TLS.

Anything else?

Open questions for discussion:

• Should the first version support only JWT/JWKS validation, or also token introspection?
• Should Fluss provide built-in token providers, or only a pluggable token provider interface?
• Should OAUTHBEARER over non-TLS be allowed with a warning, or rejected by default?

Related design: FIP-29: (m)TLS Support

Willingness to contribute

• I'm willing to submit a PR!

[polyzos]

cc @affo @morazow

[affo]

Definitely in line with the current Kafka implementation, I think it makes sense.

@litiliu is there any issue around for SCRAM as well? Are you planning that one to?

[litiliu]

Definitely in line with the current Kafka implementation, I think it makes sense.

@litiliu is there any issue around for SCRAM as well? Are you planning that one to?

Thanks for the feedback.

I haven't found a dedicated SCRAM issue yet. I agree SCRAM would also fit well into the same direction, especially if we first make the SASL mechanism layer extensible instead of hard-coding PLAIN.

My current plan is to focus on OAUTHBEARER first because it is the immediate requirement I am looking at. But I think the first step should be a small refactor that allows multiple SASL mechanisms to be plugged in cleanly. After that, SCRAM-SHA-256/SCRAM-SHA-512 could be added as another mechanism in a separate issue/PR.

I can open a separate feature issue for SCRAM as well, so the scope of this one stays focused on OAUTHBEARER.