feat: Add Kafka SASL_SSL authentication

[jouir]

Add security_protocol, sasl_mechanism, sasl_plain_username and sasl_plain_password arguments like aiokafka.AIOKafkaConsumer client.

https://aiokafka.readthedocs.io/en/stable/consumer.html

[jouir]

Same as #173 but it updates the existing kafka event source instead. Not tested with SCRAM.

[Alex-Izquierdo]

Hello @jouir thank you very much for your contribution. I have some concerns about how you are managing the options.

1. The security_protocol should not be overridden to "SSL" if its value is "SASL_SSL." It requires a different set of options (including those defined in the sasl_mechanism). Additionally, it seems to be missing an option: "SASL_PLAINTEXT."
2. The ssl_context is required for SSL and SASL_SSL options. Therefore, for our arguments, cafile is mandatory in such cases and should be enforced. Otherwise, the error returned by aiokafka could be confusing for users.
3. It would be beneficial to check all the matrices of required options on our side or capture "ValueError" exceptions from the Kafka client to provide nice and meaningful messages to the users, rather than relying on the default traceback output.

The pull request would be merged more quickly if it includes tests that deploy the required Kafka with Docker Compose, as we already do in the integration tests. Otherwise, we will have to test it manually, which will take more time.

[jouir]

Hello @Alex-Izquierdo, thank you for the review. I appreciate!

1. The security_protocol should not be overridden to "SSL" if its value is "SASL_SSL." It requires a different set of options (including those defined in the sasl_mechanism). 

The security_protocol is overridden only if there is a cafile and the value is PLAINTEXT, like before.

>>> args = {"security_protocol": "SASL_SSL"}
>>> security_protocol = args.get("security_protocol", "PLAINTEXT")
>>> cafile = args.get("cafile")
>>> if cafile or security_protocol in ["SSL", "SASL_SSL"]:
...     if security_protocol == "PLAINTEXT":
...             security_protocol = "SSL"
... 
>>> security_protocol
'SASL_SSL'

Additionally, it seems to be missing an option: "SASL_PLAINTEXT."

Good point. I will add it, thanks.

2. The ssl_context is required for SSL and SASL_SSL options. Therefore, for our arguments, cafile is mandatory in such cases and should be enforced. Otherwise, the error returned by aiokafka could be confusing for users.

The cafile is not required to create a SSLContext. We can see in the aiokafka documentation that cafile defaults to None:

If CA not specified (by either cafile, capath, cadata) default system CA will be used if found by OpenSSL.

Same with ssl:

cafile, capath, cadata represent optional CA certificates to trust for certificate verification, as in SSLContext.load_verify_locations(). If all three are None, this function can choose to trust the system’s default CA certificates instead.

Which is the case by default here. We use such setup to use very basic encryption without error from aiokafka nor ssl.

3. It would be beneficial to check all the matrices of required options on our side or capture "ValueError" exceptions from the Kafka client to provide nice and meaningful messages to the users, rather than relying on the default traceback output.

As of today, I have not encountered any ValueError exception with error messages nor traceback thrown by aiokafka because of a misconfiguration. We can provide examples to guide users, but testing all the possible cases would duplicate the work already performed by aiokafka.

The pull request would be merged more quickly if it includes tests that deploy the required Kafka with Docker Compose, as we already do in the integration tests. Otherwise, we will have to test it manually, which will take more time.

Sure. I will try to provide a Kafka broker with SASL auth matching my own case using docker-compose and associated tests.

Have a nice day

[Alex-Izquierdo]

Hello again @jouir Thanks for your quickly response! :)
You are right regarding the behavior of the SSL option. I may have misunderstood the documentation. Sorry for the confusion. I await for the SASL_PLAINTEXT and some test. Thanks a lot.

[jouir]

I have pushed a new commit including tests for SASL_PLAINTEXT and SASL_SSL authentications, with self-signed certificates.

The condition to enable the SSLContext has been changed to check for security_protocol ending with SSL. But if you prefer the old condition based on an explicit list of security protocols (like SSL, SASL_SSL) instead, I can revert the change.

Integration tests were testing the ansible.eda.kafka official source instead of the one of the repository so I've added the sources path to the CLIRunner to test current changes. I had this issue running tests locally. I don't know if there is the same issue running on the CI/CD infrastructure.

[jouir]

My new revision fixes the following issues:

• Removed useless variable assignments from the main function and reorganized the verify_mode condition because the complexity of the function was too high (> 50 statements)
• Replaced a single quote that was copy/paste from ssl documentation, because it wasn't ' or "
• Use FQCN to test the collection
• Re-ordered the imports

[Alex-Izquierdo]

Thank you very much @jouir
There is an error raised by flake, other than that it looks great to me.

[jouir]

An unused line was left over. It's now removed.