This is an automated email from the ASF dual-hosted git repository. manikumar pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new 6e164bb9ace KAFKA-14927: Add validation to be config keys in ConfigCommand tool (#14514) 6e164bb9ace is described below commit 6e164bb9ace3ea7a1a9542904d1a01c9fd3a1b48 Author: Aman Singh <103091061+singhn...@users.noreply.github.com> AuthorDate: Tue Oct 10 13:19:13 2023 +0530 KAFKA-14927: Add validation to be config keys in ConfigCommand tool (#14514) Added validation in ConfigCommand tool, only allow characters '([a-z][A-Z][0-9][._-])*' for config keys. Reviewers: Manikumar Reddy <manikumar.re...@gmail.com> --- core/src/main/scala/kafka/admin/ConfigCommand.scala | 13 +++++++++++-- .../scala/unit/kafka/admin/ConfigCommandTest.scala | 20 ++++++++++++++++++-- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index 309d95aaab8..25d400918f6 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -297,20 +297,29 @@ object ConfigCommand extends Logging { "This configuration will be ignored if the version is newer than the inter.broker.protocol.version specified in the broker or " + "if the inter.broker.protocol.version is 3.0 or newer. This configuration is deprecated and it will be removed in Apache Kafka 4.0.") } + validatePropsKey(props) props } private[admin] def parseConfigsToBeDeleted(opts: ConfigCommandOptions): Seq[String] = { if (opts.options.has(opts.deleteConfig)) { val configsToBeDeleted = opts.options.valuesOf(opts.deleteConfig).asScala.map(_.trim()) - val propsToBeDeleted = new Properties - configsToBeDeleted.foreach(propsToBeDeleted.setProperty(_, "")) configsToBeDeleted } else Seq.empty } + private def validatePropsKey(props: Properties): Unit = { + props.keySet.forEach { propsKey => + if (!propsKey.toString.matches("[a-zA-Z0-9._-]*")) { + throw new IllegalArgumentException( + s"Invalid character found for config key: ${propsKey}" + ) + } + } + } + private def processCommand(opts: ConfigCommandOptions): Unit = { val props = if (opts.options.has(opts.commandConfigOpt)) Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index 124227dd1f5..12e460a0bd3 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -304,15 +304,31 @@ class ConfigCommandTest extends Logging { createOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, shortFlag, "1", "--alter", - "--add-config", "a=b,c=,d=e,f=")) + "--add-config", "a._-c=b,c=,d=e,f=")) createOpts.checkArgs() val addedProps2 = ConfigCommand.parseConfigsToBeAdded(createOpts) assertEquals(4, addedProps2.size()) - assertEquals("b", addedProps2.getProperty("a")) + assertEquals("b", addedProps2.getProperty("a._-c")) assertEquals("e", addedProps2.getProperty("d")) assertTrue(addedProps2.getProperty("c").isEmpty) assertTrue(addedProps2.getProperty("f").isEmpty) + + var inValidCreateOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, + shortFlag, "1", + "--alter", + "--add-config", "a;c=b")) + + assertThrows(classOf[IllegalArgumentException], + () => ConfigCommand.parseConfigsToBeAdded(inValidCreateOpts)) + + inValidCreateOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, + shortFlag, "1", + "--alter", + "--add-config", "a,=b")) + + assertThrows(classOf[IllegalArgumentException], + () => ConfigCommand.parseConfigsToBeAdded(inValidCreateOpts)) } @Test