KAFKA-3396 : Unauthorized topics are returned to the user by edoardocomar · Pull Request #1428 · apache/kafka

edoardocomar · 2016-05-25T16:53:24Z

Modified KafkaApis to return Errors.UNKNOWN_TOPIC_OR_PARTITION if
principal has no Describe access to topic

Unit tests expanded

Some paths cause the client to block due to bug
https://issues.apache.org/jira/browse/KAFKA-3727?filter=-2
tests work around this by executing in separate thread

edoardocomar · 2016-05-25T16:55:45Z

Hi this work has been done with @mimaison.

Please note that we had issues with the tests because of
https://issues.apache.org/jira/browse/KAFKA-3727
"Consumer.poll() stuck in loop on non-existent topic manually assigned"
and
https://issues.apache.org/jira/browse/KAFKA-3728

ijuma · 2016-06-04T00:02:39Z

Can you please rebase this now that #1425 has been merged?

edoardocomar · 2016-06-06T11:56:00Z

Thanks @ijuma, I did;
some tests still look ugly because of https://issues.apache.org/jira/browse/KAFKA-3727

ijuma · 2016-06-06T12:03:14Z

@hachikuji Thoughts on how to handle this? It seems to me that we can't really fix this without fixing KAFKA-3727.

edoardocomar · 2016-06-06T12:26:37Z

@ijuma I'd be happy to work on KAFKA-3727, but I think we need a bit of discussion on what the expected behavior should be. My comments are in that Jira issue https://issues.apache.org/jira/browse/KAFKA-3727

hachikuji · 2016-06-06T16:34:15Z

@ijuma Seems we need to decide how the client should handle non-existent topics both before and after KIP-4. I can see a case for both retrying and raising an exception. I hate to suggest it, but perhaps the behavior should be configurable?

mimaison · 2016-06-08T10:31:09Z

@hachikuji I agree we can probably find use cases that justify both behaviours. What surprised us mostly was the difference between assign and subscribe.

Also for the looping behaviour, it would be nice to be able to control it with a max timeout or retries, as having to wakeup() from another thread is not very nice. In the end it sounds like a lot of new options so it might not be ideal.

hachikuji · 2016-06-08T16:59:00Z

@mimaison Can you clarify what surprised about the difference between assign and subscribe?

edoardocomar · 2016-06-08T17:19:00Z

@hachikuji as we wrote in https://issues.apache.org/jira/browse/KAFKA-3727

The behavior of a consumer on poll() for a non-existing topic is surprisingly different/inconsistent
between a consumer that subscribed to the topic and one that had the topic-partition manually assigned.
The "subscribed" consumer will return an empty collection
The "assigned" consumer will loop forever - this feels a bug to me.

hachikuji · 2016-06-08T18:07:26Z

@edoardocomar Ah, I missed that. I agree the behavior is a little surprising, but it makes sense when I think of the implementation. We currently block setting the initial position of each partition. When using subscribe(), it's not possible to be assigned a non-existing partition so this is never a problem. But obviously a manually assigned partition for a non-existent topic is a non-existing partition, and we'll block setting the position. So to fix the inconsistency, we'll have to think through the implications of removing that blocking.

edoardocomar · 2016-06-10T13:10:59Z

@hachikuji @ijuma why should a missing topic or partition be a retriable exception
I am thinking at least of the case where auto-create is turned off.

hachikuji · 2016-06-14T03:46:03Z

@edoardocomar Sorry for the slow response. The consumer has traditionally supported the ability to be started before any producers and that means retrying on unknown topics. Even if we had some way to inspect whether autocreate has been disabled, we probably still have to support this for compatibility. I'm also thinking about the impact of the CreateTopic API in KIP-4 (which is under discussion right now). The most likely eventual outcome is that autocreation will be taken out of the broker and pushed into the producer, so it seems that the picture doesn't really change for the consumer, but we should consider whether this outcome is ideal.

