This is an automated email from the ASF dual-hosted git repository. lordgamez pushed a commit to branch MINIFICPP-2900 in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit fb6dc4ea8f61be5652321b35ede4c44e6a6fd633 Author: Gabor Gyimesi <[email protected]> AuthorDate: Fri Sep 11 13:44:03 2026 +0200 MINIFICPP-2900 Fix leak of NodeID allocation in OPC processors --- .../opc/include/{opcbase.h => BaseOPCProcessor.h} | 2 +- .../include/{fetchopc.h => FetchOPCProcessor.h} | 6 +-- extensions/opc/include/{opc.h => OPCCommon.h} | 39 ++++++++++++++++- .../opc/include/{putopc.h => PutOPCProcessor.h} | 8 ++-- .../opc/src/{opcbase.cpp => BaseOPCProcessor.cpp} | 4 +- .../src/{fetchopc.cpp => FetchOPCProcessor.cpp} | 14 +++--- extensions/opc/src/{opc.cpp => OPCCommon.cpp} | 8 ++-- .../opc/src/{putopc.cpp => PutOPCProcessor.cpp} | 51 +++++++++++----------- ...ocessorTests.cpp => FetchOPCProcessorTests.cpp} | 2 +- ...ProcessorTests.cpp => PutOPCProcessorTests.cpp} | 12 ++--- 10 files changed, 89 insertions(+), 57 deletions(-) diff --git a/extensions/opc/include/opcbase.h b/extensions/opc/include/BaseOPCProcessor.h similarity index 99% rename from extensions/opc/include/opcbase.h rename to extensions/opc/include/BaseOPCProcessor.h index b3c02e18d..9ec4afe00 100644 --- a/extensions/opc/include/opcbase.h +++ b/extensions/opc/include/BaseOPCProcessor.h @@ -22,7 +22,7 @@ #include <utility> #include <vector> -#include "opc.h" +#include "OPCCommon.h" #include "core/ProcessorImpl.h" #include "core/ProcessSession.h" #include "minifi-cpp/core/PropertyDefinition.h" diff --git a/extensions/opc/include/fetchopc.h b/extensions/opc/include/FetchOPCProcessor.h similarity index 97% rename from extensions/opc/include/fetchopc.h rename to extensions/opc/include/FetchOPCProcessor.h index 4c3669fc0..0bf5abd3a 100644 --- a/extensions/opc/include/fetchopc.h +++ b/extensions/opc/include/FetchOPCProcessor.h @@ -22,8 +22,8 @@ #include <utility> #include <vector> -#include "opc.h" -#include "opcbase.h" +#include "OPCCommon.h" +#include "BaseOPCProcessor.h" #include "minifi-cpp/FlowFileRecord.h" #include "core/ProcessSession.h" #include "minifi-cpp/core/Property.h" @@ -149,7 +149,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor { uint64_t max_depth_ = 0; LazyModeOptions lazy_mode_ = LazyModeOptions::Off; - std::vector<UA_NodeId> translated_node_ids_; // Only used when user provides path, path->nodeid translation is only done once + std::vector<opc::NodeId> translated_node_ids_; // Only used when user provides path, path->nodeid translation is only done once }; } // namespace org::apache::nifi::minifi::processors diff --git a/extensions/opc/include/opc.h b/extensions/opc/include/OPCCommon.h similarity index 78% rename from extensions/opc/include/opc.h rename to extensions/opc/include/OPCCommon.h index 8a4973515..f193967c8 100644 --- a/extensions/opc/include/opc.h +++ b/extensions/opc/include/OPCCommon.h @@ -58,6 +58,43 @@ enum class OPCNodeDataType{ String }; +// RAII owner for a UA_NodeId that calls UA_NodeId_clear to free node id allocation +class NodeId { + public: + NodeId() = default; + explicit NodeId(UA_NodeId id) noexcept : id_(id) {} // takes ownership of an already-built node id + NodeId(const NodeId&) = delete; + NodeId& operator=(const NodeId&) = delete; + NodeId(NodeId&& other) noexcept : id_(other.id_) { other.id_ = UA_NODEID_NULL; } + NodeId& operator=(NodeId&& other) noexcept { + if (this != &other) { + UA_NodeId_clear(&id_); + id_ = other.id_; + other.id_ = UA_NODEID_NULL; + } + return *this; + } + ~NodeId() { UA_NodeId_clear(&id_); } + + static NodeId copyOf(const UA_NodeId& id) { + NodeId result; + UA_NodeId_copy(&id, &result.id_); + return result; + } + + operator const UA_NodeId&() const noexcept { return id_; } // NOLINT(google-explicit-constructor) implicit passthrough to the C API is intended + [[nodiscard]] const UA_NodeId& get() const noexcept { return id_; } + + // Returns a pointer to the (cleared) node id for an open62541 out-parameter to write a freshly created node id into. + UA_NodeId* receive() noexcept { + UA_NodeId_clear(&id_); + return &id_; + } + + private: + UA_NodeId id_ = UA_NODEID_NULL; +}; + struct NodeData; class Client; @@ -73,7 +110,7 @@ class Client { UA_ReferenceDescription * getNodeReference(UA_NodeId node_id); void traverse(UA_NodeId node_id, const std::function<NodeFoundCallBackFunc>& cb, const std::string& base_path = "", uint64_t max_depth = 0, bool fetch_root = true); bool exists(UA_NodeId node_id); - UA_StatusCode translateBrowsePathsToNodeIdsRequest(const std::string& path, std::vector<UA_NodeId>& found_node_ids, int32_t namespace_index, + UA_StatusCode translateBrowsePathsToNodeIdsRequest(const std::string& path, std::vector<NodeId>& found_node_ids, int32_t namespace_index, const std::vector<UA_UInt32>& path_reference_types, const std::shared_ptr<core::logging::Logger>& logger); template<typename T> diff --git a/extensions/opc/include/putopc.h b/extensions/opc/include/PutOPCProcessor.h similarity index 96% rename from extensions/opc/include/putopc.h rename to extensions/opc/include/PutOPCProcessor.h index 5aba702b8..2ee4bb9d8 100644 --- a/extensions/opc/include/putopc.h +++ b/extensions/opc/include/PutOPCProcessor.h @@ -22,8 +22,8 @@ #include <utility> #include <vector> -#include "opc.h" -#include "opcbase.h" +#include "OPCCommon.h" +#include "BaseOPCProcessor.h" #include "minifi-cpp/FlowFileRecord.h" #include "core/ProcessSession.h" #include "minifi-cpp/core/Property.h" @@ -118,11 +118,11 @@ class PutOPCProcessor final : public BaseOPCProcessor { private: bool readParentNodeId(); - std::expected<std::pair<bool, UA_NodeId>, std::string> configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const; + std::expected<std::pair<bool, opc::NodeId>, std::string> configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const; void updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, const std::shared_ptr<core::FlowFile>& flow_file) const; void createNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessContext& context, core::ProcessSession& session, const std::shared_ptr<core::FlowFile>& flow_file) const; - UA_NodeId parent_node_id_{}; + opc::NodeId parent_node_id_; opc::OPCNodeDataType node_data_type_{}; UA_UInt32 create_node_reference_type_ = UA_NS0ID_HASCOMPONENT; }; diff --git a/extensions/opc/src/opcbase.cpp b/extensions/opc/src/BaseOPCProcessor.cpp similarity index 99% rename from extensions/opc/src/opcbase.cpp rename to extensions/opc/src/BaseOPCProcessor.cpp index e7ff5a73e..a39cc7416 100644 --- a/extensions/opc/src/opcbase.cpp +++ b/extensions/opc/src/BaseOPCProcessor.cpp @@ -18,8 +18,8 @@ #include <memory> #include <string> -#include "opc.h" -#include "opcbase.h" +#include "OPCCommon.h" +#include "BaseOPCProcessor.h" #include "minifi-cpp/FlowFileRecord.h" #include "core/ProcessSession.h" #include "core/Core.h" diff --git a/extensions/opc/src/fetchopc.cpp b/extensions/opc/src/FetchOPCProcessor.cpp similarity index 95% rename from extensions/opc/src/fetchopc.cpp rename to extensions/opc/src/FetchOPCProcessor.cpp index 032417484..4fc4b1caa 100644 --- a/extensions/opc/src/fetchopc.cpp +++ b/extensions/opc/src/FetchOPCProcessor.cpp @@ -15,7 +15,7 @@ * limitations under the License. */ -#include "fetchopc.h" +#include "FetchOPCProcessor.h" #include <list> #include <memory> @@ -24,7 +24,7 @@ #include "minifi-cpp/core/ProcessContext.h" #include "core/ProcessSession.h" #include "core/Resource.h" -#include "opc.h" +#include "OPCCommon.h" #include "utils/Enum.h" #include "utils/StringUtils.h" #include "utils/ProcessorConfigUtils.h" @@ -77,14 +77,12 @@ void FetchOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSe }; if (id_type_ != opc::OPCNodeIDType::Path) { - UA_NodeId my_id; - my_id.namespaceIndex = namespace_idx_; + const auto namespace_index = gsl::narrow_cast<UA_UInt16>(namespace_idx_); + opc::NodeId my_id; if (id_type_ == opc::OPCNodeIDType::Int) { - my_id.identifierType = UA_NODEIDTYPE_NUMERIC; - my_id.identifier.numeric = std::stoi(node_id_); // NOLINT(cppcoreguidelines-pro-type-union-access) + my_id = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, std::stoi(node_id_))}; } else if (id_type_ == opc::OPCNodeIDType::String) { - my_id.identifierType = UA_NODEIDTYPE_STRING; - my_id.identifier.string = UA_STRING_ALLOC(node_id_.c_str()); // NOLINT(cppcoreguidelines-pro-type-union-access) + my_id = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, node_id_.c_str())}; } else { logger_->log_error("Unhandled id type: '{}'. No flowfiles are generated.", magic_enum::enum_underlying(id_type_)); context.yield(); diff --git a/extensions/opc/src/opc.cpp b/extensions/opc/src/OPCCommon.cpp similarity index 99% rename from extensions/opc/src/opc.cpp rename to extensions/opc/src/OPCCommon.cpp index b66664529..8bf836a4a 100644 --- a/extensions/opc/src/opc.cpp +++ b/extensions/opc/src/OPCCommon.cpp @@ -15,7 +15,7 @@ * limitations under the License. */ -#include "opc.h" +#include "OPCCommon.h" #include <cstdlib> #include <memory> @@ -357,7 +357,7 @@ bool Client::exists(UA_NodeId node_id) { return retval; } -UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& path, std::vector<UA_NodeId>& found_node_ids, int32_t namespace_index, +UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& path, std::vector<NodeId>& found_node_ids, int32_t namespace_index, const std::vector<UA_UInt32>& path_reference_types, const std::shared_ptr<core::logging::Logger>& logger) { logger->log_trace("Trying to find node ids for {}", path.c_str()); @@ -414,9 +414,7 @@ UA_StatusCode Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa UA_BrowsePathResult res = response.results[i]; for (size_t j = 0; j < res.targetsSize; ++j) { found_data = true; - UA_NodeId resultId; - UA_NodeId_copy(&res.targets[j].targetId.nodeId, &resultId); - found_node_ids.push_back(resultId); + found_node_ids.push_back(NodeId::copyOf(res.targets[j].targetId.nodeId)); } } diff --git a/extensions/opc/src/putopc.cpp b/extensions/opc/src/PutOPCProcessor.cpp similarity index 86% rename from extensions/opc/src/putopc.cpp rename to extensions/opc/src/PutOPCProcessor.cpp index e253d6765..b4636e0ef 100644 --- a/extensions/opc/src/putopc.cpp +++ b/extensions/opc/src/PutOPCProcessor.cpp @@ -15,7 +15,7 @@ * limitations under the License. */ -#include "putopc.h" +#include "PutOPCProcessor.h" #include <memory> #include <string> @@ -23,7 +23,7 @@ #include "minifi-cpp/core/ProcessContext.h" #include "core/ProcessSession.h" #include "core/Resource.h" -#include "opc.h" +#include "OPCCommon.h" #include "utils/StringUtils.h" #include "utils/ProcessorConfigUtils.h" @@ -60,7 +60,7 @@ void PutOPCProcessor::onSchedule(core::ProcessContext& context, core::ProcessSes bool PutOPCProcessor::readParentNodeId() { if (id_type_ == opc::OPCNodeIDType::Path) { - std::vector<UA_NodeId> translated_node_ids; + std::vector<opc::NodeId> translated_node_ids; if (connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids, namespace_idx_, path_reference_types_, logger_) != UA_STATUSCODE_GOOD) { logger_->log_error("Failed to translate {} to node id, no flow files will be put", node_id_.c_str()); @@ -69,16 +69,14 @@ bool PutOPCProcessor::readParentNodeId() { logger_->log_error("{} was translated to multiple node ids, no flow files will be put", node_id_.c_str()); return false; } else { - parent_node_id_ = translated_node_ids[0]; + parent_node_id_ = std::move(translated_node_ids[0]); } } else { - parent_node_id_.namespaceIndex = namespace_idx_; + const auto namespace_index = gsl::narrow_cast<UA_UInt16>(namespace_idx_); if (id_type_ == opc::OPCNodeIDType::Int) { - parent_node_id_.identifierType = UA_NODEIDTYPE_NUMERIC; - parent_node_id_.identifier.numeric = std::stoi(node_id_); // NOLINT(cppcoreguidelines-pro-type-union-access) + parent_node_id_ = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, std::stoi(node_id_))}; } else { // idType_ == opc::OPCNodeIDType::String - parent_node_id_.identifierType = UA_NODEIDTYPE_STRING; - parent_node_id_.identifier.string = UA_STRING_ALLOC(node_id_.c_str()); // NOLINT(cppcoreguidelines-pro-type-union-access) + parent_node_id_ = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, node_id_.c_str())}; } if (!connection_->exists(parent_node_id_)) { logger_->log_error("Parent node doesn't exist, no flow files will be put"); @@ -88,7 +86,7 @@ bool PutOPCProcessor::readParentNodeId() { return true; } -std::expected<std::pair<bool, UA_NodeId>, std::string> PutOPCProcessor::configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const { +std::expected<std::pair<bool, opc::NodeId>, std::string> PutOPCProcessor::configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const { const auto namespaceidx = context.getProperty(TargetNodeNameSpaceIndex, &flow_file).value_or(""); if (namespaceidx.empty()) { return std::unexpected{fmt::format("Flowfile {} had no target namespace index specified, routing to failure!", flow_file.getUUIDStr())}; @@ -113,24 +111,25 @@ std::expected<std::pair<bool, UA_NodeId>, std::string> PutOPCProcessor::configur flow_file.getUUIDStr(), target_id_type)}; } - UA_NodeId target_node; - target_node.namespaceIndex = nsi; + const auto namespace_index = gsl::narrow_cast<UA_UInt16>(nsi); + opc::NodeId target_node; if (target_id_type == "Int") { - target_node.identifierType = UA_NODEIDTYPE_NUMERIC; + int32_t numeric_id = 0; try { - target_node.identifier.numeric = std::stoi(target_id); // NOLINT(cppcoreguidelines-pro-type-union-access) + numeric_id = std::stoi(target_id); } catch (const std::exception&) { return std::unexpected{fmt::format("Flowfile {}: target node ID is not a valid integer: {}. Routing to failure!", flow_file.getUUIDStr(), target_id)}; } + target_node = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, numeric_id)}; } else if (target_id_type == "String") { - target_node.identifierType = UA_NODEIDTYPE_STRING; - target_node.identifier.string = UA_STRING_ALLOC(target_id.c_str()); // NOLINT(cppcoreguidelines-pro-type-union-access) + target_node = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, target_id.c_str())}; } else { return std::unexpected{fmt::format("Flowfile {}: target node ID type is invalid: {}. Routing to failure!", flow_file.getUUIDStr(), target_id_type)}; } - return std::make_pair(connection_->exists(target_node), target_node); + const bool target_node_exists = connection_->exists(target_node); + return std::make_pair(target_node_exists, std::move(target_node)); } void PutOPCProcessor::updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, const std::shared_ptr<core::FlowFile>& flow_file) const { @@ -211,31 +210,31 @@ void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string try { UA_StatusCode sc = 0; - UA_NodeId result_node; + opc::NodeId result_node; switch (node_data_type_) { case opc::OPCNodeDataType::Int64: { int64_t value = std::stoll(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::UInt64: { uint64_t value = std::stoull(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Int32: { int32_t value = std::stoi(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::UInt32: { uint32_t value = std::stoul(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Boolean: { if (auto contentstr_parsed = utils::string::toBool(contentstr)) { - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr_parsed.value(), &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr_parsed.value(), result_node.receive()); } else { throw std::runtime_error("Content cannot be converted to bool"); } @@ -243,16 +242,16 @@ void PutOPCProcessor::createNode(const UA_NodeId& target_node, const std::string } case opc::OPCNodeDataType::Float: { float value = std::stof(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::Double: { double value = std::stod(contentstr); - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, value, result_node.receive()); break; } case opc::OPCNodeDataType::String: { - sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr, &result_node); + sc = connection_->add_node(parent_node_id_, target_node, create_node_reference_type_, browse_name, contentstr, result_node.receive()); break; } default: diff --git a/extensions/opc/tests/FetchOpcProcessorTests.cpp b/extensions/opc/tests/FetchOPCProcessorTests.cpp similarity index 99% rename from extensions/opc/tests/FetchOpcProcessorTests.cpp rename to extensions/opc/tests/FetchOPCProcessorTests.cpp index 09da30ae0..d79ef9bbd 100644 --- a/extensions/opc/tests/FetchOpcProcessorTests.cpp +++ b/extensions/opc/tests/FetchOPCProcessorTests.cpp @@ -19,7 +19,7 @@ #include "unit/Catch.h" #include "OpcUaTestServer.h" #include "unit/SingleProcessorTestController.h" -#include "include/fetchopc.h" +#include "include/FetchOPCProcessor.h" #include "unit/TestUtils.h" namespace org::apache::nifi::minifi::test { diff --git a/extensions/opc/tests/PutOpcProcessorTests.cpp b/extensions/opc/tests/PutOPCProcessorTests.cpp similarity index 98% rename from extensions/opc/tests/PutOpcProcessorTests.cpp rename to extensions/opc/tests/PutOPCProcessorTests.cpp index 583dc0cf4..7197d05c7 100644 --- a/extensions/opc/tests/PutOpcProcessorTests.cpp +++ b/extensions/opc/tests/PutOPCProcessorTests.cpp @@ -19,7 +19,7 @@ #include "unit/Catch.h" #include "OpcUaTestServer.h" #include "unit/SingleProcessorTestController.h" -#include "include/putopc.h" +#include "include/PutOPCProcessor.h" #include "utils/StringUtils.h" #include "unit/TestUtils.h" #include "minifi-cpp/utils/gsl.h" @@ -39,7 +39,7 @@ struct NodeData { void verifyCreatedNode(const NodeData& expected_node, SingleProcessorTestController& controller) { auto client = minifi::opc::Client::createClient(controller.getLogger(), "", {}, {}, {}); REQUIRE(client->connect("opc.tcp://127.0.0.1:4840/") == UA_STATUSCODE_GOOD); - std::vector<UA_NodeId> found_node_ids; + std::vector<opc::NodeId> found_node_ids; std::vector<UA_UInt32> reference_types; if (!expected_node.path_reference_types.empty()) { @@ -60,14 +60,14 @@ void verifyCreatedNode(const NodeData& expected_node, SingleProcessorTestControl }, 100ms)); REQUIRE(found_node_ids.size() == 1); - REQUIRE(found_node_ids[0].namespaceIndex == expected_node.namespace_index); - REQUIRE(found_node_ids[0].identifierType == UA_NODEIDTYPE_NUMERIC); - REQUIRE(found_node_ids[0].identifier.numeric == expected_node.node_id); // NOLINT(cppcoreguidelines-pro-type-union-access) + REQUIRE(found_node_ids[0].get().namespaceIndex == expected_node.namespace_index); + REQUIRE(found_node_ids[0].get().identifierType == UA_NODEIDTYPE_NUMERIC); + REQUIRE(found_node_ids[0].get().identifier.numeric == expected_node.node_id); // NOLINT(cppcoreguidelines-pro-type-union-access) UA_ReferenceDescription ref_desc; ref_desc.isForward = true; ref_desc.referenceTypeId = UA_NODEID_NUMERIC(0, UA_NODEIDTYPE_NUMERIC); - ref_desc.nodeId.nodeId = found_node_ids[0]; + ref_desc.nodeId.nodeId = found_node_ids[0].get(); ref_desc.browseName = UA_QUALIFIEDNAME_ALLOC(expected_node.namespace_index, expected_node.browse_name.c_str()); ref_desc.displayName = UA_LOCALIZEDTEXT_ALLOC("en-US", expected_node.browse_name.c_str()); const auto ref_desc_guard = gsl::finally([&ref_desc] {
