Skip to content

KAFKA-12394: Return TOPIC_AUTHORIZATION_FAILED in delete topic response if no describe permission #10223

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 3 commits into from
Mar 2, 2021
Merged

KAFKA-12394: Return TOPIC_AUTHORIZATION_FAILED in delete topic response if no describe permission #10223

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef340d3
MINOR: Do not expose topic name in `DeleteTopic` response if no descr…
hachikuji Feb 26, 2021
76589c1
Use `TOPIC_AUTHORIZATION_FAILED` when no describe permission
hachikuji Mar 2, 2021
7dffa6d
Prefer explicit `contains` call
hachikuji Mar 2, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions core/src/main/scala/kafka/controller/ControllerContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,10 @@ class ControllerContext {
}.keySet
}

def topicName(topicId: Uuid): Option[String] = {
topicNames.get(topicId)
}

def clearPartitionLeadershipInfo(): Unit = partitionLeadershipInfo.clear()

def partitionWithLeadersCount: Int = partitionLeadershipInfo.size
Expand Down
37 changes: 22 additions & 15 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1874,7 +1874,7 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topic.name() != null && topic.topicId() != Uuid.ZERO_UUID)
throw new InvalidRequestException("Topic name and topic ID can not both be specified.")
val name = if (topic.topicId() == Uuid.ZERO_UUID) topic.name()
else zkSupport.controller.controllerContext.topicNames.getOrElse(topic.topicId(), null)
else zkSupport.controller.controllerContext.topicName(topic.topicId).orNull
Comment thread
chia7712 marked this conversation as resolved.
results.add(new DeletableTopicResult()
.setName(name)
.setTopicId(topic.topicId()))
Expand All @@ -1884,20 +1884,27 @@ class KafkaApis(val requestChannel: RequestChannel,
val authorizedDeleteTopics = authHelper.filterByAuthorized(request.context, DELETE, TOPIC,
results.asScala.filter(result => result.name() != null))(_.name)
results.forEach { topic =>
val unresolvedTopicId = !(topic.topicId() == Uuid.ZERO_UUID) && topic.name() == null
if (!config.usesTopicId && topicIdsFromRequest.contains(topic.topicId)) {
topic.setErrorCode(Errors.UNSUPPORTED_VERSION.code)
topic.setErrorMessage("Topic IDs are not supported on the server.")
} else if (unresolvedTopicId)
topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
else if (topicIdsFromRequest.contains(topic.topicId) && !authorizedDescribeTopics(topic.name))
topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
else if (!authorizedDeleteTopics.contains(topic.name))
topic.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
else if (!metadataCache.contains(topic.name))
topic.setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
else
toDelete += topic.name
val unresolvedTopicId = topic.topicId() != Uuid.ZERO_UUID && topic.name() == null
if (!config.usesTopicId && topicIdsFromRequest.contains(topic.topicId)) {
topic.setErrorCode(Errors.UNSUPPORTED_VERSION.code)
topic.setErrorMessage("Topic IDs are not supported on the server.")
} else if (unresolvedTopicId) {
topic.setErrorCode(Errors.UNKNOWN_TOPIC_ID.code)
} else if (topicIdsFromRequest.contains(topic.topicId) && !authorizedDescribeTopics.contains(topic.name)) {

// Because the client does not have Describe permission, the name should
// not be returned in the response. Note, however, that we do not consider
// the topicId itself to be sensitive, so there is no reason to obscure
// this case with `UNKNOWN_TOPIC_ID`.
topic.setName(null)

@dajac dajac Feb 27, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just noticed that the Name in the schema [1] used mapKey: true. I wonder if we should remove it now that Name is nullable. Usually, when we use mapKey, we expect to be able to query with the name and this is not always the case now. It seems that we don't rely on it in the admin client but we do in tests. What do you think?

[1] https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/DeleteTopicsResponse.json#L41

@hachikuji hachikuji Feb 27, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a good question. I agree using mapKey makes less sense now. Let me try it out. If the diff is not too bad, we can do it here. Otherwise, we can do it separately.

@hachikuji hachikuji Mar 1, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have one dependence here which makes this a little more work than I wanted to do here: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L1909. I filed a separate JIRA so that we don't forget about it: https://issues.apache.org/jira/browse/KAFKA-12395.

topic.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
} else if (!authorizedDeleteTopics.contains(topic.name)) {

@chia7712 chia7712 Mar 2, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

according to #10184 (comment), should it handle the case name provided, topic missing, describable => UNKNOWN_TOPIC_OR_PARTITION?

@hachikuji hachikuji Mar 2, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess there are really two sub-cases. Here's how they are currently handled:

name provided, topic missing, describable, deletable => UNKNOWN_TOPIC_OR_PARTITION
name provided, topic missing, describable, undeletable => TOPIC_AUTHORIZATION_FAILED

This seems defensible to me. The UNKNOWN_TOPIC_OR_PARTITION error will cause the client to retry because of the possibility of stale metadata, but it can't delete the topic anyway because of the authorization failure. It seems better to fail fast?

@chia7712 chia7712 Mar 2, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems defensible to me. The UNKNOWN_TOPIC_OR_PARTITION error will cause the client to retry because of the possibility of stale metadata, but it can't delete the topic anyway because of the authorization failure. It seems better to fail fast?

That makes sense to me. We have to make sure this rule is applied to #10184 :)

topic.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
} else if (!metadataCache.contains(topic.name)) {
topic.setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code)
} else {
toDelete += topic.name
}
}
// If no authorized topics return immediately
if (toDelete.isEmpty)
Expand Down
132 changes: 49 additions & 83 deletions core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@ import org.apache.kafka.common.{ElectionType, IsolationLevel, Node, TopicPartiti
import org.apache.kafka.test.{TestUtils => JTestUtils}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource

import scala.annotation.nowarn
import scala.collection.mutable
Expand Down Expand Up @@ -245,9 +247,14 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
})
)

