Kafka notifications: user and passwd (authentication)

[fgalan]

As continuation of issue #4666, implement the user and passwd fields in kafka and kafkaCustom notifications, in a similar way they work for MQTT notifications.

Modifiy orion-api.md to include them in the corresponding sections (look to similar doc in MQTT part).

Look for FIXME #4714 marks in the code. They could be useful.

CC: @orianar

[GregorioBlazquez]

Maybe this information can be useful:

In this PR of Kafnus, kafka security has been activated github.com/telefonicaid/kafnus/pull/205

Here the docker-compose file can be checked:
https://github.com/telefonicaid/kafnus/blob/4e6cce18babe9e21d1f639d05c7ad18170d3bcb9/docker/docker-…

There is some documentation about this:
https://github.com/telefonicaid/kafnus/blob/main/doc/AdvancedTopics/02_security.md#required-broker-…

[orianar]

To address #4714 I’m going to implement the user and passwd fields for kafka and kafkaCustom notifications, following the same approach as MQTT (optional fields and passwd is obfuscated when the subscription is retrieved).

Case (Orion)	What the user sends in the subscription	What Orion does	Key librdkafka config
1) No authentication (legacy, as today)	url, topic (no user/passwd)	Connects without SASL/TLS (no behavior change)	bootstrap.servers=<url> (no security.protocol, no sasl.*)
2) Auth + TLS with SASL/PLAIN	url, topic, user, passwd, saslMechanism:"PLAIN" (no securityProtocol, default SASL_SSL)	Uses username/password and encrypts the channel	security.protocol=SASL_SSL + sasl.mechanism=PLAIN + sasl.username=<user> + sasl.password=<passwd>
3) Auth + TLS with SASL/SCRAM-SHA-256	url, topic, user, passwd, saslMechanism:"SCRAM-SHA-256" (no securityProtocol, default SASL_SSL)	Same, SCRAM 256 mechanism	security.protocol=SASL_SSL + sasl.mechanism=SCRAM-SHA-256 + sasl.username + sasl.password
4) Auth + TLS with SASL/SCRAM-SHA-512	url, topic, user, passwd, saslMechanism:"SCRAM-SHA-512" (no securityProtocol, default SASL_SSL)	Same, SCRAM 512 mechanism	security.protocol=SASL_SSL + sasl.mechanism=SCRAM-SHA-512 + sasl.username + sasl.password
5) Auth without TLS (dev/test only, explicit override)	url, topic, user, passwd, saslMechanism:"PLAIN", securityProtocol:"SASL_PLAINTEXT"	Uses username/password without encrypting the channel	security.protocol=SASL_PLAINTEXT + sasl.mechanism=PLAIN + sasl.username=<user> + sasl.password=<passwd>

Default rule: if user/passwd are present and securityProtocol is not specified, Orion will use SASL_SSL.

Notes:

saslMechanism per subscription is needed to support mixed environments (different brokers, or operational changes between PLAIN and SCRAM).

Connection cache keying: we should not key connections only by url; the key should include at least url + securityProtocol + saslMechanism + user to avoid reusing producers across different credentials/mechanisms.

TLS/CA: with SASL_SSL, if the broker presents a certificate signed by an internal/non-public CA, the connection may fail due to certificate verification. This is usually fixed without changing Orion, by installing the CA in the host/image/container trust store (e.g., adding ca.pem to the system and updating certificates). Detailed TLS/certificate management is considered out of scope for this issue (we’ll just document it to avoid surprises).

Security note: SASL_PLAINTEXT sends both credentials and payload in clear text; it should only be used in isolated dev/test environments.

CI/testing: we’ll add an integration test for SASL_PLAINTEXT + PLAIN (the simplest setup). Full end-to-end testing for SASL_SSL with real TLS certificates/CA will be left as a manual test or a follow-up issue to avoid expanding the scope of this issue.

Cc: @fgalan

[fgalan]

Have a look to the -insecureNotif CLI / ORION_INSECURE_NOTIF env var (initially though to disable authenticated with unknown CA certificates in HTTPS case, but I think also could apply here)