This is an automated email from the ASF dual-hosted git repository.
FrankYang0529 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2904f5c4928 KAFKA-18998: Move AuthHelper to server module (#22307)
2904f5c4928 is described below
commit 2904f5c4928cba71b2eba9135bee9e31458a920c
Author: PoAn Yang <[email protected]>
AuthorDate: Wed Jun 24 19:13:50 2026 +0900
KAFKA-18998: Move AuthHelper to server module (#22307)
Move AuthHelper from core to server module and rewrite in Java.
Reviewers: Mickael Maison <[email protected]>, Sushant Mahajan
<[email protected]>, Chia-Ping Tsai <[email protected]>
---------
Signed-off-by: PoAn Yang <[email protected]>
---
.../DescribeTopicPartitionsRequestHandler.java | 2 +-
core/src/main/scala/kafka/server/AclApis.scala | 1 +
core/src/main/scala/kafka/server/AuthHelper.scala | 176 -----------
.../src/main/scala/kafka/server/ConfigHelper.scala | 1 +
.../main/scala/kafka/server/ControllerApis.scala | 44 +--
core/src/main/scala/kafka/server/KafkaApis.scala | 116 ++++----
.../DescribeTopicPartitionsRequestHandlerTest.java | 6 +-
.../scala/unit/kafka/server/AuthHelperTest.scala | 261 -----------------
.../unit/kafka/server/ControllerApisTest.scala | 38 +--
.../java/org/apache/kafka/server/AuthHelper.java | 266 +++++++++++++++++
.../org/apache/kafka/server/AuthHelperTest.java | 325 +++++++++++++++++++++
11 files changed, 705 insertions(+), 531 deletions(-)
diff --git
a/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java
b/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java
index 101f09c2a71..70f7f408378 100644
---
a/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java
+++
b/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java
@@ -17,7 +17,6 @@
package kafka.server.handlers;
-import kafka.server.AuthHelper;
import kafka.server.KafkaConfig;
import org.apache.kafka.common.Uuid;
@@ -31,6 +30,7 @@ import
org.apache.kafka.common.requests.DescribeTopicPartitionsRequest;
import org.apache.kafka.common.resource.Resource;
import org.apache.kafka.metadata.MetadataCache;
import org.apache.kafka.network.Request;
+import org.apache.kafka.server.AuthHelper;
import java.util.HashSet;
import java.util.List;
diff --git a/core/src/main/scala/kafka/server/AclApis.scala
b/core/src/main/scala/kafka/server/AclApis.scala
index 10c7f6bb4d7..9c8113c8ffd 100644
--- a/core/src/main/scala/kafka/server/AclApis.scala
+++ b/core/src/main/scala/kafka/server/AclApis.scala
@@ -31,6 +31,7 @@ import org.apache.kafka.common.resource.Resource.CLUSTER_NAME
import org.apache.kafka.common.resource.ResourceType
import org.apache.kafka.network.Request
import org.apache.kafka.security.authorizer.AuthorizerUtils
+import org.apache.kafka.server.AuthHelper
import org.apache.kafka.server.ProcessRole
import org.apache.kafka.server.authorizer._
import org.apache.kafka.server.purgatory.DelayedFuturePurgatory
diff --git a/core/src/main/scala/kafka/server/AuthHelper.scala
b/core/src/main/scala/kafka/server/AuthHelper.scala
deleted file mode 100644
index ea76650da68..00000000000
--- a/core/src/main/scala/kafka/server/AuthHelper.scala
+++ /dev/null
@@ -1,176 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.server
-
-import java.lang.{Byte => JByte}
-import java.util.Collections
-import org.apache.kafka.clients.admin.EndpointType
-import org.apache.kafka.common.acl.AclOperation
-import org.apache.kafka.common.acl.AclOperation.DESCRIBE
-import org.apache.kafka.common.errors.ClusterAuthorizationException
-import org.apache.kafka.common.internals.Plugin
-import org.apache.kafka.common.message.DescribeClusterResponseData
-import
org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBrokerCollection
-import org.apache.kafka.network.Request
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.{DescribeClusterRequest,
RequestContext}
-import org.apache.kafka.common.resource.Resource.CLUSTER_NAME
-import org.apache.kafka.common.resource.ResourceType.CLUSTER
-import org.apache.kafka.common.resource.{PatternType, Resource,
ResourcePattern, ResourceType}
-import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.security.authorizer.AclEntry
-import org.apache.kafka.server.authorizer.{Action, AuthorizationResult,
Authorizer}
-
-import scala.collection.Seq
-import scala.jdk.CollectionConverters._
-
-class AuthHelper(authorizer: Option[Plugin[Authorizer]]) {
- def authorize(requestContext: RequestContext,
- operation: AclOperation,
- resourceType: ResourceType,
- resourceName: String,
- logIfAllowed: Boolean = true,
- logIfDenied: Boolean = true,
- refCount: Int = 1): Boolean = {
- authorizer.forall { authZ =>
- val resource = new ResourcePattern(resourceType, resourceName,
PatternType.LITERAL)
- val actions = Collections.singletonList(new Action(operation, resource,
refCount, logIfAllowed, logIfDenied))
- authZ.get.authorize(requestContext, actions).get(0) ==
AuthorizationResult.ALLOWED
- }
- }
-
- def authorizeClusterOperation(request: Request, operation: AclOperation):
Unit = {
- if (!authorize(request.context, operation, CLUSTER, CLUSTER_NAME))
- throw new ClusterAuthorizationException(s"Request $request needs
$operation permission.")
- }
-
- def authorizedOperations(request: Request, resource: Resource): Int = {
- val supportedOps =
AclEntry.supportedOperations(resource.resourceType).asScala.toList
- val authorizedOps = authorizer match {
- case Some(authZ) =>
- val resourcePattern = new ResourcePattern(resource.resourceType,
resource.name, PatternType.LITERAL)
- val actions = supportedOps.map { op => new Action(op, resourcePattern,
1, false, false) }
- authZ.get.authorize(request.context, actions.asJava).asScala
- .zip(supportedOps)
- .filter(_._1 == AuthorizationResult.ALLOWED)
- .map(_._2).toSet
- case None =>
- supportedOps.toSet
- }
- Utils.to32BitField(authorizedOps.map(operation =>
operation.code.asInstanceOf[JByte]).asJava)
- }
-
- def authorizeByResourceType(requestContext: RequestContext, operation:
AclOperation,
- resourceType: ResourceType): Boolean = {
- authorizer.forall { authZ =>
- authZ.get.authorizeByResourceType(requestContext, operation,
resourceType) == AuthorizationResult.ALLOWED
- }
- }
-
- def partitionSeqByAuthorized[T](requestContext: RequestContext,
- operation: AclOperation,
- resourceType: ResourceType,
- resources: Seq[T],
- logIfAllowed: Boolean = true,
- logIfDenied: Boolean = true)(resourceName: T
=> String): (Seq[T], Seq[T]) = {
- authorizer match {
- case Some(_) =>
- val authorizedResourceNames = filterByAuthorized(requestContext,
operation, resourceType,
- resources, logIfAllowed, logIfDenied)(resourceName)
- resources.partition(resource =>
authorizedResourceNames.contains(resourceName(resource)))
- case None => (resources, Seq.empty)
- }
- }
-
- def filterByAuthorized[T](requestContext: RequestContext,
- operation: AclOperation,
- resourceType: ResourceType,
- resources: Iterable[T],
- logIfAllowed: Boolean = true,
- logIfDenied: Boolean = true)(resourceName: T =>
String): Set[String] = {
- authorizer match {
- case Some(authZ) =>
- val resourceNameToCount = resources.groupMapReduce(resourceName)(_ =>
1)(_ + _)
- val actions = resourceNameToCount.iterator.map { case (resourceName,
count) =>
- val resource = new ResourcePattern(resourceType, resourceName,
PatternType.LITERAL)
- new Action(operation, resource, count, logIfAllowed, logIfDenied)
- }.toBuffer
- authZ.get.authorize(requestContext, actions.asJava).asScala
- .zip(resourceNameToCount.keySet)
- .collect { case (authzResult, resourceName) if authzResult ==
AuthorizationResult.ALLOWED =>
- resourceName
- }.toSet
- case None => resources.iterator.map(resourceName).toSet
- }
- }
-
- def computeDescribeClusterResponse(
- request: Request,
- expectedEndpointType: EndpointType,
- clusterId: String,
- getNodes: () => DescribeClusterBrokerCollection,
- getControllerId: () => Int
- ): DescribeClusterResponseData = {
- val describeClusterRequest = request.body(classOf[DescribeClusterRequest])
- val requestEndpointType =
EndpointType.fromId(describeClusterRequest.data().endpointType())
- if (requestEndpointType.equals(EndpointType.UNKNOWN)) {
- return new DescribeClusterResponseData().
- setErrorCode(if (request.header.data().requestApiVersion() == 0) {
- Errors.INVALID_REQUEST.code()
- } else {
- Errors.UNSUPPORTED_ENDPOINT_TYPE.code()
- }).
- setErrorMessage("Unsupported endpoint type " +
describeClusterRequest.data().endpointType().toInt)
- } else if (!expectedEndpointType.equals(requestEndpointType)) {
- return new DescribeClusterResponseData().
- setErrorCode(if (request.header.data().requestApiVersion() == 0) {
- Errors.INVALID_REQUEST.code()
- } else {
- Errors.MISMATCHED_ENDPOINT_TYPE.code()
- }).
- setErrorMessage("The request was sent to an endpoint of type " +
expectedEndpointType +
- ", but we wanted an endpoint of type " + requestEndpointType)
- }
- var clusterAuthorizedOperations = Int.MinValue // Default value in the
schema
- // get cluster authorized operations
- if (describeClusterRequest.data.includeClusterAuthorizedOperations) {
- if (authorize(request.context, DESCRIBE, CLUSTER, CLUSTER_NAME))
- clusterAuthorizedOperations = authorizedOperations(request,
Resource.CLUSTER)
- else
- clusterAuthorizedOperations = 0
- }
- // Get the node list and the controller ID.
- val nodes = getNodes()
- val controllerId = getControllerId()
- // If the provided controller ID is not in the node list, return -1 instead
- // to avoid confusing the client. This could happen in a case where we know
- // the controller ID, but we don't yet have KIP-919 information about that
- // controller.
- val effectiveControllerId = if (nodes.find(controllerId) == null) {
- -1
- } else {
- controllerId
- }
- new DescribeClusterResponseData().
- setClusterId(clusterId).
- setControllerId(effectiveControllerId).
- setClusterAuthorizedOperations(clusterAuthorizedOperations).
- setBrokers(nodes).
- setEndpointType(expectedEndpointType.id())
- }
-}
diff --git a/core/src/main/scala/kafka/server/ConfigHelper.scala
b/core/src/main/scala/kafka/server/ConfigHelper.scala
index 304d8c15541..7e4de8eb6fc 100644
--- a/core/src/main/scala/kafka/server/ConfigHelper.scala
+++ b/core/src/main/scala/kafka/server/ConfigHelper.scala
@@ -33,6 +33,7 @@ import
org.apache.kafka.common.resource.ResourceType.{CLUSTER, GROUP, TOPIC}
import org.apache.kafka.coordinator.group.GroupConfig
import org.apache.kafka.metadata.{ConfigRepository, MetadataCache}
import org.apache.kafka.network.Request
+import org.apache.kafka.server.AuthHelper
import org.apache.kafka.server.ConfigHelperUtils.createResponseConfig
import org.apache.kafka.server.config.{DynamicBrokerConfig,
ServerTopicConfigSynonyms}
import org.apache.kafka.server.logger.LoggingController
diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala
b/core/src/main/scala/kafka/server/ControllerApis.scala
index 77ca2559866..2957d5770fb 100644
--- a/core/src/main/scala/kafka/server/ControllerApis.scala
+++ b/core/src/main/scala/kafka/server/ControllerApis.scala
@@ -55,12 +55,12 @@ import org.apache.kafka.metadata.{BrokerHeartbeatReply,
BrokerRegistrationReply,
import org.apache.kafka.network.Request
import org.apache.kafka.raft.RaftManager
import org.apache.kafka.security.DelegationTokenManager
-import org.apache.kafka.server.{ApiVersionManager, EnvelopeUtils, ProcessRole}
+import org.apache.kafka.server.{ApiVersionManager, AuthHelper, EnvelopeUtils,
ProcessRole}
import org.apache.kafka.server.authorizer.Authorizer
import org.apache.kafka.server.common.{ApiMessageAndVersion, RequestLocal}
import org.apache.kafka.server.quota.ControllerMutationQuota
-import scala.jdk.CollectionConverters._
+import scala.jdk.javaapi.OptionConverters
/**
@@ -81,7 +81,7 @@ class ControllerApis(
) extends ApiRequestHandler with Logging {
this.logIdent = s"[ControllerApis nodeId=${config.nodeId}] "
- val authHelper = new AuthHelper(authorizerPlugin)
+ val authHelper = new AuthHelper(OptionConverters.toJava(authorizerPlugin))
val configHelper = new ConfigHelper(metadataCache, config, metadataCache)
val requestHelper = new RequestHandlerHelper(requestChannel, quotas, time)
val runtimeLoggerManager = new RuntimeLoggerManager(config.nodeId,
logger.underlying)
@@ -204,9 +204,9 @@ class ControllerApis(
val future = deleteTopics(context,
deleteTopicsRequest.data,
request.context.apiVersion,
- authHelper.authorize(request.context, DELETE, CLUSTER, CLUSTER_NAME,
logIfDenied = false),
- names => authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
names)(n => n),
- names => authHelper.filterByAuthorized(request.context, DELETE, TOPIC,
names)(n => n))
+ authHelper.authorize(request.context, DELETE, CLUSTER, CLUSTER_NAME,
true, false, 1),
+ names => authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
names, (n: String) => n),
+ names => authHelper.filterByAuthorized(request.context, DELETE, TOPIC,
names, (n: String) => n))
future.handle[Unit] { (results, exception) =>
val response = if (exception != null) {
deleteTopicsRequest.getErrorResponse(exception)
@@ -224,8 +224,8 @@ class ControllerApis(
request: DeleteTopicsRequestData,
apiVersion: Int,
hasClusterAuth: Boolean,
- getDescribableTopics: Iterable[String] => Set[String],
- getDeletableTopics: Iterable[String] => Set[String]
+ getDescribableTopics: lang.Iterable[String] => util.Set[String],
+ getDeletableTopics: lang.Iterable[String] => util.Set[String]
): CompletableFuture[util.List[DeletableTopicResult]] = {
// Check if topic deletion is enabled at all.
if (!config.deleteTopicEnable) {
@@ -298,11 +298,10 @@ class ControllerApis(
}
// Get the list of deletable topics (those we can delete) and the list
of describable
// topics.
- val topicsToAuthenticate = toAuthenticate.asScala
val (describable, deletable) = if (hasClusterAuth) {
- (topicsToAuthenticate.toSet, topicsToAuthenticate.toSet)
+ (toAuthenticate, toAuthenticate)
} else {
- (getDescribableTopics(topicsToAuthenticate),
getDeletableTopics(topicsToAuthenticate))
+ (getDescribableTopics(toAuthenticate),
getDeletableTopics(toAuthenticate))
}
// For each topic that was provided by ID, check if authentication
failed.
// If so, remove it from the idToName map and create an error response
for it.
@@ -367,10 +366,10 @@ class ControllerApis(
controllerMutationQuotaRecorderFor(controllerMutationQuota))
val future = createTopics(context,
createTopicsRequest.data,
- authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME,
logIfDenied = false),
- names => authHelper.filterByAuthorized(request.context, CREATE, TOPIC,
names)(identity),
+ authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME,
true, false, 1),
+ names => authHelper.filterByAuthorized(request.context, CREATE, TOPIC,
names, (n: String) => n),
names => authHelper.filterByAuthorized(request.context,
DESCRIBE_CONFIGS, TOPIC,
- names, logIfDenied = false)(identity),
+ names, true, false, (n: String) => n),
request.isForwarded)
future.handle[Unit] { (result, exception) =>
val response = if (exception != null) {
@@ -392,8 +391,8 @@ class ControllerApis(
context: ControllerRequestContext,
request: CreateTopicsRequestData,
hasClusterAuth: Boolean,
- getCreatableTopics: Iterable[String] => Set[String],
- getDescribableTopics: Iterable[String] => Set[String],
+ getCreatableTopics: lang.Iterable[String] => util.Set[String],
+ getDescribableTopics: lang.Iterable[String] => util.Set[String],
forwarded: Boolean
): CompletableFuture[CreateTopicsResponseData] = {
val topicNames = new util.HashSet[String]()
@@ -407,14 +406,15 @@ class ControllerApis(
}
}
- val allowedTopicNames =
topicNames.asScala.diff(Set(Topic.CLUSTER_METADATA_TOPIC_NAME))
+ val allowedTopicNames = new util.HashSet[String](topicNames)
+ allowedTopicNames.remove(Topic.CLUSTER_METADATA_TOPIC_NAME)
val authorizedTopicNames = if (hasClusterAuth) {
allowedTopicNames
} else {
getCreatableTopics.apply(allowedTopicNames)
}
- val describableTopicNames =
getDescribableTopics.apply(allowedTopicNames).asJava
+ val describableTopicNames = getDescribableTopics.apply(allowedTopicNames)
val effectiveRequest = request.duplicate()
val iterator = effectiveRequest.topics().iterator()
while (iterator.hasNext) {
@@ -798,8 +798,8 @@ class ControllerApis(
}
private def handleCreatePartitions(request: Request):
CompletableFuture[Unit] = {
- def filterAlterAuthorizedTopics(topics: Iterable[String]): Set[String] = {
- authHelper.filterByAuthorized(request.context, ALTER, TOPIC, topics)(n
=> n)
+ def filterAlterAuthorizedTopics(topics: lang.Iterable[String]):
util.Set[String] = {
+ authHelper.filterByAuthorized(request.context, ALTER, TOPIC, topics, (n:
String) => n)
}
val createPartitionsRequest =
request.body(classOf[CreatePartitionsRequest])
val controllerMutationQuota =
quotas.controllerMutation.newQuotaFor(request.session, request.header, 3)
@@ -830,7 +830,7 @@ class ControllerApis(
def createPartitions(
context: ControllerRequestContext,
request: CreatePartitionsRequestData,
- getAlterAuthorizedTopics: Iterable[String] => Set[String]
+ getAlterAuthorizedTopics: lang.Iterable[String] => util.Set[String]
): CompletableFuture[util.List[CreatePartitionsTopicResult]] = {
val responses = new util.ArrayList[CreatePartitionsTopicResult]()
val duplicateTopicNames = new util.HashSet[String]()
@@ -848,7 +848,7 @@ class ControllerApis(
setErrorMessage("Duplicate topic name."))
topicNames.remove(topicName)
}
- val authorizedTopicNames = getAlterAuthorizedTopics(topicNames.asScala)
+ val authorizedTopicNames = getAlterAuthorizedTopics(topicNames)
val topics = new util.ArrayList[CreatePartitionsTopic]
topicNames.forEach { topicName =>
if (authorizedTopicNames.contains(topicName)) {
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala
b/core/src/main/scala/kafka/server/KafkaApis.scala
index 3bf0861d847..36948c34fb8 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -64,7 +64,7 @@ import
org.apache.kafka.coordinator.transaction.InitProducerIdResult
import org.apache.kafka.metadata.{ConfigRepository, MetadataCache}
import org.apache.kafka.network.Request
import org.apache.kafka.security.DelegationTokenManager
-import org.apache.kafka.server.{ApiVersionManager, AutoTopicCreationManager,
ClientMetricsManager, FetchManager, ForwardingManager, ProcessRole}
+import org.apache.kafka.server.{ApiVersionManager, AuthHelper,
AutoTopicCreationManager, ClientMetricsManager, FetchManager,
ForwardingManager, ProcessRole}
import org.apache.kafka.server.authorizer._
import org.apache.kafka.server.common.{GroupVersion, RequestLocal,
ShareVersion, StreamsVersion, TransactionVersion}
import org.apache.kafka.server.quota.{ReplicaQuota, ReplicationQuotaManager}
@@ -119,7 +119,7 @@ class KafkaApis(val requestChannel: RequestChannel,
type ProduceResponseStats = Map[TopicIdPartition, RecordValidationStats]
this.logIdent = "[KafkaApi-%d] ".format(brokerId)
val configHelper = new ConfigHelper(metadataCache, config, configRepository)
- val authHelper = new AuthHelper(authorizerPlugin)
+ val authHelper = new AuthHelper(OptionConverters.toJava(authorizerPlugin))
val requestHelper = new RequestHandlerHelper(requestChannel, quotas, time)
val aclApis = new AclApis(authHelper, authorizerPlugin, requestHelper,
ProcessRole.BrokerRole, config)
val configManager = new ConfigAdminManager(brokerId, config,
configRepository)
@@ -298,8 +298,9 @@ class KafkaApis(val requestChannel: RequestChannel,
request.context,
READ,
TOPIC,
- offsetCommitRequest.data.topics.asScala
- )(_.name)
+ offsetCommitRequest.data.topics,
+ (t: OffsetCommitRequestData.OffsetCommitRequestTopic) => t.name
+ )
val responseBuilder = OffsetCommitResponse.newBuilder(useTopicIds)
val authorizedTopicsRequest = new
mutable.ArrayBuffer[OffsetCommitRequestData.OffsetCommitRequestTopic]()
@@ -430,7 +431,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
// cache the result to avoid redundant authorization calls
- val authorizedTopics = authHelper.filterByAuthorized(request.context,
WRITE, TOPIC, topicIdToPartitionData)(_._1.topic)
+ val authorizedTopics = authHelper.filterByAuthorized(request.context,
WRITE, TOPIC, topicIdToPartitionData.asJava, (t: (TopicIdPartition,
ProduceRequestData.PartitionProduceData)) => t._1.topic)
topicIdToPartitionData.foreach { case (topicIdPartition, partition) =>
// This caller assumes the type is MemoryRecords and that is true on
current serialization
@@ -610,7 +611,7 @@ class KafkaApis(val requestChannel: RequestChannel,
else
partitionDatas += topicIdPartition -> partitionData
}
- val authorizedTopics = authHelper.filterByAuthorized(request.context,
READ, TOPIC, partitionDatas)(_._1.topicPartition.topic)
+ val authorizedTopics = authHelper.filterByAuthorized(request.context,
READ, TOPIC, partitionDatas.asJava, (t: (TopicIdPartition,
FetchRequest.PartitionData)) => t._1.topicPartition.topic)
partitionDatas.foreach { case (topicIdPartition, data) =>
if (!authorizedTopics.contains(topicIdPartition.topic))
erroneous += topicIdPartition ->
FetchResponse.partitionResponse(topicIdPartition,
Errors.TOPIC_AUTHORIZATION_FAILED)
@@ -794,8 +795,10 @@ class KafkaApis(val requestChannel: RequestChannel,
.setOffset(ListOffsetsResponse.UNKNOWN_OFFSET)
}
- val (authorizedRequestInfo, unauthorizedRequestInfo) =
authHelper.partitionSeqByAuthorized(request.context,
- DESCRIBE, TOPIC, offsetRequest.topics.asScala.toSeq)(_.name)
+ val partitionResult = authHelper.partitionByAuthorized(request.context,
+ DESCRIBE, TOPIC, offsetRequest.topics, (t:
ListOffsetsRequestData.ListOffsetsTopic) => t.name)
+ val authorizedRequestInfo = partitionResult.authorized.asScala
+ val unauthorizedRequestInfo = partitionResult.unauthorized.asScala
val unauthorizedResponseStatus = unauthorizedRequestInfo.map(topic =>
new ListOffsetsTopicResponse()
@@ -917,16 +920,16 @@ class KafkaApis(val requestChannel: RequestChannel,
metadataRequest.topics.asScala.toSet
val authorizedForDescribeTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
- topics, logIfDenied = !metadataRequest.isAllTopics)(identity)
+ topics.asJava, true, !metadataRequest.isAllTopics, (t: String) => t)
var (authorizedTopics, unauthorizedForDescribeTopics) =
topics.partition(authorizedForDescribeTopics.contains)
var unauthorizedForCreateTopics = Set[String]()
if (authorizedTopics.nonEmpty) {
val nonExistingTopics =
authorizedTopics.filterNot(metadataCache.contains)
if (metadataRequest.allowAutoTopicCreation &&
config.autoCreateTopicsEnable && nonExistingTopics.nonEmpty) {
- if (!authHelper.authorize(request.context, CREATE, CLUSTER,
CLUSTER_NAME, logIfDenied = false)) {
+ if (!authHelper.authorize(request.context, CREATE, CLUSTER,
CLUSTER_NAME, true, false, 1)) {
val authorizedForCreateTopics =
authHelper.filterByAuthorized(request.context, CREATE, TOPIC,
- nonExistingTopics)(identity)
+ nonExistingTopics.asJava, (t: String) => t).asScala
unauthorizedForCreateTopics =
nonExistingTopics.diff(authorizedForCreateTopics)
authorizedTopics = authorizedTopics.diff(unauthorizedForCreateTopics)
}
@@ -1083,8 +1086,9 @@ class KafkaApis(val requestChannel: RequestChannel,
requestContext,
DESCRIBE,
TOPIC,
- groupFetchResponse.topics.asScala
- )(_.name)
+ groupFetchResponse.topics,
+ (t: OffsetFetchResponseData.OffsetFetchResponseTopics) => t.name
+ )
val topics = new
mutable.ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseTopics]
groupFetchResponse.topics.forEach { topic =>
@@ -1130,8 +1134,9 @@ class KafkaApis(val requestChannel: RequestChannel,
requestContext,
DESCRIBE,
TOPIC,
- groupFetchRequest.topics.asScala
- )(_.name)
+ groupFetchRequest.topics,
+ (t: OffsetFetchRequestData.OffsetFetchRequestTopics) => t.name
+ )
val authorizedTopics = new
mutable.ArrayBuffer[OffsetFetchRequestData.OffsetFetchRequestTopics]
val errorTopics = new
mutable.ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseTopics]
@@ -1356,7 +1361,7 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleListGroupsRequest(request: Request): CompletableFuture[Unit] = {
val listGroupsRequest = request.body(classOf[ListGroupsRequest])
- val hasClusterDescribe = authHelper.authorize(request.context, DESCRIBE,
CLUSTER, CLUSTER_NAME, logIfDenied = false)
+ val hasClusterDescribe = authHelper.authorize(request.context, DESCRIBE,
CLUSTER, CLUSTER_NAME, true, false, 1)
groupCoordinator.listGroups(
request.context,
@@ -1371,7 +1376,7 @@ class KafkaApis(val requestChannel: RequestChannel,
} else {
// Otherwise, only groups with described group are returned.
val authorizedGroups = response.groups.asScala.filter { group =>
- authHelper.authorize(request.context, DESCRIBE, GROUP,
group.groupId, logIfDenied = false)
+ authHelper.authorize(request.context, DESCRIBE, GROUP,
group.groupId, true, false, 1)
}
new ListGroupsResponse(response.setGroups(authorizedGroups.asJava))
}
@@ -1439,8 +1444,9 @@ class KafkaApis(val requestChannel: RequestChannel,
val deleteGroupsRequest = request.body(classOf[DeleteGroupsRequest])
val groups = deleteGroupsRequest.data.groupsNames.asScala.distinct
- val (authorizedGroups, unauthorizedGroups) =
- authHelper.partitionSeqByAuthorized(request.context, DELETE, GROUP,
groups)(identity)
+ val deleteGroupsPartition =
authHelper.partitionByAuthorized(request.context, DELETE, GROUP, groups.asJava,
(g: String) => g)
+ val authorizedGroups = deleteGroupsPartition.authorized.asScala
+ val unauthorizedGroups = deleteGroupsPartition.unauthorized.asScala
groupCoordinator.deleteGroups(
request.context,
@@ -1573,7 +1579,7 @@ class KafkaApis(val requestChannel: RequestChannel,
val authorizedForDeleteTopicOffsets = mutable.Map[TopicPartition, Long]()
val topics = deleteRecordsRequest.data.topics.asScala
- val authorizedTopics = authHelper.filterByAuthorized(request.context,
DELETE, TOPIC, topics)(_.name)
+ val authorizedTopics = authHelper.filterByAuthorized(request.context,
DELETE, TOPIC, deleteRecordsRequest.data.topics, (t:
DeleteRecordsRequestData.DeleteRecordsTopic) => t.name)
val deleteTopicPartitions = topics.flatMap { deleteTopic =>
deleteTopic.partitions.asScala.map { deletePartition =>
new TopicPartition(deleteTopic.name, deletePartition.partitionIndex)
-> deletePartition.offset
@@ -1643,7 +1649,7 @@ class KafkaApis(val requestChannel: RequestChannel,
requestHelper.sendErrorResponseMaybeThrottle(request,
Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED.exception)
return
}
- } else if (!authHelper.authorize(request.context, IDEMPOTENT_WRITE,
CLUSTER, CLUSTER_NAME, true, false)
+ } else if (!authHelper.authorize(request.context, IDEMPOTENT_WRITE,
CLUSTER, CLUSTER_NAME, true, false, 1)
&& !authHelper.authorizeByResourceType(request.context,
AclOperation.WRITE, ResourceType.TOPIC)) {
requestHelper.sendErrorResponseMaybeThrottle(request,
Errors.CLUSTER_AUTHORIZATION_FAILED.exception)
return
@@ -1741,7 +1747,7 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleWriteTxnMarkersRequest(request: Request, requestLocal:
RequestLocal): Unit = {
// We are checking for AlterCluster permissions first. If it is not
present, we are authorizing cluster operation
// The latter will throw an exception if it is denied.
- if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME,
logIfDenied = false)) {
+ if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME,
true, false, 1)) {
authHelper.authorizeClusterOperation(request, CLUSTER_ACTION)
}
val writeTxnMarkersRequest = request.body(classOf[WriteTxnMarkersRequest])
@@ -1942,7 +1948,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// Only request versions less than 4 need write authorization since
they come from clients.
val authorizedTopics =
if (version < 4)
- authHelper.filterByAuthorized(request.context, WRITE, TOPIC,
partitionsToAdd.filterNot(tp => Topic.isInternal(tp.topic)))(_.topic)
+ authHelper.filterByAuthorized(request.context, WRITE, TOPIC,
partitionsToAdd.filterNot(tp => Topic.isInternal(tp.topic)).asJava, (tp:
TopicPartition) => tp.topic).asScala
else
partitionsToAdd.map(_.topic).toSet
for (topicPartition <- partitionsToAdd) {
@@ -2095,8 +2101,9 @@ class KafkaApis(val requestChannel: RequestChannel,
request.context,
READ,
TOPIC,
- txnOffsetCommitRequest.data.topics.asScala
- )(_.name)
+ txnOffsetCommitRequest.data.topics,
+ (t: TxnOffsetCommitRequestData.TxnOffsetCommitRequestTopic) => t.name
+ )
val responseBuilder = TxnOffsetCommitResponse.newBuilder(useTopicIds)
val authorizedTopicCommittedOffsets = new
mutable.ArrayBuffer[TxnOffsetCommitRequestData.TxnOffsetCommitRequestTopic]()
@@ -2186,9 +2193,12 @@ class KafkaApis(val requestChannel: RequestChannel,
// cluster permission. With KIP-320, the consumer now also uses this API
to check for log truncation
// following a leader change, so we also allow topic describe permission.
val (authorizedTopics, unauthorizedTopics) =
- if (authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER,
CLUSTER_NAME, logIfDenied = false))
+ if (authHelper.authorize(request.context, CLUSTER_ACTION, CLUSTER,
CLUSTER_NAME, true, false, 1))
(topics, Seq.empty[OffsetForLeaderTopic])
- else authHelper.partitionSeqByAuthorized(request.context, DESCRIBE,
TOPIC, topics)(_.topic)
+ else {
+ val partitionResult =
authHelper.partitionByAuthorized(request.context, DESCRIBE, TOPIC,
topics.asJava, (t: OffsetForLeaderTopic) => t.topic)
+ (partitionResult.authorized.asScala.toSeq,
partitionResult.unauthorized.asScala.toSeq)
+ }
val endOffsetsForAuthorizedPartitions =
replicaManager.lastOffsetForLeaderEpoch(authorizedTopics)
val endOffsetsForUnauthorizedPartitions = unauthorizedTopics.map {
offsetForLeaderTopic =>
@@ -2432,8 +2442,9 @@ class KafkaApis(val requestChannel: RequestChannel,
request.context,
READ,
TOPIC,
- offsetDeleteRequest.data.topics.asScala
- )(_.name)
+ offsetDeleteRequest.data.topics,
+ (t: OffsetDeleteRequestData.OffsetDeleteRequestTopic) => t.name
+ )
val responseBuilder = new OffsetDeleteResponse.Builder
val authorizedTopicPartitions = new
OffsetDeleteRequestData.OffsetDeleteRequestTopicCollection()
@@ -2675,7 +2686,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// Clients are not allowed to see topics that are not authorized for
Describe.
val subscribedTopicSet =
consumerGroupHeartbeatRequest.data.subscribedTopicNames.asScala.toSet
val authorizedTopics = authHelper.filterByAuthorized(request.context,
DESCRIBE, TOPIC,
- subscribedTopicSet)(identity)
+ subscribedTopicSet.asJava, (t: String) => t)
if (authorizedTopics.size < subscribedTopicSet.size) {
val responseData = new ConsumerGroupHeartbeatResponseData()
.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
@@ -2757,7 +2768,7 @@ class KafkaApis(val requestChannel: RequestChannel,
.collect(Collectors.toSet[String])
.asScala
val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
- topicsToCheck)(identity)
+ topicsToCheck.asJava, (t: String) => t)
val updatedGroups = response.groups.stream().map { group =>
val hasUnauthorizedTopic = group.members.stream()
.flatMap(member => util.stream.Stream.of(member.assignment,
member.targetAssignment))
@@ -2838,7 +2849,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
if (requiredTopics.nonEmpty) {
- val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
requiredTopics)(identity)
+ val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
requiredTopics.asJava, (t: String) => t)
if (authorizedTopics.size < requiredTopics.size) {
val responseData = new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
requestHelper.sendMaybeThrottle(request, new
StreamsGroupHeartbeatResponse(responseData))
@@ -2859,8 +2870,8 @@ class KafkaApis(val requestChannel: RequestChannel,
if (!topicsToCreate.isEmpty) {
val createTopicUnauthorized =
- if(!authHelper.authorize(request.context, CREATE, CLUSTER,
CLUSTER_NAME, logIfDenied = false))
- authHelper.partitionSeqByAuthorized(request.context, CREATE,
TOPIC, topicsToCreate.keySet.asScala.toSeq)(identity[String])._2
+ if (!authHelper.authorize(request.context, CREATE, CLUSTER,
CLUSTER_NAME, true, false, 1))
+ authHelper.partitionByAuthorized(request.context, CREATE,
TOPIC, topicsToCreate.keySet.stream().toList, (t: String) =>
t).unauthorized.asScala
else Set.empty
if (createTopicUnauthorized.nonEmpty) {
@@ -3017,7 +3028,7 @@ class KafkaApis(val requestChannel: RequestChannel,
.asScala
val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
- topicsToCheck)(identity)
+ topicsToCheck.asJava, (t: String) => t)
val updatedGroups = response.groups.stream.map { group =>
val hasUnauthorizedTopic = if (group.topology == null) false
else
@@ -3149,7 +3160,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// Clients are not allowed to see topics that are not authorized for
Describe.
val subscribedTopicSet =
shareGroupHeartbeatRequest.data.subscribedTopicNames.asScala.toSet
val authorizedTopics = authHelper.filterByAuthorized(request.context,
DESCRIBE, TOPIC,
- subscribedTopicSet)(identity)
+ subscribedTopicSet.asJava, (t: String) => t)
if (authorizedTopics.size < subscribedTopicSet.size) {
val responseData = new ShareGroupHeartbeatResponseData()
.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
@@ -3229,7 +3240,7 @@ class KafkaApis(val requestChannel: RequestChannel,
.collect(Collectors.toSet[String])
.asScala
val authorizedTopics =
authHelper.filterByAuthorized(request.context, DESCRIBE, TOPIC,
- topicsToCheck)(identity)
+ topicsToCheck.asJava, (t: String) => t)
val updatedGroups = response.groups.stream().map { group =>
val hasUnauthorizedTopic = group.members.stream()
.flatMap(member => member.assignment.topicPartitions.stream)
@@ -3363,8 +3374,9 @@ class KafkaApis(val requestChannel: RequestChannel,
request.context,
READ,
TOPIC,
- topicIdPartitionSeq
- )(_.topicPartition.topic)
+ topicIdPartitionSeq.asJava,
+ (tp: TopicIdPartition) => tp.topicPartition.topic
+ ).asScala
// Variable to store the topic partition wise result of piggybacked
acknowledgements.
var acknowledgeResult: CompletableFuture[Map[TopicIdPartition,
ShareAcknowledgeResponseData.PartitionData]] =
@@ -3688,8 +3700,9 @@ class KafkaApis(val requestChannel: RequestChannel,
request.context,
READ,
TOPIC,
- topicIdPartitionSeq
- )(_.topicPartition.topic)
+ topicIdPartitionSeq.asJava,
+ (tp: TopicIdPartition) => tp.topicPartition.topic
+ ).asScala
val erroneous = mutable.Map[TopicIdPartition,
ShareAcknowledgeResponseData.PartitionData]()
val acknowledgementDataFromRequest =
getAcknowledgeBatchesFromShareAcknowledgeRequest(shareAcknowledgeRequest,
topicIdNames, erroneous)
@@ -3888,13 +3901,15 @@ class KafkaApis(val requestChannel: RequestChannel,
.setErrorMessage(error.message)
} else {
// Clients are not allowed to see offsets for topics that are not
authorized for Describe.
- val (authorizedOffsets, _) = authHelper.partitionSeqByAuthorized(
+ val authorizedOffsetsPartition = authHelper.partitionByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
- groupDescribeOffsetsResponse.topics.asScala
- )(_.topicName)
- groupDescribeOffsetsResponse.setTopics(authorizedOffsets.asJava)
+ groupDescribeOffsetsResponse.topics,
+ (t:
DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseTopic)
=> t.topicName
+ )
+ val authorizedOffsets = authorizedOffsetsPartition.authorized
+ groupDescribeOffsetsResponse.setTopics(authorizedOffsets)
}
}
}
@@ -3903,18 +3918,21 @@ class KafkaApis(val requestChannel: RequestChannel,
groupDescribeOffsetsRequest:
DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup
):
CompletableFuture[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup]
= {
// Clients are not allowed to see offsets for topics that are not
authorized for Describe.
- val (authorizedTopics, unauthorizedTopics) =
authHelper.partitionSeqByAuthorized(
+ val partitionResult = authHelper.partitionByAuthorized(
requestContext,
DESCRIBE,
TOPIC,
- groupDescribeOffsetsRequest.topics.asScala
- )(_.topicName)
+ groupDescribeOffsetsRequest.topics,
+ (t:
DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestTopic) =>
t.topicName
+ )
+ val authorizedTopics = partitionResult.authorized
+ val unauthorizedTopics = partitionResult.unauthorized.asScala
groupCoordinator.describeShareGroupOffsets(
requestContext,
new
DescribeShareGroupOffsetsRequestData.DescribeShareGroupOffsetsRequestGroup()
.setGroupId(groupDescribeOffsetsRequest.groupId)
- .setTopics(authorizedTopics.asJava)
+ .setTopics(authorizedTopics)
).handle[DescribeShareGroupOffsetsResponseData.DescribeShareGroupOffsetsResponseGroup]
{ (groupDescribeOffsetsResponse, exception) =>
if (exception != null) {
val error = Errors.forException(exception)
diff --git
a/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java
b/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java
index 3f20bcae6c3..8f62356eb7e 100644
---
a/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java
+++
b/core/src/test/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandlerTest.java
@@ -17,7 +17,6 @@
package kafka.server.handlers;
-import kafka.server.AuthHelper;
import kafka.server.KafkaConfig;
import kafka.utils.TestUtils;
@@ -62,6 +61,7 @@ import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.network.metrics.RequestChannelMetrics;
import org.apache.kafka.raft.KRaftConfigs;
import org.apache.kafka.raft.QuorumConfig;
+import org.apache.kafka.server.AuthHelper;
import org.apache.kafka.server.authorizer.Action;
import org.apache.kafka.server.authorizer.AuthorizationResult;
import org.apache.kafka.server.authorizer.Authorizer;
@@ -192,7 +192,7 @@ class DescribeTopicPartitionsRequestHandlerTest {
KRaftMetadataCache metadataCache = new KRaftMetadataCache(0, () ->
KRaftVersion.KRAFT_VERSION_1);
updateKraftMetadataCache(metadataCache, records);
DescribeTopicPartitionsRequestHandler handler =
- new DescribeTopicPartitionsRequestHandler(metadataCache, new
AuthHelper(scala.Option.apply(authorizerPlugin)), createKafkaDefaultConfig());
+ new DescribeTopicPartitionsRequestHandler(metadataCache, new
AuthHelper(Optional.ofNullable(authorizerPlugin)), createKafkaDefaultConfig());
// 3.1 Basic test
DescribeTopicPartitionsRequest describeTopicPartitionsRequest = new
DescribeTopicPartitionsRequest(
@@ -390,7 +390,7 @@ class DescribeTopicPartitionsRequestHandlerTest {
KRaftMetadataCache metadataCache = new KRaftMetadataCache(0, () ->
KRaftVersion.KRAFT_VERSION_1);
updateKraftMetadataCache(metadataCache, records);
DescribeTopicPartitionsRequestHandler handler =
- new DescribeTopicPartitionsRequestHandler(metadataCache, new
AuthHelper(scala.Option.apply(authorizerPlugin)), createKafkaDefaultConfig());
+ new DescribeTopicPartitionsRequestHandler(metadataCache, new
AuthHelper(Optional.ofNullable(authorizerPlugin)), createKafkaDefaultConfig());
// 3.1 With cursor point to the first one
DescribeTopicPartitionsRequest describeTopicPartitionsRequest = new
DescribeTopicPartitionsRequest(new DescribeTopicPartitionsRequestData()
diff --git a/core/src/test/scala/unit/kafka/server/AuthHelperTest.scala
b/core/src/test/scala/unit/kafka/server/AuthHelperTest.scala
deleted file mode 100644
index 1f08046a113..00000000000
--- a/core/src/test/scala/unit/kafka/server/AuthHelperTest.scala
+++ /dev/null
@@ -1,261 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.server
-
-import java.net.InetAddress
-import java.util
-import org.apache.kafka.clients.admin.EndpointType
-import org.apache.kafka.common.acl.AclOperation
-import org.apache.kafka.common.internals.Plugin
-import org.apache.kafka.common.message.{DescribeClusterRequestData,
DescribeClusterResponseData}
-import
org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBrokerCollection
-import org.apache.kafka.common.network.{ClientInformation, ListenerName}
-import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{DescribeClusterRequest,
RequestContext, RequestHeader}
-import org.apache.kafka.common.resource.{PatternType, ResourcePattern,
ResourceType}
-import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.network.Request
-import org.apache.kafka.server.authorizer.{Action, AuthorizationResult,
Authorizer}
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.Test
-import org.mockito.ArgumentMatchers.argThat
-import org.mockito.ArgumentMatchers
-import org.mockito.Mockito.{mock, verify, when}
-
-import scala.collection.Seq
-import scala.jdk.CollectionConverters._
-
-object AuthHelperTest {
- def newMockDescribeClusterRequest(
- data: DescribeClusterRequestData,
- requestVersion: Int
- ): Request = {
- val requestContext = new RequestContext(
- new RequestHeader(ApiKeys.DESCRIBE_CLUSTER, requestVersion.toShort, "",
0),
- "",
- InetAddress.getLocalHost,
- KafkaPrincipal.ANONYMOUS,
- new ListenerName("PLAINTEXT"),
- SecurityProtocol.PLAINTEXT,
- ClientInformation.EMPTY,
- false)
- val request: Request = mock(classOf[Request])
- when(request.body(classOf[DescribeClusterRequest])).thenReturn(
- new DescribeClusterRequest(data, requestVersion.toShort))
- when(request.context).thenReturn(requestContext)
- when(request.header).thenReturn(requestContext.header)
- request
- }
-}
-
-class AuthHelperTest {
- import AuthHelperTest.newMockDescribeClusterRequest
-
- private val clientId = ""
-
- @Test
- def testAuthorize(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
-
- val operation = AclOperation.WRITE
- val resourceType = ResourceType.TOPIC
- val resourceName = "topic-1"
- val requestHeader = new RequestHeader(ApiKeys.PRODUCE,
ApiKeys.PRODUCE.latestVersion, clientId, 0)
- val requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost,
- KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
- SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false)
-
- val expectedActions = Seq(
- new Action(operation, new ResourcePattern(resourceType, resourceName,
PatternType.LITERAL),
- 1, true, true)
- )
-
- when(authorizer.authorize(requestContext, expectedActions.asJava))
- .thenReturn(Seq(AuthorizationResult.ALLOWED).asJava)
-
- val result = new AuthHelper(Some(authorizerPlugin)).authorize(
- requestContext, operation, resourceType, resourceName)
-
- verify(authorizer).authorize(requestContext, expectedActions.asJava)
-
- assertEquals(true, result)
- }
-
- @Test
- def testFilterByAuthorized(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
-
- val operation = AclOperation.WRITE
- val resourceType = ResourceType.TOPIC
- val resourceName1 = "topic-1"
- val resourceName2 = "topic-2"
- val resourceName3 = "topic-3"
- val requestHeader = new RequestHeader(ApiKeys.PRODUCE,
ApiKeys.PRODUCE.latestVersion,
- clientId, 0)
- val requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost,
- KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
- SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false)
-
- val expectedActions = Seq(
- new Action(operation, new ResourcePattern(resourceType, resourceName1,
PatternType.LITERAL),
- 2, true, true),
- new Action(operation, new ResourcePattern(resourceType, resourceName2,
PatternType.LITERAL),
- 1, true, true),
- new Action(operation, new ResourcePattern(resourceType, resourceName3,
PatternType.LITERAL),
- 1, true, true),
- )
-
- when(authorizer.authorize(
- ArgumentMatchers.eq(requestContext), argThat((t: java.util.List[Action])
=> t.containsAll(expectedActions.asJava))
- )).thenAnswer { invocation =>
- val actions =
invocation.getArgument(1).asInstanceOf[util.List[Action]].asScala
- actions.map { action =>
- if (Set(resourceName1,
resourceName3).contains(action.resourcePattern.name))
- AuthorizationResult.ALLOWED
- else
- AuthorizationResult.DENIED
- }.asJava
- }
-
- val result = new AuthHelper(Some(authorizerPlugin)).filterByAuthorized(
- requestContext,
- operation,
- resourceType,
- // Duplicate resource names should not trigger multiple calls to
authorize
- Seq(resourceName1, resourceName2, resourceName1, resourceName3)
- )(identity)
-
- verify(authorizer).authorize(
- ArgumentMatchers.eq(requestContext), argThat((t: java.util.List[Action])
=> t.containsAll(expectedActions.asJava))
- )
-
- assertEquals(Set(resourceName1, resourceName3), result)
- }
-
- @Test
- def testComputeDescribeClusterResponseV1WithUnknownEndpointType(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new DescribeClusterRequestData().setEndpointType(123.toByte), 1)
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.BROKER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => new DescribeClusterBrokerCollection(),
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setErrorCode(Errors.UNSUPPORTED_ENDPOINT_TYPE.code()).
- setErrorMessage("Unsupported endpoint type 123"), responseData)
- }
-
- @Test
- def testComputeDescribeClusterResponseV0WithUnknownEndpointType(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new DescribeClusterRequestData().setEndpointType(123.toByte), 0)
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.BROKER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => new DescribeClusterBrokerCollection(),
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setErrorCode(Errors.INVALID_REQUEST.code()).
- setErrorMessage("Unsupported endpoint type 123"), responseData)
- }
-
- @Test
- def testComputeDescribeClusterResponseV1WithUnexpectedEndpointType(): Unit =
{
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new
DescribeClusterRequestData().setEndpointType(EndpointType.BROKER.id()), 1)
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.CONTROLLER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => new DescribeClusterBrokerCollection(),
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setErrorCode(Errors.MISMATCHED_ENDPOINT_TYPE.code()).
- setErrorMessage("The request was sent to an endpoint of type CONTROLLER,
but we wanted an endpoint of type BROKER"), responseData)
- }
-
- @Test
- def testComputeDescribeClusterResponseV0WithUnexpectedEndpointType(): Unit =
{
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new
DescribeClusterRequestData().setEndpointType(EndpointType.BROKER.id()), 0)
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.CONTROLLER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => new DescribeClusterBrokerCollection(),
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setErrorCode(Errors.INVALID_REQUEST.code()).
- setErrorMessage("The request was sent to an endpoint of type CONTROLLER,
but we wanted an endpoint of type BROKER"), responseData)
- }
-
- @Test
- def testComputeDescribeClusterResponseWhereControllerIsNotFound(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new
DescribeClusterRequestData().setEndpointType(EndpointType.CONTROLLER.id()), 1)
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.CONTROLLER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => new DescribeClusterBrokerCollection(),
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setClusterId("ltCWoi9wRhmHSQCIgAznEg").
- setControllerId(-1).
- setClusterAuthorizedOperations(Int.MinValue).
- setEndpointType(2.toByte), responseData)
- }
-
- @Test
- def testComputeDescribeClusterResponseSuccess(): Unit = {
- val authorizer: Authorizer = mock(classOf[Authorizer])
- val authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name")
- val authHelper = new AuthHelper(Some(authorizerPlugin))
- val request = newMockDescribeClusterRequest(
- new
DescribeClusterRequestData().setEndpointType(EndpointType.CONTROLLER.id()), 1)
- val nodes = new DescribeClusterBrokerCollection(
-
java.util.Arrays.asList[DescribeClusterResponseData.DescribeClusterBroker](
- new
DescribeClusterResponseData.DescribeClusterBroker().setBrokerId(1)))
- val responseData = authHelper.computeDescribeClusterResponse(request,
- EndpointType.CONTROLLER,
- "ltCWoi9wRhmHSQCIgAznEg",
- () => nodes,
- () => 1)
- assertEquals(new DescribeClusterResponseData().
- setClusterId("ltCWoi9wRhmHSQCIgAznEg").
- setControllerId(1).
- setClusterAuthorizedOperations(Int.MinValue).
- setBrokers(nodes).
- setEndpointType(2.toByte), responseData)
- }
-}
diff --git a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
index 027cea189c0..fd189d1c95e 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
@@ -738,8 +738,8 @@ class ControllerApisTest {
setErrorMessage(s"Creation of internal topic
${Topic.CLUSTER_METADATA_TOPIC_NAME} is prohibited."))
assertEquals(expectedResponse,
controllerApis.createTopics(ANONYMOUS_CONTEXT, request,
hasClusterAuth = false,
- _ => Set("baz", "indescribable"),
- _ => Set("baz"),
+ _ => util.Set.of("baz", "indescribable"),
+ _ => util.Set.of("baz"),
forwarded = false).get().topics().asScala.toSet)
}
@@ -788,8 +788,8 @@ class ControllerApisTest {
assertEquals(expectedResponse,
controllerApis.deleteTopics(ANONYMOUS_CONTEXT, request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = true,
- _ => Set.empty,
- _ => Set.empty).get().asScala.toSet)
+ _ => util.Set.of[String](),
+ _ => util.Set.of[String]()).get().asScala.toSet)
}
@Test
@@ -814,8 +814,8 @@ class ControllerApisTest {
assertEquals(response, controllerApis.deleteTopics(ANONYMOUS_CONTEXT,
request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = true,
- _ => Set.empty,
- _ => Set.empty).get().asScala.toSet)
+ _ => util.Set.of[String](),
+ _ => util.Set.of[String]()).get().asScala.toSet)
}
@Test
@@ -856,8 +856,8 @@ class ControllerApisTest {
assertEquals(response, controllerApis.deleteTopics(ANONYMOUS_CONTEXT,
request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = false,
- names => names.toSet,
- names => names.toSet).get().asScala.toSet)
+ names => names.asScala.toSet.asJava,
+ names => names.asScala.toSet.asJava).get().asScala.toSet)
}
@Test
@@ -892,8 +892,8 @@ class ControllerApisTest {
assertEquals(response, controllerApis.deleteTopics(ANONYMOUS_CONTEXT,
request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = false,
- _ => Set("foo", "baz"),
- _ => Set.empty).get().asScala.toSet)
+ _ => util.Set.of("foo", "baz"),
+ _ => util.Set.of[String]()).get().asScala.toSet)
}
@Test
@@ -917,8 +917,8 @@ class ControllerApisTest {
assertEquals(expectedResponse,
controllerApis.deleteTopics(ANONYMOUS_CONTEXT, request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = false,
- _ => Set("foo"),
- _ => Set.empty).get().asScala.toSet)
+ _ => util.Set.of("foo"),
+ _ => util.Set.of[String]()).get().asScala.toSet)
}
@Test
@@ -936,8 +936,8 @@ class ControllerApisTest {
classOf[ExecutionException], () =>
controllerApis.deleteTopics(ANONYMOUS_CONTEXT, request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = false,
- _ => Set("foo", "bar"),
- _ => Set("foo", "bar")).get()).getCause.getClass)
+ _ => util.Set.of("foo", "bar"),
+ _ => util.Set.of("foo", "bar")).get()).getCause.getClass)
}
@Test
@@ -954,14 +954,14 @@ class ControllerApisTest {
TestUtils.assertFutureThrows(classOf[TopicDeletionDisabledException],
controllerApis.deleteTopics(ANONYMOUS_CONTEXT, request,
ApiKeys.DELETE_TOPICS.latestVersion().toInt,
hasClusterAuth = false,
- _ => Set("foo", "bar"),
- _ => Set("foo", "bar")))
+ _ => util.Set.of("foo", "bar"),
+ _ => util.Set.of("foo", "bar")))
TestUtils.assertFutureThrows(classOf[InvalidRequestException],
controllerApis.deleteTopics(ANONYMOUS_CONTEXT, request,
1,
hasClusterAuth = false,
- _ => Set("foo", "bar"),
- _ => Set("foo", "bar")))
+ _ => util.Set.of("foo", "bar"),
+ _ => util.Set.of("foo", "bar")))
}
@ParameterizedTest
@@ -999,7 +999,7 @@ class ControllerApisTest {
setErrorCode(TOPIC_AUTHORIZATION_FAILED.code()).
setErrorMessage(null)),
controllerApis.createPartitions(ANONYMOUS_CONTEXT, request,
- _ => Set("foo", "bar")).get().asScala.toSet)
+ _ => util.Set.of("foo", "bar")).get().asScala.toSet)
}
@Test
diff --git a/server/src/main/java/org/apache/kafka/server/AuthHelper.java
b/server/src/main/java/org/apache/kafka/server/AuthHelper.java
new file mode 100644
index 00000000000..8a3d2d6983f
--- /dev/null
+++ b/server/src/main/java/org/apache/kafka/server/AuthHelper.java
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server;
+
+import org.apache.kafka.clients.admin.EndpointType;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.internals.Plugin;
+import org.apache.kafka.common.message.DescribeClusterResponseData;
+import
org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBrokerCollection;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.DescribeClusterRequest;
+import org.apache.kafka.common.requests.RequestContext;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.Resource;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.network.Request;
+import org.apache.kafka.security.authorizer.AclEntry;
+import org.apache.kafka.server.authorizer.Action;
+import org.apache.kafka.server.authorizer.AuthorizationResult;
+import org.apache.kafka.server.authorizer.Authorizer;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public class AuthHelper {
+
+ private final Optional<Plugin<Authorizer>> authorizer;
+
+ public AuthHelper(Optional<Plugin<Authorizer>> authorizer) {
+ this.authorizer = authorizer;
+ }
+
+ public boolean authorize(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType,
+ String resourceName,
+ boolean logIfAllowed,
+ boolean logIfDenied,
+ int refCount
+ ) {
+ if (authorizer.isEmpty()) {
+ return true;
+ }
+ ResourcePattern resource = new ResourcePattern(resourceType,
resourceName, PatternType.LITERAL);
+ List<Action> actions = List.of(
+ new Action(operation, resource, refCount, logIfAllowed,
logIfDenied)
+ );
+ return authorizer.get().get().authorize(requestContext,
actions).get(0) == AuthorizationResult.ALLOWED;
+ }
+
+ public boolean authorize(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType,
+ String resourceName
+ ) {
+ return authorize(requestContext, operation, resourceType,
resourceName, true, true, 1);
+ }
+
+ public void authorizeClusterOperation(Request request, AclOperation
operation) {
+ if (!authorize(request.context(), operation, ResourceType.CLUSTER,
Resource.CLUSTER_NAME)) {
+ throw new ClusterAuthorizationException("Request " + request + "
needs " + operation + " permission.");
+ }
+ }
+
+ public int authorizedOperations(Request request, Resource resource) {
+ List<AclOperation> supportedOps = new
ArrayList<>(AclEntry.supportedOperations(resource.resourceType()));
+ Set<AclOperation> authorizedOps;
+ if (authorizer.isPresent()) {
+ ResourcePattern resourcePattern = new
ResourcePattern(resource.resourceType(), resource.name(), PatternType.LITERAL);
+ List<Action> actions = supportedOps.stream()
+ .map(op -> new Action(op, resourcePattern, 1, false, false))
+ .toList();
+ List<AuthorizationResult> results =
authorizer.get().get().authorize(request.context(), actions);
+ authorizedOps = new HashSet<>();
+ // Authorizer.authorize returns one result per action in the same
order, so the i-th
+ // result corresponds to the i-th supported operation. Iterate up
to the smaller size to
+ // stay within bounds.
+ int count = Math.min(results.size(), supportedOps.size());
+ for (int i = 0; i < count; i++) {
+ if (results.get(i) == AuthorizationResult.ALLOWED) {
+ authorizedOps.add(supportedOps.get(i));
+ }
+ }
+ } else {
+ authorizedOps = new HashSet<>(supportedOps);
+ }
+ Set<Byte> opCodes = authorizedOps.stream()
+ .map(AclOperation::code)
+ .collect(Collectors.toSet());
+ return Utils.to32BitField(opCodes);
+ }
+
+ public boolean authorizeByResourceType(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType
+ ) {
+ return authorizer.map(authorizerPlugin ->
authorizerPlugin.get().authorizeByResourceType(requestContext, operation,
resourceType) == AuthorizationResult.ALLOWED).orElse(true);
+ }
+
+ public <T> Set<String> filterByAuthorized(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType,
+ Iterable<T> resources,
+ boolean logIfAllowed,
+ boolean logIfDenied,
+ Function<T, String> resourceName
+ ) {
+ if (authorizer.isEmpty()) {
+ Set<String> result = new HashSet<>();
+ for (T resource : resources) {
+ result.add(resourceName.apply(resource));
+ }
+ return result;
+ }
+ // Count occurrences of each resource name
+ Map<String, Integer> resourceNameToCount = new HashMap<>();
+ for (T resource : resources) {
+ String name = resourceName.apply(resource);
+ resourceNameToCount.merge(name, 1, Integer::sum);
+ }
+ if (resourceNameToCount.isEmpty()) {
+ return Set.of();
+ }
+
+ List<String> names = new ArrayList<>(resourceNameToCount.keySet());
+ List<Action> actions = names.stream()
+ .map(name -> new Action(
+ operation,
+ new ResourcePattern(resourceType, name, PatternType.LITERAL),
+ resourceNameToCount.get(name),
+ logIfAllowed,
+ logIfDenied
+ ))
+ .toList();
+ List<AuthorizationResult> results =
authorizer.get().get().authorize(requestContext, actions);
+ Set<String> authorized = new HashSet<>();
+ // Authorizer.authorize returns one result per action in the same
order, so the i-th
+ // result corresponds to the i-th resource name. Iterate up to the
smaller size to
+ // stay within bounds.
+ int count = Math.min(results.size(), names.size());
+ for (int i = 0; i < count; i++) {
+ if (results.get(i) == AuthorizationResult.ALLOWED) {
+ authorized.add(names.get(i));
+ }
+ }
+ return authorized;
+ }
+
+ public <T> Set<String> filterByAuthorized(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType,
+ Iterable<T> resources,
+ Function<T, String> resourceName
+ ) {
+ return filterByAuthorized(requestContext, operation, resourceType,
resources, true, true, resourceName);
+ }
+
+ public record PartitionResult<T>(List<T> authorized, List<T> unauthorized)
{
+ }
+
+ public <T> PartitionResult<T> partitionByAuthorized(
+ RequestContext requestContext,
+ AclOperation operation,
+ ResourceType resourceType,
+ List<T> resources,
+ Function<T, String> resourceName
+ ) {
+ if (authorizer.isEmpty()) {
+ return new PartitionResult<>(resources, List.of());
+ }
+ Set<String> authorizedResourceNames = filterByAuthorized(
+ requestContext, operation, resourceType, resources, true, true,
resourceName
+ );
+ List<T> authorized = new ArrayList<>();
+ List<T> unauthorized = new ArrayList<>();
+ for (T resource : resources) {
+ if
(authorizedResourceNames.contains(resourceName.apply(resource))) {
+ authorized.add(resource);
+ } else {
+ unauthorized.add(resource);
+ }
+ }
+ return new PartitionResult<>(authorized, unauthorized);
+ }
+
+ public DescribeClusterResponseData computeDescribeClusterResponse(
+ Request request,
+ EndpointType expectedEndpointType,
+ String clusterId,
+ Supplier<DescribeClusterBrokerCollection> getNodes,
+ Supplier<Integer> getControllerId
+ ) {
+ DescribeClusterRequest describeClusterRequest =
request.body(DescribeClusterRequest.class);
+ EndpointType requestEndpointType =
EndpointType.fromId(describeClusterRequest.data().endpointType());
+ if (requestEndpointType.equals(EndpointType.UNKNOWN)) {
+ return new DescribeClusterResponseData()
+ .setErrorCode(request.header().data().requestApiVersion() == 0
+ ? Errors.INVALID_REQUEST.code()
+ : Errors.UNSUPPORTED_ENDPOINT_TYPE.code())
+ .setErrorMessage("Unsupported endpoint type " + (int)
describeClusterRequest.data().endpointType());
+ } else if (!expectedEndpointType.equals(requestEndpointType)) {
+ return new DescribeClusterResponseData()
+ .setErrorCode(request.header().data().requestApiVersion() == 0
+ ? Errors.INVALID_REQUEST.code()
+ : Errors.MISMATCHED_ENDPOINT_TYPE.code())
+ .setErrorMessage("The request was sent to an endpoint of type
" + expectedEndpointType +
+ ", but we wanted an endpoint of type " +
requestEndpointType);
+ }
+
+ int clusterAuthorizedOperations = Integer.MIN_VALUE; // Default value
in the schema
+ // get cluster authorized operations
+ if
(describeClusterRequest.data().includeClusterAuthorizedOperations()) {
+ if (authorize(request.context(), AclOperation.DESCRIBE,
ResourceType.CLUSTER, Resource.CLUSTER_NAME)) {
+ clusterAuthorizedOperations = authorizedOperations(request,
Resource.CLUSTER);
+ } else {
+ clusterAuthorizedOperations = 0;
+ }
+ }
+
+ // Get the node list and the controller ID.
+ DescribeClusterBrokerCollection nodes = getNodes.get();
+ int controllerId = getControllerId.get();
+ // If the provided controller ID is not in the node list, return -1
instead
+ // to avoid confusing the client. This could happen in a case where we
know
+ // the controller ID, but we don't yet have KIP-919 information about
that controller.
+ int effectiveControllerId = (nodes.find(controllerId) == null) ? -1 :
controllerId;
+
+ return new DescribeClusterResponseData()
+ .setClusterId(clusterId)
+ .setControllerId(effectiveControllerId)
+ .setClusterAuthorizedOperations(clusterAuthorizedOperations)
+ .setBrokers(nodes)
+ .setEndpointType(expectedEndpointType.id());
+ }
+}
diff --git a/server/src/test/java/org/apache/kafka/server/AuthHelperTest.java
b/server/src/test/java/org/apache/kafka/server/AuthHelperTest.java
new file mode 100644
index 00000000000..7d947f75d80
--- /dev/null
+++ b/server/src/test/java/org/apache/kafka/server/AuthHelperTest.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.server;
+
+import org.apache.kafka.clients.admin.EndpointType;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.internals.Plugin;
+import org.apache.kafka.common.message.DescribeClusterRequestData;
+import org.apache.kafka.common.message.DescribeClusterResponseData;
+import
org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBroker;
+import
org.apache.kafka.common.message.DescribeClusterResponseData.DescribeClusterBrokerCollection;
+import org.apache.kafka.common.network.ClientInformation;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.DescribeClusterRequest;
+import org.apache.kafka.common.requests.RequestContext;
+import org.apache.kafka.common.requests.RequestHeader;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.network.Request;
+import org.apache.kafka.server.authorizer.Action;
+import org.apache.kafka.server.authorizer.AuthorizationResult;
+import org.apache.kafka.server.authorizer.Authorizer;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class AuthHelperTest {
+
+ private Authorizer authorizer;
+ private Plugin<Authorizer> authorizerPlugin;
+
+ @BeforeEach
+ public void setUp() {
+ authorizer = mock(Authorizer.class);
+ authorizerPlugin = Plugin.wrapInstance(authorizer, null,
"authorizer.class.name");
+ }
+
+ private static Request
newMockDescribeClusterRequest(DescribeClusterRequestData data, int
requestVersion)
+ throws UnknownHostException {
+ RequestContext requestContext = new RequestContext(
+ new RequestHeader(ApiKeys.DESCRIBE_CLUSTER, (short)
requestVersion, "", 0),
+ "",
+ InetAddress.getLocalHost(),
+ KafkaPrincipal.ANONYMOUS,
+ new ListenerName("PLAINTEXT"),
+ SecurityProtocol.PLAINTEXT,
+ ClientInformation.EMPTY,
+ false);
+ Request request = mock(Request.class);
+ when(request.body(DescribeClusterRequest.class)).thenReturn(
+ new DescribeClusterRequest(data, (short) requestVersion));
+ when(request.context()).thenReturn(requestContext);
+ when(request.header()).thenReturn(requestContext.header);
+ return request;
+ }
+
+ @Test
+ public void testAuthorize() throws UnknownHostException {
+ AclOperation operation = AclOperation.WRITE;
+ ResourceType resourceType = ResourceType.TOPIC;
+ String resourceName = "topic-1";
+ RequestHeader requestHeader = new RequestHeader(ApiKeys.PRODUCE,
ApiKeys.PRODUCE.latestVersion(), "", 0);
+ RequestContext requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost(),
+ KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
+ SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false);
+
+ List<Action> expectedActions = List.of(
+ new Action(operation, new ResourcePattern(resourceType,
resourceName, PatternType.LITERAL), 1, true, true)
+ );
+
+ when(authorizer.authorize(requestContext, expectedActions))
+ .thenReturn(List.of(AuthorizationResult.ALLOWED));
+
+ boolean result = new
AuthHelper(Optional.of(authorizerPlugin)).authorize(
+ requestContext, operation, resourceType, resourceName);
+
+ verify(authorizer).authorize(requestContext, expectedActions);
+ assertTrue(result);
+ }
+
+ @Test
+ public void testFilterByAuthorized() throws UnknownHostException {
+ AclOperation operation = AclOperation.WRITE;
+ ResourceType resourceType = ResourceType.TOPIC;
+ String resourceName1 = "topic-1";
+ String resourceName2 = "topic-2";
+ String resourceName3 = "topic-3";
+ RequestHeader requestHeader = new RequestHeader(ApiKeys.PRODUCE,
ApiKeys.PRODUCE.latestVersion(), "", 0);
+ RequestContext requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost(),
+ KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
+ SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false);
+
+ List<Action> expectedActions = List.of(
+ new Action(operation, new ResourcePattern(resourceType,
resourceName1, PatternType.LITERAL), 2, true, true),
+ new Action(operation, new ResourcePattern(resourceType,
resourceName2, PatternType.LITERAL), 1, true, true),
+ new Action(operation, new ResourcePattern(resourceType,
resourceName3, PatternType.LITERAL), 1, true, true)
+ );
+
+ when(authorizer.authorize(
+ ArgumentMatchers.eq(requestContext),
+ argThat(t -> t.containsAll(expectedActions))
+ )).thenAnswer(invocation -> {
+ List<Action> actions = invocation.getArgument(1);
+ return actions.stream().map(action -> {
+ String name = action.resourcePattern().name();
+ if (name.equals(resourceName1) || name.equals(resourceName3))
+ return AuthorizationResult.ALLOWED;
+ else
+ return AuthorizationResult.DENIED;
+ }).toList();
+ });
+
+ // Duplicate resource names should not trigger multiple calls to
authorize
+ Set<String> result = new
AuthHelper(Optional.of(authorizerPlugin)).filterByAuthorized(
+ requestContext,
+ operation,
+ resourceType,
+ List.of(resourceName1, resourceName2, resourceName1,
resourceName3),
+ s -> s
+ );
+
+ verify(authorizer).authorize(
+ ArgumentMatchers.eq(requestContext),
+ argThat(t -> t.containsAll(expectedActions))
+ );
+
+ assertEquals(Set.of(resourceName1, resourceName3), result);
+ }
+
+ @Test
+ public void
testFilterByAuthorizedIsResilientToMismatchedAuthorizeResults() throws
UnknownHostException {
+ AclOperation operation = AclOperation.WRITE;
+ ResourceType resourceType = ResourceType.TOPIC;
+ RequestHeader requestHeader = new RequestHeader(ApiKeys.PRODUCE,
ApiKeys.PRODUCE.latestVersion(), "", 0);
+ RequestContext requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost(),
+ KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
+ SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false);
+
+ // A single distinct resource produces a single action, but a
misbehaving authorizer may return
+ // a result list of a different size. The original Scala
implementation used `zip`, which tolerated
+ // this by truncating to the shorter sequence. Verify we keep that
behavior instead of failing with
+ // IndexOutOfBoundsException.
+ when(authorizer.authorize(ArgumentMatchers.eq(requestContext),
ArgumentMatchers.any()))
+ .thenReturn(List.of(AuthorizationResult.ALLOWED,
AuthorizationResult.DENIED));
+
+ Set<String> result = new
AuthHelper(Optional.of(authorizerPlugin)).filterByAuthorized(
+ requestContext, operation, resourceType, List.of("topic-1"), s ->
s);
+
+ assertEquals(Set.of("topic-1"), result);
+ }
+
+ @Test
+ public void testPartitionByAuthorized() throws UnknownHostException {
+ AclOperation operation = AclOperation.DESCRIBE;
+ ResourceType resourceType = ResourceType.TOPIC;
+ RequestHeader requestHeader = new RequestHeader(ApiKeys.METADATA,
ApiKeys.METADATA.latestVersion(), "", 0);
+ RequestContext requestContext = new RequestContext(requestHeader, "1",
InetAddress.getLocalHost(),
+ KafkaPrincipal.ANONYMOUS,
ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
+ SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false);
+
+ when(authorizer.authorize(ArgumentMatchers.eq(requestContext),
ArgumentMatchers.any()))
+ .thenAnswer(invocation -> {
+ List<Action> actions = invocation.getArgument(1);
+ return actions.stream().map(action -> {
+ String name = action.resourcePattern().name();
+ if (name.equals("topic-1") || name.equals("topic-3"))
+ return AuthorizationResult.ALLOWED;
+ else
+ return AuthorizationResult.DENIED;
+ }).toList();
+ });
+
+ AuthHelper.PartitionResult<String> result = new
AuthHelper(Optional.of(authorizerPlugin)).partitionByAuthorized(
+ requestContext, operation, resourceType, List.of("topic-1",
"topic-2", "topic-3"), s -> s);
+
+ // The original input order should be preserved within each partition.
+ assertEquals(List.of("topic-1", "topic-3"), result.authorized());
+ assertEquals(List.of("topic-2"), result.unauthorized());
+ }
+
+ @Test
+ public void testPartitionByAuthorizedWithoutAuthorizer() {
+ List<String> resources = List.of("topic-1", "topic-2", "topic-3");
+ // The request context is unused when there is no authorizer, so it
can be null here.
+ AuthHelper.PartitionResult<String> result = new
AuthHelper(Optional.empty()).partitionByAuthorized(
+ null, AclOperation.DESCRIBE, ResourceType.TOPIC, resources, s ->
s);
+
+ // Without an authorizer everything is authorized.
+ assertEquals(resources, result.authorized());
+ assertTrue(result.unauthorized().isEmpty());
+ }
+
+ @Test
+ public void testComputeDescribeClusterResponseV1WithUnknownEndpointType()
throws UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new DescribeClusterRequestData().setEndpointType((byte) 123), 1);
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.BROKER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ DescribeClusterBrokerCollection::new,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setErrorCode(Errors.UNSUPPORTED_ENDPOINT_TYPE.code())
+ .setErrorMessage("Unsupported endpoint type 123"), responseData);
+ }
+
+ @Test
+ public void testComputeDescribeClusterResponseV0WithUnknownEndpointType()
throws UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new DescribeClusterRequestData().setEndpointType((byte) 123), 0);
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.BROKER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ DescribeClusterBrokerCollection::new,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setErrorCode(Errors.INVALID_REQUEST.code())
+ .setErrorMessage("Unsupported endpoint type 123"), responseData);
+ }
+
+ @Test
+ public void
testComputeDescribeClusterResponseV1WithUnexpectedEndpointType() throws
UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new
DescribeClusterRequestData().setEndpointType(EndpointType.BROKER.id()), 1);
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.CONTROLLER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ DescribeClusterBrokerCollection::new,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setErrorCode(Errors.MISMATCHED_ENDPOINT_TYPE.code())
+ .setErrorMessage("The request was sent to an endpoint of type
CONTROLLER, but we wanted an endpoint of type BROKER"),
+ responseData);
+ }
+
+ @Test
+ public void
testComputeDescribeClusterResponseV0WithUnexpectedEndpointType() throws
UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new
DescribeClusterRequestData().setEndpointType(EndpointType.BROKER.id()), 0);
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.CONTROLLER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ DescribeClusterBrokerCollection::new,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setErrorCode(Errors.INVALID_REQUEST.code())
+ .setErrorMessage("The request was sent to an endpoint of type
CONTROLLER, but we wanted an endpoint of type BROKER"),
+ responseData);
+ }
+
+ @Test
+ public void testComputeDescribeClusterResponseWhereControllerIsNotFound()
throws UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new
DescribeClusterRequestData().setEndpointType(EndpointType.CONTROLLER.id()), 1);
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.CONTROLLER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ DescribeClusterBrokerCollection::new,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setClusterId("ltCWoi9wRhmHSQCIgAznEg")
+ .setControllerId(-1)
+ .setClusterAuthorizedOperations(Integer.MIN_VALUE)
+ .setEndpointType((byte) 2), responseData);
+ }
+
+ @Test
+ public void testComputeDescribeClusterResponseSuccess() throws
UnknownHostException {
+ AuthHelper authHelper = new AuthHelper(Optional.of(authorizerPlugin));
+ Request request = newMockDescribeClusterRequest(
+ new
DescribeClusterRequestData().setEndpointType(EndpointType.CONTROLLER.id()), 1);
+ DescribeClusterBrokerCollection nodes = new
DescribeClusterBrokerCollection(
+ List.of(new DescribeClusterBroker().setBrokerId(1)).iterator());
+ DescribeClusterResponseData responseData =
authHelper.computeDescribeClusterResponse(request,
+ EndpointType.CONTROLLER,
+ "ltCWoi9wRhmHSQCIgAznEg",
+ () -> nodes,
+ () -> 1);
+ assertEquals(new DescribeClusterResponseData()
+ .setClusterId("ltCWoi9wRhmHSQCIgAznEg")
+ .setControllerId(1)
+ .setClusterAuthorizedOperations(Integer.MIN_VALUE)
+ .setBrokers(nodes)
+ .setEndpointType((byte) 2), responseData);
+ }
+}
