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`<class.name`.</li>
 *   <li>Broker configures and starts authorizer instance. Authorizer implememtationimplementation starts loading its metadata.</li>
 *   <li>Broker starts SocketServer to accept connections and process requests<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 #startFuture(String, SecurityProtocol#start(AuthorizerServerInfo)}
 *       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
 *       each actionactions 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 Configurable, Closeable {

    /**
     * Returns a future that can be used to wait until all metadata for authorizing
     * requests on the specified listener are available. Each listener will be started
     * only after its metadata is available and authorizer is ready to start authorizing
     * requests on that listener.
     *
     * @param listenerName      Name of listener
     * @param securityProtocol  Security protocol used by the listener
     * @return CompletableFuture that completes when authorizer is ready to start authorizing
     *         requests on the specified listener
     */
    default CompletableFuture<Void> startFuture(String listenerName, SecurityProtocol securityProtocol) {
        return CompletableFuture.completedFuture(null);
    }   <li>All authorizer operations including authorization and ACL updates must be thread-safe.</li>
 *   <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 {

    /**
     * AuthorizesStarts theloading specifiedauthorization action.metadata Additionaland metadatareturns forfutures thethat actioncan isbe specified
used to    * in `requestContext`.wait until
     *
 metadata for authorizing requests on *each @paramlistener requestContextis Requestavailable. contextEach includinglistener requestwill type,be
 security protocol and listener name
* started only after its *metadata @paramis actionsavailable Actionsand beingauthorizer authorizedis includingready resourceto andstart operationauthorizing
     * @returnrequests Liston ofthat authorizationlistener.
 results for each action in*
 the same order as the* provided@param actions
serverInfo Metadata for the broker */
including broker id and List<AuthorizationResult> authorize(KafkaRequestContext requestContext, List<Action> actions);

    /**
     * Creates a new ACL binding.listener endpoints
     * @return CompletionStage for each endpoint that completes when authorizer is ready to
     *
     * @param requestContext Request contextstart ifauthorizing therequests ACLon is being created by a broker to handlethat listener.
     */
    Map<Endpoint, *? extends CompletionStage<Void>> start(AuthorizerServerInfo serverInfo);

    a/**
 client request to create ACL.* ThisAuthorizes maythe bespecified nullaction. ifAdditional ACLmetadata isfor createdthe directlyaction inis ZooKeeperspecified
     * in `requestContext`.
     * using AclCommand.<p>
     * This @paramis a resourcePatternsynchronous ResourceAPI patterndesigned for whichuse ACLswith arelocally beingcached added
ACLs. Since this method is *invoked @paramon accessControlEntriesthe
 List of access control entries* torequest addthread forwhile resource
processing each request, implementations of *
this method should avoid time-consuming
 * @return Create result for* eachremote accesscommunication controlthat entrymay inblock therequest samethreads.
 order as in the input list*
     */
 @param requestContext Request context List<AclCreateResult>including createAcls(KafkaRequestContextrequest requestContexttype,
 security protocol and listener name
     * @param actions Actions being authorized including resource and operation for                 ResourcePattern resourcePattern,each action
     * @return List of authorization results for each action in the same order as the provided actions
     */
    List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext,     List<AccessControlEntry> accessControlEntriesList<Action> actions);

    /**
     * DeletesCreates the specifiednew ACL bindingbindings.
     * <p>
     * @paramThis requestContextis Request context ifan asynchronous API that enables the caller ACLto isavoid beingblocking deletedduring bythe aupdate. brokerImplementations toof handlethis
     * API can return completed futures using {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
   a client request* to deleteprocess ACL.the Thisupdate maysynchronously beon nullthe ifrequest ACLthread.
 is deleted directly in ZooKeeper*
     * @param requestContext Request context if the ACL usingis AclCommand.
being created by a broker *to @paramhandle
 resourcePattern Resource pattern for which* ACLs are being remove
    a *client @paramrequest accessControlEntriesto Setcreate ofACLs. accessThis controlmay entriesbe tonull removeif forACLs resource
are created directly in  *ZooKeeper
     * @return    Delete result for eachusing accessAclCommand.
 control entry in the same* order@param asaclBindings inACL thebindings inputto listcreate
     */
    List<AclDeleteResult> deleteAcls(KafkaRequestContext requestContext,
                 * @return Create result for each ACL binding in the same order as in the input list. Each result
     *         is returned as a CompletionStage that ResourcePatterncompletes resourcePattern,
when the result is available.
     */
    List<? extends CompletionStage<AclCreateResult>> createAcls(AuthorizableRequestContext requestContext, List<AclBinding> aclBindings);

    /**
     * Deletes all ACL bindings that match the provided List<AccessControlEntry> accessControlEntries);

filters.
     /** <p>
     * Deletes all ACL bindings for the specified resource pattern. Only ACLs explicitly configuredThis is an asynchronous API that enables the caller to avoid blocking during the update. Implementations of this
     * API can return completed futures using {@link java.util.concurrent.CompletableFuture#completedFuture(Object)}
     * withto thisprocess resourcethe patternupdate aresynchronously deleted,on nothe matchingrequest is performedthread.
     *
     * @param requestContext Request context if the ACLsACL areis 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 aclBindingFilters Filters resourcePatternto Resourcematch patternACL whosebindings ACLsthat 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, Set<AccessControlEntry>> acls();

a
     /**
      * 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<p>
 acls defined for the specified* patternThis areis returned.
a synchronous API designed for *
use with locally cached ACLs. *This @returnmethod ACLis bindingsinvoked foron resourcethe pattern.request
     */
 thread while processing DescribeAcls Set<AccessControlEntry>requests acls(ResourcePattern resourcePattern);
}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.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 requests.
     */
    String https://kafka.apache.org/protocol#protocol_api_keys for request types.
     */
    int 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.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
InterfaceStability;

/**
 * Runtime broker configuration metadata provided to authorizers during start up.
 */
@InterfaceStability.Evolving
public classinterface ActionAuthorizerServerInfo {

    private/**
 final ResourcePattern resourcePattern;
  * Returns privatecluster finalmetadata AclOperationfor operation;
the broker running this privateauthorizer finalincluding AuthorizationModecluster authorizationMode;id.
    private final int count;

 */
    publicClusterResource Action(AclOperation operation,clusterResource();

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

   String resourceName,/**
     * Returns endpoints for all listeners including the advertised host and port to AuthorizationMode authorizationMode,which
     * the listener is bound.
     */
    intCollection<Endpoint> count) {endpoints();

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

    /**
     * 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 + '\'' +
            ')';
    }
}