package org.apache.kafka.server.authorizer;

import java.io.Closeable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.KafkaRequestContext;

/**
 *
 * Pluggable authorizer interface for Kafka brokers.
 *
 * Startup sequence in brokers:
 * <ol>
 *   <li>Broker creates authorizer instance if configured in `authorizer.class` or `authorizer.class.name`.</li>
 *   <li>Broker configures and starts authorizer instance. Authorizer implementation starts loading its metadata.</li>
 *   <li>Broker starts SocketServer to accept connections and process requests.</li>
 *   <li>For each listener, SocketServer waits for authorization metadata to be available in the
 *       authorizer before accepting connections. The future returned by {@link #start(String, SecurityProtocol)}
 *       must return only when authorizer is ready to authorize requests on the listener.</li>
 *   <li>Broker accepts connections. For each connection, broker performs authentication and then accepts Kafka requests.
 *       For each request, broker invokes {@link #authorize(KafkaRequestContext, List)} to authorize
 *       actions performed by the request.</li>
 * </ol>
 *
 * Authorizer implementation class may optionally implement the following interfaces:
 * <ul>
 *   <li>@{@link org.apache.kafka.common.Reconfigurable} to enable dynamic reconfiguration
 *   without restarting the broker.</li>
 *   <li>{@link org.apache.kafka.common.ClusterResourceListener} to obtain cluster id</li>
 * </ul>
 */
@InterfaceStability.Evolving
public interface Authorizer extends Configurable, Closeable {

    /**
     * Starts loading authorization metadata and returns futures that can be used to wait until
     * metadata for authorizing requests on each listener is available. Each listener will be
     * started only after its metadata is available and authorizer is ready to start authorizing
     * requests on that listener.
     *
     * @param listeners Listener names with their security protocols
     * @return CompletableFutures for each listener that completes when authorizer is ready to
     *         start authorizing requests on that listener. Returned map contains one future
     *         for each listener name in the input `listeners` map.
     */
    Map<String, CompletableFuture<Void>> start(Map<String, SecurityProtocol> listeners);

    /**
     * Authorizes the specified action. Additional metadata for the action is specified
     * in `requestContext`.
     *
     * @param requestContext Request context including request type, security protocol and listener name
     * @param actions Actions being authorized including resource and operation for each action
     * @return List of authorization results for each action in the same order as the provided actions
     */
    List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);

    /**
     * Creates a new ACL binding.
     *
     * @param requestContext Request context if the ACL is being created by a broker to handle
     *        a client request to create ACLs. This may be null if ACLs are created directly in ZooKeeper
     *        using AclCommand.
     * @param resourcePattern Resource pattern for which ACLs are being added
     * @param accessControlEntries List of access control entries to add for resource
     *
     * @return Create result for each access control entry in the same order as in the input list
     */
    List<AclCreateResult> createAcls(KafkaRequestContext requestContext,
                                     ResourcePattern resourcePattern,
                                     List<AccessControlEntry> accessControlEntries);

    /**
     * Deletes the specified ACL binding.
     *
     * @param requestContext Request context if the ACL is being deleted by a broker to handle
     *        a client request to delete ACLs. This may be null if ACLs are deleted directly in ZooKeeper
     *        using AclCommand.
     * @param resourcePattern Resource pattern for which ACLs are being removed
     * @param accessControlEntries Set of access control entries to remove for resource
     *
     * @return Delete result for each access control entry in the same order as in the input list.
     *         Each result indicates if the entry was actually deleted.
     */
    List<AclDeleteResult> deleteAcls(KafkaRequestContext requestContext,
                                     ResourcePattern resourcePattern,
                                     List<AccessControlEntry> accessControlEntries);

    /**
     * Deletes all ACL bindings for the specified resource pattern. Only ACLs explicitly configured
     * with this resource pattern are deleted, no matching is performed.
     *
     * @param requestContext Request context if the ACLs are being deleted by a broker to handle
     *        a client request to delete ACLs. This may be null if ACLs are deleted directly in ZooKeeper
     *        using AclCommand.
     * @param resourcePattern Resource pattern whose ACLs are deleted
     * @return true if at least one ACL binding was deleted, false otherwise
     */
    boolean deleteAcls(KafkaRequestContext requestContext, ResourcePattern resourcePattern);

    /**
     * Returns all ACL bindings.
     *
     * @return Access control entries of all ACL bindings grouped by resource pattern.
     */
    Map<ResourcePattern, Set<AccessControlEntry>> acls();

    /**
     * Returns access control entries defined for the specified resource pattern.
     *
     * @param resourcePattern Resource pattern for which ACLs are returned. No matching is performed,
     *        only ACLs defined for the specified pattern are returned.
     *
     * @return ACL bindings for resource pattern.
     */
    Set<AccessControlEntry> acls(ResourcePattern resourcePattern);
}

