Skip to content

KAFKA-6342: Move workaround for JSON parsing of non-escaped strings #8591

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 13 commits into from
May 6, 2020
Merged

KAFKA-6342: Move workaround for JSON parsing of non-escaped strings #8591

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
e7c4346
Initial Commit
umesh9794 Dec 13, 2017
5b48c95
Fixed the applicability of new method based on trunk
umesh9794 Jan 9, 2018
752d831
Merge branch 'trunk'
umesh9794 Jan 9, 2018
1e941b9
Rebased and applied review comments
umesh9794 Jan 15, 2018
a14e6cb
Rebased and applied review comments
umesh9794 Jan 22, 2018
ae659d1
Changed the AclJson to have the incorrect format for principals
umesh9794 Jan 24, 2018
1c308a8
Applied the review comments
umesh9794 Jun 5, 2018
438d010
Rebase changes + fixing test
viktorsomogyi Apr 30, 2020
fec67e0
Address review comments
viktorsomogyi Apr 30, 2020
755634f
Use jdk.CollectionConverters
viktorsomogyi May 5, 2020
b9c70ff
Get rid of parseBytesWithAclFallback
viktorsomogyi May 6, 2020
bdfa380
Optimize imports
viktorsomogyi May 6, 2020
802e255
Revert AclEntry import changes
viktorsomogyi May 6, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added core/src/main/scala/kafka/consumer/TopicCount.scala
View file
Open in desktop
Empty file.
24 changes: 22 additions & 2 deletions core/src/main/scala/kafka/security/authorizer/AclEntry.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,12 @@

package kafka.security.authorizer

import java.nio.charset.StandardCharsets

