You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "tinaselenge (via GitHub)" <gi...@apache.org> on 2024/02/15 16:15:05 UTC
[PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
tinaselenge opened a new pull request, #15377:
URL: https://github.com/apache/kafka/pull/15377
This makes the test simpler to enable KRaft later.
It's unclear why a custom authoriser had to be used in the first place based on the commit made this change and tests seem to work without it. The initial ACLs required by the tests are created with a super admin client instead of `AclAuthorizationAdmin` which is part of the custom authorizer.
`IDEMPOTENT_WRITE` was added to the initial ACLs created in the SASL setup, because SSL and BaseAdmin tests that being extended or extending this class do not override the `configuredClusterPermissions` method therefore end up not matching the expected permissions set on the cluster initially. Overriding this method does not seem necessary otherwise.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1549736126
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
Review Comment:
This was necessary because admin client is created with different security configurations for SSL tests than SASL and Plain. We could remove this and override the createConfig() method instead but the override is needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1581029496
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
Review Comment:
So I initially thought the `createConfig()` in BaseAdminIntegrationTest is creating different security properties by calling `TestUtils.adminClientSecurityConfigs()` than the security properties created as part of the set up in IntegrationTestHarness. However following all the properties `adminClientSecurityConfigs()` set, they are exactly the same except the `certAlias`. So basically BaseAdminIntegrationTest mostly just setting security properties twice for no good reason. This seems to upset SSL admin test, because of the mismatch in the certAlias.
Removing the unnecessary call to `adminClientSecurityConfigs()` in BaseAdminIntegrationTest worked for all of the tests.
Thanks for pointing this out!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1560822590
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
Review Comment:
I'm still not sure I understand. The `createConfig()` method is in `BaseAdminIntegrationTest` so it should create the right configuration for the running test automatically. Why is it able to correctly create the configuration when running with SASL_SSL but not with SSL?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1601900093
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1776,6 +1776,33 @@ object TestUtils extends Logging {
45000)
}
+
+ def waitAndVerifyAcl(expected: AccessControlEntry,
+ authorizer: JAuthorizer,
+ resource: ResourcePattern,
+ accessControlEntryFilter: AccessControlEntryFilter = AccessControlEntryFilter.ANY): Unit = {
+ val newLine = scala.util.Properties.lineSeparator
+
+ val filter = new AclBindingFilter(resource.toFilter, accessControlEntryFilter)
+ waitUntilTrue(() => authorizer.acls(filter).asScala.map(_.entry).toSet.contains(expected),
+ s"expected to contain acl: " + expected +
Review Comment:
We can do `s"expected to contain acl: $expected "`, otherwise remove the `s` interpolator in front of the string.
Same in `waitAndVerifyRemovedAcl()` below
##########
core/src/test/scala/unit/kafka/utils/TestUtils.scala:
##########
@@ -1776,6 +1776,33 @@ object TestUtils extends Logging {
45000)
}
+
+ def waitAndVerifyAcl(expected: AccessControlEntry,
+ authorizer: JAuthorizer,
+ resource: ResourcePattern,
+ accessControlEntryFilter: AccessControlEntryFilter = AccessControlEntryFilter.ANY): Unit = {
Review Comment:
The only caller never specifies this argument, maybe we can remove it? Same below in `waitAndVerifyRemovedAcl()`
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -12,63 +12,89 @@
*/
package kafka.api
-import java.util
import kafka.security.authorizer.AclAuthorizer
import kafka.server.KafkaConfig
-import kafka.utils.{CoreUtils, JaasTestUtils, TestUtils}
import kafka.utils.TestUtils._
+import kafka.utils.{JaasTestUtils, TestUtils}
import org.apache.kafka.clients.admin._
import org.apache.kafka.common.Uuid
-import org.apache.kafka.common.acl._
-import org.apache.kafka.common.acl.AclOperation.{ALL, ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE}
+import org.apache.kafka.common.acl.AclOperation._
import org.apache.kafka.common.acl.AclPermissionType.{ALLOW, DENY}
-import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
+import org.apache.kafka.common.acl._
+import org.apache.kafka.common.config.{ConfigResource, SaslConfigs, TopicConfig}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidRequestException, TopicAuthorizationException, UnknownTopicOrPartitionException}
import org.apache.kafka.common.resource.PatternType.LITERAL
import org.apache.kafka.common.resource.ResourceType.{GROUP, TOPIC}
-import org.apache.kafka.common.resource.{PatternType, Resource, ResourcePattern, ResourcePatternFilter, ResourceType}
+import org.apache.kafka.common.resource._
Review Comment:
Let's keep the explicit imports, same below
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1534098042
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -13,66 +13,94 @@
package kafka.api
import java.util
-import kafka.security.authorizer.AclAuthorizer
+import kafka.security.authorizer.{AclAuthorizer}
import kafka.security.authorizer.AclEntry.{WildcardHost, WildcardPrincipalString}
import kafka.server.KafkaConfig
-import kafka.utils.{CoreUtils, JaasTestUtils, TestUtils}
+import kafka.utils.{JaasTestUtils, TestUtils}
import kafka.utils.TestUtils._
import org.apache.kafka.clients.admin._
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.acl._
-import org.apache.kafka.common.acl.AclOperation.{ALL, ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE}
+import org.apache.kafka.common.acl.AclOperation.{ALL, ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE, IDEMPOTENT_WRITE}
import org.apache.kafka.common.acl.AclPermissionType.{ALLOW, DENY}
-import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
+import org.apache.kafka.common.config.{ConfigResource, SaslConfigs, TopicConfig}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidRequestException, TopicAuthorizationException, UnknownTopicOrPartitionException}
import org.apache.kafka.common.resource.PatternType.LITERAL
import org.apache.kafka.common.resource.ResourceType.{GROUP, TOPIC}
import org.apache.kafka.common.resource.{PatternType, Resource, ResourcePattern, ResourcePatternFilter, ResourceType}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.server.authorizer.Authorizer
+import org.apache.kafka.common.utils.Utils
import org.apache.kafka.storage.internals.log.LogConfig
import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo, Timeout}
-import java.util.Collections
import scala.jdk.CollectionConverters._
import scala.collection.Seq
import scala.compat.java8.OptionConverters._
import scala.concurrent.ExecutionException
import scala.util.{Failure, Success, Try}
+@Timeout(120)
class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetup {
val clusterResourcePattern = new ResourcePattern(ResourceType.CLUSTER, Resource.CLUSTER_NAME, PatternType.LITERAL)
- val authorizationAdmin = new AclAuthorizationAdmin(classOf[AclAuthorizer], classOf[AclAuthorizer])
+ val aclAuthorizerClassName = classOf[AclAuthorizer].getName
Review Comment:
Let's add the type for a public field:
```
val aclAuthorizerClassName: String = classOf[AclAuthorizer].getName
```
##########
core/src/test/scala/integration/kafka/api/BaseAdminIntegrationTest.scala:
##########
@@ -230,6 +230,13 @@ abstract class BaseAdminIntegrationTest extends IntegrationTestHarness with Logg
config
}
+ def createAdminClient: Admin = {
Review Comment:
Could we update PlaintextAdminIntegrationTest to use this method as well?
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -13,66 +13,94 @@
package kafka.api
import java.util
-import kafka.security.authorizer.AclAuthorizer
+import kafka.security.authorizer.{AclAuthorizer}
Review Comment:
We don't need the brackets around `AclAuthorizer`
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
Review Comment:
Could we get rid of this override? We don't have one in the SaslSsl test.
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
+ println("should be passing here")
Review Comment:
This can be deleted now
##########
core/src/test/scala/integration/kafka/api/SaslSslAdminIntegrationTest.scala:
##########
@@ -13,66 +13,94 @@
package kafka.api
import java.util
-import kafka.security.authorizer.AclAuthorizer
+import kafka.security.authorizer.{AclAuthorizer}
import kafka.security.authorizer.AclEntry.{WildcardHost, WildcardPrincipalString}
import kafka.server.KafkaConfig
-import kafka.utils.{CoreUtils, JaasTestUtils, TestUtils}
+import kafka.utils.{JaasTestUtils, TestUtils}
import kafka.utils.TestUtils._
import org.apache.kafka.clients.admin._
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.acl._
-import org.apache.kafka.common.acl.AclOperation.{ALL, ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE}
+import org.apache.kafka.common.acl.AclOperation.{ALL, ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE, IDEMPOTENT_WRITE}
import org.apache.kafka.common.acl.AclPermissionType.{ALLOW, DENY}
-import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
+import org.apache.kafka.common.config.{ConfigResource, SaslConfigs, TopicConfig}
import org.apache.kafka.common.errors.{ClusterAuthorizationException, InvalidRequestException, TopicAuthorizationException, UnknownTopicOrPartitionException}
import org.apache.kafka.common.resource.PatternType.LITERAL
import org.apache.kafka.common.resource.ResourceType.{GROUP, TOPIC}
import org.apache.kafka.common.resource.{PatternType, Resource, ResourcePattern, ResourcePatternFilter, ResourceType}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.server.authorizer.Authorizer
+import org.apache.kafka.common.utils.Utils
import org.apache.kafka.storage.internals.log.LogConfig
import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo}
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo, Timeout}
-import java.util.Collections
import scala.jdk.CollectionConverters._
import scala.collection.Seq
import scala.compat.java8.OptionConverters._
import scala.concurrent.ExecutionException
import scala.util.{Failure, Success, Try}
+@Timeout(120)
class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetup {
val clusterResourcePattern = new ResourcePattern(ResourceType.CLUSTER, Resource.CLUSTER_NAME, PatternType.LITERAL)
- val authorizationAdmin = new AclAuthorizationAdmin(classOf[AclAuthorizer], classOf[AclAuthorizer])
+ val aclAuthorizerClassName = classOf[AclAuthorizer].getName
+ def kafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KafkaServerPrincipalUnqualifiedName)
- this.serverConfig.setProperty(KafkaConfig.ZkEnableSecureAclsProp, "true")
+ var superUserAdmin: Admin = _
override protected def securityProtocol = SecurityProtocol.SASL_SSL
override protected lazy val trustStoreFile = Some(TestUtils.tempFile("truststore", ".jks"))
- override def generateConfigs: Seq[KafkaConfig] = {
- this.serverConfig.setProperty(KafkaConfig.AuthorizerClassNameProp, authorizationAdmin.authorizerClassName)
- super.generateConfigs
- }
-
- override def configureSecurityBeforeServersStart(testInfo: TestInfo): Unit = {
- authorizationAdmin.initializeAcls()
- }
-
@BeforeEach
override def setUp(testInfo: TestInfo): Unit = {
+ this.serverConfig.setProperty(KafkaConfig.AuthorizerClassNameProp, aclAuthorizerClassName)
+ this.serverConfig.setProperty(KafkaConfig.ZkEnableSecureAclsProp, "true")
+ this.serverConfig.setProperty(AclAuthorizer.SuperUsersProp, kafkaPrincipal.toString)
+
setUpSasl()
super.setUp(testInfo)
+ setInitialAcls()
}
def setUpSasl(): Unit = {
startSasl(jaasSections(Seq("GSSAPI"), Some("GSSAPI"), Both, JaasTestUtils.KafkaServerContextName))
+
+ val loginContext = jaasAdminLoginModule("GSSAPI")
+ superuserClientConfig.put(SaslConfigs.SASL_JAAS_CONFIG, loginContext)
+ }
+
+ private def setInitialAcls(): Unit = {
+ superUserAdmin = createSuperuserAdminClient()
+ val ace = new AccessControlEntry(WildcardPrincipalString, WildcardHost, ALL, ALLOW)
+ superUserAdmin.createAcls(List(new AclBinding(new ResourcePattern(TOPIC, "*", LITERAL), ace)).asJava)
+ superUserAdmin.createAcls(List(new AclBinding(new ResourcePattern(GROUP, "*", LITERAL), ace)).asJava)
+
+ val clusterAcls = List(clusterAcl(ALLOW, CREATE),
+ clusterAcl(ALLOW, DELETE),
+ clusterAcl(ALLOW, CLUSTER_ACTION),
+ clusterAcl(ALLOW, ALTER_CONFIGS),
+ clusterAcl(ALLOW, ALTER),
+ clusterAcl(ALLOW, IDEMPOTENT_WRITE))
+
+ superUserAdmin.createAcls(clusterAcls.map(ace => new AclBinding(clusterResourcePattern, ace)).asJava)
+
+ brokers.foreach { b =>
+ TestUtils.waitAndVerifyAcls(Set(ace), b.dataPlaneRequestProcessor.authorizer.get, new ResourcePattern(TOPIC, "*", LITERAL))
+ TestUtils.waitAndVerifyAcls(Set(ace), b.dataPlaneRequestProcessor.authorizer.get, new ResourcePattern(GROUP, "*", LITERAL))
+ TestUtils.waitAndVerifyAcls(clusterAcls.toSet, b.dataPlaneRequestProcessor.authorizer.get, clusterResourcePattern)
+ }
+ }
+
+ private def clusterAcl(permissionType: AclPermissionType, operation: AclOperation): AccessControlEntry = {
+ new AccessControlEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "*").toString,
+ WildcardHost, operation, permissionType)
}
@AfterEach
override def tearDown(): Unit = {
super.tearDown()
+ if (superUserAdmin != null) Utils.closeQuietly(superUserAdmin, "superUserAdminClient")
Review Comment:
`createSuperuserAdminClient()` adds the client to `IntegrationTestHarness.adminClients` so I think it should already get closed automatically.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on PR #15377:
URL: https://github.com/apache/kafka/pull/15377#issuecomment-2127382470
@mimaison thank you so much for reviewing the PR again. I think I have addressed the comments now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1560828250
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -258,4 +275,21 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
+ val config = new Properties()
+ config.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "40000")
+ config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "20000")
+ val client = createAdminClient(configOverrides = config)
+ adminClients += client
Review Comment:
The call on the line above already adds the client to `adminClients` from a super class so this is not needed. Same for `adminClients.foreach(_.close())` in`tearDown()`
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -74,16 +82,33 @@ object SslAdminIntegrationTest {
futures.asJava
}
}
+
+ class TestPrincipalBuilder extends DefaultKafkaPrincipalBuilder(null, null) {
+ private val Pattern = "O=A (.*?),CN=(.*?)".r
+
+ // Use fields from DN as server principal to grant authorisation for servers and super admin client
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ val peerPrincipal = context.asInstanceOf[SslAuthenticationContext].session.getPeerPrincipal.getName
+ peerPrincipal match {
+ case Pattern(name, cn) =>
+ val principal = if ((name == "server") || (cn == superuserCn)) "server" else KafkaPrincipal.ANONYMOUS.getName
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, principal)
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
}
class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
- override val authorizationAdmin = new AclAuthorizationAdmin(classOf[SslAdminIntegrationTest.TestableAclAuthorizer], classOf[AclAuthorizer])
-
- this.serverConfig.setProperty(ZkConfigs.ZK_ENABLE_SECURE_ACLS_CONFIG, "true")
+ override val aclAuthorizerClassName: String = classOf[SslAdminIntegrationTest.TestableAclAuthorizer].getName
+ this.serverConfig.setProperty(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required")
+ this.serverConfig.setProperty(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, classOf[SslAdminIntegrationTest.TestPrincipalBuilder].getName)
override protected def securityProtocol = SecurityProtocol.SSL
override protected lazy val trustStoreFile = Some(TestUtils.tempFile("truststore", ".jks"))
Review Comment:
I think this override can be deleted, the super class is already setting it to the same value.
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -74,16 +82,33 @@ object SslAdminIntegrationTest {
futures.asJava
}
}
+
+ class TestPrincipalBuilder extends DefaultKafkaPrincipalBuilder(null, null) {
+ private val Pattern = "O=A (.*?),CN=(.*?)".r
+
+ // Use fields from DN as server principal to grant authorisation for servers and super admin client
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ val peerPrincipal = context.asInstanceOf[SslAuthenticationContext].session.getPeerPrincipal.getName
+ peerPrincipal match {
+ case Pattern(name, cn) =>
+ val principal = if ((name == "server") || (cn == superuserCn)) "server" else KafkaPrincipal.ANONYMOUS.getName
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, principal)
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
}
class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
- override val authorizationAdmin = new AclAuthorizationAdmin(classOf[SslAdminIntegrationTest.TestableAclAuthorizer], classOf[AclAuthorizer])
-
- this.serverConfig.setProperty(ZkConfigs.ZK_ENABLE_SECURE_ACLS_CONFIG, "true")
+ override val aclAuthorizerClassName: String = classOf[SslAdminIntegrationTest.TestableAclAuthorizer].getName
+ this.serverConfig.setProperty(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required")
+ this.serverConfig.setProperty(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, classOf[SslAdminIntegrationTest.TestPrincipalBuilder].getName)
override protected def securityProtocol = SecurityProtocol.SSL
override protected lazy val trustStoreFile = Some(TestUtils.tempFile("truststore", ".jks"))
private val adminClients = mutable.Buffer.empty[Admin]
Review Comment:
This can be removed too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "tinaselenge (via GitHub)" <gi...@apache.org>.
tinaselenge commented on code in PR #15377:
URL: https://github.com/apache/kafka/pull/15377#discussion_r1581029496
##########
core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala:
##########
@@ -259,4 +275,22 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
assertTrue(metrics.nonEmpty, s"Unable to find metric $name: allMetrics: ${allMetrics.keySet.map(_.getMBeanName)}")
metrics.map(_.asInstanceOf[Gauge[Int]].value).sum
}
+
+ override def createAdminClient: Admin = {
Review Comment:
So I initially thought the `createConfig()` in BaseAdminIntegrationTest is creating different security properties by calling `TestUtils.adminClientSecurityConfigs()` than the security properties created as part of the set up in IntegrationTestHarness. However following all the properties `adminClientSecurityConfigs()` set, they are exactly the same except the `certAlias`. So basically BaseAdminIntegrationTest mostly just setting security properties twice for no good reason. This seems to upset SSL admin test, because of the mismatch in the certAlias.
Removing the unnecessary call to `adminClientSecurityConfigs()` in BaseAdminIntegrationTest worked for all of the tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactor SSL/SASL admin integration tests to not use a custom authorizer [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison merged PR #15377:
URL: https://github.com/apache/kafka/pull/15377
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org