val requestKeysToErrorWithIds = (id: Uuid) => Map[ApiKeys, Nothing => Errors](
ApiKeys.DELETE_TOPICS -> ((resp: requests.DeleteTopicsResponse) => Errors.forCode(resp.data.responses.asScala.find(_.topicId == id).get.errorCode()))
)
def findErrorForTopicId(id: Uuid, response: AbstractResponse): Errors = {
response match {
case res: DeleteTopicsResponse =>
Errors.forCode(res.data.responses.asScala.find(_.topicId == id).get.errorCode)
case _ =>
fail(s"Unexpected response type $response")
}
}

val requestKeysToAcls = Map[ApiKeys, Map[ResourcePattern, Set[AccessControlEntry]]](
ApiKeys.METADATA -> topicDescribeAcl,
Expand Down Expand Up @@ -534,12 +541,12 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
.setTimeoutMs(5000)).build()
}

private def deleteTopicsWithIdsRequest(id: Uuid = getTopicIds()(topic)): DeleteTopicsRequest = {
private def deleteTopicsWithIdsRequest(topicId: Uuid): DeleteTopicsRequest = {
new DeleteTopicsRequest.Builder(
new DeleteTopicsRequestData()
.setTopics(Collections.singletonList(
new DeleteTopicsRequestData.DeleteTopicState()
.setTopicId(id)))
.setTopicId(topicId)))
.setTimeoutMs(5000)).build()
}

Expand Down Expand Up @@ -737,29 +744,53 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
sendRequests(requestKeyToRequest)
}

@Test
def testAuthorizationDeleteTopicsIdWithTopicExisting(): Unit = {
sendRequests(mutable.Map(ApiKeys.CREATE_TOPICS -> createTopicsRequest))
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def testTopicIdAuthorization(withTopicExisting: Boolean): Unit = {
val topicId = if (withTopicExisting) {
createTopic(topic)
getTopicIds()(topic)
} else {
Uuid.randomUuid()
}

val requestKeyToRequest = mutable.LinkedHashMap[ApiKeys, AbstractRequest](
ApiKeys.DELETE_TOPICS -> deleteTopicsWithIdsRequest(topicId)
)

val id = getTopicIds()(topic)
def sendAndVerify(
request: AbstractRequest,
isAuthorized: Boolean,
isDescribeAuthorized: Boolean
): Unit = {
val response = connectAndReceive[AbstractResponse](request)
val error = findErrorForTopicId(topicId, response)
if (!withTopicExisting) {
assertEquals(Errors.UNKNOWN_TOPIC_ID, error)
} else if (!isDescribeAuthorized || !isAuthorized) {
assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED, error)
}
}

for ((key, request) <- mutable.Map(ApiKeys.DELETE_TOPICS -> deleteTopicsWithIdsRequest())) {
for ((key, request) <- requestKeyToRequest) {
removeAllClientAcls()
val resources = requestKeysToAcls(key).map(_._1.resourceType).toSet
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = false, topicExists = true, describeAuthorized = false, id = id)
sendAndVerify(request, isAuthorized = false, isDescribeAuthorized = false)

val describeAcls = topicDescribeAcl(topicResource)
addAndVerifyAcls(describeAcls, topicResource)

val resourceToAcls = requestKeysToAcls(key)
resourceToAcls.get(topicResource).foreach { acls =>
val describeAcls = topicDescribeAcl(topicResource)
val isAuthorized = describeAcls == acls
addAndVerifyAcls(describeAcls, topicResource)
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = isAuthorized, topicExists = true, describeAuthorized = true, id = id)
removeAllClientAcls()
sendAndVerify(request, isAuthorized = isAuthorized, isDescribeAuthorized = true)
}

for ((resource, acls) <- resourceToAcls)
removeAllClientAcls()
for ((resource, acls) <- resourceToAcls) {
addAndVerifyAcls(acls, resource)
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = true, topicExists = true, describeAuthorized = true, id = id)
}

