Introduce authenticate in Fluss.

[loserwang1024]

Purpose

Linked issue: close #484 . This Pr is depended on #531 .

Brief change log

Introduce authenticate in Fluss.

Tests

com.alibaba.fluss.rpc.netty.authenticate.AuthenticationTest

API and Format

Config

• server.authenticate.protocol.map: A map defining the authentication protocol for each listener. The format is 'listenerName1:protocol1,listenerName2:protocol2', e.g., 'INTERNAL:PLAINTEXT,CLIENT:GSSAPI'. Each listener can be associated with a specific authentication protocol. Listeners not included in the map will use PLAINTEXT by default, which does not require authentication.
• client.authenticate.protocol:The authentication protocol used to authenticate the client.

The config of one protocol:

• client.authenticate.${protocol}.xxx
• server.authenticate.${protocol}.xxx

Interface

AuthenticationPlugin for client and server:

/**
 * The AuthenticationPlugin interface defines a contract for authentication mechanisms in the Fluss
 * RPC system. Implementations of this interface provide specific authentication protocols (e.g.,
 * PLAINTEXT, AK-SK) and must be registered via the Service Provider Interface (SPI) mechanism to be
 * discoverable by the system.
 *
 * <p>Authentication plugins are used by {@link AuthenticatorLoader} to create client and server
 * authenticators based on configuration settings such as {@link
 * com.alibaba.fluss.config.ConfigOptions#CLIENT_AUTHENTICATE_PROTOCOL} and {@link
 * com.alibaba.fluss.config.ConfigOptions#SERVER_AUTHENTICATE_PROTOCOL_MAP}.
 *
 * <p>To implement this interface:
 *
 * <ol>
 *   <li>Define the supported authentication protocol via {@link #authProtocol()}.
 *   <li>Implement the required logic for client/server authentication (via subclasses like {@link
 *       ClientAuthenticationPlugin}).
 *   <li>Register the plugin implementation in {@code
 *       META-INF/services/com.alibaba.fluss.rpc.authenticate.AuthenticationPlugin}.
 * </ol>
 *
 * @since 0.1
 */
public interface AuthenticationPlugin extends Plugin {
    /**
     * Returns the authentication protocol identifier for this plugin (e.g., "PLAINTEXT", "AK-SK").
     * This identifier is used by the system to match plugins with configuration settings and select
     * the appropriate authentication mechanism.
     *
     * <p>It is typically configured via:
     *
     * <ul>
     *   <li>{@link com.alibaba.fluss.config.ConfigOptions#CLIENT_AUTHENTICATE_PROTOCOL} for
     *       client-side authentication.
     *   <li>{@link com.alibaba.fluss.config.ConfigOptions#SERVER_AUTHENTICATE_PROTOCOL_MAP} for
     *       server listener-specific authentication.
     * </ul>
     *
     * @return The protocol name (e.g., "PLAINTEXT").
     */
    String authProtocol();
}


public interface ClientAuthenticationPlugin extends AuthenticationPlugin {

    ClientAuthenticator createClientAuthenticator(Configuration configuration);
}

public interface ServerAuthenticationPlugin extends AuthenticationPlugin {

    ServerAuthenticator createServerAuthenticator(Configuration configuration);
}

Authenticator for client and server:

public interface ServerAuthenticator {

    String protocol();

    byte[] evaluateResponse(byte[] token) throws AuthenticationException;

    /**
     * Create principal from authenticated token for later authorization.(this can only invoke if is
     * complete).
     */
    FlussPrincipal createPrincipal();

    boolean isComplete();
}

public interface ClientAuthenticator {

    String protocol();

    byte[] authenticate(byte[] data);

    boolean isComplete();
}

Principal to identify user:

/**
 * Represents a security principal in Fluss, defined by a {@code name} and {@code type}.
 *
 * <p>The principal type indicates the category of the principal (e.g., "User", "Group", "Role"),
 * while the name identifies the specific entity within that category. By default, the simple
 * authorizer uses "User" as the principal type, but custom authorizers can extend this to support
 * role-based or group-based access control lists (ACLs).
 *
 * <p>Example usage:
 *
 * <ul>
 *   <li>{@code new FlussPrincipal("admin", "User")} – A standard user principal.
 *   <li>{@code new FlussPrincipal("admins", "Group")} – A group-based principal for authorization.
 * </ul>
 *
 * @since 0.1
 */
public class FlussPrincipal implements Principal {
    public static final FlussPrincipal ANONYMOUS = new FlussPrincipal("ANONYMOUS", "User");

    private final String name;
    private final String type;

    public FlussPrincipal(String name, String type) {
        this.name = name;
        this.type = type;
    }

    @Override
    public String getName() {
        return name;
    }

    public String getType() {
        return type;
    }

    @Override
    public boolean equals(Object o) {
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        FlussPrincipal that = (FlussPrincipal) o;
        return Objects.equals(name, that.name) && Objects.equals(type, that.type);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name, type);
    }
}

Documentation

[loserwang1024]

@wuchong , I have rebased this pr into main, would you like to help review it?

[wuchong]

Flink119CatalogITCase>FlinkCatalogITCase.testAuthentication:449 is failed.

[wuchong]

I'm going to review this PR, could you rebase to the main branch?

[loserwang1024]

could you rebase to the main branch

@wuchong , Done it

[wuchong]

LGTM. I appended a commit to address the following minor comemtns.

[wuchong]

should evaluateResponse first then check authenticator.isCompleted()?

[wuchong]

not used

[wuchong]

not used

[wuchong]

We may use uppercase when developing the auth plugin(such as: RAM, SSL, SASL), but we can loose this to equalsIgnoreCase to allow users use lowercase in client side:

client.security.protocol: ram
client.security.ram.ak: xxxx

instead of having to use

client.security.protocol: RAM
client.security.RAM.ak: xxxx

[wuchong]

@loserwang1024 please check the appended commit again.