/**
 * 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.7
 */
@PublicEvolving
public class FlussPrincipal implements Principal {
    public static final FlussPrincipal ANONYMOUS = new FlussPrincipal("ANONYMOUS", "User");
    /** The wildcard principal, which represents all principals. */
    public static final FlussPrincipal WILD_CARD_PRINCIPAL = new FlussPrincipal("*", "*");

    /** The wildcard principal, which is only used in filter to matches all principals. */
    public static final FlussPrincipal ANY = new FlussPrincipal(null, null);

    private final String name;
    private final String type;

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

/** The connection session of a request. */
public class Session implements Serializable {
    private final short apiVersion;
    private final String listenerName;
    private final boolean isInternal;
    private final InetAddress inetAddress;
    private final FlussPrincipal principal;
}

/** Authenticator for client side. */
@PublicEvolving
public interface ClientAuthenticator extends Closeable {

    /** The protocol name of the authenticator, which will send in the AuthenticateRequest. */
    String protocol();

    /** Initialize the authenticator. */
    default void initialize(AuthenticateContext context) throws AuthenticationException {}

    /**
     * Determines whether the client authenticator should proactively send an initial token to the
     * server.
     *
     * <p>When this method returns {@code true}, it indicates that the client is the initiator of
     * the authentication exchange and should actively call {@link #authenticate(byte[])
     * authenticate(new byte[0])} to generate and send the initial token without waiting for a
     * challenge from the server.
     *
     * @return {@code true} if the client should initiate authentication by sending an initial
     *     token; {@code false} if the client expects to receive the first token or challenge from
     *     the server.
     */
    default boolean hasInitialTokenResponse() {
        return true;
    }

    /**
     * * Generates the initial token or calculates a token based on the server's challenge, then
     * sends it back to the server. This method sets the client authentication status as complete if
     * the authentication succeeds. <br>
     * If this method returns `null`, it indicates that both the client and server have completed
     * authentication, and no further token exchange is required.
     *
     * <p>Below are examples illustrating the design rationale:
     *
     * <p>1. **Username and Password Authentication (One-Way Authentication):** <br>
     * - Client → Server: Sends an initial token containing the username and password, marking the
     * client authentication as complete. <br>
     * - Server verifies the token and sets its status as complete. <br>
     * - Server → Client: Responds with success or failure.
     *
     * <p>2. **GSS-KRB5 Authentication (Two-Way Authentication with a Third-Party Authentication
     * Server):** <br>
     * - Client → Server: Sends an initial token calculated using the client's ticket. <br>
     * - Server verifies the client's ticket and generates a challenge based on the client's token
     * and the server's ticket. <br>
     * - Server → Client: Sends the challenge. <br>
     * - Client verifies the server's ticket, sets its status as complete, and calculates a response
     * token. <br>
     * - Client → Server: Sends the response token. <br>
     * - Server verifies the token, sets its status as complete, and responds with success or
     * failure.
     *
     * <p>3. **SCRAM-SHA-256 Authentication (Two-Way Authentication without a Third-Party
     * Authentication Server):** <br>
     * - Client → Server: Sends an initial token containing a random string. <br>
     * - Server verifies the token format and responds with a salt value. <br>
     * - Server → Client: Sends the salt value. <br>
     * - Client → Server: Encrypts the password with the salt and sends the result. <br>
     * - Server verifies the token, sets its status as complete, and sends a signature challenge.
     * <br>
     * - Server → Client: Sends the server's signature. <br>
     * - Client verifies the signature, sets its status as complete, and returns `null` to indicate
     * no further token exchange is needed.
     *
     * @param data The initial token or server's challenge.
     * @return The token to send back to the server, or `null` if authentication is complete.
     */
    @Nullable
    byte[] authenticate(byte[] data) throws AuthenticationException;

    /** Checks if the authentication from client side is completed. */
    boolean isCompleted();

    default void close() {}

    /** The context of the authentication process. */
    interface AuthenticateContext {
        String ipAddress();
    }
}

**
 * Authenticator for server side.
 *
 * @since 0.7
 */
@PublicEvolving
public interface ServerAuthenticator extends Closeable {

