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.CompletableFutureCompletionStage;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AccessControlEntryEndpoint;
import org.apache.kafka.common.annotationacl.InterfaceStabilityAclBinding;
import org.apache.kafka.common.resourceacl.ResourcePatternAclBindingFilter;
import org.apache.kafka.common.securityannotation.auth.SecurityProtocolInterfaceStability;
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(MapAuthorizerServerInfo)}
 *       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 the following interfaces:
 * <ul>
 *   <li>@{@{@link org.apache.kafka.common.Reconfigurable}
 * to enable dynamic reconfiguration
 *   without restarting the broker.</li>
 * <p>
 * <li>{@link org.apache.kafka.common.ClusterResourceListener} to obtain cluster id</li><b>Threading model:</b>
 * </ul><ul>
 */
@InterfaceStability.Evolving
public interface Authorizer extends<li>All Configurable,authorizer Closeableoperations {

including authorization and ACL /**
updates must be thread-safe.</li>
 * * Starts loading<li>ACL authorizationupdate metadatamethods andare returnsasynchronous. futuresImplementations thatwith canlow beupdate usedlatency tomay waitreturn untila
 *    * metadata for authorizingcompleted requestsfuture onusing 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`.{@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
     * @parammetadata actionsfor Actionsauthorizing beingrequests on 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 List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);

    /**listener.
     *
     * Creates@param aserverInfo newMetadata ACLfor binding.
the broker including broker id and listener *endpoints
     * @param@return requestContextCompletionStage Requestfor contexteach ifendpoint thethat ACLcompletes iswhen beingauthorizer created by a brokeris ready to handle
     *        a start clientauthorizing requestrequests toon createthat ACLslistener.
    This may*/
 be null if ACLsMap<Endpoint, are? createdextends directlyCompletionStage<Void>> in ZooKeeperstart(AuthorizerServerInfo serverInfo);

     /**
     * Authorizes the usingspecified AclCommandaction.
     * @param resourcePattern Resource pattern for which ACLs are being addedAdditional metadata for the action is specified
     * @paramin accessControlEntries`requestContext`.
 List of access control entries to add for resource* <p>
     *
 This is a synchronous *API @returndesigned Createfor resultuse forwith eachlocally accesscached controlACLs. entrySince inthis themethod sameis orderinvoked as inon the input list
     */
 request thread while List<AclCreateResult>processing createAcls(KafkaRequestContexteach requestContext,
    request, implementations of this method should avoid time-consuming
     * remote communication that may block request threads.
     *
     * @param requestContext Request context including request type, security protocol and ResourcePatternlistener resourcePattern,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 List<AccessControlEntry>provided accessControlEntries);actions

     */**
    List<AuthorizationResult> * Deletes the specified ACL binding.authorize(AuthorizableRequestContext requestContext, List<Action> actions);

     /**
     * @param requestContext Request context if theCreates new ACL isbindings.
 being deleted by a broker* to handle<p>
     * This is an asynchronous API that enables athe clientcaller request to deleteavoid ACLs.blocking Thisduring maythe beupdate. nullImplementations if ACLs are deleted directly in ZooKeeperof this
     * API can return completed futures using  using AclCommand.{@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
     * @param resourcePattern Resource pattern for which ACLs are being removed to process the update synchronously on the request thread.
     *
     * @param accessControlEntriesrequestContext SetRequest ofcontext accessif controlthe entriesACL tois removebeing forcreated resource
by a broker to  *handle
     * @return Delete result for each access control entrya inclient therequest sameto ordercreate asACLs. inThis themay inputbe list.
null if ACLs are created *directly in ZooKeeper
     *  Each result indicates if the entry was actuallyusing deletedAclCommand.
     */
 @param aclBindings ACL List<AclDeleteResult>bindings deleteAcls(KafkaRequestContext requestContext,to create
     *
     * @return Create result for each ACL binding in the same order as in the input list. Each result
     *    ResourcePattern resourcePattern,
    is returned as a CompletionStage that completes when the result is available.
     */
    List<? extends CompletionStage<AclCreateResult>> createAcls(AuthorizableRequestContext requestContext,         List<AccessControlEntry> accessControlEntriesList<AclBinding> aclBindings);

    /**
     * Deletes all ACL bindings that formatch the provided specifiedfilters.
 resource pattern. Only ACLs explicitly* configured<p>
     * with this resource pattern are deleted, no matching is performed.
     *This is an asynchronous API that enables the caller to avoid blocking during the update. Implementations of this
     * @paramAPI requestContextcan Requestreturn contextcompleted iffutures theusing ACLs are{@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
     * to process the update synchronously on the request thread.
     *
     * @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 whose ACLs are aclBindingFilters Filters to match ACL bindings that are to be deleted
     *
 @return true if at least* @return oneDelete ACLresult bindingfor waseach deleted,filter falsein otherwise
the same order as in */
the input list.
  boolean deleteAcls(KafkaRequestContext requestContext, ResourcePattern* resourcePattern);

    /**
    Each *result Returnsindicates allwhich ACL bindings.
 were actually deleted as well *as any
     * @return  Access control entries of all ACL bindings groupedthat bymatched resourcebut pattern.
could not be deleted. Each */
result is returned as Map<ResourcePattern,a
 Set<AccessControlEntry>> acls();

   * /**
     * Returns access controlCompletionStage entriesthat definedcompletes forwhen the specifiedresult resourceis patternavailable.
     */
    List<? *extends @paramCompletionStage<AclDeleteResult>> resourcePattern Resource pattern for which ACLs are returned. No matching is performed,
     *    deleteAcls(AuthorizableRequestContext requestContext, List<AclBindingFilter> aclBindingFilters);

    /**
     * Returns ACL bindings which match the provided filter.
    only ACLs defined * <p>
     * This is a synchronous API designed for theuse specifiedwith patternlocally arecached returned.ACLs. This method is invoked on the request
     *
     * @return ACL bindings for resource pattern thread while processing DescribeAcls requests and should avoid time-consuming remote communication that may
     * block request threads.
     */
    Set<AccessControlEntry> acls(ResourcePattern resourcePattern* @return Iterator for ACL bindings, which may be populated lazily.
     */
    Iterable<AclBinding> acls(AclBindingFilter filter);
}

package org.apache.kafka.commonserver.authorizer;

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 KafkaRequestContextAuthorizableRequestContext {

    /**
     * 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 16-bit API key of the request type (ApiKey) from the request header. For fetch requests,See
     * the metrics names FetchFollower and FetchConsumer will be used to distinguish between
     * replica fetch requests and client fetch requestshttps://kafka.apache.org/protocol#protocol_api_keys for request types.
     */
    Stringint 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 orgjava.apache.kafka.common.acl.AclOperationutil.Collection;
import org.apache.kafka.common.annotation.InterfaceStabilityClusterResource;
import org.apache.kafka.common.resource.PatternTypeEndpoint;
import org.apache.kafka.common.resourceannotation.ResourcePatternInterfaceStability;
import org.apache.kafka.common.resource.ResourceType;

@InterfaceStability.Evolving
public class Action {

    private final ResourcePattern resourcePattern;
    private final AclOperation operation;
/**
 * Runtime broker configuration metadata provided to authorizers during start up.
 */
@InterfaceStability.Evolving
public interface AuthorizerServerInfo {

    /**
    private final* AuthorizationModeReturns authorizationMode;
cluster metadata for the privatebroker finalrunning int count;

    public Action(AclOperation operation,this authorizer including cluster id.
     */
    ClusterResource clusterResource();

    /**
    ResourceType resourceType,
* Returns broker id. This may be a generated broker id if `broker.id` was not configured.
    String resourceName,*/
    int brokerId();

    /**
     * Returns endpoints for AuthorizationModeall authorizationMode,
listeners including the advertised host and port to which
     * the listener   int count) {is bound.
     */
   this.operation =Collection<Endpoint> operationendpoints();

    /**
    this.resourcePattern =* new ResourcePattern(resourceType, resourceName, PatternType.LITERAL);
        this.authorizationMode = authorizationMode;Returns the inter-broker endpoint. This is one of the endpoints returned by {@link #endpoints()}.
     */
   this.count =Endpoint countinterBrokerEndpoint();
    }

    /**
     * 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}

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

    /**
     * Indicates if audit logs tracking ALLOWED access should include this action if result is
     * 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.
     */
    public boolean logIfAllowed() {
        return logIfAllowed;
    }

    /**
     * Indicates if audit logs tracking DENIED access should include this action if result is
     * DENIED. The flag is true if access to a resource was explicitly requested and request
     * is denied as a result of this authorization request. The flag is false if request was
     * filtering out authorized resources (e.g. to subscribe to regex pattern). The flag is also
     * false if this is an optional authorization where an alternative resource authorization is
     * applied if this fails (e.g. Cluster:Create which is subsequently overridden by Topic:Create).
     */
    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 + '\'' +
            ')';
    }
}