sendAndVerify(request, isAuthorized = true, isDescribeAuthorized = true)
}
}

Expand Down Expand Up @@ -789,33 +820,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
sendRequests(requestKeyToRequest, false)
}

@Test
def testAuthorizationDeleteTopicsIdWithTopicNotExisting(): Unit = {
val id = Uuid.randomUuid()
val requestKeyToRequest = mutable.LinkedHashMap[ApiKeys, AbstractRequest](
ApiKeys.DELETE_TOPICS -> deleteTopicsWithIdsRequest(id),
)

for ((key, request) <- requestKeyToRequest) {
removeAllClientAcls()
val resources = requestKeysToAcls(key).map(_._1.resourceType).toSet
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = false, topicExists = false, describeAuthorized = false, id = id)

val resourceToAcls = requestKeysToAcls(key)
resourceToAcls.get(topicResource).foreach { acls =>
val describeAcls = topicDescribeAcl(topicResource)
val isAuthorized = describeAcls == acls
addAndVerifyAcls(describeAcls, topicResource)
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = isAuthorized, topicExists = false, describeAuthorized = true, id = id)
removeAllClientAcls()
}

for ((resource, acls) <- resourceToAcls)
addAndVerifyAcls(acls, resource)
sendRequestWithIdAndVerifyResponseError(request, resources, isAuthorized = true, topicExists = false, describeAuthorized = true, id = id)
}
}

@Test
def testCreateTopicAuthorizationWithClusterCreate(): Unit = {
removeAllClientAcls()
Expand Down Expand Up @@ -2058,44 +2062,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
}
}

private def sendRequestWithIdAndVerifyResponseError(request: AbstractRequest,
resources: Set[ResourceType],
isAuthorized: Boolean,
topicExists: Boolean,
describeAuthorized: Boolean,
id: Uuid): AbstractResponse = {
val apiKey = request.apiKey
val response = connectAndReceive[AbstractResponse](request)
val error = requestKeysToErrorWithIds(id)(apiKey).asInstanceOf[AbstractResponse => Errors](response)

val authorizationErrors = resources.flatMap { resourceType =>
if (resourceType == TOPIC) {
if (isAuthorized)
Set(Errors.UNKNOWN_TOPIC_ID, AclEntry.authorizationError(ResourceType.TOPIC))
else if (describeAuthorized)
Set(AclEntry.authorizationError(ResourceType.TOPIC))
else
Set(Errors.UNKNOWN_TOPIC_ID)
} else {
Set(AclEntry.authorizationError(resourceType))
}
}

if (topicExists)
if (isAuthorized)
assertFalse(authorizationErrors.contains(error), s"$apiKey should be allowed. Found unexpected authorization error $error")
else
assertTrue(authorizationErrors.contains(error), s"$apiKey should be forbidden. Found error $error but expected one of $authorizationErrors")
else if (resources == Set(TOPIC))
if (isAuthorized)
assertEquals(Errors.UNKNOWN_TOPIC_ID, error, s"$apiKey had an unexpected error")
else {
assertEquals(Errors.UNKNOWN_TOPIC_ID, error, s"$apiKey had an unexpected error")
}

response
}

private def sendRequestAndVerifyResponseError(request: AbstractRequest,
resources: Set[ResourceType],
isAuthorized: Boolean,
Expand Down
Loading