import kafka.utils.Json
import kafka.utils.json.JsonValue
import org.apache.kafka.common.acl.{AccessControlEntry, AclOperation, AclPermissionType}
import org.apache.kafka.common.acl.AclOperation.{READ, WRITE, CREATE, DESCRIBE, DELETE, ALTER, DESCRIBE_CONFIGS, ALTER_CONFIGS, CLUSTER_ACTION, IDEMPOTENT_WRITE}
import org.apache.kafka.common.acl.AclOperation.{ALTER, ALTER_CONFIGS, CLUSTER_ACTION, CREATE, DELETE, DESCRIBE, DESCRIBE_CONFIGS, IDEMPOTENT_WRITE, READ, WRITE}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.resource.{ResourcePattern, ResourceType}
import org.apache.kafka.common.security.auth.KafkaPrincipal
Expand Down Expand Up @@ -79,7 +82,7 @@ object AclEntry {
if (bytes == null || bytes.isEmpty)
return collection.immutable.Set.empty[AclEntry]

Json.parseBytes(bytes).map(_.asJsonObject).map { js =>
parseBytesWithAclFallback(bytes).map(_.asJsonObject).map { js =>
//the acl json version.
require(js(VersionKey).to[Int] == CurrentVersion)
js(AclsKey).asJsonArray.iterator.map(_.asJsonObject).map { itemJs =>
Expand All @@ -92,6 +95,23 @@ object AclEntry {
}.getOrElse(Set.empty)
}

/**
* Parse a JSON string into a JsonValue if possible. `None` is returned if `input` is not valid JSON. This method is currently used
* to read the already stored invalid ACLs JSON which was persisted using older versions of Kafka (prior to Kafka 1.1.0). KAFKA-6319
*/
private def parseBytesWithAclFallback(input: Array[Byte]): Option[JsonValue] = {
// Before 1.0.1, Json#encode did not escape backslash or any other special characters. SSL principals
// stored in ACLs may contain backslash as an escape char, making the JSON generated in earlier versions invalid.
// Escape backslash and retry to handle these strings which may have been persisted in ZK.
// Note that this does not handle all special characters (e.g. non-escaped double quotes are not supported)
Json.tryParseBytes(input) match {
case Left(e) =>
Comment thread
viktorsomogyi marked this conversation as resolved.
Outdated
val escapedInput = new String(input, StandardCharsets.UTF_8).replaceAll("\\\\", "\\\\\\\\")
Json.parseFull(escapedInput)
case Right(v) => Some(v)
}
}

def toJsonCompatibleMap(acls: Set[AclEntry]): Map[String, Any] = {
Map(AclEntry.VersionKey -> AclEntry.CurrentVersion, AclEntry.AclsKey -> acls.map(acl => acl.toMap.asJava).toList.asJava)
}
Expand Down
11 changes: 1 addition & 10 deletions core/src/main/scala/kafka/utils/Json.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,16 +35,7 @@ object Json {
*/
def parseFull(input: String): Option[JsonValue] =

@hachikuji hachikuji May 1, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm probably missing something obvious, but I found the following callers of this function: DeleteRecordsCommand, LeaderElectionCommand, PreferredReplicaLeaderElectionCommand, and ReassignPartitionsCommand. As far as I can tell, none of these uses involve the parsing of ACLs. Did we lose this compatibility handling at some point or is there some magic invocation?

@ijuma ijuma May 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The compatibility fix was only required for ACLs, but we added it here to keep the fix small for backporting. The idea was to move to the appropriate place soon after, but it got stuck for a while. Do you have some concerns?

@ijuma ijuma May 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Discussed this offline with @hachikuji. I had misunderstood his point. It looks like the compatibility code has not been used since 1.1.0:

9b44c3e#diff-81f2ec590f4e047be7e3a7e5267cbc9aR60

@ijuma ijuma May 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe we should just drop this code.

@viktorsomogyi viktorsomogyi May 5, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I looks like parseFull is used in a couple of places as Json mentioned but I can perhaps just delegate this to parseBytes like this below and can continue using that in the mentioned callers.

def parseFull(input: String): Option[JsonValue] = parseBytes(input.getBytes(Charset.defaultCharset()))

Or shall I just get rid of this code and use parseBytes everywhere?

@ijuma ijuma May 5, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This code is fine as it is. We are suggesting that we don't need parseBytesWithAclFallback in AclEntry.

@viktorsomogyi viktorsomogyi May 6, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, then I misunderstood it too. :)
However isn't it possible that this issue resurfaces again if we remove this? I suspect the user who raised it may still use 1.0 where it got fixed and when they upgrade they'd run into this again. Or am I missing something else too?

@ijuma ijuma May 6, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@viktorsomogyi I think the argument is that if no-one complained since 1.1 which was released more than 2 years ago, then maybe this issue is very rare.

@viktorsomogyi viktorsomogyi May 6, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, that's fair. I've uploaded the changes.

try Option(mapper.readTree(input)).map(JsonValue(_))
catch {
case _: JsonProcessingException =>
// Before 1.0.1, Json#encode did not escape backslash or any other special characters. SSL principals
// stored in ACLs may contain backslash as an escape char, making the JSON generated in earlier versions invalid.
// Escape backslash and retry to handle these strings which may have been persisted in ZK.
// Note that this does not handle all special characters (e.g. non-escaped double quotes are not supported)
val escapedInput = input.replaceAll("\\\\", "\\\\\\\\")
try Option(mapper.readTree(escapedInput)).map(JsonValue(_))
catch { case _: JsonProcessingException => None }
}
catch { case _: JsonProcessingException => None }

/**
* Parse a JSON string into either a generic type T, or a JsonProcessingException in the case of
Expand Down
21 changes: 11 additions & 10 deletions core/src/test/scala/unit/kafka/security/authorizer/AclEntryTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,27 +23,28 @@ import org.apache.kafka.common.acl.AclOperation.READ
import org.apache.kafka.common.acl.AclPermissionType.{ALLOW, DENY}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.junit.{Assert, Test}
import org.scalatestplus.junit.JUnitSuite

import scala.jdk.CollectionConverters._
import scala.collection.JavaConverters._
Comment thread
viktorsomogyi marked this conversation as resolved.
Outdated

class AclEntryTest extends JUnitSuite {
class AclEntryTest {

val AclJson = "{\"version\": 1, \"acls\": [{\"host\": \"host1\",\"permissionType\": \"Deny\",\"operation\": \"READ\", \"principal\": \"User:alice\" }, " +
"{ \"host\": \"*\" , \"permissionType\": \"Allow\", \"operation\": \"Read\", \"principal\": \"User:bob\" }, " +
"{ \"host\": \"host1\", \"permissionType\": \"Deny\", \"operation\": \"Read\" , \"principal\": \"User:bob\"} ]}"
val AclJson = """{"version": 1, "acls": [{"host": "host1","permissionType": "Deny","operation": "READ", "principal": "User:alice" },
{ "host": "*" , "permissionType": "Allow", "operation": "Read", "principal": "User:bob" },
{ "host": "host1", "permissionType": "Deny", "operation": "Read" , "principal": "User:bob"},
{ "host": "host1", "permissionType": "Deny", "operation": "Read" , "principal": "User:\,bob"},
{ "host": "host1", "permissionType": "Deny", "operation": "Read" , "principal": "User:bob1\,bob2"}]}"""

@Test
def testAclJsonConversion(): Unit = {
val acl1 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "alice"), DENY, "host1" , READ)
val acl2 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "bob"), ALLOW, "*", READ)
val acl3 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "bob"), DENY, "host1", READ)
val acl4 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "\\,bob"), DENY, "host1", READ)
val acl5 = AclEntry(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "bob1\\,bob2"), DENY, "host1", READ)

val acls = Set[AclEntry](acl1, acl2, acl3)
val jsonAcls = Json.encodeAsBytes(AclEntry.toJsonCompatibleMap(acls).asJava)
val acls = Set[AclEntry](acl1, acl2, acl3, acl4, acl5)

Assert.assertEquals(acls, AclEntry.fromBytes(jsonAcls))
Assert.assertEquals(acls, AclEntry.fromBytes(Json.encodeAsBytes(AclEntry.toJsonCompatibleMap(acls).asJava)))
Assert.assertEquals(acls, AclEntry.fromBytes(AclJson.getBytes(UTF_8)))
}

}
4 changes: 0 additions & 4 deletions core/src/test/scala/unit/kafka/utils/JsonTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,10 +59,6 @@ class JsonTest {
val encoded = Json.legacyEncodeAsString(map)
val decoded = Json.parseFull(encoded)
assertEquals(Json.parseFull("""{"foo1":"bar1\\,bar2", "foo2":"\\bar"}"""), decoded)

// Test strings with non-escaped backslash and quotes. This is to verify that ACLs
// containing non-escaped chars persisted using 1.0 can be parsed.
assertEquals(decoded, Json.parseFull("""{"foo1":"bar1\,bar2", "foo2":"\bar"}"""))
}

@Test
Expand Down