package org.apache.kafka.common;

import java.net.InetAddress;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.SecurityProtocol;

/**
 * Request context interface that provides data from request header as well as connection
 * and authentication information to plugins.
 */
@InterfaceStability.Evolving
public interface KafkaRequestContext {

    /**
     * Returns name of listener on which request was received.
     */
    String listenerName();

    /**
     * Returns the security protocol for the listener on which request was received.
     */
    SecurityProtocol securityProtocol();

    /**
     * Returns authenticated principal for the connection on which request was received.
     */
    KafkaPrincipal principal();

    /**
     * Returns client IP address from which request was sent.
     */
    InetAddress clientAddress();

    /**
     * Returns the request type (ApiKey) from the request header. For fetch requests,
     * the metrics names FetchFollower and FetchConsumer will be used to distinguish between
     * replica fetch requests and client fetch requests.
     */
    String requestType();

    /**
     * Returns the request version from the request header.
     */
    int requestVersion();

    /**
     * Returns the client id from the request header.
     */
    String clientId();

    /**
     * Returns the correlation id from the request header.
     */
    int correlationId();
}

package org.apache.kafka.server.authorizer;

import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;

@InterfaceStability.Evolving
public class Action {

    private final ResourcePattern resourcePattern;
    private final AclOperation operation;
    private final AuthorizationMode authorizationMode;
    private final int count;

    public Action(AclOperation operation,
                  ResourceType resourceType,
                  String resourceName,
                  AuthorizationMode authorizationMode,
                  int count) {
        this.operation = operation;
        this.resourcePattern = new ResourcePattern(resourceType, resourceName, PatternType.LITERAL);
        this.authorizationMode = authorizationMode;
        this.count = count;
    }

    /**
     * Resource on which action is being performed.
     */
    public ResourcePattern resourcePattern() {
        return resourcePattern;
    }

    /**
     * Operation being performed.
     */
    public AclOperation operation() {
        return operation;
    }

    /**
     * Authorization mode to enable authorization logs to distinguish between attempts
     * to access unauthorized resources and other filtering operations performed by the broker.
     */
    public AuthorizationMode authorizationMode() {
        return authorizationMode;
    }

    /**
     * Number of times the authorization result is used. For example, a single topic
     * authorization result may be used with `count` partitions of the topic within a single
     * request.
     */
    public int count() {
        return count;
    }

package org.apache.kafka.server.authorizer;

public enum AuthorizationMode {
    /**
     * Access was requested to resource. If authorization result is ALLOWED, access is granted to
     * the resource to perform the request. If DENIED, request is failed with authorization failure.
     */
    MANDATORY,

    /**
     * Access was requested to resource. If authorization result is ALLOWED, access is granted to
     * the resource to perform the request. If DENIED, alternative authorization rules are applied
     * to determine if access is allowed.
     */
    OPTIONAL,

    /**
     * Access was requested to authorized resources (e.g. to subscribe to regex pattern).
     * Request is performed on resources whose authorization result is ALLOWED and the rest of
     * the resources are filtered out.
     */
    FILTER,

    /**
     * Request to list authorized operations. No access is actually performed by this request
     * based on the authorization result.
     */
    LIST_AUTHORIZED
}

package org.apache.kafka.server.authorizer;

public enum AuthorizationResult {
    ALLOWED,
    DENIED
}

import org.apache.kafka.common.annotation.InterfaceStability;
@InterfaceStability.Evolving
public class AclCreateResult {
    private final Throwable exception;

    public AclCreateResult() {
        this(null);
    }

    public AclCreateResult(Throwable exception) {
        this.exception = exception;
    }

    /**
     * Returns any exception during create. If exception is null, the request has succeeded.
     */
    public Throwable exception() {
        return exception;
    }

    /**
     * Returns true if the request failed.
     */
    public boolean failed() {
        return exception != null;
    }
}

import org.apache.kafka.common.annotation.InterfaceStability;
@InterfaceStability.Evolving
public class AclDeleteResult {
    private final Throwable exception;
    private final boolean deleted;

    public AclDeleteResult(Throwable exception) {
        this(false, exception);
    }

    public AclDeleteResult(boolean deleted) {
        this(deleted, null);
    }

    private AclDeleteResult(boolean deleted, Throwable exception) {
        this.deleted = deleted;
        this.exception = exception;
    }

    /**
     * Returns any exception while attempting to delete an ACL.
     */
    public Throwable exception() {
        return exception;
    }

    /**
     * Returns true if request has failed with an exception. {@link #deleted()} indicates if
     * an ACL was deleted.
     */
    public boolean failed() {
        return exception != null;
    }

    /**
     * Returns true if a delete was performed.
     */
    public boolean deleted() {
        return deleted;
    }
}