package org.apache.kafka.server.authorizer;

import java.io.Closeable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFutureCompletionStage;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AclBindingEndpoint;
import org.apache.kafka.common.acl.AclBindingFilterAclBinding;
import org.apache.kafka.common.annotationacl.InterfaceStabilityAclBindingFilter;
import org.apache.kafka.common.annotation.KafkaRequestContextInterfaceStability;
import org.apache.kafka.server.ServerInfo;
import org.apache.kafka.server.ServerInfo.Endpoint;

/**
 *
 *
/**
 *
 * Pluggable authorizer interface for Kafka brokers.
 *
 * Startup sequence in brokers:
 * <ol>
 *   <li>Broker creates authorizer instance if configured in `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(ServerInfoAuthorizerServerInfo)}
 *       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(KafkaRequestContextAuthorizableRequestContext, List)} to authorize
 *       actions performed by the request.</li>
 * </ol>
 *
 * Authorizer implementation class may optionally implement @{@link org.apache.kafka.common.Reconfigurable}
 * to enable dynamic reconfiguration without restarting the broker.
 * <p>
 * <b>Thread<b>Threading safetymodel:</b> All
 * <ul>
 *   <li>All authorizer operations including authorization and ACL updates must be thread-safe.
 * </p>li>
 */
@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 serverInfo Metadata for the broker including broker id and listener endpoints
     * @return CompletableFutures for each endpoint that completes when authorizer is ready to
     *         start authorizing requests on that listener.
     */
    Map<Endpoint, CompletableFuture<Void>> start(ServerInfo serverInfo);

    /**
     * Authorizes the specified action. Additional metadata for the action is specified
     * in `requestContext`.   <li>ACL update methods are asynchronous. Implementations with low update latency may return a
 *       completed future using {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}.
 *       This ensures that the request will be handled synchronously by the caller without using a
 *       purgatory to wait for the result. If ACL updates require remote communication which may block,
 *       return a future that is completed asynchronously when the remote operation completes. This enables
 *       the caller to process other requests on the request threads without blocking.</li>
 *   <li>Any threads or thread pools used for processing remote operations asynchronously can be started during
 *       {@link #start(AuthorizerServerInfo)}. These threads must be shutdown during {@link Authorizer#close()}.</li>
 * </ul>
 * </p>
 */
@InterfaceStability.Evolving
public interface Authorizer extends Configurable, Closeable {

    /**
     *
 Starts loading authorization metadata *and @paramreturns requestContextfutures Requestthat contextcan includingbe requestused type,to security protocol and listener namewait until
     * metadata @paramfor actionsauthorizing Actionsrequests beingon authorizedeach includinglistener resourceis andavailable. operationEach forlistener eachwill actionbe
     * @returnstarted Listonly ofafter authorizationits resultsmetadata foris eachavailable actionand inauthorizer theis sameready orderto as the provided actionsstart authorizing
     */
 requests on that listener.
 List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);*

     /**
 @param serverInfo Metadata for *the Createsbroker newincluding ACLbroker bindings.
id and listener   *endpoints
     * @param@return requestContextCompletionStage Requestfor contexteach ifendpoint thethat ACLcompletes iswhen beingauthorizer created by a brokeris ready to handle
     *         astart clientauthorizing requestrequests toon createthat ACLslistener.
 This may be null if ACLs are created directly in ZooKeeper*/
    Map<Endpoint, ? extends CompletionStage<Void>> start(AuthorizerServerInfo serverInfo);

     /**
     * Authorizes the usingspecified AclCommandaction.
 Additional metadata for the *action @paramis aclBindingsspecified
 ACL bindings to create
 * in   *`requestContext`.
     * @return<p>
 Create result for each ACL* bindingThis inis thea samesynchronous orderAPI asdesigned infor theuse inputwith list
locally cached ACLs. Since this */
    List<AclCreateResult> createAcls(KafkaRequestContext requestContext, List<AclBinding> aclBindings);

    /**method is invoked on the
     * request thread while processing each request, implementations of this method should avoid time-consuming
     * Deletes all ACL bindingsremote communication that matchmay theblock providedrequest filtersthreads.
     *
     * @param requestContext Request context ifincluding therequest ACLtype, issecurity beingprotocol deletedand by a broker to handlelistener name
     * @param actions Actions being authorized including  a client request to delete ACLs. This may be null if ACLs are deleted directly in ZooKeeper
     * resource and operation for each action
     * @return List of authorization results for each action in the same order as the provided actions
     */
  using AclCommand.
 List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext,  * @param aclBindingFilters Filters to match ACL bindings that are to be deletedList<Action> actions);

    /**
     * Creates new ACL bindings.
     * <p>
     * @returnThis Deleteis resultan forasynchronous eachAPI filterthat inenables the caller sameto orderavoid asblocking induring the input list.update. Implementations of this
     * API can return completed futures using {@link  Each result indicates which ACL bindings were actually deleted as well as anyjava.util.concurrent.CompletableFuture#completedFuture(Object)}
     * to process the update synchronously on the request thread.
     *
     * @param requestContext Request bindingscontext thatif matchedthe butACL couldis not be deleted.being created by a broker to handle
     */
    List<AclDeleteResult> deleteAcls(KafkaRequestContext requestContext, List<AclBindingFilter> aclBindingFilters);

