Kafka source: support SASL/SCRAM mechanisms

[franky-m]

Is your feature request related to a problem? Please describe.
Currently, Kafka source only supports SASL/PLAIN authentication mechanism, but apparently no SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512.

Describe the solution you'd like

Extend Data Prepper’s authentication options to include mechanisms such as SCRAM-SHA-512.

Example:

pipeline:
  name: kafka-pipeline
  source:
    kafka:
      bootstrap_servers:
        - 127.0.0.1:9093
      topics:
        - name: topic1
          group_id: groupID1
      authentication:
        sasl:
          plaintext:
            mechanism: SCRAM-SHA-512 # or SCRAM-SHA-256 or plain
            username: your_kafka_username
            password: your_kafka_password

Additional context

Many Kafka deployments rely on SCRAM mechanisms for improved security. Users who require SCRAM-SHA-512 authentication need this feature to seamlessly integrate Data Prepper into their existing Kafka infrastructure.

[dlvenable]

@franky-m , This is a great idea. Would you be able to contribute a PR to help with this? We could give you some pointers in the code.

[franky-m]

Hi could you give me the pointers you mentioned? I would try to implement the SASL/SCRAM support myself and if I succeed I would open a PR

[burandobata]

+1

[franky-m]

Hi @dlvenable! Did you have time to collect the pointers you mentioned?

[dlvenable]

@franky-m ,

Yes, I have some references.

First, I think that SASL/SCRAM is different from SASL/PLAIN. So the configuration should probably be a little different in the YAML.

It should have the following structure instead.

authentication:
  sasl:
    scram:
      username: your_kafka_username
      password: your_kafka_password

You can see where we add the current configuraiton in this block:

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/AuthConfig.java

Lines 21 to 24 in 64445fa

	public static class SaslAuthConfig {
	@JsonAlias("plain")
	@JsonProperty("plaintext")
	private PlainTextAuthConfig plainTextAuthConfig;

You can add something like the following below there to add the scram option.

        @JsonProperty("scram")
        private ScramAuthConfig scramAuthConfig;

Here is code where we set the plain configuration into the Kafka properties:

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java

Lines 120 to 132 in af7d1b5

	private static void setPlainTextAuthProperties(final Properties properties, final PlainTextAuthConfig plainTextAuthConfig,
	final EncryptionConfig encryptionConfig) {
	final String username = plainTextAuthConfig.getUsername();
	final String password = plainTextAuthConfig.getPassword();
	properties.put(SASL_MECHANISM, "PLAIN");
	properties.put(SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";");
	if (checkEncryptionType(encryptionConfig, EncryptionType.SSL)) {
	properties.put(SECURITY_PROTOCOL, "SASL_SSL");
	setSecurityProtocolSSLProperties(properties, encryptionConfig);
	} else { // EncryptionType.NONE
	properties.put(SECURITY_PROTOCOL, "SASL_PLAINTEXT");
	}
	}

And this is the code where we call it.

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java

Lines 341 to 342 in af7d1b5

	} else if (Objects.nonNull(plainTextAuthConfig) && Objects.nonNull(kafkaClusterAuthConfig.getEncryptionConfig())) {
	setPlainTextAuthProperties(properties, plainTextAuthConfig, kafkaClusterAuthConfig.getEncryptionConfig());

You could add a new condition that would look somewhat like:

...
} else if(Objects.nonNull(saslAuthConfig.getScramAuthConfig())) {
  setScramAuthProperties(properties, saslAuthConfig.getScramAuthConfig());  // new method; maybe it needs the encryption config too
} else ...

[dlvenable]

@franky-m , We did make some changes to support dynamically updating the password if it changes in the underlying source (e.g. AWS Secrets Manager). It isn't necessary to have that implemented, but would be nice.

@chenqi0805 , Can you provide any guidance on how that would be implemented?