You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2016/07/31 09:02:41 UTC
kafka git commit: KAFKA-3950;
Only throw authorization exception if pattern subscription matches
topic
Repository: kafka
Updated Branches:
refs/heads/0.10.0 77ced78f1 -> 44ad3ec77
KAFKA-3950; Only throw authorization exception if pattern subscription matches topic
Author: Manikumar Reddy O <ma...@gmail.com>
Author: Ismael Juma <gi...@juma.me.uk>
Reviewers: Ashish Singh <as...@cloudera.com>, Sriharsha Chintalapani <sc...@hortonworks.com>, Jason Gustafson <ja...@confluent.io>, Ismael Juma <is...@juma.me.uk>
Closes #1687 from omkreddy/KAFKA-3950
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/44ad3ec7
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/44ad3ec7
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/44ad3ec7
Branch: refs/heads/0.10.0
Commit: 44ad3ec77364fab869cb5e01a182341bd0ae2698
Parents: 77ced78
Author: Manikumar Reddy O <ma...@gmail.com>
Authored: Sun Jul 31 10:01:45 2016 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Sun Jul 31 10:01:45 2016 +0100
----------------------------------------------------------------------
.../consumer/internals/ConsumerCoordinator.java | 22 ++++-
.../kafka/api/AuthorizerIntegrationTest.scala | 98 +++++++++++++++++++-
2 files changed, 110 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/44ad3ec7/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 2880efc..59f3250 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -146,20 +146,27 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
this.metadata.addListener(new Metadata.Listener() {
@Override
public void onMetadataUpdate(Cluster cluster) {
- // if we encounter any unauthorized topics, raise an exception to the user
- if (!cluster.unauthorizedTopics().isEmpty())
- throw new TopicAuthorizationException(new HashSet<>(cluster.unauthorizedTopics()));
if (subscriptions.hasPatternSubscription()) {
+
+ Set<String> unauthorizedTopics = new HashSet<String>();
+ for (String topic : cluster.unauthorizedTopics()) {
+ if (filterTopic(topic))
+ unauthorizedTopics.add(topic);
+ }
+ if (!unauthorizedTopics.isEmpty())
+ throw new TopicAuthorizationException(unauthorizedTopics);
+
final List<String> topicsToSubscribe = new ArrayList<>();
for (String topic : cluster.topics())
- if (subscriptions.getSubscribedPattern().matcher(topic).matches() &&
- !(excludeInternalTopics && TopicConstants.INTERNAL_TOPICS.contains(topic)))
+ if (filterTopic(topic))
topicsToSubscribe.add(topic);
subscriptions.changeSubscription(topicsToSubscribe);
metadata.setTopics(subscriptions.groupSubscription());
+ } else if (!cluster.unauthorizedTopics().isEmpty()) {
+ throw new TopicAuthorizationException(new HashSet<>(cluster.unauthorizedTopics()));
}
// check if there are any changes to the metadata which should trigger a rebalance
@@ -175,6 +182,11 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
});
}
+ private boolean filterTopic(String topic) {
+ return subscriptions.getSubscribedPattern().matcher(topic).matches() &&
+ !(excludeInternalTopics && TopicConstants.INTERNAL_TOPICS.contains(topic));
+ }
+
private PartitionAssignor lookupAssignor(String name) {
for (PartitionAssignor assignor : this.assignors) {
if (assignor.name().equals(name))
http://git-wip-us.apache.org/repos/asf/kafka/blob/44ad3ec7/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
index 60eb74c..59ead2f 100644
--- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
@@ -16,6 +16,7 @@ import java.io.{DataInputStream, DataOutputStream}
import java.net.Socket
import java.nio.ByteBuffer
import java.util.concurrent.ExecutionException
+import java.util.regex.Pattern
import java.util.{ArrayList, Collections, Properties}
import kafka.cluster.EndPoint
@@ -25,7 +26,8 @@ import kafka.integration.KafkaServerTestHarness
import kafka.security.auth._
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
-import org.apache.kafka.clients.consumer.{Consumer, ConsumerRecord, KafkaConsumer, OffsetAndMetadata}
+import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener
+import org.apache.kafka.clients.consumer._
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.errors._
import org.apache.kafka.common.protocol.{ApiKeys, Errors, SecurityProtocol}
@@ -42,6 +44,7 @@ import org.apache.kafka.common.internals.TopicConstants
class AuthorizerIntegrationTest extends KafkaServerTestHarness {
val topic = "topic"
+ val topicPattern = "topic.*"
val part = 0
val brokerId: Integer = 0
val correlationId = 0
@@ -353,7 +356,7 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness {
consumeRecords(this.consumers.head)
Assert.fail("should have thrown exception")
} catch {
- case e: TopicAuthorizationException => assertEquals(Collections.singleton(topic), e.unauthorizedTopics());
+ case e: TopicAuthorizationException => assertEquals(Collections.singleton(topic), e.unauthorizedTopics())
}
}
@@ -370,7 +373,7 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness {
consumeRecords(this.consumers.head)
Assert.fail("should have thrown exception")
} catch {
- case e: TopicAuthorizationException => assertEquals(Collections.singleton(topic), e.unauthorizedTopics());
+ case e: TopicAuthorizationException => assertEquals(Collections.singleton(topic), e.unauthorizedTopics())
}
}
@@ -388,7 +391,7 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness {
Assert.fail("should have thrown exception")
} catch {
case e: TopicAuthorizationException =>
- assertEquals(Collections.singleton(topic), e.unauthorizedTopics());
+ assertEquals(Collections.singleton(topic), e.unauthorizedTopics())
}
}
@@ -405,6 +408,91 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness {
}
@Test
+ def testPatternSubscriptionWithNoTopicAccess() {
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), topicResource)
+ sendRecords(1, tp)
+ removeAllAcls()
+
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+ try {
+ this.consumers.head.subscribe(Pattern.compile(topicPattern), new NoOpConsumerRebalanceListener)
+ this.consumers.head.poll(50)
+ Assert.fail("should have thrown exception")
+ } catch {
+ case e: TopicAuthorizationException => assertEquals(Collections.singleton(topic), e.unauthorizedTopics())
+ }
+ }
+
+ @Test
+ def testPatternSubscriptionWithTopicAndGroupRead() {
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), topicResource)
+ sendRecords(1, tp)
+
+ //create a unmatched topic
+ val unmatchedTopic = "unmatched"
+ TestUtils.createTopic(zkUtils, unmatchedTopic, 1, 1, this.servers)
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), new Resource(Topic, unmatchedTopic))
+ sendRecords(1, new TopicPartition(unmatchedTopic, part))
+ removeAllAcls()
+
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), topicResource)
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+ val consumer = consumers.head
+ consumer.subscribe(Pattern.compile(topicPattern), new NoOpConsumerRebalanceListener)
+ consumeRecords(consumer)
+
+ // 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
+ consumer.subscribe(Pattern.compile(TopicConstants.GROUP_METADATA_TOPIC_NAME), new NoOpConsumerRebalanceListener)
+ assertTrue(consumer.poll(50).isEmpty)
+ }
+
+
+ @Test
+ def testPatternSubscriptionMatchingInternalTopicWithNoPermission() {
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), topicResource)
+ sendRecords(1, tp)
+ removeAllAcls()
+
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), topicResource)
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+
+ val consumerConfig = new Properties
+ consumerConfig.put(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG, "false")
+ val consumer = TestUtils.createNewConsumer(TestUtils.getBrokerListStrFromServers(servers), groupId = group,
+ securityProtocol = SecurityProtocol.PLAINTEXT, props = Some(consumerConfig))
+ try {
+ consumer.subscribe(Pattern.compile(".*"), new NoOpConsumerRebalanceListener)
+ consumeRecords(consumer)
+ fail("should have thrown exception")
+ }
+ catch {
+ case e: TopicAuthorizationException =>
+ assertEquals(Collections.singleton(TopicConstants.GROUP_METADATA_TOPIC_NAME), e.unauthorizedTopics())
+ } finally consumer.close()
+ }
+
+ @Test
+ def testPatternSubscriptionNotMatchingInternalTopic() {
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), topicResource)
+ sendRecords(1, tp)
+ removeAllAcls()
+
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), topicResource)
+ addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Read)), groupResource)
+
+ val consumerConfig = new Properties
+ consumerConfig.put(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG, "false")
+ val consumer = TestUtils.createNewConsumer(TestUtils.getBrokerListStrFromServers(servers), groupId = group,
+ securityProtocol = SecurityProtocol.PLAINTEXT, props = Some(consumerConfig))
+ try {
+ consumer.subscribe(Pattern.compile(topicPattern), new NoOpConsumerRebalanceListener)
+ consumeRecords(consumer)
+ } finally consumer.close()
+ }
+
+ @Test
def testCreatePermissionNeededToReadFromNonExistentTopic() {
val newTopic = "newTopic"
val topicPartition = new TopicPartition(newTopic, 0)
@@ -418,7 +506,7 @@ class AuthorizerIntegrationTest extends KafkaServerTestHarness {
Assert.fail("should have thrown exception")
} catch {
case e: TopicAuthorizationException =>
- assertEquals(Collections.singleton(newTopic), e.unauthorizedTopics());
+ assertEquals(Collections.singleton(newTopic), e.unauthorizedTopics())
}
addAndVerifyAcls(Set(new Acl(KafkaPrincipal.ANONYMOUS, Allow, Acl.WildCardHost, Write)), newTopicResource)