a client request to /**
     * Returns ACL bindings which match the provided filter.create ACLs. This may be null if ACLs are created directly in ZooKeeper
     *        using AclCommand.
     * @return@param Iterator foraclBindings ACL bindings, which may be populated lazily.to create
     */
    Iterable<AclBinding> acls(AclBindingFilter filter);
}

deleteAcls(AuthorizableRequestContext requestContext, List<AclBindingFilter> aclBindingFilters);

    /**
     * Returns ACL bindings which match the provided filter.
     * <p>
     * This is a synchronous API designed for use with locally cached ACLs. This method is invoked on the request
     * thread while processing DescribeAcls requests and should avoid time-consuming remote communication that may
     * block request threads.
     *
     * @return Iterator for ACL bindings, which may be populated lazily.
     */
    Iterable<AclBinding> acls(AclBindingFilter filter);
}

package org.apache.kafka.server.authorizer;

import java.utilnet.ObjectsInetAddress;
import org.apache.kafka.common.aclannotation.AclOperationInterfaceStability;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.resource.PatternTypesecurity.auth.KafkaPrincipal;
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 int resourceReferenceCount;
    private final boolean logIfAllowed;
    private final boolean logIfDenied;

    public Action(AclOperation operation,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 AuthorizableRequestContext {

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

    /**
    ResourcePattern resourcePattern,
* Returns the security protocol for the listener on which request was received.
      int resourceReferenceCount,*/
    SecurityProtocol securityProtocol();

    /**
     * Returns authenticated principal booleanfor logIfAllowed,
the connection on which request was received.
     */
       boolean logIfDenied) {KafkaPrincipal principal();

    /**
    this.operation =* operation;
Returns client IP address from which request was this.resourcePattern = resourcePattern;sent.
     */
   this.logIfAllowed =InetAddress logIfAllowedclientAddress();

    /**
    this.logIfDenied = logIfDenied;
  * 16-bit API key of the request from the request header. See
      this.resourceReferenceCount = resourceReferenceCount;
* https://kafka.apache.org/protocol#protocol_api_keys for request types.
     */
    int }requestType();

    /**
     * Returns Resourcethe onrequest whichversion actionfrom isthe beingrequest performedheader.
     */
    publicint ResourcePattern resourcePatternrequestVersion() {;

    /**
    return resourcePattern;
* Returns the client }

id from the  /**
     * Operation being performedrequest header.
     */
    publicString AclOperation operationclientId() {
        return operation;
    }

    /**
     * IndicatesReturns ifthe auditcorrelation logsid trackingfrom ALLOWEDthe access should include this action if result isrequest header.
     */
 ALLOWED. The flag is true if access to a resource is granted while processing the request as a
     * result of this authorization. The flag is false only for requests used to describe access where
     * no operation on the resource is actually performed based on the authorization result.int correlationId();
}

package org.apache.kafka.server.authorizer;

import java.util.CollectionsObjects;
import java.util.Collectionorg.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.aclannotation.AclBindingInterfaceStability;
import org.apache.kafka.common.errorsresource.ApiExceptionPatternType;

public class AclDeleteResult {
    import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;

@InterfaceStability.Evolving
public class Action {

    private final ApiExceptionResourcePattern exceptionresourcePattern;
    private final Collection<DeletionResult>AclOperation deletionResultsoperation;

    publicprivate AclDeleteResult(ApiException exception) {final int resourceReferenceCount;
    private final boolean logIfAllowed;
  this(Collections.emptySet(), exception);
 private final boolean }logIfDenied;

    public AclDeleteResultAction(Collection<DeletionResult> deleteResults) {
AclOperation operation,
         this(deleteResults, null);
    }

    private AclDeleteResult(Collection<DeletionResult> deleteResults, ApiException exception) {
ResourcePattern resourcePattern,
           this.deletionResults = deleteResults;
     int resourceReferenceCount,
  this.exception = exception;
    }

    /**
     * Returnsboolean anylogIfAllowed,
 exception while attempting to match ACL filter to delete ACLs.
     */
    public ApiException exception(boolean logIfDenied) {
        this.operation return= exceptionoperation;
     }

   this.resourcePattern = /**resourcePattern;
     * Returns delete resultthis.logIfAllowed for each matching ACL binding.= logIfAllowed;
     */
   this.logIfDenied public= Collection<DeletionResult> deletionResults() {
logIfDenied;
        this.resourceReferenceCount = return deletionResultsresourceReferenceCount;
    }


    /**
     * DeleteResource resulton forwhich eachaction ACLis binding that matched a delete filterbeing performed.
     */
    public static class DeletionResultResourcePattern resourcePattern() {
        private final AclBinding aclBindingreturn resourcePattern;
    }

    private/**
 final ApiException exception;

   * Operation being performed.
     */
    public AclOperation DeletionResultoperation(AclBinding aclBinding) {
            this(aclBinding, null)return operation;
    }

    }/**

     * Indicates if publicaudit DeletionResult(AclBinding aclBinding, ApiException exception) {
      logs tracking ALLOWED access should include this action if result is
     * thisALLOWED.aclBinding =The aclBinding;
flag is true if access to a resource is granted while  this.exception = exception;processing the request as a
     * result of }

this authorization. The flag is false only for /**
requests used to describe access where
     * Returnsno ACLoperation bindingon that matched the deleteresource filter.is {@link #deleted()} indicates if
         *actually performed based on the bindingauthorization was deletedresult.
         */
        public AclBindingboolean aclBindinglogIfAllowed() {
            return aclBindinglogIfAllowed;
        }

        /**
     * Indicates if audit *logs Returnstracking exceptionDENIED thataccess resultedshould ininclude failurethis toaction deleteif ACLresult binding.is
     * DENIED. The flag */
is true if access to a resource was publicexplicitly ApiExceptionrequested exception()and {request
     * is denied as a result of returnthis exception;
authorization request. The flag is false if request }was

     * filtering out /**
authorized resources (e.g. to subscribe to regex pattern). The *flag Returnsis truealso
 if ACL binding was deleted,* false otherwise.
if this is an optional authorization where an  */alternative resource authorization is
     * applied if publicthis booleanfails deleted() {
      e.g. Cluster:Create which is subsequently overridden by Topic:Create).
     */
 return exception == null;
    public boolean logIfDenied() {
        return logIfDenied;
    }
    }
}

    /**
     * Number of times the resource being authorized is referenced within the request. For example, a single
     * request may reference `n` topic partitions of the same topic. Brokers will authorize the topic once
     * with `resourceReferenceCount=n`. Authorizers may include the count in audit logs.
     */
    public int resourceReferenceCount() {
        return resourceReferenceCount;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof Action)) {
            return false;
        }

        Action that = (Action) o;
        return Objects.equals(this.resourcePattern, that.resourcePattern) &&
            Objects.equals(this.operation, that.operation) &&
            this.resourceReferenceCount == that.resourceReferenceCount &&
            this.logIfAllowed == that.logIfAllowed &&
            this.logIfDenied == that.logIfDenied;

    }

    @Override
    public int hashCode() {
        return Objects.hash(resourcePattern, operation, resourceReferenceCount, logIfAllowed, logIfDenied);
    }

    @Override
    public String toString() {
        return "Action(" +
            ", resourcePattern='" + resourcePattern + '\'' +
            ", operation='" + operation + '\'' +
            ", resourceReferenceCount='" + resourceReferenceCount + '\'' +
            ", logIfAllowed='" + logIfAllowed + '\'' +
            ", logIfDenied='" + logIfDenied + '\'' +
            ')';
    }
}