    String protocol();

    default void matchProtocol(String protocol) throws AuthenticationException {
        if (!protocol().equalsIgnoreCase(protocol)) {
            throw new AuthenticationException(
                    String.format(
                            "Authenticate protocol not match: protocol of server is '%s' while protocol of client is '%s'",
                            protocol(), protocol));
        }
    }

    /** Initialize the authenticator. */
    default void initialize(AuthenticateContext context) {}

    /**
     * * Generates the challenge based on the client's token, then sends it back to the client. This
     * method sets the server authentication status as complete if the authentication succeeds.
     *
     * <p>Below are examples illustrating the design rationale:
     *
     * <p>1. **Username and Password Authentication (One-Way Authentication):** <br>
     * - Client → Server: Sends an initial token containing the username and password, marking the
     * client authentication as complete. <br>
     * - Server verifies the token and sets its status as complete. <br>
     * - Server → Client: Responds with success or failure.
     *
     * <p>2. **GSS-KRB5 Authentication (Two-Way Authentication with a Third-Party Authentication
     * Server):** <br>
     * - Client → Server: Sends an initial token calculated using the client's ticket. <br>
     * - Server verifies the client's ticket and generates a challenge based on the client's token
     * and the server's ticket. <br>
     * - Server → Client: Sends the challenge. <br>
     * - Client verifies the server's ticket, sets its status as complete, and calculates a response
     * token. <br>
     * - Client → Server: Sends the response token. <br>
     * - Server verifies the token, sets its status as complete, and responds with success or
     * failure.
     *
     * <p>3. **SCRAM-SHA-256 Authentication (Two-Way Authentication without a Third-Party
     * Authentication Server):** <br>
     * - Client → Server: Sends an initial token containing a random string. <br>
     * - Server verifies the token format and responds with a salt value. <br>
     * - Server → Client: Sends the salt value. <br>
     * - Client → Server: Encrypts the password with the salt and sends the result. <br>
     * - Server verifies the token, sets its status as complete, and sends a signature challenge.
     * <br>
     * - Server → Client: Sends the server's signature. <br>
     * - Client verifies the signature, sets its status as complete, and returns `null` to indicate
     * no further token exchange is needed.
     *
     * @param token the token sent by the client.
     * @return The challenge to send back to the server.
     */
    byte[] evaluateResponse(byte[] token) throws AuthenticationException;

    /** Checks if the authentication from server side is completed. */
    boolean isCompleted();

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

    /** Close the authenticator. */
    default void close() throws IOException {}

    /** The context of the authentication process. */
    interface AuthenticateContext {
        String ipAddress();

        String listenerName();

        String protocol();
    }
}

/**
 * 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.
 *
 * @since 0.7
 */
@PublicEvolving
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_SECURITY_PROTOCOL} for client-side
     *       authentication.
     *   <li>{@link com.alibaba.fluss.config.ConfigOptions#SERVER_SECURITY_PROTOCOL_MAP} for server
     *       listener-specific authentication.
     * </ul>
     *
     * @return The protocol name (e.g., "PLAINTEXT").
     */
    String authProtocol();
}

/**
 * Defines the client-side authentication plugin interface for implementing protocol-specific
 * identity validation logic.
 *
 * <p>The protocol type is specified via {@link
 * com.alibaba.fluss.config.ConfigOptions#CLIENT_SECURITY_PROTOCOL}, and configuration parameters
 * must be prefixed with {@code client.security.${protocol}}.
 */
@PublicEvolving
public interface ClientAuthenticationPlugin extends AuthenticationPlugin {
    /**
     * Creates a client-side authenticator instance for this authentication protocol.
     *
     * @return A new client authenticator instance implementing the protocol defined by {@link
     *     #authProtocol()}
     */
    ClientAuthenticator createClientAuthenticator(Configuration configuration);
}

