Properly mask all sensitive fields in Consumer and Producer settings

[mdedetrich]

Resolves: #85

This PR reuses the upstream Kafka mechanism that is used to filter/mask passwords which means its 100% correct (the .properties fields being printed in .toString refers to configuration that is passed to proper Kafka). In more detail, Kafka has its own org.apache.kafka.common.config.AbstractConfig which actually maintains types for keys. One of those types is kafka.common.config.ConfigDef.Type.PASSWORD which is how upstream Kafka determines whether any sensitive field needs to be masked (see https://github.com/apache/kafka/blob/51bc41031b91b16ffcbd95832e8a20a00d3b6f81/clients/src/main/java/org/apache/kafka/common/config/types/Password.java#L54-L57 for more info).

Also note that this project already has tests for determining if the filtering is working (see https://github.com/mdedetrich/incubator-pekko-connectors-kafka/blob/0a55b158d2d7fd8c050ab147d80d259a79e3ffe2/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala#L107-L117 and https://github.com/mdedetrich/incubator-pekko-connectors-kafka/blob/0a55b158d2d7fd8c050ab147d80d259a79e3ffe2/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala#L91-L101) so I can confirm that it works (i.e. there is no regression).

[gmethvin]

lgtm

[pjfanning]

would it be possible to add a unit test or 2?

[mdedetrich]

would it be possible to add a unit test or 2?

They already exist, see

https://github.com/mdedetrich/incubator-pekko-connectors-kafka/blob/0a55b158d2d7fd8c050ab147d80d259a79e3ffe2/tests/src/test/scala/org/apache/pekko/kafka/ConsumerSettingsSpec.scala#L107-L117

and

https://github.com/mdedetrich/incubator-pekko-connectors-kafka/blob/0a55b158d2d7fd8c050ab147d80d259a79e3ffe2/tests/src/test/scala/org/apache/pekko/kafka/ProducerSettingsSpec.scala#L91-L101

Now what the CVE is alluding to is the fact that there are more fields than the original implementation and hence what is just being tested in those unit tests. Should I try and find another field that is considered sensitive but doesn't have a key containing password and create a unit test for it?

[pjfanning]

@mdedetrich the CVE relates to (sasl.jaas.config,org.apache.kafka.common.security.plain.PlainLoginModule required username='FOOBAR' password='FOOBAR';) - would it be possible to add something like one of the existing tests but that uses sasl.jaas.config,org.apache.kafka.common.security.plain.PlainLoginModule?

[mdedetrich]

@pjfanning I have updated the already existing unit tests to check for an additional sensitive config key called ssl.keystore.key. ssl.keystore.key is considered sensitive according to Kafka (see https://kafka.apache.org/documentation/#producerconfigs_ssl.keystore.key and https://kafka.apache.org/documentation/#consumerconfigs_ssl.keystore.key) and the test only passes with these changes in the PR

[mdedetrich]

@mdedetrich the CVE relates to (sasl.jaas.config,org.apache.kafka.common.security.plain.PlainLoginModule required username='FOOBAR' password='FOOBAR';) - would it be possible to add something like one of the existing tests but that uses sasl.jaas.config,org.apache.kafka.common.security.plain.PlainLoginModule?

sasl.jaas.config is a config with type Password so it will get masked like every other key that is of type Password (see https://kafka.apache.org/documentation/#connectconfigs_sasl.jaas.config). Thats how this PR works, every single thing which is of type Password gets filtered.

I can add a unit test for this if you really want but then I have to manually construct the value so it satisfies to the config parser and with the modified unit test I did its really not necessary

[pjfanning]

lgtm

[mdedetrich]

All tests pass, merging PR.