This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a0cf4b386 MINIFICPP-2925 Fix reading SSL data in PublishKafka (#2284)
a0cf4b386 is described below
commit a0cf4b386f5487fc68750b06b5e385d962f35720
Author: Gabor Gyimesi <[email protected]>
AuthorDate: Tue Oct 13 16:45:11 2026 +0200
MINIFICPP-2925 Fix reading SSL data in PublishKafka (#2284)
- Fix std::expected return value handling
- Make SSL kafka tests more robust by not relying on default CA cert paths
- Probe system default CA cert paths for SSL in Kafka processors if no
CA cert is defined for SSL connection in both source and conan builds
---
extensions/kafka/KafkaProcessorBase.cpp | 9 ++++--
extensions/kafka/PublishKafka.cpp | 2 +-
extensions/kafka/tests/CMakeLists.txt | 1 +
extensions/kafka/tests/PublishKafkaTests.cpp | 40 ++++++++++++++++++++++++++
extensions/kafka/tests/features/environment.py | 1 +
5 files changed, 50 insertions(+), 3 deletions(-)
diff --git a/extensions/kafka/KafkaProcessorBase.cpp
b/extensions/kafka/KafkaProcessorBase.cpp
index e4e224015..e0a70a5d8 100644
--- a/extensions/kafka/KafkaProcessorBase.cpp
+++ b/extensions/kafka/KafkaProcessorBase.cpp
@@ -33,13 +33,16 @@ void
KafkaProcessorBase::setKafkaAuthenticationParameters(api::core::ProcessCont
utils::setKafkaConfigurationField(*config, "security.protocol",
std::string{magic_enum::enum_name(security_protocol_)});
logger_->log_debug("Kafka security.protocol [{}]",
magic_enum::enum_name(security_protocol_));
if (security_protocol_ == kafka::SecurityProtocolOption::ssl ||
security_protocol_ == kafka::SecurityProtocolOption::sasl_ssl) {
+ // "probe" makes librdkafka look for the system CA certificates in the
standard locations, regardless of how OpenSSL was built
+ std::string ca_location = "probe";
if (auto ssl_data = getSslData(context)) {
if (ssl_data->ca_loc.empty() && ssl_data->cert_loc.empty() &&
ssl_data->key_loc.empty() && ssl_data->key_pw.empty()) {
logger_->log_warn("Security protocol is set to {}, but no valid
security parameters are set in the properties or in the SSL Context Service.",
magic_enum::enum_name(security_protocol_));
} else {
- utils::setKafkaConfigurationField(*config, "ssl.ca.location",
ssl_data->ca_loc.string());
- logger_->log_debug("Kafka ssl.ca.location [{}]", ssl_data->ca_loc);
+ if (!ssl_data->ca_loc.empty()) {
+ ca_location = ssl_data->ca_loc.string();
+ }
utils::setKafkaConfigurationField(*config, "ssl.certificate.location",
ssl_data->cert_loc.string());
logger_->log_debug("Kafka ssl.certificate.location [{}]",
ssl_data->cert_loc);
utils::setKafkaConfigurationField(*config, "ssl.key.location",
ssl_data->key_loc.string());
@@ -48,6 +51,8 @@ void
KafkaProcessorBase::setKafkaAuthenticationParameters(api::core::ProcessCont
logger_->log_debug("Kafka ssl.key.password was set");
}
}
+ utils::setKafkaConfigurationField(*config, "ssl.ca.location", ca_location);
+ logger_->log_debug("Kafka ssl.ca.location [{}]", ca_location);
}
auto sasl_mechanism =
api::utils::parseEnumProperty<kafka::SASLMechanismOption>(context,
SASLMechanism);
diff --git a/extensions/kafka/PublishKafka.cpp
b/extensions/kafka/PublishKafka.cpp
index b934675ca..104900143 100644
--- a/extensions/kafka/PublishKafka.cpp
+++ b/extensions/kafka/PublishKafka.cpp
@@ -595,7 +595,7 @@ bool PublishKafka::createNewTopic(const
api::core::ProcessContext& context, cons
}
std::optional<api::utils::net::SslData>
PublishKafka::getSslData(api::core::ProcessContext& context) const {
- if (auto result = context.getSslData(SSLContextService); result) { return
*result; }
+ if (auto result = context.getSslData(SSLContextService); result && *result)
{ return *result; }
api::utils::net::SslData ssl_data;
if (auto security_ca = context.getProperty(SecurityCA, nullptr)) {
ssl_data.ca_loc = *security_ca; }
diff --git a/extensions/kafka/tests/CMakeLists.txt
b/extensions/kafka/tests/CMakeLists.txt
index e299e73a1..01cbd1a2f 100644
--- a/extensions/kafka/tests/CMakeLists.txt
+++ b/extensions/kafka/tests/CMakeLists.txt
@@ -25,6 +25,7 @@ FOREACH (testfile ${KAFKA_INTEGRATION_TESTS})
add_minifi_executable("${testfilename}" "${testfile}")
target_include_directories(${testfilename} BEFORE PRIVATE
"${CMAKE_SOURCE_DIR}/extensions/kafka")
target_compile_definitions(${testfilename} PRIVATE
"MINIFI_EXTENSION_GROUP_NAME=org.apache.nifi.minifi.test")
+ target_compile_definitions(${testfilename} PRIVATE
TEST_RESOURCES="${TEST_RESOURCES}")
set_target_properties(${testfilename} PROPERTIES RUNTIME_OUTPUT_DIRECTORY
"${CMAKE_BINARY_DIR}/bin")
target_link_libraries(${testfilename} Catch2WithMain)
target_link_libraries(${testfilename} minifi-kafka)
diff --git a/extensions/kafka/tests/PublishKafkaTests.cpp
b/extensions/kafka/tests/PublishKafkaTests.cpp
index f0de8ec9f..92b82e5ee 100644
--- a/extensions/kafka/tests/PublishKafkaTests.cpp
+++ b/extensions/kafka/tests/PublishKafkaTests.cpp
@@ -15,6 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+#include <algorithm>
+#include <filesystem>
+#include <string>
+#include <string_view>
+
#include "MockLogger.h"
#include "MockProcessContext.h"
#include "MockProcessSession.h"
@@ -55,4 +60,39 @@ TEST_CASE("Trigger without valid broker", "[PublishKafka]") {
CHECK_FALSE(logger->logs_.empty());
}
+namespace {
+bool hasDebugLog(const mock::MockLogger& logger, std::string_view message) {
+ const auto debug_logs = logger.logs_.find(core::logging::LOG_LEVEL::debug);
+ return debug_logs != logger.logs_.end() &&
std::ranges::any_of(debug_logs->second, [&](const auto& log) { return
log.contains(message); });
+}
+} // namespace
+
+TEST_CASE("Security properties are used for SSL connection when no SSL Context
Service is set", "[PublishKafka]") {
+ const auto logger = std::make_shared<mock::MockLogger>();
+ auto publish_kafka = PublishKafka(mock::metadataWithLogger(logger));
+ auto context = mock::MockProcessContext{};
+ const auto ca_cert_path = (std::filesystem::path(TEST_RESOURCES) /
"nifi-cert.pem").string();
+ context.properties_.emplace(PublishKafka::ClientName.name, "test_client");
+ context.properties_.emplace(PublishKafka::SeedBrokers.name,
"test_seedbroker");
+ context.properties_.emplace(PublishKafka::SecurityProtocol.name, "ssl");
+ context.properties_.emplace(PublishKafka::SecurityCA.name, ca_cert_path);
+
+ REQUIRE_NOTHROW(publish_kafka.onScheduleImpl(context));
+
+ CHECK(hasDebugLog(*logger, "Kafka ssl.ca.location [" + ca_cert_path + "]"));
+}
+
+TEST_CASE("System CA certificate locations are probed when no CA certificate
is set for SSL connection", "[PublishKafka]") {
+ const auto logger = std::make_shared<mock::MockLogger>();
+ auto publish_kafka = PublishKafka(mock::metadataWithLogger(logger));
+ auto context = mock::MockProcessContext{};
+ context.properties_.emplace(PublishKafka::ClientName.name, "test_client");
+ context.properties_.emplace(PublishKafka::SeedBrokers.name,
"test_seedbroker");
+ context.properties_.emplace(PublishKafka::SecurityProtocol.name, "ssl");
+
+ REQUIRE_NOTHROW(publish_kafka.onScheduleImpl(context));
+
+ CHECK(hasDebugLog(*logger, "Kafka ssl.ca.location [probe]"));
+}
+
} // namespace org::apache::nifi::minifi::processors::test
diff --git a/extensions/kafka/tests/features/environment.py
b/extensions/kafka/tests/features/environment.py
index 0d22bd6d2..6d75bc586 100644
--- a/extensions/kafka/tests/features/environment.py
+++ b/extensions/kafka/tests/features/environment.py
@@ -31,6 +31,7 @@ def before_all(context):
def before_scenario(context, scenario):
common_before_scenario(context, scenario)
+ context.override_default_ca_cert_files = False
def after_scenario(context, scenario):