**
 * Defines the server-side authentication plugin interface for implementing protocol-specific
 * identity validation logic.
 *
 * <p>The protocol type is specified via {@link
 * com.alibaba.fluss.config.ConfigOptions#SERVER_SECURITY_PROTOCOL_MAP}, and configuration
 * parameters must be prefixed with {@code security.${protocol}}.
 *
 * @since 0.7
 */
@PublicEvolving
public interface ServerAuthenticationPlugin extends AuthenticationPlugin {

    /**
     * Creates a server-side authenticator instance for this authentication protocol.
     *
     * @return A new server authenticator instance implementing the protocol defined by {@link
     *     #authProtocol()}
     */
    ServerAuthenticator createServerAuthenticator(Configuration configuration);
}

{"version": 1, 
 "acls": [ 
{ 
      "principals": ["user:alice", "group:kafka-devs"], 
      "permissionType": "ALLOW", 
      "operations": [ "READ", "WRITE" ], 
      "hosts": [ "host1", "host2" ] 
    }, 
   { 
     "principal": ["user:bob", "user:*"] ,
     "permissionType": "ALLOW", 
     "operations": [ "READ" ], 
     "hosts": [ "*" ] 
   }, 
   { 
     "principal": "user:bob", 
     "permissionType": "DENY", 
     "operations": [ "READ" ], 
     "hosts": [ "host1", "host2" ] 
   } 
 ] 
}

public enum ApiKeys {
 	CREATE_ACLS(1039, 0, 0, PUBLIC),
    LIST_ACLS(1040, 0, 0, PUBLIC),
    DROP_ACLS(1041, 0, 0, PUBLIC),
}

@PublicEvolving
public interface Admin extends AutoCloseable {
 /**
     * Retrieves ACL entries filtered by principal for the specified resource.
     *
     * <p>1. Validates the user has 'describe' permission on the resource. 2. Returns entries
     * matching the principal if permitted; throws an exception otherwise.
     *
     * @return A CompletableFuture containing the filtered ACL entries.
     */
    CompletableFuture<Collection<AclBinding>> listAcls(AclBindingFilter aclBindingFilter);

    /**
     * Creates multiple ACL entries in a single atomic operation.
     *
     * <p>1. Validates the user has 'alter' permission on the resource. 2. Creates the ACL entries
     * if valid and permitted.
     *
     * <p>Each entry in {@code aclBindings} must have a valid principal, operation and permission.
     *
     * @param aclBindings List of ACL entries to create.
     * @return A CompletableFuture indicating completion of the operation.
     */
    CreateAclsResult createAcls(Collection<AclBinding> aclBindings);

    /**
     * Removes multiple ACL entries in a single atomic operation.
     *
     * <p>1. Validates the user has 'alter' permission on the resource. 2. Removes entries only if
     * they exactly match the provided entries (principal, operation, permission). 3. Does not
     * remove entries if any of the ACL entries do not exist.
     *
     * @param filters List of ACL entries to remove.
     * @return A CompletableFuture indicating completion of the operation.
     */
    DropAclsResult dropAcls(Collection<AclBindingFilter> filters);
}

**
 * The {@code Authorizer} interface defines the contract for managing and enforcing access control
 * in a system. It provides methods for authorization, adding or removing ACLs (Access Control
 * Lists), and listing existing ACL bindings.
 */
public interface Authorizer extends Closeable {

    /**
     * Initializes the authorizer. This method should be called before any other methods are used.
     *
     * @throws Exception if an error occurs during initialization
     */
    void startup() throws Exception;

    /** Closes the authorizer and releases any associated resources. */
    void close();

    /**
     * Checks if a given session is authorized to perform a specific operation on a resource. This
     * method is used for authorization checks and returns a boolean result indicating whether the
     * session has the required permissions. It does not throw an exception if the session is
     * unauthorized.
     *
     * @param session the session associated with the request
     * @param operationType the type of operation being checked
     * @param resource the resource on which the operation is being performed
     * @return true if the session is authorized, false otherwise
     */
    boolean isAuthorized(Session session, OperationType operationType, Resource resource);

    /**
     * Checks if a given session is authorized to perform a specific operation on a resource. Unlike
     * {@link #isAuthorized(Session, OperationType, Resource)}, this method enforces authorization
     * by throwing an {@link AuthenticationException} if the session is not authorized to perform
     * the operation. It is intended for scenarios where access control must be strictly enforced.
     *
     * @param session the session associated with the request
     * @param operationType the type of operation being checked
     * @param resource the resource on which the operation is being performed
     * @throws AuthorizationException if the session is not authorized to perform the operation
     */
    void authorize(Session session, OperationType operationType, Resource resource)
            throws AuthorizationException;

    /**
     * Filters a collection of resource names based on the provided session, operation, resources.
     */
    Collection<Resource> filterByAuthorized(
            Session session, OperationType operation, List<Resource> resources);

    /**
     * Adds multiple ACL bindings to the system after verifying that the session has the required
     * 'alter' permission on the associated resources.
     *
     * @param session the session associated with the request
     * @param aclBindings a list of ACL bindings to add
     * @return a list of results indicating the outcome of each ACL creation attempt
     * @throws SecurityException if the session does not have the required 'alter' permission
     */
    List<AclCreateResult> addAcls(Session session, List<AclBinding> aclBindings);

    /**
     * Removes ACL bindings from the system based on the provided filters, after verifying that the
     * session has the required 'alter' permission on the associated resources.
     *
     * @param session the session associated with the request
     * @param filters a list of filters specifying which ACL bindings to remove
     * @return a list of results indicating the outcome of each ACL deletion attempt
     * @throws SecurityException if the session does not have the required 'alter' permission
     */
    List<AclDeleteResult> dropAcls(Session session, List<AclBindingFilter> filters);

    /**
     * Lists all ACL bindings that match the specified filter.
     *
     * @param filter the filter to apply when listing ACL bindings
     * @return a collection of matching ACL bindings
     */
    Collection<AclBinding> listAcls(Session session, AclBindingFilter filter);
}

/** AuthorizePlugin. */
public interface AuthorizationPlugin extends Plugin {

    String identifier();

    Authorizer createAuthorizer(Context context);

    /** Provides session information describing the authorizer to be accessed. */
    @PublicEvolving
    interface Context {

        /** Get configuration of fluss server to authorize. */
        Configuration getConfiguration();

        /** Get zookeeper client to store authorization information. */
        Optional<ZooKeeperClient> getZooKeeperClient();
    }
}

**
 * Represents a resource object with type and name, used for ACL (Access Control List) management.
 *
 * @since 0.7
 */
public class Resource {
    public static final String WILDCARD_RESOURCE = "*";
    public static final String TABLE_SPLITTER = "\\.";

    public static final String FLUSS_CLUSTER = "fluss-cluster";
    private final ResourceType type;
    private final String name;

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

    public stat

**
 * Enumeration representing resource types used in ACL management. Permission hierarchies are
 * defined as follows:
 *
 * <p>1. CLUSTER permissions implicitly include all own database permissions under the same
 * operation. 2. DATABASE permissions implicitly include all own tables permissions under the same
 * operation 3. ANY represents global permissions covering all resource types, this only used for
 * lookup.
 *
 * <p>This hierarchy ensures that higher-level permissions automatically grant access to lower-level
 * resources within the same operation.
 *
 * @since 0.7
 */
@PublicEvolving
public enum ResourceType {
    /** In a filter, matches any ResourceType. */
    ANY((byte) 1),
    CLUSTER((byte) 2),
    DATABASE((byte) 3),
    TABLE((byte) 4);

    private final byte code;

    ResourceType(byte code) {
        this.code = code;
    }

    public byte getCode() {
        return code;
    }

    public static ResourceType fromCode(byte code) {
        for (ResourceType resourceType : values()) {
            if (resourceType.code == code) {
                return resourceType;
            }
        }
        throw new IllegalArgumentException("Unknown resource type code " + code);
    }

    public static ResourceType fromName(String name) {
        for (ResourceType resourceType : values()) {
            if (resourceType.name().equalsIgnoreCase(name)) {
                return resourceType;
            }
        }
        throw new IllegalArgumentException("Unknown resource type name " + name);
    }
}

/**
 * Enumeration representing operation types used in ACL (Access Control List) systems.
 *
 * <p><b>Permission Inheritance Rules:</b>
 *
 * <p>1. {@link #ALL} grants permission for all operations
 *
 * <p>2. {@link #READ}, {@link #WRITE}, {@link #CREATE}, {@link #DROP}, and {@link #ALTER}
 * implicitly include {@link #DESCRIBE}
 *
 * @since 0.7
 */
@PublicEvolving
public enum OperationType {
    /** In a filter, matches any OperationType. */
    ANY((byte) 1),
    ALL((byte) 2),
    READ((byte) 3),
    WRITE((byte) 4),
    CREATE((byte) 5),
    DROP((byte) 6),
    ALTER((byte) 7),
    DESCRIBE((byte) 8);

    private final byte code;

    OperationType(byte code) {
        this.code = code;
    }

    public byte getCode() {
        return code;
    }

**
 * Enumeration representing permission types used in ACL.
 *
 * @since 0.7
 */
public enum PermissionType {
    /** In a filter, matches any PermissionType. */
    ANY((byte) 1),

    /**
     * Permission type indicating allowed access. Grants explicit permission for specified
     * operations on resources.
     */
    ALLOW((byte) 2);

    // todo: Will introduce DENY type in the future.

    private final byte code;
}

**
 * Represents an Access Control List (ACL) entry defining permissions for a principal on a specific
 * resource operation.
 *
 * <p>This class encapsulates the core elements of an ACL entry:
 *
 * <p>- Principal: User/role identifier
 *
 * <p>- Permission: Allow or deny access
 *
 * <p>- Host: Source host restriction (can be "*" for any host)
 *
 * <p>- Operation: Specific operation type (e.g., read/write)
 *
 * @since 0.7
 */
@PublicEvolving
public class AccessControlEntry {
    public static final String WILD_CARD_HOST = "*";

    private final FlussPrincipal principal;
    private final PermissionType permissionType;
    private final String host;
    private final OperationType operationType;

    public AccessControlEntry(
            FlussPrincipal principal,
            String host,
            OperationType operationType,
            PermissionType permissionType) {
        this.principal = checkNotNull(principal);
        this.host = checkNotNull(host);
        this.permissionType = checkNotNull(permissionType);
        this.operationType = checkNotNull(operationType);
    }

    public FlussPrincipal getPrincipal() {
        return principal;
    }

    public PermissionType getPermissionType() {
        return permissionType;
    }

    public String getHost() {
        return host;
    }

    public OperationType getOperationType() {
        return operationType;
    }

    @Override
    public boolean equals(Object o) {
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        AccessControlEntry that = (AccessControlEntry) o;
        return Objects.equals(principal, that.principal)
                && permissionType == that.permissionType
                && Objects.equals(host, that.host)
                && operationType == that.operationType;
    }

**
 * Represents a filter which matches access control entries.
 *
 * <p>The API for this class is still evolving and we may break compatibility in minor releases, if
 * necessary.
 *
 * @since 0.7
 */
@PublicEvolving
public class AccessControlEntryFilter {
    @Nullable private final FlussPrincipal principal;
    private final PermissionType permissionType;
    @Nullable private final String host;
    private final OperationType operationType;

    public static final AccessControlEntryFilter ANY =
            new AccessControlEntryFilter(
                    FlussPrincipal.ANY, null, OperationType.ANY, PermissionType.ANY);

    public AccessControlEntryFilter(
            @Nullable FlussPrincipal principal,
            @Nullable String host,
            OperationType operation,
            PermissionType permissionType) {
        this.principal = principal;
        this.host = host;
        this.operationType = checkNotNull(operation);
        this.permissionType = checkNotNull(permissionType);
    }
}

**
 * Represents an Access Control List (ACL) binding that associates a resource with specific access
 * control rules.
 *
 * <p>An {@code AclBinding} encapsulates the relationship between a {@link Resource} and an {@link
 * AccessControlEntry}, defining which principal (user/role) has what permissions on the resource.
 *
 * @since 0.7
 */
@PublicEvolving
public class AclBinding {
    private final Resource resource;
    private final AccessControlEntry accessControlEntry;

    public AclBinding(Resource resource, AccessControlEntry accessControlEntry) {
        this.resource = resource;
        this.accessControlEntry = accessControlEntry;
    }
}