One option for the problem at hand is to keep the current behavior (i.e. do nothing). Users can use the partitionsFor() API to verify the existence of any needed topics and raise an exception themselves if they expect it to exist ahead of time. Alternatively, we would probably need a separate configuration that lets the consumer raise an exception instead of retrying. This would call for a new KIP. Topic deletion may be an interesting case to think about as well since the partitionsFor workaround may not help (I'm not too sure).

All that said, I don't think we need to block this issue on a resolution to that problem. It doesn't seem like a big difference to me whether we retry indefinitely because the user is not authorized to the topic or because we are waiting for a topic that may never be created. In either case, the user ultimately has to monitor the progress of the application and take steps when it is failing or falling behind.

ijuma · 2016-06-14T07:48:47Z

My concern is that configuring security is challenging enough as it is. Configuring authorization incorrectly is very common and harder to diagnose than a topic that hasn't been created (even the latter sometimes causes people to be confused and waste time).

Replacing an exception with indefinite blocking is making the user experience during development even worse than it is at the moment (which is already suboptimal in a number of situations). Ideally, we would be moving in the other direction.

If we decide to do this, then at the very least, we need to think about how users will diagnose the problem and update the authorization documentation with the information.

hachikuji · 2016-06-14T17:30:08Z

@ijuma At the very least, we can change the consumer/producer log messages to warn the user that the topic might not exist or they might not have access to it. I agree this is less than ideal, but I don't see an alternative other than what I suggested above (changing how the client handles unknown topics). That is, unless we think this patch is not worth having at all. I was a little torn on this initially, but it does seem better to avoid leaking topic information.

ijuma · 2016-06-14T17:43:43Z

@hachikuji The log warning you have suggested seems like our best bet at the moment.

edoardocomar · 2016-06-22T13:52:13Z

Hi @ijuma @hachikuji
we have made a PR #1535
to address the non existent topic or partition issue https://issues.apache.org/jira/browse/KAFKA-3727

ijuma · 2016-07-14T07:25:30Z

core/src/main/scala/kafka/server/KafkaApis.scala

I think we need to handle the all topics case specially:

val topics = // Handle old metadata request logic. Version 0 has no way to specify "no topics". if (requestVersion == 0) { if (metadataRequest.topics() == null || metadataRequest.topics().isEmpty) metadataCache.getAllTopics() else metadataRequest.topics.asScala.toSet } else { if (metadataRequest.isAllTopics) metadataCache.getAllTopics() else metadataRequest.topics.asScala.toSet }

For that case, we want to filter the topics that the user doesn't have DESCRIBE access to.

hachikuji · 2016-08-01T17:18:07Z

core/src/main/scala/kafka/server/KafkaApis.scala

Not part of this patch, but the authorize function seems to already handle this, so could we replace this and the if below with a call to authorize(request.session, Create, Resource.ClusterResource)?

Can you fix this please? Is there any reason not to?

done thanks

mimaison · 2016-08-17T13:28:06Z

Thanks for all the feedback ! I've updated the PR

mimaison · 2016-09-08T09:49:20Z

@ijuma @hachikuji I believe we've addressed all the comments, can you have another look ?

ijuma · 2016-09-08T09:51:31Z

Sorry for the delay @mimaison and thanks for the reminder. Will try to look at this soon.

ijuma · 2016-09-20T09:45:47Z

@mimaison, with the feature freeze in place, this is one of the main bugs we want to get in for 0.10.1.0. Could you please merge trunk into this branch? KIP-79 made some changes to the list offsets request so it may need some updates. Sorry about that. :(

edoardocomar · 2016-09-20T15:11:58Z

@ijuma @hachikuji - in hindsight, shouldn't we also handle an offsetRequest not to return TOPIC_AUTHORIZATION_FAILED where Describe auth is checked. e.g.

private def handleOffsetRequestV0(request : RequestChannel.Request) ... {
...
    val (authorizedRequestInfo, unauthorizedRequestInfo) = offsetRequest.offsetData.asScala.partition {
      case (topicPartition, _) => authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic))
    }

    val unauthorizedResponseStatus = unauthorizedRequestInfo.mapValues(_ =>
      new PartitionData(Errors.TOPIC_AUTHORIZATION_FAILED.code, List[JLong]().asJava)
    )

it seems we had missed it

Modified KafkaApis to return Errors.UNKNOWN_TOPIC_OR_PARTITION if principal has no Describe access to topic Unit tests expanded Some paths cause the client to block due to bug https://issues.apache.org/jira/browse/KAFKA-3727?filter=-2 tests work around this by executing in separate thread

- Added tests from 44ad3ec - Small refactorings

Rebased after kip-79 changes. Fixing leak of topic for LIST_OFFSETS when unauthorized. Added tests.

edoardocomar · 2016-09-21T12:38:05Z

@ijuma @hachikuji the Jenkins build failure
https://builds.apache.org/job/kafka-trunk-git-pr-jdk7/5818/
looks unrelated to our changes. Can you please review ? thanks

I'm forcing a new rebuild by rebasing on trunk rather than 0.10.1

ijuma · 2016-09-21T12:51:16Z

@edoardocomar, we should, yes. I think that was "missed" because that code was moved as part of KIP-79, but I haven't checked in detail yet. Will check the PR today or tomorrow. Thanks!

edoardocomar · 2016-09-21T16:18:45Z

@ijuma Jenkins has failed with a timeout, the tests locally pass for me can you please kick Jenkins again ? thanks

edoardocomar · 2016-09-22T21:46:40Z

@ijuma @hachikuji can you please take a look at the revised PR? thanks

hachikuji

Thanks for the patch! I left some comments (mostly nitpicks). I do think it's worthwhile treating UNKNOWN_TOPIC_OR_PARTITION explicitly in OffsetFetch and OffsetCommit response handlers on the client though.

hachikuji · 2016-09-22T22:19:19Z

core/src/main/scala/kafka/server/KafkaApis.scala

-      // filter non-existent topics
-      val invalidRequestsInfo = offsetCommitRequest.offsetData.asScala.filter { case (topicPartition, _) =>
-        !metadataCache.contains(topicPartition.topic)
+


nit: unneeded newline?

hachikuji · 2016-09-22T22:22:47Z

core/src/main/scala/kafka/server/KafkaApis.scala

-        !metadataCache.contains(topicPartition.topic)
+
+      val (existingOrAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = offsetCommitRequest.offsetData.asScala.toMap.partition {
+        case (topicPartition, _) => metadataCache.contains(topicPartition.topic) && authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic))


nit: not much difference here, but it seems more natural to check authorization first.

MetadataCache has two methods that do the same thing: contains and hasTopicMetadata. We should remove one of them to avoid confusion. Also, we call hasTopicMetadata in the block for if (header.apiVersion == 0), which seems unnecessary as we have already done it here (it doesn't prevent potential race conditions either although it does narrow the window a little).

hachikuji · 2016-09-22T22:36:57Z

core/src/main/scala/kafka/server/KafkaApis.scala

+      new MetadataResponse.TopicMetadata(Errors.TOPIC_AUTHORIZATION_FAILED, topic, common.Topic.isInternal(topic),
        java.util.Collections.emptyList()))

+    // do not disclose the existence of unauthorized topics


nit: should be obvious, but we may as well emphasize unauthorized for Describe.

hachikuji · 2016-09-22T22:38:53Z

core/src/main/scala/kafka/server/KafkaApis.scala

+    // do not disclose the existence of unauthorized topics
+    val unauthorizedForDescribeTopicMetadata =
+      // In case of all topics, don't include unauthorized topics
+      if ((requestVersion == 0 && (metadataRequest.topics() == null || metadataRequest.topics().isEmpty)) || (metadataRequest.isAllTopics))


nit: Doesn't seem like the parenthesis around metadataRequest.isAllTopics are necessary.

hachikuji · 2016-09-22T23:12:15Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+      if (resourceType == Topic)
+        if (apiKey == ApiKeys.PRODUCE || apiKey == ApiKeys.FETCH ) {
+          //Only allowing TOPIC_AUTHORIZATION_FAILED as an error code
+          Seq(resourceType.errorCode)


Might be clearer to reference the TOPIC_AUTHORIZATION error code directly?

I think this is not right. Both fetch and produce should also return UNKNOWN_TOPIC_OR_PARTITION if given a topic that the user doesn't have DESCRIBE permission for, right? It seems like neither checks if the user has DESCRIBE permission, which seems to defeat the point of this PR.

addressed thanks

hachikuji · 2016-09-22T23:21:24Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

-      for (record <- consumer.poll(50).asScala) {
-        records.add(record)
+
+    val future = Future {


It's a little weird here to bound this loop by the number of iterations. Could we refactor it to use waitUntil? For example:

TestUtils.waitUntilTrue(() => { for (record <- consumer.poll(50).asScala) records.add(record) records.size >= numRecords })

Either that or use the timeout directly in the loop instead of maxIters?

hachikuji · 2016-09-22T23:24:59Z

core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala


-  /**
+  @Test
+  def testNoProduceAclWithDescribeAcl {


nit: The first "Acl" seems out of place. Should this be testNoProduceWithDescribeAcl? A few more like this below.

hachikuji · 2016-09-22T23:28:36Z

core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala

+    }
+  }
+
+  private def noConsumeAclWithDescribeAclProlog {


nit: What is "Prolog"?

a Setup step. renamed

hachikuji · 2016-09-22T23:29:43Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

    val producerProps = props.getOrElse(new Properties)
    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
    producerProps.put(ProducerConfig.ACKS_CONFIG, acks.toString)
-    producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs.toString)


Why was this removed?

To be able to set it to a low value in EndToEndAuthorizationTest
to allow some test to finish quickly

This change causes the parameter of createNewProducer to be ignored affecting a bunch of other tests though. It seems like the easiest thing is to add a couple of protected methods in IntegrationTestHarness: createProducer and createConsumer. You can then override that in EndToEndAuthorizationTest.

good point, done

hachikuji · 2016-09-22T23:30:25Z

core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala

-      for (record <- consumer.poll(50).asScala) {
-        records.add(record)
+
+    val future = Future {


Same as above. Seems like we could make this a more conventional timeout loop.

mimaison · 2016-09-23T15:46:56Z

Thanks for the feedback, we'll update the PR on Monday.

ijuma

Thanks for your patience. I left a few more comments. The most important ones are related to how we don't seem to check DESCRIBE permissions for fetch and produce.

ijuma · 2016-09-24T22:02:55Z

core/src/main/scala/kafka/server/KafkaApis.scala

-      val invalidRequestsInfo = offsetCommitRequest.offsetData.asScala.filter { case (topicPartition, _) =>
-        !metadataCache.contains(topicPartition.topic)
+
+      val (existingOrAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = offsetCommitRequest.offsetData.asScala.toMap.partition {


The first variable should be existingAndAuthorizedForDescribeTopics.

ijuma · 2016-09-24T22:17:17Z

core/src/main/scala/kafka/server/KafkaApis.scala

      // the callback for sending an offset commit response
      def sendResponseCallback(commitStatus: immutable.Map[TopicPartition, Short]) {
-        val mergedCommitStatus = commitStatus ++ unauthorizedRequestInfo.mapValues(_ => Errors.TOPIC_AUTHORIZATION_FAILED.code)
+        var combinedCommitStatus = commitStatus.mapValues(new JShort(_)) ++


Should be a val.

ijuma · 2016-09-24T22:36:29Z

core/src/main/scala/kafka/server/KafkaApis.scala

-        !metadataCache.contains(topicPartition.topic)
+
+      val (existingOrAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = offsetCommitRequest.offsetData.asScala.toMap.partition {
+        case (topicPartition, _) => metadataCache.contains(topicPartition.topic) && authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic))


MetadataCache has two methods that do the same thing: contains and hasTopicMetadata. We should remove one of them to avoid confusion. Also, we call hasTopicMetadata in the block for if (header.apiVersion == 0), which seems unnecessary as we have already done it here (it doesn't prevent potential race conditions either although it does narrow the window a little).

ijuma · 2016-09-24T23:38:42Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+
+    addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+    this.consumers.head.subscribe(Pattern.compile(topicPattern), new NoOpConsumerRebalanceListener)
+    this.consumers.head.poll(50)


No need for () in subscription and isEmpty.

ijuma · 2016-09-24T23:44:58Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

    consumeRecords(this.consumers.head)
  }

+  @Test


We should extend the pattern subscription tests to also include the case where the user has DESCRIBE, but not READ permission. In those cases, we should still get TopicAuthorizationException.

done thanks

ijuma · 2016-09-24T23:55:06Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+
+    // set the subscription pattern to an internal topic that the consumer has no read permission for, but since
+    // `exclude.internal.topics` is true by default, the subscription should be empty and no authorization exception
+    // should be thrown


Given the implementation in trunk, this comment isn't really helpful anymore. I would add an additional case where the consumer has DESCRIBE, but no READ permission for the internal topic. And then move this comment there (and tweak it slightly).

ijuma · 2016-09-24T23:55:57Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+    try {
+      consumer.subscribe(Pattern.compile(".*"), new NoOpConsumerRebalanceListener)
+      consumeRecords(consumer)
+    } catch {


Why do we get this instead of an empty subscription?

oops, fixed test

ijuma · 2016-09-25T00:13:20Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

+      if (resourceType == Topic)
+        if (apiKey == ApiKeys.PRODUCE || apiKey == ApiKeys.FETCH ) {
+          //Only allowing TOPIC_AUTHORIZATION_FAILED as an error code
+          Seq(resourceType.errorCode)


I think this is not right. Both fetch and produce should also return UNKNOWN_TOPIC_OR_PARTITION if given a topic that the user doesn't have DESCRIBE permission for, right? It seems like neither checks if the user has DESCRIBE permission, which seems to defeat the point of this PR.

edoardocomar · 2016-09-26T18:15:00Z

@ijuma We have handled the FETCH and PRODUCE commands

edoardocomar · 2016-09-26T18:15:39Z

As I have squashed some commits, I am opening a new PR

Reopening of #1428 Author: Edoardo Comar <ecomar@uk.ibm.com> Author: Mickael Maison <mickael.maison@gmail.com> Reviewers: Grant Henke <granthenke@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io> Closes #1908 from edoardocomar/KAFKA-3396

Reopening of #1428 Author: Edoardo Comar <ecomar@uk.ibm.com> Author: Mickael Maison <mickael.maison@gmail.com> Reviewers: Grant Henke <granthenke@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io> Closes #1908 from edoardocomar/KAFKA-3396 (cherry picked from commit 8124f6e) Signed-off-by: Jason Gustafson <jason@confluent.io>

edoardocomar force-pushed the KAFKA-3396 branch 2 times, most recently from b188a46 to 0b12cb4 Compare May 26, 2016 19:51

edoardocomar force-pushed the KAFKA-3396 branch from 0b12cb4 to 1e67a4f Compare June 6, 2016 11:54

ijuma reviewed Jul 14, 2016
View reviewed changes

hachikuji reviewed Aug 1, 2016
View reviewed changes

mimaison force-pushed the KAFKA-3396 branch from 184726a to 2f7a96c Compare August 17, 2016 13:23

edoardocomar force-pushed the KAFKA-3396 branch from 2f7a96c to 2de248f Compare September 21, 2016 10:52

edoardocomar and others added 4 commits September 21, 2016 13:24

KAFKA-3396: Updates and cleanups following the feedback

cce5726

KAFKA-3396: More updates + rebased off trunk

c7aefe2

- Added tests from 44ad3ec - Small refactorings

KAFKA-3396 : Unauthorized topics are returned to the user

cf464fa

Rebased after kip-79 changes. Fixing leak of topic for LIST_OFFSETS when unauthorized. Added tests.

edoardocomar force-pushed the KAFKA-3396 branch from 2de248f to cf464fa Compare September 21, 2016 12:43

hachikuji reviewed Sep 22, 2016

View reviewed changes

edoardocomar closed this Sep 23, 2016

ijuma reviewed Sep 25, 2016

View reviewed changes

edoardocomar mentioned this pull request Sep 26, 2016

KAFKA-3396 : Unauthorized topics are returned to the user #1908

Closed

efeg added a commit to efeg/kafka that referenced this pull request May 29, 2024

Add 2021 to LICENSE (apache#1428)

3ec095e

Conversation

edoardocomar commented May 25, 2016

Uh oh!

edoardocomar commented May 25, 2016

Uh oh!

ijuma commented Jun 4, 2016

Uh oh!

edoardocomar commented Jun 6, 2016

Uh oh!

ijuma commented Jun 6, 2016

Uh oh!

edoardocomar commented Jun 6, 2016

Uh oh!

hachikuji commented Jun 6, 2016

Uh oh!

mimaison commented Jun 8, 2016

Uh oh!

hachikuji commented Jun 8, 2016

Uh oh!

edoardocomar commented Jun 8, 2016

Uh oh!

hachikuji commented Jun 8, 2016

Uh oh!

edoardocomar commented Jun 10, 2016

Uh oh!

hachikuji commented Jun 14, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ijuma commented Jun 14, 2016

Uh oh!

hachikuji commented Jun 14, 2016

Uh oh!

ijuma commented Jun 14, 2016

Uh oh!

edoardocomar commented Jun 22, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Sep 15, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mimaison commented Aug 17, 2016

Uh oh!

mimaison commented Sep 8, 2016

Uh oh!

ijuma commented Sep 8, 2016

Uh oh!

ijuma commented Sep 20, 2016

Uh oh!

edoardocomar commented Sep 20, 2016

Uh oh!

edoardocomar commented Sep 21, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ijuma commented Sep 21, 2016

Uh oh!

edoardocomar commented Sep 21, 2016

Uh oh!

edoardocomar commented Sep 22, 2016

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Sep 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma Sep 24, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Jun 14, 2016 •

edited

Loading

hachikuji Sep 15, 2016 •

edited

Loading

edoardocomar commented Sep 21, 2016 •

edited

Loading

hachikuji Sep 22, 2016 •

edited

Loading

ijuma Sep 24, 2016 •

edited

Loading

hachikuji Sep 22, 2016 •

edited

Loading

hachikuji Sep 22, 2016 •

edited

Loading