From 9449f055c7a0b340a8d69d7365c5817464b2f6ed Mon Sep 17 00:00:00 2001 From: Dhruvil Shah Date: Fri, 20 Jul 2018 12:35:12 -0700 Subject: [PATCH] KAFKA-7185: Allow empty resource name when matching ACLs (#5400) Reviewers: Ismael Juma , Rajini Sivaram --- .../kafka/security/auth/SimpleAclAuthorizer.scala | 2 +- .../security/auth/SimpleAclAuthorizerTest.scala | 13 +++++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala index 55352584c26..e77656d748c 100644 --- a/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala +++ b/core/src/main/scala/kafka/security/auth/SimpleAclAuthorizer.scala @@ -238,7 +238,7 @@ class SimpleAclAuthorizer extends Authorizer with Logging { val prefixed = aclCache.range( Resource(resourceType, resourceName, PatternType.PREFIXED), - Resource(resourceType, resourceName.substring(0, 1), PatternType.PREFIXED) + Resource(resourceType, resourceName.take(1), PatternType.PREFIXED) ) .filterKeys(resource => resourceName.startsWith(resource.name)) .flatMap { case (resource, versionedAcls) => versionedAcls.acls } diff --git a/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala b/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala index 5b65a7f2586..5461413871b 100644 --- a/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala +++ b/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala @@ -92,6 +92,19 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness { simpleAclAuthorizer.authorize(session, Read, Resource(Topic, "something", PREFIXED)) } + @Test + def testAuthorizeWithEmptyResourceName(): Unit = { + assertFalse(simpleAclAuthorizer.authorize(session, Read, Resource(Group, "", LITERAL))) + simpleAclAuthorizer.addAcls(Set[Acl](allowReadAcl), Resource(Group, WildCardResource, LITERAL)) + assertTrue(simpleAclAuthorizer.authorize(session, Read, Resource(Group, "", LITERAL))) + } + + // Authorizing the empty resource is not supported because we create a znode with the resource name. + @Test(expected = classOf[IllegalArgumentException]) + def testEmptyAclThrowsException(): Unit = { + simpleAclAuthorizer.addAcls(Set[Acl](allowReadAcl), Resource(Group, "", LITERAL)) + } + @Test def testTopicAcl() { val user1 = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)