You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ab...@apache.org on 2020/12/04 16:04:44 UTC
[nifi-minifi-cpp] branch main updated: MINIFICPP-1402 - Encrypt
flow configuration and change encryption key
This is an automated email from the ASF dual-hosted git repository.
aboda 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 35999e0 MINIFICPP-1402 - Encrypt flow configuration and change encryption key
35999e0 is described below
commit 35999e0d52cff2fce060c2820c3682aff70fc1d1
Author: Adam Debreceni <ad...@protonmail.com>
AuthorDate: Fri Dec 4 16:29:06 2020 +0100
MINIFICPP-1402 - Encrypt flow configuration and change encryption key
Signed-off-by: Arpad Boda <ab...@apache.org>
This closes #937
---
README.md | 5 +
controller/Controller.h | 6 +-
encrypt-config/ArgParser.cpp | 173 ++++++++
encrypt-config/{EncryptConfig.h => ArgParser.h} | 49 ++-
.../{ConfigFileEncryptor.h => CommandException.h} | 8 +-
encrypt-config/ConfigFileEncryptor.cpp | 46 ++-
encrypt-config/ConfigFileEncryptor.h | 3 +
encrypt-config/EncryptConfig.cpp | 122 +++---
encrypt-config/EncryptConfig.h | 22 +-
encrypt-config/EncryptConfigMain.cpp | 26 +-
encrypt-config/{ConfigFileEncryptor.h => Utils.h} | 7 +-
extensions/coap/tests/CoapIntegrationBase.h | 4 +-
extensions/http-curl/protocols/RESTSender.cpp | 6 +-
extensions/http-curl/tests/C2ConfigEncryption.cpp | 56 +++
extensions/http-curl/tests/CMakeLists.txt | 1 +
.../tests/ControllerServiceIntegrationTests.cpp | 2 +-
extensions/http-curl/tests/HTTPHandlers.h | 18 +-
extensions/http-curl/tests/HTTPIntegrationBase.h | 2 +-
.../http-curl/tests/HttpGetIntegrationTest.cpp | 8 +-
.../http-curl/tests/VerifyInvokeHTTPTest.cpp | 2 +-
.../tests/integration/TestExecuteProcess.cpp | 4 +-
.../include/Defaults.h | 30 +-
libminifi/include/FlowController.h | 90 +----
libminifi/include/c2/C2Agent.h | 41 +-
libminifi/include/c2/C2Client.h | 88 ++++
libminifi/include/c2/C2Payload.h | 10 +-
libminifi/include/core/ConfigurationFactory.h | 27 +-
libminifi/include/core/Flow.h | 59 +++
libminifi/include/core/FlowConfiguration.h | 48 ++-
libminifi/include/core/state/UpdateController.h | 2 +-
libminifi/include/core/yaml/YamlConfiguration.h | 44 +-
libminifi/include/properties/Configuration.h | 1 +
libminifi/include/properties/Decryptor.h | 31 +-
.../include/utils/CollectionUtils.h | 46 ++-
.../include/utils/EncryptionProvider.h | 35 +-
libminifi/include/utils/EncryptionUtils.h | 15 +
libminifi/include/utils/StringUtils.h | 119 +++---
.../Decryptor.h => utils/file/FileSystem.h} | 38 +-
libminifi/include/utils/file/FileUtils.h | 8 +
libminifi/include/utils/file/PathUtils.h | 17 +
libminifi/src/FlowController.cpp | 445 ++-------------------
libminifi/src/c2/C2Agent.cpp | 399 ++++++++----------
libminifi/src/c2/C2Client.cpp | 324 +++++++++++++++
libminifi/src/core/ConfigurationFactory.cpp | 23 +-
libminifi/src/core/FlowConfiguration.cpp | 21 +
libminifi/src/core/yaml/YamlConfiguration.cpp | 11 +
.../EncryptionProvider.cpp} | 36 +-
libminifi/src/utils/EncryptionUtils.cpp | 18 +-
libminifi/src/utils/StringUtils.cpp | 15 +-
libminifi/src/utils/file/FileSystem.cpp | 81 ++++
libminifi/test/integration/IntegrationBase.h | 33 +-
.../test/integration/ProvenanceReportingTest.cpp | 6 +-
.../PersistableKeyValueStoreServiceTest.cpp | 2 +-
libminifi/test/nodefs/NoYamlConfiguration.cpp | 3 +-
.../test/persistence-tests/PersistenceTests.cpp | 6 +-
libminifi/test/resources/decrypted.config.yml | 41 ++
libminifi/test/rocksdb-tests/RepoTests.cpp | 3 +-
libminifi/test/unit/CollectionUtilsTests.cpp | 62 +++
libminifi/test/unit/ControllerTests.cpp | 28 +-
libminifi/test/unit/DecryptorTests.cpp | 8 +-
libminifi/test/unit/FileSystemTests.cpp | 93 +++++
libminifi/test/unit/ProvenanceTestHelper.h | 4 -
libminifi/test/unit/StringUtilsTests.cpp | 40 ++
main/MainHelper.h | 17 +-
main/MiNiFiMain.cpp | 14 +-
65 files changed, 1924 insertions(+), 1128 deletions(-)
diff --git a/README.md b/README.md
index a6ef548..411fe0b 100644
--- a/README.md
+++ b/README.md
@@ -32,6 +32,7 @@ MiNiFi is a child project effort of Apache NiFi. This repository is for a nativ
- [Running](#running)
- [Deploying](#deploying)
- [Extensions](#extensions)
+ - [Security](#security)
- [Operations](#operations)
- [Issue Tracking](#issue-tracking)
- [Documentation](#documentation)
@@ -546,6 +547,10 @@ On Windows it is suggested that MSI be used for installation.
Please see [Extensions.md](Extensions.md) on how to build and run conditionally built dependencies and extensions.
+### Security
+
+For securing a MiNiFi agent's configuration files it comes with a tool called `encrypt-config`. Its documentation is available [here](https://cwiki.apache.org/confluence/display/MINIFI/Securing+MiNiFi+configuration+files).
+
### Recommended Antivirus Exclusions
Antivirus software can take a long time to scan directories and the files within them. Additionally, if the antivirus software locks files or directories during a scan, those resources are unavailable to MiNiFi processes, causing latency or unavailability of these resources in a MiNiFi instance. To prevent these performance and reliability issues from occurring, it is highly recommended to configure your antivirus software to skip scans on the following MiNiFi C++ directories:
diff --git a/controller/Controller.h b/controller/Controller.h
index ca4a634..6b93173 100644
--- a/controller/Controller.h
+++ b/controller/Controller.h
@@ -267,7 +267,8 @@ std::shared_ptr<core::controller::ControllerService> getControllerService(const
std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
- std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(prov_repo, flow_repo, content_repo, configuration, stream_factory, nifi_configuration_class_name);
+ std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(
+ prov_repo, flow_repo, content_repo, configuration, stream_factory, nifi_configuration_class_name);
std::shared_ptr<minifi::FlowController> controller = std::unique_ptr<minifi::FlowController>(
new minifi::FlowController(prov_repo, flow_repo, configuration, std::move(flow_configuration), content_repo));
@@ -324,7 +325,8 @@ void printManifest(const std::shared_ptr<minifi::Configure> &configuration) {
std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
- std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(prov_repo, flow_repo, content_repo, configuration, stream_factory, nifi_configuration_class_name);
+ std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(
+ prov_repo, flow_repo, content_repo, configuration, stream_factory, nifi_configuration_class_name);
std::shared_ptr<minifi::FlowController> controller = std::unique_ptr<minifi::FlowController>(
new minifi::FlowController(prov_repo, flow_repo, configuration, std::move(flow_configuration), content_repo, "manifest", false));
diff --git a/encrypt-config/ArgParser.cpp b/encrypt-config/ArgParser.cpp
new file mode 100644
index 0000000..2b98637
--- /dev/null
+++ b/encrypt-config/ArgParser.cpp
@@ -0,0 +1,173 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <string>
+#include <set>
+#include <iostream>
+#include <algorithm>
+#include "ArgParser.h"
+#include "utils/OptionalUtils.h"
+#include "utils/StringUtils.h"
+#include "utils/CollectionUtils.h"
+#include "CommandException.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace encrypt_config {
+
+const std::vector<Argument> Arguments::registered_args_{
+ {std::set<std::string>{"--minifi-home", "-m"},
+ true,
+ "minifi home",
+ "Specifies the home directory used by the minifi agent"}
+};
+
+const std::vector<Flag> Arguments::registered_flags_{
+ {std::set<std::string>{"--help", "-h"},
+ "Prints this help message"},
+ {std::set<std::string>{"--encrypt-flow-config"},
+ "If set, the flow configuration file (as specified in minifi.properties) is also encrypted."}
+};
+
+std::string Arguments::getHelp() {
+ std::stringstream ss;
+ ss << "Usage: " << "encrypt-config";
+ for (const auto& arg : registered_args_) {
+ ss << " ";
+ if (!arg.required) {
+ ss << "[";
+ }
+ ss << utils::StringUtils::join("|", arg.names)
+ << " <" << arg.value_name << ">";
+ if (!arg.required) {
+ ss << "]";
+ }
+ }
+ for (const auto& flag : registered_flags_) {
+ ss << " [" << utils::StringUtils::join("|", flag.names) << "]";
+ }
+ ss << "\n";
+ for (const auto& arg : registered_args_) {
+ ss << "\t";
+ ss << utils::StringUtils::join("|", arg.names) << " : ";
+ if (arg.required) {
+ ss << "(required)";
+ } else {
+ ss << "(optional)";
+ }
+ ss << " " << arg.description;
+ ss << "\n";
+ }
+ for (const auto& flag : registered_flags_) {
+ ss << "\t" << utils::StringUtils::join("|", flag.names) << " : "
+ << flag.description << "\n";
+ }
+ return ss.str();
+}
+
+void Arguments::set(const std::string& key, const std::string& value) {
+ if (get(key)) {
+ throw CommandException("Key is specified more than once \"" + key + "\"");
+ }
+ args_[key] = value;
+}
+
+void Arguments::set(const std::string& flag) {
+ if (isSet(flag)) {
+ throw CommandException("Flag is specified more than once \"" + flag + "\"");
+ }
+ flags_.insert(flag);
+}
+
+utils::optional<std::string> Arguments::get(const std::string &key) const {
+ return getArg(key) | utils::flatMap([&] (const Argument& arg) {return get(arg);});
+}
+
+utils::optional<std::string> Arguments::get(const Argument& arg) const {
+ for (const auto& name : arg.names) {
+ auto it = args_.find(name);
+ if (it != args_.end()) {
+ return it->second;
+ }
+ }
+ return {};
+}
+
+bool Arguments::isSet(const std::string &flag) const {
+ utils::optional<Flag> opt_flag = getFlag(flag);
+ if (!opt_flag) {
+ return false;
+ }
+ return utils::haveCommonItem(opt_flag->names, flags_);
+}
+
+Arguments Arguments::parse(int argc, char* argv[]) {
+ Arguments args;
+ for (int argIdx = 1; argIdx < argc; ++argIdx) {
+ std::string key{argv[argIdx]};
+ if (getFlag(key)) {
+ args.set(key);
+ continue;
+ }
+ if (!getArg(key)) {
+ throw CommandException("Unrecognized option: \"" + key + "\"");
+ }
+ if (argIdx == argc - 1) {
+ throw CommandException("No value specified for key \"" + key + "\"");
+ }
+ ++argIdx;
+ std::string value{argv[argIdx]};
+ args.set(key, value);
+ }
+ if (args.isSet("-h")) {
+ std::cout << getHelp();
+ std::exit(0);
+ }
+ for (const auto& arg : registered_args_) {
+ if (arg.required && !args.get(arg)) {
+ throw CommandException("Missing required option " + utils::StringUtils::join("|", arg.names));
+ }
+ }
+ return args;
+}
+
+utils::optional<Flag> Arguments::getFlag(const std::string &name) {
+ for (const auto& flag : registered_flags_) {
+ if (flag.names.count(name) > 0) {
+ return flag;
+ }
+ }
+ return {};
+}
+
+utils::optional<Argument> Arguments::getArg(const std::string &key) {
+ for (const auto& arg : registered_args_) {
+ if (arg.names.count(key) > 0) {
+ return arg;
+ }
+ }
+ return {};
+}
+
+} // namespace encrypt_config
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
+
diff --git a/encrypt-config/EncryptConfig.h b/encrypt-config/ArgParser.h
similarity index 52%
copy from encrypt-config/EncryptConfig.h
copy to encrypt-config/ArgParser.h
index dd8b351..d33d3ee 100644
--- a/encrypt-config/EncryptConfig.h
+++ b/encrypt-config/ArgParser.h
@@ -17,8 +17,10 @@
#pragma once
#include <string>
-
-#include "utils/EncryptionUtils.h"
+#include <set>
+#include <vector>
+#include <map>
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -26,24 +28,43 @@ namespace nifi {
namespace minifi {
namespace encrypt_config {
-class EncryptConfig {
+struct Argument {
+ std::set<std::string> names;
+ bool required;
+ std::string value_name;
+ std::string description;
+};
+
+struct Flag {
+ std::set<std::string> names;
+ std::string description;
+};
+
+class Arguments {
+ static const std::vector<Argument> registered_args_;
+ static const std::vector<Flag> registered_flags_;
+
+ void set(const std::string& key, const std::string& value);
+
+ void set(const std::string& bool_key);
+
+ static utils::optional<Argument> getArg(const std::string& key);
+ static utils::optional<Flag> getFlag(const std::string& flag);
+
public:
- EncryptConfig(int argc, char* argv[]);
- void encryptSensitiveProperties() const;
+ static Arguments parse(int argc, char* argv[]);
- private:
- std::string bootstrapFilePath() const;
- std::string propertiesFilePath() const;
+ static std::string getHelp();
- static std::string parseMinifiHomeFromTheOptions(int argc, char* argv[]);
+ utils::optional<std::string> get(const std::string& key) const;
- utils::crypto::Bytes getEncryptionKey() const;
- std::string hexDecodeAndValidateKey(const std::string& key) const;
- void writeEncryptionKeyToBootstrapFile(const utils::crypto::Bytes& encryption_key) const;
+ bool isSet(const std::string& flag) const;
- void encryptSensitiveProperties(const utils::crypto::Bytes& encryption_key) const;
+ private:
+ utils::optional<std::string> get(const Argument& key) const;
- const std::string minifi_home_;
+ std::map<std::string, std::string> args_;
+ std::set<std::string> flags_;
};
} // namespace encrypt_config
diff --git a/encrypt-config/ConfigFileEncryptor.h b/encrypt-config/CommandException.h
similarity index 85%
copy from encrypt-config/ConfigFileEncryptor.h
copy to encrypt-config/CommandException.h
index 169edc1..1de4494 100644
--- a/encrypt-config/ConfigFileEncryptor.h
+++ b/encrypt-config/CommandException.h
@@ -16,8 +16,8 @@
*/
#pragma once
-#include "ConfigFile.h"
-#include "utils/EncryptionUtils.h"
+#include <exception>
+#include <string>
namespace org {
namespace apache {
@@ -25,7 +25,9 @@ namespace nifi {
namespace minifi {
namespace encrypt_config {
-uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes& encryption_key);
+struct CommandException : std::logic_error {
+ using std::logic_error::logic_error;
+};
} // namespace encrypt_config
} // namespace minifi
diff --git a/encrypt-config/ConfigFileEncryptor.cpp b/encrypt-config/ConfigFileEncryptor.cpp
index a62d2ad..128931b 100644
--- a/encrypt-config/ConfigFileEncryptor.cpp
+++ b/encrypt-config/ConfigFileEncryptor.cpp
@@ -28,7 +28,15 @@ namespace nifi {
namespace minifi {
namespace encrypt_config {
-uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes& encryption_key) {
+bool isEncrypted(const utils::optional<std::string>& encryption_type) {
+ return encryption_type && !encryption_type->empty() && *encryption_type != "plaintext";
+}
+
+uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes & encryption_key) {
+ return encryptSensitivePropertiesInFile(config_file, EncryptionKeys{{}, encryption_key});
+}
+
+uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const EncryptionKeys& keys) {
int num_properties_encrypted = 0;
for (const auto& property_key : config_file.getSensitiveProperties()) {
@@ -38,20 +46,36 @@ uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::
std::string encryption_type_key = property_key + ".protected";
utils::optional<std::string> encryption_type = config_file.getValue(encryption_type_key);
- if (!encryption_type || encryption_type->empty() || *encryption_type == "plaintext") {
- std::string encrypted_property_value = utils::crypto::encrypt(*property_value, encryption_key);
+ std::string raw_value = *property_value;
+ if (isEncrypted(encryption_type)) {
+ try {
+ utils::crypto::decrypt(raw_value, keys.encryption_key);
+ std::cout << "Property \"" << property_key << "\" is already properly encrypted.\n";
+ continue;
+ } catch (const std::exception&) {}
+ if (!keys.old_key) {
+ throw std::runtime_error("No old encryption key is provided to attempt decryption of property \"" + property_key + "\"");
+ }
+ try {
+ raw_value = utils::crypto::decrypt(raw_value, *keys.old_key);
+ std::cout << "Successfully decrypted property \"" << property_key << "\" using old key.\n";
+ } catch (const std::exception&) {
+ throw std::runtime_error("Couldn't decrypt property \"" + property_key + "\" using the old key.");
+ }
+ }
- config_file.update(property_key, encrypted_property_value);
+ std::string encrypted_property_value = utils::crypto::encrypt(raw_value, keys.encryption_key);
- if (encryption_type) {
- config_file.update(encryption_type_key, utils::crypto::EncryptionType::name());
- } else {
- config_file.insertAfter(property_key, encryption_type_key, utils::crypto::EncryptionType::name());
- }
+ config_file.update(property_key, encrypted_property_value);
- std::cout << "Encrypted property: " << property_key << '\n';
- ++num_properties_encrypted;
+ if (encryption_type) {
+ config_file.update(encryption_type_key, utils::crypto::EncryptionType::name());
+ } else {
+ config_file.insertAfter(property_key, encryption_type_key, utils::crypto::EncryptionType::name());
}
+
+ std::cout << "Encrypted property: " << property_key << '\n';
+ ++num_properties_encrypted;
}
return num_properties_encrypted;
diff --git a/encrypt-config/ConfigFileEncryptor.h b/encrypt-config/ConfigFileEncryptor.h
index 169edc1..d58563f 100644
--- a/encrypt-config/ConfigFileEncryptor.h
+++ b/encrypt-config/ConfigFileEncryptor.h
@@ -18,6 +18,7 @@
#include "ConfigFile.h"
#include "utils/EncryptionUtils.h"
+#include "Utils.h"
namespace org {
namespace apache {
@@ -27,6 +28,8 @@ namespace encrypt_config {
uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes& encryption_key);
+uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const EncryptionKeys& keys);
+
} // namespace encrypt_config
} // namespace minifi
} // namespace nifi
diff --git a/encrypt-config/EncryptConfig.cpp b/encrypt-config/EncryptConfig.cpp
index 2d6dea6..743674f 100644
--- a/encrypt-config/EncryptConfig.cpp
+++ b/encrypt-config/EncryptConfig.cpp
@@ -25,14 +25,12 @@
#include "ConfigFileEncryptor.h"
#include "utils/file/FileUtils.h"
#include "utils/OptionalUtils.h"
+#include "Defaults.h"
namespace {
-constexpr const char* CONF_DIRECTORY_NAME = "conf";
-constexpr const char* BOOTSTRAP_FILE_NAME = "bootstrap.conf";
-constexpr const char* MINIFI_PROPERTIES_FILE_NAME = "minifi.properties";
+constexpr const char* OLD_KEY_PROPERTY_NAME = "nifi.bootstrap.sensitive.key.old";
constexpr const char* ENCRYPTION_KEY_PROPERTY_NAME = "nifi.bootstrap.sensitive.key";
-constexpr const char* USAGE_STRING = "Usage: encrypt-config --minifi-home <your-minifi-home>";
} // namespace
@@ -42,81 +40,115 @@ namespace nifi {
namespace minifi {
namespace encrypt_config {
-EncryptConfig::EncryptConfig(int argc, char* argv[]) : minifi_home_(parseMinifiHomeFromTheOptions(argc, argv)) {
+EncryptConfig::EncryptConfig(const std::string& minifi_home) : minifi_home_(minifi_home) {
if (sodium_init() < 0) {
throw std::runtime_error{"Could not initialize the libsodium library!"};
}
+ // encryption/decryption depends on the libsodium library which needs to be initialized
+ keys_ = getEncryptionKeys();
}
-std::string EncryptConfig::parseMinifiHomeFromTheOptions(int argc, char* argv[]) {
- if (argc >= 2) {
- for (int i = 1; i < argc; ++i) {
- std::string argstr(argv[i]);
- if ((argstr == "-h") || (argstr == "--help")) {
- std::cout << USAGE_STRING << std::endl;
- std::exit(0);
- }
- }
+EncryptConfig::EncryptionType EncryptConfig::encryptSensitiveProperties() const {
+ encryptSensitiveProperties(keys_);
+ if (keys_.old_key) {
+ return EncryptionType::RE_ENCRYPT;
}
+ return EncryptionType::ENCRYPT;
+}
- if (argc >= 3) {
- for (int i = 1; i < argc; ++i) {
- std::string argstr(argv[i]);
- if ((argstr == "-m") || (argstr == "--minifi-home")) {
- if (i+1 < argc) {
- return std::string(argv[i+1]);
- }
- }
- }
+void EncryptConfig::encryptFlowConfig() const {
+ encrypt_config::ConfigFile properties_file{std::ifstream{propertiesFilePath()}};
+ utils::optional<std::string> config_path = properties_file.getValue(Configure::nifi_flow_configuration_file);
+ if (!config_path) {
+ config_path = utils::file::PathUtils::resolve(minifi_home_, "conf/config.yml");
+ std::cout << "Couldn't find path of configuration file, using default: \"" << *config_path << "\"\n";
+ } else {
+ config_path = utils::file::PathUtils::resolve(minifi_home_, *config_path);
+ std::cout << "Encrypting flow configuration file: \"" << *config_path << "\"\n";
+ }
+ std::string config_content;
+ try {
+ std::ifstream config_file{*config_path, std::ios::binary};
+ config_file.exceptions(std::ios::failbit | std::ios::badbit);
+ config_content = std::string{std::istreambuf_iterator<char>(config_file), {}};
+ } catch (...) {
+ throw std::runtime_error("Error while reading flow configuration file \"" + *config_path + "\"");
}
+ try {
+ utils::crypto::decrypt(config_content, keys_.encryption_key);
+ std::cout << "Flow config file is already properly encrypted.\n";
+ return;
+ } catch (const std::exception&) {}
- throw std::runtime_error{USAGE_STRING};
-}
+ if (utils::crypto::isEncrypted(config_content)) {
+ if (!keys_.old_key) {
+ throw std::runtime_error("Config file is encrypted, but no old encryption key is set.");
+ }
+ std::cout << "Trying to decrypt flow config file using the old key ...\n";
+ try {
+ config_content = utils::crypto::decrypt(config_content, *keys_.old_key);
+ } catch (const std::exception&) {
+ throw std::runtime_error("Flow config is encrypted, but couldn't be decrypted.");
+ }
+ } else {
+ std::cout << "Flow config file is not encrypted, using as-is.\n";
+ }
-void EncryptConfig::encryptSensitiveProperties() const {
- utils::crypto::Bytes encryption_key = getEncryptionKey();
- encryptSensitiveProperties(encryption_key);
+ std::string encrypted_content = utils::crypto::encrypt(config_content, keys_.encryption_key);
+ try {
+ std::ofstream encrypted_file{*config_path, std::ios::binary};
+ encrypted_file.exceptions(std::ios::failbit | std::ios::badbit);
+ encrypted_file << encrypted_content;
+ } catch (...) {
+ throw std::runtime_error("Error while writing encrypted flow configuration file \"" + *config_path + "\"");
+ }
+ std::cout << "Successfully encrypted flow configuration file: \"" << *config_path << "\"\n";
}
std::string EncryptConfig::bootstrapFilePath() const {
- return utils::file::concat_path(
- utils::file::concat_path(minifi_home_, CONF_DIRECTORY_NAME),
- BOOTSTRAP_FILE_NAME);
+ return utils::file::concat_path(minifi_home_, DEFAULT_BOOTSTRAP_FILE);
}
std::string EncryptConfig::propertiesFilePath() const {
- return utils::file::concat_path(
- utils::file::concat_path(minifi_home_, CONF_DIRECTORY_NAME),
- MINIFI_PROPERTIES_FILE_NAME);
+ return utils::file::concat_path(minifi_home_, DEFAULT_NIFI_PROPERTIES_FILE);
}
-utils::crypto::Bytes EncryptConfig::getEncryptionKey() const {
+EncryptionKeys EncryptConfig::getEncryptionKeys() const {
encrypt_config::ConfigFile bootstrap_file{std::ifstream{bootstrapFilePath()}};
- utils::optional<std::string> key_from_bootstrap_file = bootstrap_file.getValue(ENCRYPTION_KEY_PROPERTY_NAME);
+ utils::optional<std::string> decryption_key_hex = bootstrap_file.getValue(OLD_KEY_PROPERTY_NAME);
+ utils::optional<std::string> encryption_key_hex = bootstrap_file.getValue(ENCRYPTION_KEY_PROPERTY_NAME);
+
+ EncryptionKeys keys;
+ if (decryption_key_hex && !decryption_key_hex->empty()) {
+ std::string binary_key = hexDecodeAndValidateKey(*decryption_key_hex, OLD_KEY_PROPERTY_NAME);
+ std::cout << "Old encryption key found in " << bootstrapFilePath() << "\n";
+ keys.old_key = utils::crypto::stringToBytes(binary_key);
+ }
- if (key_from_bootstrap_file && !key_from_bootstrap_file->empty()) {
- std::string binary_key = hexDecodeAndValidateKey(*key_from_bootstrap_file);
+ if (encryption_key_hex && !encryption_key_hex->empty()) {
+ std::string binary_key = hexDecodeAndValidateKey(*encryption_key_hex, ENCRYPTION_KEY_PROPERTY_NAME);
std::cout << "Using the existing encryption key found in " << bootstrapFilePath() << '\n';
- return utils::crypto::stringToBytes(binary_key);
+ keys.encryption_key = utils::crypto::stringToBytes(binary_key);
} else {
std::cout << "Generating a new encryption key...\n";
utils::crypto::Bytes encryption_key = utils::crypto::generateKey();
writeEncryptionKeyToBootstrapFile(encryption_key);
std::cout << "Wrote the new encryption key to " << bootstrapFilePath() << '\n';
- return encryption_key;
+ keys.encryption_key = encryption_key;
}
+ return keys;
}
-std::string EncryptConfig::hexDecodeAndValidateKey(const std::string& key) const {
+std::string EncryptConfig::hexDecodeAndValidateKey(const std::string& key, const std::string& key_name) const {
// Note: from_hex() allows [and skips] non-hex characters
std::string binary_key = utils::StringUtils::from_hex(key);
if (binary_key.size() == utils::crypto::EncryptionType::keyLength()) {
return binary_key;
} else {
std::stringstream error;
- error << "The encryption key " << ENCRYPTION_KEY_PROPERTY_NAME << " in the bootstrap file\n"
+ error << "The encryption key \"" << key_name << "\" in the bootstrap file\n"
<< " " << bootstrapFilePath() << '\n'
- << "is invalid; delete it to generate a new key.";
+ << "is invalid.";
throw std::runtime_error{error.str()};
}
}
@@ -134,13 +166,13 @@ void EncryptConfig::writeEncryptionKeyToBootstrapFile(const utils::crypto::Bytes
bootstrap_file.writeTo(bootstrapFilePath());
}
-void EncryptConfig::encryptSensitiveProperties(const utils::crypto::Bytes& encryption_key) const {
+void EncryptConfig::encryptSensitiveProperties(const EncryptionKeys& keys) const {
encrypt_config::ConfigFile properties_file{std::ifstream{propertiesFilePath()}};
if (properties_file.size() == 0) {
throw std::runtime_error{"Properties file " + propertiesFilePath() + " not found!"};
}
- uint32_t num_properties_encrypted = encryptSensitivePropertiesInFile(properties_file, encryption_key);
+ uint32_t num_properties_encrypted = encryptSensitivePropertiesInFile(properties_file, keys);
if (num_properties_encrypted == 0) {
std::cout << "Could not find any (new) sensitive properties to encrypt in " << propertiesFilePath() << '\n';
return;
diff --git a/encrypt-config/EncryptConfig.h b/encrypt-config/EncryptConfig.h
index dd8b351..f17878e 100644
--- a/encrypt-config/EncryptConfig.h
+++ b/encrypt-config/EncryptConfig.h
@@ -18,7 +18,7 @@
#include <string>
-#include "utils/EncryptionUtils.h"
+#include "Utils.h"
namespace org {
namespace apache {
@@ -28,22 +28,28 @@ namespace encrypt_config {
class EncryptConfig {
public:
- EncryptConfig(int argc, char* argv[]);
- void encryptSensitiveProperties() const;
+ enum class EncryptionType {
+ ENCRYPT,
+ RE_ENCRYPT
+ };
+
+ explicit EncryptConfig(const std::string& minifi_home);
+ EncryptionType encryptSensitiveProperties() const;
+
+ void encryptFlowConfig() const;
private:
std::string bootstrapFilePath() const;
std::string propertiesFilePath() const;
- static std::string parseMinifiHomeFromTheOptions(int argc, char* argv[]);
-
- utils::crypto::Bytes getEncryptionKey() const;
- std::string hexDecodeAndValidateKey(const std::string& key) const;
+ EncryptionKeys getEncryptionKeys() const;
+ std::string hexDecodeAndValidateKey(const std::string& key, const std::string& key_name) const;
void writeEncryptionKeyToBootstrapFile(const utils::crypto::Bytes& encryption_key) const;
- void encryptSensitiveProperties(const utils::crypto::Bytes& encryption_key) const;
+ void encryptSensitiveProperties(const EncryptionKeys& encryption_key) const;
const std::string minifi_home_;
+ EncryptionKeys keys_;
};
} // namespace encrypt_config
diff --git a/encrypt-config/EncryptConfigMain.cpp b/encrypt-config/EncryptConfigMain.cpp
index 6622e51..ae59303 100644
--- a/encrypt-config/EncryptConfigMain.cpp
+++ b/encrypt-config/EncryptConfigMain.cpp
@@ -19,15 +19,33 @@
#include <typeinfo>
#include "EncryptConfig.h"
+#include "ArgParser.h"
+#include "CommandException.h"
+
+using org::apache::nifi::minifi::encrypt_config::Arguments;
+using org::apache::nifi::minifi::encrypt_config::EncryptConfig;
+using org::apache::nifi::minifi::encrypt_config::CommandException;
int main(int argc, char* argv[]) try {
- org::apache::nifi::minifi::encrypt_config::EncryptConfig encrypt_config{argc, argv};
- encrypt_config.encryptSensitiveProperties();
+ Arguments args = Arguments::parse(argc, argv);
+ EncryptConfig encrypt_config{args.get("-m").value()};
+ EncryptConfig::EncryptionType type = encrypt_config.encryptSensitiveProperties();
+ if (args.isSet("--encrypt-flow-config")) {
+ encrypt_config.encryptFlowConfig();
+ } else if (type == EncryptConfig::EncryptionType::RE_ENCRYPT) {
+ std::cout << "WARNING: you did not request the flow config to be updated, "
+ << "if it is currently encrypted and the old key is removed, "
+ << "you won't be able to recover the flow config.\n";
+ }
return 0;
+} catch (const CommandException& c_ex) {
+ std::cerr << c_ex.what() << "\n";
+ std::cerr << Arguments::getHelp() << "\n";
+ return 1;
} catch (const std::exception& ex) {
std::cerr << ex.what() << "\n(" << typeid(ex).name() << ")\n";
- return 1;
+ return 2;
} catch (...) {
std::cerr << "Unknown error\n";
- return 2;
+ return 3;
}
diff --git a/encrypt-config/ConfigFileEncryptor.h b/encrypt-config/Utils.h
similarity index 87%
copy from encrypt-config/ConfigFileEncryptor.h
copy to encrypt-config/Utils.h
index 169edc1..23a521e 100644
--- a/encrypt-config/ConfigFileEncryptor.h
+++ b/encrypt-config/Utils.h
@@ -16,8 +16,8 @@
*/
#pragma once
-#include "ConfigFile.h"
#include "utils/EncryptionUtils.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -25,7 +25,10 @@ namespace nifi {
namespace minifi {
namespace encrypt_config {
-uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes& encryption_key);
+struct EncryptionKeys {
+ utils::optional<utils::crypto::Bytes> old_key;
+ utils::crypto::Bytes encryption_key;
+};
} // namespace encrypt_config
} // namespace minifi
diff --git a/extensions/coap/tests/CoapIntegrationBase.h b/extensions/coap/tests/CoapIntegrationBase.h
index ddf6660..2f2f920 100644
--- a/extensions/coap/tests/CoapIntegrationBase.h
+++ b/extensions/coap/tests/CoapIntegrationBase.h
@@ -44,7 +44,7 @@ class CoapIntegrationBase : public IntegrationBase {
server.reset();
}
- virtual void run(std::string test_file_location) override {
+ void run(const std::string& test_file_location, const utils::optional<std::string>& bootstrap_file = {}) override {
testSetup();
std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
@@ -61,7 +61,7 @@ class CoapIntegrationBase : public IntegrationBase {
core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location);
- std::shared_ptr<core::ProcessGroup> pg{ yaml_config.getRoot(test_file_location).release() };
+ std::shared_ptr<core::ProcessGroup> pg{ yaml_config.getRoot().release() };
queryRootProcessGroup(pg);
diff --git a/extensions/http-curl/protocols/RESTSender.cpp b/extensions/http-curl/protocols/RESTSender.cpp
index 649910e..37c6a2c 100644
--- a/extensions/http-curl/protocols/RESTSender.cpp
+++ b/extensions/http-curl/protocols/RESTSender.cpp
@@ -99,7 +99,7 @@ const C2Payload RESTSender::sendPayload(const std::string url, const Direction d
std::unique_ptr<utils::HTTPUploadCallback> callback = nullptr;
// Callback for transfer. Declared in order to destruct in proper order - take care!
- std::unique_ptr<utils::FileOutputCallback> file_callback = nullptr;
+ std::unique_ptr<utils::ByteOutputCallback> file_callback = nullptr;
utils::HTTPReadCallback read;
// Client declared last to make sure calbacks are still available when client is destructed
@@ -129,9 +129,7 @@ const C2Payload RESTSender::sendPayload(const std::string url, const Direction d
}
if (payload.getOperation() == TRANSFER) {
- utils::file::FileManager file_man;
- auto file = file_man.unique_file(true);
- file_callback = std::unique_ptr<utils::FileOutputCallback>(new utils::FileOutputCallback(file));
+ file_callback = std::unique_ptr<utils::ByteOutputCallback>(new utils::ByteOutputCallback(std::numeric_limits<size_t>::max()));
read.pos = 0;
read.ptr = file_callback.get();
client.setReadCallback(&read);
diff --git a/extensions/http-curl/tests/C2ConfigEncryption.cpp b/extensions/http-curl/tests/C2ConfigEncryption.cpp
new file mode 100644
index 0000000..d122b81
--- /dev/null
+++ b/extensions/http-curl/tests/C2ConfigEncryption.cpp
@@ -0,0 +1,56 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#undef NDEBUG
+#include <iterator>
+#include <fstream>
+#include "HTTPIntegrationBase.h"
+#include "HTTPHandlers.h"
+#include "utils/IntegrationTestUtils.h"
+#include "utils/EncryptionProvider.h"
+
+int main(int argc, char **argv) {
+ const cmd_args args = parse_cmdline_args(argc, argv, "update");
+ TestController controller;
+ // copy config file to temporary location as it will get overridden
+ char tmp_format[] = "/var/tmp/c2.XXXXXX";
+ std::string home_path = controller.createTempDirectory(tmp_format);
+ std::string live_config_file = utils::file::FileUtils::concat_path(home_path, "config.yml");
+ utils::file::FileUtils::copy_file(args.test_file, live_config_file);
+ // the C2 server will update the flow with the contents of args.test_file
+ // which will be encrypted and persisted to the temporary live_config_file
+ C2UpdateHandler handler(args.test_file);
+ VerifyC2Update harness(10000);
+ harness.getConfiguration()->set(minifi::Configure::nifi_flow_configuration_encrypt, "true");
+ harness.setKeyDir(args.key_dir);
+ harness.setUrl(args.url, &handler);
+ handler.setC2RestResponse(harness.getC2RestUrl(), "configuration", "true");
+
+ harness.run(live_config_file, args.key_dir);
+
+ auto encryptor = utils::crypto::EncryptionProvider::create(args.key_dir);
+ assert(encryptor);
+
+ std::ifstream encrypted_file{live_config_file, std::ios::binary};
+ std::string decrypted_config = encryptor->decrypt(std::string(std::istreambuf_iterator<char>(encrypted_file), {}));
+
+ std::ifstream original_file{args.test_file, std::ios::binary};
+ std::string original_config{std::istreambuf_iterator<char>(original_file), {}};
+
+ assert(decrypted_config == original_config);
+}
diff --git a/extensions/http-curl/tests/CMakeLists.txt b/extensions/http-curl/tests/CMakeLists.txt
index 490f701..2378862 100644
--- a/extensions/http-curl/tests/CMakeLists.txt
+++ b/extensions/http-curl/tests/CMakeLists.txt
@@ -71,6 +71,7 @@ add_test(NAME HTTPStreamingCallbackTests COMMAND "HTTPStreamingCallbackTests" WO
add_test(NAME HttpGetIntegrationTest COMMAND HttpGetIntegrationTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
add_test(NAME C2UpdateTest COMMAND C2UpdateTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
+add_test(NAME C2ConfigEncryption COMMAND C2ConfigEncryption "${TEST_RESOURCES}/decrypted.config.yml" "${TEST_RESOURCES}/")
add_test(NAME C2JstackTest COMMAND C2JstackTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
add_test(NAME C2DescribeManifestTest COMMAND C2DescribeManifestTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/")
add_test(NAME C2DescribeCoreComponentStateTest COMMAND C2DescribeCoreComponentStateTest "${TEST_RESOURCES}/TestC2DescribeCoreComponentState.yml" "${TEST_RESOURCES}/")
diff --git a/extensions/http-curl/tests/ControllerServiceIntegrationTests.cpp b/extensions/http-curl/tests/ControllerServiceIntegrationTests.cpp
index 1ada032..efb6dcd 100644
--- a/extensions/http-curl/tests/ControllerServiceIntegrationTests.cpp
+++ b/extensions/http-curl/tests/ControllerServiceIntegrationTests.cpp
@@ -81,7 +81,7 @@ int main(int argc, char **argv) {
core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, args.test_file);
- std::shared_ptr<core::ProcessGroup> pg(yaml_config.getRoot(args.test_file));
+ std::shared_ptr<core::ProcessGroup> pg(yaml_config.getRoot());
std::shared_ptr<core::controller::StandardControllerServiceProvider> provider = std::make_shared<core::controller::StandardControllerServiceProvider>(map, pg, std::make_shared<minifi::Configure>());
std::shared_ptr<core::controller::ControllerServiceNode> mockNode = pg->findControllerService("MockItLikeIts1995");
diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h
index aff9000..706dbc5 100644
--- a/extensions/http-curl/tests/HTTPHandlers.h
+++ b/extensions/http-curl/tests/HTTPHandlers.h
@@ -482,11 +482,19 @@ class C2UpdateHandler : public ServerAwareHandler {
return true;
}
- virtual void setC2RestResponse(const std::string& url, const std::string& name) {
- response_ = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {"
- "\"operation\" : \"update\", "
- "\"operationid\" : \"8675309\", "
- "\"name\": \"" + name + "\", \"content\": { \"location\": \"" + url + "\"}}]}";
+ void setC2RestResponse(const std::string& url, const std::string& name, const utils::optional<std::string>& persist = {}) {
+ std::string content = "{\"location\": \"" + url + "\"";
+ if (persist) {
+ content += ", \"persist\": \"" + *persist + "\"";
+ }
+ content += "}";
+ response_ =
+ "{\"operation\" : \"heartbeat\", "
+ "\"requested_operations\": [ {"
+ "\"operation\" : \"update\", "
+ "\"operationid\" : \"8675309\", "
+ "\"name\": \"" + name + "\", "
+ "\"content\": " + content + "}]}";
}
std::atomic<size_t> calls_{0};
diff --git a/extensions/http-curl/tests/HTTPIntegrationBase.h b/extensions/http-curl/tests/HTTPIntegrationBase.h
index 9e1940a..83afcbf 100644
--- a/extensions/http-curl/tests/HTTPIntegrationBase.h
+++ b/extensions/http-curl/tests/HTTPIntegrationBase.h
@@ -203,7 +203,7 @@ public:
void runAssertions() override {
using org::apache::nifi::minifi::utils::verifyLogLinePresenceInPollTime;
- assert(verifyLogLinePresenceInPollTime(std::chrono::seconds(10), "Invalid configuration payload", "update failed."));
+ assert(verifyLogLinePresenceInPollTime(std::chrono::seconds(10), "Invalid configuration payload", "update failed"));
}
void cleanup() override {
diff --git a/extensions/http-curl/tests/HttpGetIntegrationTest.cpp b/extensions/http-curl/tests/HttpGetIntegrationTest.cpp
index 9891ab8..caf5ebc 100644
--- a/extensions/http-curl/tests/HttpGetIntegrationTest.cpp
+++ b/extensions/http-curl/tests/HttpGetIntegrationTest.cpp
@@ -92,14 +92,12 @@ int main(int argc, char **argv) {
new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, args.test_file));
std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo);
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr),
- content_repo,
- DEFAULT_ROOT_GROUP_NAME,
- true);
+ std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(
+ test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, args.test_file);
- std::shared_ptr<core::Processor> proc = yaml_config.getRoot(args.test_file)->findProcessorByName("invoke");
+ std::shared_ptr<core::Processor> proc = yaml_config.getRoot()->findProcessorByName("invoke");
assert(proc != nullptr);
const auto inv = std::dynamic_pointer_cast<minifi::processors::InvokeHTTP>(proc);
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp b/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp
index a0bf58b..42b1b9c 100644
--- a/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp
+++ b/extensions/http-curl/tests/VerifyInvokeHTTPTest.cpp
@@ -76,7 +76,7 @@ class VerifyInvokeHTTP : public HTTPIntegrationBase {
setProperties(processorController->getProcessor());
}
- void run(std::string flow_yml_path) override {
+ void run(const std::string& flow_yml_path, const utils::optional<std::string>& bootstrap_file = {}) override {
setupFlow(flow_yml_path);
startFlowController();
diff --git a/extensions/standard-processors/tests/integration/TestExecuteProcess.cpp b/extensions/standard-processors/tests/integration/TestExecuteProcess.cpp
index 359f032..2455ef5 100644
--- a/extensions/standard-processors/tests/integration/TestExecuteProcess.cpp
+++ b/extensions/standard-processors/tests/integration/TestExecuteProcess.cpp
@@ -54,8 +54,8 @@ int main(int argc, char **argv) {
std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
std::shared_ptr<TestRepository> repo =
std::static_pointer_cast<TestRepository>(test_repo);
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<
- TestFlowController>(test_repo, test_repo, content_repo);
+ std::shared_ptr<minifi::FlowController> controller =
+ std::make_shared<TestFlowController>(test_repo, test_repo, content_repo);
utils::Identifier processoruuid = processor->getUUID();
assert(processoruuid);
diff --git a/encrypt-config/ConfigFileEncryptor.h b/libminifi/include/Defaults.h
similarity index 56%
copy from encrypt-config/ConfigFileEncryptor.h
copy to libminifi/include/Defaults.h
index 169edc1..f1efd52 100644
--- a/encrypt-config/ConfigFileEncryptor.h
+++ b/libminifi/include/Defaults.h
@@ -14,21 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#pragma once
-
-#include "ConfigFile.h"
-#include "utils/EncryptionUtils.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace encrypt_config {
-
-uint32_t encryptSensitivePropertiesInFile(ConfigFile& config_file, const utils::crypto::Bytes& encryption_key);
+#pragma once
-} // namespace encrypt_config
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+#ifdef WIN32
+#define DEFAULT_NIFI_CONFIG_YML "\\conf\\config.yml"
+#define DEFAULT_NIFI_PROPERTIES_FILE "\\conf\\minifi.properties"
+#define DEFAULT_LOG_PROPERTIES_FILE "\\conf\\minifi-log.properties"
+#define DEFAULT_UID_PROPERTIES_FILE "\\conf\\minifi-uid.properties"
+#define DEFAULT_BOOTSTRAP_FILE "\\conf\\bootstrap.conf"
+#else
+#define DEFAULT_NIFI_CONFIG_YML "./conf/config.yml"
+#define DEFAULT_NIFI_PROPERTIES_FILE "./conf/minifi.properties"
+#define DEFAULT_LOG_PROPERTIES_FILE "./conf/minifi-log.properties"
+#define DEFAULT_UID_PROPERTIES_FILE "./conf/minifi-uid.properties"
+#define DEFAULT_BOOTSTRAP_FILE "./conf/bootstrap.conf"
+#endif
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index b566ff8..9a41134 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -46,6 +46,7 @@
#include "core/state/nodes/FlowInformation.h"
#include "core/state/nodes/MetricsBase.h"
#include "core/state/UpdateController.h"
+#include "c2/C2Client.h"
#include "CronDrivenSchedulingAgent.h"
#include "EventDrivenSchedulingAgent.h"
#include "FlowControlProtocol.h"
@@ -53,6 +54,7 @@
#include "properties/Configure.h"
#include "TimerDrivenSchedulingAgent.h"
#include "utils/Id.h"
+#include "utils/file/FileSystem.h"
namespace org {
namespace apache {
@@ -66,27 +68,19 @@ namespace minifi {
* Flow Controller class. Generally used by FlowController factory
* as a singleton.
*/
-class FlowController : public core::controller::ControllerServiceProvider, public state::response::NodeReporter, public state::StateMonitor, public std::enable_shared_from_this<FlowController> {
+class FlowController : public core::controller::ControllerServiceProvider, public state::StateMonitor, public c2::C2Client, public std::enable_shared_from_this<FlowController> {
public:
- /**
- * Flow controller constructor
- */
- explicit FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure,
- std::unique_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<core::ContentRepository> content_repo, std::string name, bool headless_mode);
-
- explicit FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure,
- std::unique_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<core::ContentRepository> content_repo)
- : FlowController(std::move(provenance_repo), std::move(flow_file_repo), std::move(configure), std::move(flow_configuration), std::move(content_repo), DEFAULT_ROOT_GROUP_NAME, false) {
- }
-
- explicit FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure,
- std::unique_ptr<core::FlowConfiguration> flow_configuration)
+ FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<Configure> configure, std::unique_ptr<core::FlowConfiguration> flow_configuration,
+ std::shared_ptr<core::ContentRepository> content_repo, std::string name = DEFAULT_ROOT_GROUP_NAME,
+ bool headless_mode = false, std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>());
+
+ FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<Configure> configure, std::unique_ptr<core::FlowConfiguration> flow_configuration,
+ std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<utils::file::FileSystem> filesystem)
: FlowController(std::move(provenance_repo), std::move(flow_file_repo), std::move(configure), std::move(flow_configuration),
- std::make_shared<core::repository::FileSystemRepository>(), DEFAULT_ROOT_GROUP_NAME, false) {
- content_repo_->initialize(configuration_);
- }
+ std::move(content_repo), DEFAULT_ROOT_GROUP_NAME, false, std::move(filesystem)) {}
- // Destructor
~FlowController() override;
// Get the provenance repository
@@ -118,7 +112,7 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
}
// Unload the current flow YAML, clean the root process group and all its children
int16_t stop() override;
- int16_t applyUpdate(const std::string &source, const std::string &configuration) override;
+ int16_t applyUpdate(const std::string &source, const std::string &configuration, bool persist) override;
int16_t drainRepositories() override {
return -1;
}
@@ -169,6 +163,9 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
}
utils::Identifier getComponentUUID() const override {
+ if (!root_) {
+ return {};
+ }
return root_->getUUID();
}
@@ -180,6 +177,10 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
return "0";
}
+ utils::Identifier getControllerUUID() const override {
+ return getUUID();
+ }
+
/**
* Creates a controller service through the controller service provider impl.
* @param type class name
@@ -294,19 +295,6 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
void disableAllControllerServices() override;
/**
- * Retrieves metrics node
- * @return metrics response node
- */
- std::shared_ptr<state::response::ResponseNode> getMetricsNode(const std::string& metricsClass) const override;
-
- /**
- * Retrieves root nodes configured to be included in heartbeat
- * @param includeManifest -- determines if manifest is to be included
- * @return a list of response nodes
- */
- std::vector<std::shared_ptr<state::response::ResponseNode>> getHeartbeatNodes(bool includeManifest) const override;
-
- /**
* Retrieves the agent manifest to be sent as a response to C2 DESCRIBE manifest
* @return the agent manifest response node
*/
@@ -316,23 +304,11 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
std::vector<BackTrace> getTraces() override;
- void initializeC2();
- void stopC2();
-
protected:
- void loadC2ResponseConfiguration();
- void loadC2ResponseConfiguration(const std::string &prefix);
- std::shared_ptr<state::response::ResponseNode> loadC2ResponseConfiguration(const std::string &prefix, std::shared_ptr<state::response::ResponseNode>);
-
// function to load the flow file repo.
void loadFlowRepo();
void initializeExternalComponents();
- /**
- * Initializes flow controller paths.
- */
- virtual void initializePaths(const std::string &adjustedFilename);
-
utils::optional<std::chrono::milliseconds> loadShutdownTimeoutFromConfiguration();
private:
@@ -347,24 +323,12 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
// flow controller mutex
std::recursive_mutex mutex_;
- // Root Process Group
- std::shared_ptr<core::ProcessGroup> root_;
// Whether it is running
std::atomic<bool> running_;
std::atomic<bool> updating_;
- // conifiguration filename
- std::string configuration_filename_;
- std::atomic<bool> c2_initialized_;
- std::atomic<bool> flow_update_;
- std::atomic<bool> c2_enabled_;
// Whether it has already been initialized (load the flow XML already)
std::atomic<bool> initialized_;
- // Provenance Repo
- std::shared_ptr<core::Repository> provenance_repo_;
- // FlowFile Repo
- std::shared_ptr<core::Repository> flow_file_repo_;
- std::shared_ptr<core::ContentRepository> content_repo_;
// Thread pool for schedulers
utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_;
// Flow Timer Scheduler
@@ -375,29 +339,15 @@ class FlowController : public core::controller::ControllerServiceProvider, publi
std::shared_ptr<CronDrivenSchedulingAgent> cron_scheduler_;
// FlowControl Protocol
std::unique_ptr<FlowControlProtocol> protocol_;
- std::shared_ptr<Configure> configuration_;
std::shared_ptr<core::controller::ControllerServiceMap> controller_service_map_;
std::shared_ptr<core::controller::ControllerServiceProvider> controller_service_provider_;
- // flow configuration object.
- std::unique_ptr<core::FlowConfiguration> flow_configuration_;
// metrics information
std::chrono::steady_clock::time_point start_time_;
- mutable std::mutex metrics_mutex_;
- // root_nodes cache
- std::map<std::string, std::shared_ptr<state::response::ResponseNode>> root_response_nodes_;
- // metrics cache
- std::map<std::string, std::shared_ptr<state::response::ResponseNode>> device_information_;
- // metrics cache
- std::map<std::string, std::shared_ptr<state::response::ResponseNode>> component_metrics_;
- std::map<uint8_t, std::vector<std::shared_ptr<state::response::ResponseNode>>> component_metrics_by_id_;
- // metrics last run
- std::chrono::steady_clock::time_point last_metrics_capture_;
private:
std::chrono::milliseconds shutdown_check_interval_{1000};
std::shared_ptr<logging::Logger> logger_;
std::string serial_number_;
- std::unique_ptr<state::UpdateController> c2_agent_;
};
} // namespace minifi
diff --git a/libminifi/include/c2/C2Agent.h b/libminifi/include/c2/C2Agent.h
index 5f38e48..6333635 100644
--- a/libminifi/include/c2/C2Agent.h
+++ b/libminifi/include/c2/C2Agent.h
@@ -15,8 +15,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_C2_C2AGENT_H_
-#define LIBMINIFI_INCLUDE_C2_C2AGENT_H_
+
+#pragma once
#include <map>
#include <string>
@@ -40,6 +40,7 @@
#include "utils/Id.h"
#include "utils/MinifiConcurrentQueue.h"
#include "utils/ThreadPool.h"
+#include "utils/file/FileSystem.h"
namespace org {
namespace apache {
@@ -48,7 +49,6 @@ namespace minifi {
namespace c2 {
-#define C2_AGENT_UPDATE_NAME "C2UpdatePolicy"
/**
* Purpose and Justification: C2 agent will be the mechanism that will abstract the protocol to do the work.
*
@@ -62,9 +62,12 @@ namespace c2 {
*/
class C2Agent : public state::UpdateController {
public:
+ static constexpr const char* UPDATE_NAME = "C2UpdatePolicy";
+
C2Agent(core::controller::ControllerServiceProvider* controller,
const std::shared_ptr<state::StateMonitor> &updateSink,
- const std::shared_ptr<Configure> &configure);
+ const std::shared_ptr<Configure> &configure,
+ const std::shared_ptr<utils::file::FileSystem>& filesystem = std::make_shared<utils::file::FileSystem>());
virtual ~C2Agent() noexcept {
delete protocol_.load();
}
@@ -85,14 +88,8 @@ class C2Agent : public state::UpdateController {
}
protected:
- /**
- * Restarts this agent.
- */
void restart_agent();
- /**
- * Update agent per the provided C2 update from c2 server or triggers
- */
void update_agent();
/**
@@ -101,9 +98,6 @@ class C2Agent : public state::UpdateController {
*/
void checkTriggers();
- /**
- * Configure the C2 agent
- */
void configure(const std::shared_ptr<Configure> &configure, bool reconfigure = true);
/**
@@ -118,12 +112,6 @@ class C2Agent : public state::UpdateController {
* Extract the payload
* @param resp payload to be moved into the function.
*/
- void extractPayload(const C2Payload &&resp);
-
- /**
- * Extract the payload
- * @param payload reference.
- */
void extractPayload(const C2Payload &resp);
/**
@@ -167,6 +155,15 @@ class C2Agent : public state::UpdateController {
*/
C2Payload prepareConfigurationOptions(const C2ContentResponse &resp) const;
+ private:
+ utils::TaskRescheduleInfo produce();
+ utils::TaskRescheduleInfo consume();
+
+ bool handleConfigurationUpdate(const C2ContentResponse &resp);
+
+ utils::optional<std::string> fetchFlow(const std::string& uri) const;
+
+ protected:
std::timed_mutex metrics_mutex_;
std::map<std::string, std::shared_ptr<state::response::ResponseNode>> metrics_map_;
@@ -198,9 +195,6 @@ class C2Agent : public state::UpdateController {
// function that performs the heartbeat
std::function<utils::TaskRescheduleInfo()> c2_producer_;
- // function that acts upon the
- std::function<utils::TaskRescheduleInfo()> c2_consumer_;
-
// reference to the update sink, against which we will execute updates.
std::shared_ptr<state::StateMonitor> update_sink_;
@@ -215,6 +209,8 @@ class C2Agent : public state::UpdateController {
// shared pointer to the configuration of this agent
std::shared_ptr<Configure> configuration_;
+ std::shared_ptr<utils::file::FileSystem> filesystem_;
+
// shared pointer to the running c2 configuration.
std::shared_ptr<Configure> running_c2_configuration;
@@ -251,4 +247,3 @@ class C2Agent : public state::UpdateController {
} // namespace apache
} // namespace org
-#endif // LIBMINIFI_INCLUDE_C2_C2AGENT_H_
diff --git a/libminifi/include/c2/C2Client.h b/libminifi/include/c2/C2Client.h
new file mode 100644
index 0000000..7f81761
--- /dev/null
+++ b/libminifi/include/c2/C2Client.h
@@ -0,0 +1,88 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <vector>
+#include <map>
+#include <string>
+#include <mutex>
+#include "core/state/UpdateController.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include "properties/Configure.h"
+#include "core/logging/Logger.h"
+#include "core/state/nodes/MetricsBase.h"
+#include "core/Repository.h"
+#include "core/ContentRepository.h"
+#include "core/ProcessGroup.h"
+#include "core/Flow.h"
+#include "utils/file/FileSystem.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace c2 {
+
+class C2Client : public core::Flow, public state::response::NodeReporter {
+ public:
+ C2Client(
+ std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
+ std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::ContentRepository> content_repo,
+ std::unique_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<utils::file::FileSystem> filesystem,
+ std::shared_ptr<logging::Logger> logger = logging::LoggerFactory<C2Client>::getLogger());
+
+ void initialize(core::controller::ControllerServiceProvider* controller, const std::shared_ptr<state::StateMonitor> &update_sink);
+
+ std::shared_ptr<state::response::ResponseNode> getMetricsNode(const std::string& metrics_class) const override;
+
+ std::vector<std::shared_ptr<state::response::ResponseNode>> getHeartbeatNodes(bool include_manifest) const override;
+
+ void stopC2();
+
+ protected:
+ bool isC2Enabled() const;
+
+ private:
+ void initializeComponentMetrics();
+ void loadC2ResponseConfiguration(const std::string &prefix);
+ std::shared_ptr<state::response::ResponseNode> loadC2ResponseConfiguration(const std::string &prefix, std::shared_ptr<state::response::ResponseNode> prev_node);
+
+ protected:
+ std::shared_ptr<Configure> configuration_;
+ std::shared_ptr<utils::file::FileSystem> filesystem_;
+ std::unique_ptr<state::UpdateController> c2_agent_;
+
+ private:
+ std::atomic_bool initialized_{false};
+ std::shared_ptr<logging::Logger> logger_;
+
+ mutable std::mutex metrics_mutex_;
+ std::map<std::string, std::shared_ptr<state::response::ResponseNode>> root_response_nodes_;
+ std::map<std::string, std::shared_ptr<state::response::ResponseNode>> component_metrics_;
+
+ protected:
+ std::atomic<bool> flow_update_{false};
+};
+
+} // namespace c2
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
diff --git a/libminifi/include/c2/C2Payload.h b/libminifi/include/c2/C2Payload.h
index d46e1a2..0201e86 100644
--- a/libminifi/include/c2/C2Payload.h
+++ b/libminifi/include/c2/C2Payload.h
@@ -125,9 +125,6 @@ class C2Payload : public state::Update {
*/
bool validate() override { return true; }
- /**
- * Get content responses from this payload.
- */
const std::vector<C2ContentResponse> &getContent() const noexcept { return content_; }
/**
@@ -164,10 +161,9 @@ class C2Payload : public state::Update {
bool isContainer() const noexcept { return is_container_; }
void setContainer(bool is_container) noexcept { is_container_ = is_container; }
- /**
- * Get nested payloads.
- */
- const std::vector<C2Payload> &getNestedPayloads() const noexcept { return payloads_; }
+ const std::vector<C2Payload> &getNestedPayloads() const & noexcept { return payloads_; }
+
+ std::vector<C2Payload>&& getNestedPayloads() && noexcept {return std::move(payloads_);}
void reservePayloads(size_t new_capacity) { payloads_.reserve(new_capacity); }
diff --git a/libminifi/include/core/ConfigurationFactory.h b/libminifi/include/core/ConfigurationFactory.h
index 14117a3..02aa931 100644
--- a/libminifi/include/core/ConfigurationFactory.h
+++ b/libminifi/include/core/ConfigurationFactory.h
@@ -32,27 +32,32 @@ namespace minifi {
namespace core {
template<typename T>
-typename std::enable_if<!class_operations<T>::value, T*>::type instantiate(const std::shared_ptr<core::Repository> &repo, const std::shared_ptr<core::Repository> &flow_file_repo,
- const std::shared_ptr<core::ContentRepository> &content_repo, std::shared_ptr<Configure> configuration,
- const std::string path) {
+typename std::enable_if<!class_operations<T>::value, T*>::type instantiate(
+ const std::shared_ptr<core::Repository> &repo, const std::shared_ptr<core::Repository> &flow_file_repo,
+ const std::shared_ptr<core::ContentRepository> &content_repo, std::shared_ptr<Configure> configuration,
+ const utils::optional<std::string>& path, const std::shared_ptr<utils::file::FileSystem>& filesystem) {
throw std::runtime_error("Cannot instantiate class");
}
template<typename T>
-typename std::enable_if<class_operations<T>::value, T*>::type instantiate(const std::shared_ptr<core::Repository> &repo, const std::shared_ptr<core::Repository> &flow_file_repo,
- const std::shared_ptr<core::ContentRepository> &content_repo, const std::shared_ptr<io::StreamFactory> &stream_factory,
- std::shared_ptr<Configure> configuration, const std::string path) {
- return new T(repo, flow_file_repo, content_repo, stream_factory, configuration, path);
+typename std::enable_if<class_operations<T>::value, T*>::type instantiate(
+ const std::shared_ptr<core::Repository> &repo, const std::shared_ptr<core::Repository> &flow_file_repo,
+ const std::shared_ptr<core::ContentRepository> &content_repo, const std::shared_ptr<io::StreamFactory> &stream_factory,
+ std::shared_ptr<Configure> configuration, const utils::optional<std::string>& path,
+ const std::shared_ptr<utils::file::FileSystem>& filesystem) {
+ return new T(repo, flow_file_repo, content_repo, stream_factory, configuration, path, filesystem);
}
/**
* Configuration factory is used to create a new FlowConfiguration
* object.
*/
-std::unique_ptr<core::FlowConfiguration> createFlowConfiguration(std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo,
- std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
- std::shared_ptr<io::StreamFactory> stream_factory, const std::string configuration_class_name, const std::string path = "",
- bool fail_safe = false);
+std::unique_ptr<core::FlowConfiguration> createFlowConfiguration(
+ std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
+ std::shared_ptr<io::StreamFactory> stream_factory, const std::string& configuration_class_name,
+ const utils::optional<std::string>& path = {}, std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(),
+ bool fail_safe = false);
} // namespace core
} // namespace minifi
diff --git a/libminifi/include/core/Flow.h b/libminifi/include/core/Flow.h
new file mode 100644
index 0000000..2857aad
--- /dev/null
+++ b/libminifi/include/core/Flow.h
@@ -0,0 +1,59 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <utility>
+#include "core/ProcessGroup.h"
+#include "core/Repository.h"
+#include "core/ContentRepository.h"
+#include "core/FlowConfiguration.h"
+#include "utils/Id.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace core {
+
+class Flow {
+ protected:
+ Flow(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<core::FlowConfiguration> flow_configuration)
+ : provenance_repo_(std::move(provenance_repo)),
+ flow_file_repo_(std::move(flow_file_repo)),
+ content_repo_(std::move(content_repo)),
+ flow_configuration_(std::move(flow_configuration)) {}
+
+ virtual ~Flow() = default;
+
+ virtual utils::Identifier getControllerUUID() const = 0;
+
+ std::shared_ptr<core::ProcessGroup> root_;
+ std::unique_ptr<core::FlowConfiguration> flow_configuration_;
+ std::shared_ptr<core::Repository> provenance_repo_;
+ std::shared_ptr<core::Repository> flow_file_repo_;
+ std::shared_ptr<core::ContentRepository> content_repo_;
+};
+
+} // namespace core
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
diff --git a/libminifi/include/core/FlowConfiguration.h b/libminifi/include/core/FlowConfiguration.h
index 974535e..685cb67 100644
--- a/libminifi/include/core/FlowConfiguration.h
+++ b/libminifi/include/core/FlowConfiguration.h
@@ -21,6 +21,7 @@
#include <memory>
#include <string>
#include <vector>
+#include <utility>
#include "core/Core.h"
#include "Connection.h"
@@ -38,6 +39,8 @@
#include "core/ProcessGroup.h"
#include "io/StreamFactory.h"
#include "core/state/nodes/FlowInformation.h"
+#include "utils/file/FileSystem.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -63,14 +66,16 @@ class FlowConfiguration : public CoreComponent {
* Constructor that will be used for configuring
* the flow controller.
*/
- explicit FlowConfiguration(std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::ContentRepository> content_repo,
- std::shared_ptr<io::StreamFactory> stream_factory, std::shared_ptr<Configure> configuration, const std::string path)
+ explicit FlowConfiguration(std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<io::StreamFactory> stream_factory,
+ std::shared_ptr<Configure> configuration, const utils::optional<std::string>& path,
+ std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>())
: CoreComponent(core::getClassName<FlowConfiguration>()),
- config_path_(path),
- flow_file_repo_(flow_file_repo),
- content_repo_(content_repo),
- stream_factory_(stream_factory),
+ flow_file_repo_(std::move(flow_file_repo)),
+ content_repo_(std::move(content_repo)),
+ stream_factory_(std::move(stream_factory)),
configuration_(configuration),
+ filesystem_(std::move(filesystem)),
logger_(logging::LoggerFactory<FlowConfiguration>::getLogger()) {
controller_services_ = std::make_shared<core::controller::ControllerServiceMap>();
service_provider_ = std::make_shared<core::controller::StandardControllerServiceProvider>(controller_services_, nullptr, configuration);
@@ -79,11 +84,21 @@ class FlowConfiguration : public CoreComponent {
configuration->get(Configure::nifi_c2_flow_url, flowUrl);
flow_version_ = std::make_shared<state::response::FlowVersion>(flowUrl, bucket_id, flowId);
+ if (!path) {
+ logger_->log_error("Configuration path is not specified.");
+ } else {
+ config_path_ = utils::file::PathUtils::canonicalize(*path);
+ if (!config_path_) {
+ logger_->log_error("Couldn't find config file \"%s\".", *path);
+ config_path_ = path;
+ }
+ }
+
// it is okay if this has already been called
initialize_static_functions();
}
- virtual ~FlowConfiguration();
+ ~FlowConfiguration() override;
// Create Processor (Node/Input/Output Port) based on the name
std::shared_ptr<core::Processor> createProcessor(std::string name, utils::Identifier &uuid);
@@ -110,16 +125,18 @@ class FlowConfiguration : public CoreComponent {
return configuration_;
}
+ bool persist(const std::string& configuration);
+
/**
* Returns the configuration path string
* @return config_path_
*/
- const std::string &getConfigurationPath() {
+ const utils::optional<std::string> &getConfigurationPath() {
return config_path_;
}
virtual std::unique_ptr<core::ProcessGroup> getRoot() {
- return getRoot(config_path_);
+ return nullptr;
}
std::unique_ptr<core::ProcessGroup> updateFromPayload(const std::string &source, const std::string &yamlConfigPayload);
@@ -128,15 +145,6 @@ class FlowConfiguration : public CoreComponent {
return nullptr;
}
- /**
- * Base implementation that returns a null root pointer.
- * @return Extensions should return a non-null pointer in order to
- * properly configure flow controller.
- */
- virtual std::unique_ptr<core::ProcessGroup> getRoot(const std::string &from_config) {
- return nullptr;
- }
-
std::shared_ptr<core::controller::StandardControllerServiceProvider> &getControllerServiceProvider() {
return service_provider_;
}
@@ -168,7 +176,7 @@ class FlowConfiguration : public CoreComponent {
// based, shared controller service map.
std::shared_ptr<core::controller::ControllerServiceMap> controller_services_;
// configuration path
- std::string config_path_;
+ utils::optional<std::string> config_path_;
// flow file repo
std::shared_ptr<core::Repository> flow_file_repo_;
// content repository.
@@ -178,6 +186,8 @@ class FlowConfiguration : public CoreComponent {
std::shared_ptr<Configure> configuration_;
std::shared_ptr<state::response::FlowVersion> flow_version_;
+ std::shared_ptr<utils::file::FileSystem> filesystem_;
+
private:
std::shared_ptr<logging::Logger> logger_;
};
diff --git a/libminifi/include/core/state/UpdateController.h b/libminifi/include/core/state/UpdateController.h
index 2c93657..cb44748 100644
--- a/libminifi/include/core/state/UpdateController.h
+++ b/libminifi/include/core/state/UpdateController.h
@@ -206,7 +206,7 @@ class StateMonitor : public StateController {
* < 0 is an error code
* 0 is success
*/
- virtual int16_t applyUpdate(const std::string & source, const std::string &configuration) = 0;
+ virtual int16_t applyUpdate(const std::string & source, const std::string &configuration, bool persist = false) = 0;
/**
* Apply an update that the agent must decode. This is useful for certain operations
diff --git a/libminifi/include/core/yaml/YamlConfiguration.h b/libminifi/include/core/yaml/YamlConfiguration.h
index 0674124..4ff9a41 100644
--- a/libminifi/include/core/yaml/YamlConfiguration.h
+++ b/libminifi/include/core/yaml/YamlConfiguration.h
@@ -30,6 +30,8 @@
#include "sitetosite/SiteToSite.h"
#include "utils/Id.h"
#include "utils/StringUtils.h"
+#include "utils/file/FileSystem.h"
+#include "utils/OptionalUtils.h"
#include "yaml-cpp/yaml.h"
namespace org {
@@ -38,7 +40,6 @@ namespace nifi {
namespace minifi {
namespace core {
-#define DEFAULT_FLOW_YAML_FILE_NAME "conf/config.yml"
#define CONFIG_YAML_FLOW_CONTROLLER_KEY "Flow Controller"
#define CONFIG_YAML_PROCESSORS_KEY "Processors"
#define CONFIG_YAML_CONNECTIONS_KEY "Connections"
@@ -56,31 +57,36 @@ namespace core {
class YamlConfiguration : public FlowConfiguration {
public:
- explicit YamlConfiguration(std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::ContentRepository> content_repo,
- std::shared_ptr<io::StreamFactory> stream_factory, std::shared_ptr<Configure> configuration, const std::string path = DEFAULT_FLOW_YAML_FILE_NAME)
- : FlowConfiguration(repo, flow_file_repo, content_repo, stream_factory, configuration, path),
- logger_(logging::LoggerFactory<YamlConfiguration>::getLogger()) {
- stream_factory_ = stream_factory;
- if (IsNullOrEmpty(config_path_)) {
- config_path_ = DEFAULT_FLOW_YAML_FILE_NAME;
- }
- }
+ explicit YamlConfiguration(const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_file_repo,
+ const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<io::StreamFactory>& stream_factory,
+ const std::shared_ptr<Configure>& configuration, const utils::optional<std::string>& path = {},
+ const std::shared_ptr<utils::file::FileSystem>& filesystem = std::make_shared<utils::file::FileSystem>());
- virtual ~YamlConfiguration() = default;
+ ~YamlConfiguration() override = default;
/**
* Returns a shared pointer to a ProcessGroup object containing the
- * flow configuration. The yamlConfigFile argument is the location
- * of a YAML file containing the flow configuration.
+ * flow configuration.
*
- * @param yamlConfigFile a string holding the location of the YAML file
- * to be loaded into a flow configuration tree
* @return the root ProcessGroup node of the flow
* configuration tree
*/
- virtual std::unique_ptr<core::ProcessGroup> getRoot(const std::string &yamlConfigFile) {
- YAML::Node rootYamlNode = YAML::LoadFile(yamlConfigFile);
- return getYamlRoot(&rootYamlNode);
+ std::unique_ptr<core::ProcessGroup> getRoot() override {
+ if (!config_path_) {
+ logger_->log_error("Cannot instantiate flow, no config file is set.");
+ throw Exception(ExceptionType::FLOW_EXCEPTION, "No config file specified");
+ }
+ utils::optional<std::string> configuration = filesystem_->read(config_path_.value());
+ if (!configuration) {
+ return nullptr;
+ }
+ try {
+ YAML::Node rootYamlNode = YAML::Load(configuration.value());
+ return getYamlRoot(&rootYamlNode);
+ } catch(...) {
+ logger_->log_error("Invalid yaml configuration file");
+ throw;
+ }
}
/**
@@ -116,7 +122,7 @@ class YamlConfiguration : public FlowConfiguration {
* @return the root ProcessGroup node of the flow
* configuration tree
*/
- std::unique_ptr<core::ProcessGroup> getRootFromPayload(const std::string &yamlConfigPayload) {
+ std::unique_ptr<core::ProcessGroup> getRootFromPayload(const std::string &yamlConfigPayload) override {
try {
YAML::Node rootYamlNode = YAML::Load(yamlConfigPayload);
return getYamlRoot(&rootYamlNode);
diff --git a/libminifi/include/properties/Configuration.h b/libminifi/include/properties/Configuration.h
index c15df6c..85cbf73 100644
--- a/libminifi/include/properties/Configuration.h
+++ b/libminifi/include/properties/Configuration.h
@@ -52,6 +52,7 @@ class Configuration : public Properties {
// nifi.flow.configuration.file
static constexpr const char *nifi_default_directory = "nifi.default.directory";
static constexpr const char *nifi_flow_configuration_file = "nifi.flow.configuration.file";
+ static constexpr const char *nifi_flow_configuration_encrypt = "nifi.flow.configuration.encrypt";
static constexpr const char *nifi_flow_configuration_file_exit_failure = "nifi.flow.configuration.file.exit.onfailure";
static constexpr const char *nifi_flow_configuration_file_backup_update = "nifi.flow.configuration.backup.on.update";
static constexpr const char *nifi_flow_engine_threads = "nifi.flow.engine.threads";
diff --git a/libminifi/include/properties/Decryptor.h b/libminifi/include/properties/Decryptor.h
index 06bf830..d344d3c 100644
--- a/libminifi/include/properties/Decryptor.h
+++ b/libminifi/include/properties/Decryptor.h
@@ -17,9 +17,11 @@
#pragma once
#include <string>
+#include <utility>
#include "utils/EncryptionUtils.h"
#include "utils/OptionalUtils.h"
+#include "utils/EncryptionProvider.h"
namespace org {
namespace apache {
@@ -28,27 +30,26 @@ namespace minifi {
class Decryptor {
public:
- explicit Decryptor(const utils::crypto::Bytes& encryption_key);
+ explicit Decryptor(utils::crypto::EncryptionProvider provider)
+ : provider_(std::move(provider)) {}
- static bool isValidEncryptionMarker(const utils::optional<std::string>& encryption_marker);
- std::string decrypt(const std::string& encrypted_text) const;
+ static bool isValidEncryptionMarker(const utils::optional<std::string>& encryption_marker) {
+ return encryption_marker && *encryption_marker == utils::crypto::EncryptionType::name();
+ }
- static utils::optional<Decryptor> create(const std::string& minifi_home);
+ std::string decrypt(const std::string& encrypted_text) const {
+ return provider_.decrypt(encrypted_text);
+ }
+
+ static utils::optional<Decryptor> create(const std::string& minifi_home) {
+ return utils::crypto::EncryptionProvider::create(minifi_home)
+ | utils::map([](const utils::crypto::EncryptionProvider& provider) {return Decryptor{provider};});
+ }
private:
- const utils::crypto::Bytes encryption_key_;
+ const utils::crypto::EncryptionProvider provider_;
};
-inline Decryptor::Decryptor(const utils::crypto::Bytes& encryption_key) : encryption_key_(encryption_key) {}
-
-inline bool Decryptor::isValidEncryptionMarker(const utils::optional<std::string>& encryption_marker) {
- return encryption_marker && *encryption_marker == utils::crypto::EncryptionType::name();
-}
-
-inline std::string Decryptor::decrypt(const std::string& encrypted_text) const {
- return utils::crypto::decrypt(encrypted_text, encryption_key_);
-}
-
} // namespace minifi
} // namespace nifi
} // namespace apache
diff --git a/encrypt-config/EncryptConfig.h b/libminifi/include/utils/CollectionUtils.h
similarity index 52%
copy from encrypt-config/EncryptConfig.h
copy to libminifi/include/utils/CollectionUtils.h
index dd8b351..46b3f4f 100644
--- a/encrypt-config/EncryptConfig.h
+++ b/libminifi/include/utils/CollectionUtils.h
@@ -14,39 +14,45 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#pragma once
-#include <string>
+#pragma once
-#include "utils/EncryptionUtils.h"
+#include <algorithm>
+#include <utility>
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
-namespace encrypt_config {
-
-class EncryptConfig {
- public:
- EncryptConfig(int argc, char* argv[]);
- void encryptSensitiveProperties() const;
+namespace utils {
- private:
- std::string bootstrapFilePath() const;
- std::string propertiesFilePath() const;
+namespace internal {
- static std::string parseMinifiHomeFromTheOptions(int argc, char* argv[]);
+template<typename T, typename Arg, typename = void>
+struct find_in_range {
+ static auto call(const T& range, const Arg& arg) -> decltype(std::find(range.begin(), range.end(), arg)) {
+ return std::find(range.begin(), range.end(), arg);
+ }
+};
- utils::crypto::Bytes getEncryptionKey() const;
- std::string hexDecodeAndValidateKey(const std::string& key) const;
- void writeEncryptionKeyToBootstrapFile(const utils::crypto::Bytes& encryption_key) const;
+template<typename T, typename Arg>
+struct find_in_range<T, Arg, decltype(std::declval<const T&>().find(std::declval<const Arg&>()), void())> {
+ static auto call(const T& range, const Arg& arg) -> decltype(range.find(arg)) {
+ return range.find(arg);
+ }
+};
- void encryptSensitiveProperties(const utils::crypto::Bytes& encryption_key) const;
+} // namespace internal
- const std::string minifi_home_;
-};
+template<typename T, typename U>
+bool haveCommonItem(const T& a, const U& b) {
+ using Item = typename T::value_type;
+ return std::any_of(a.begin(), a.end(), [&] (const Item& item) {
+ return internal::find_in_range<U, Item>::call(b, item) != b.end();
+ });
+}
-} // namespace encrypt_config
+} // namespace utils
} // namespace minifi
} // namespace nifi
} // namespace apache
diff --git a/encrypt-config/EncryptConfig.h b/libminifi/include/utils/EncryptionProvider.h
similarity index 61%
copy from encrypt-config/EncryptConfig.h
copy to libminifi/include/utils/EncryptionProvider.h
index dd8b351..33815f1 100644
--- a/encrypt-config/EncryptConfig.h
+++ b/libminifi/include/utils/EncryptionProvider.h
@@ -14,39 +14,42 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
#pragma once
+#include <utility>
#include <string>
-
#include "utils/EncryptionUtils.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
-namespace encrypt_config {
+namespace utils {
+namespace crypto {
-class EncryptConfig {
+class EncryptionProvider {
public:
- EncryptConfig(int argc, char* argv[]);
- void encryptSensitiveProperties() const;
+ explicit EncryptionProvider(Bytes encryption_key)
+ : encryption_key_(std::move(encryption_key)) {}
- private:
- std::string bootstrapFilePath() const;
- std::string propertiesFilePath() const;
+ static utils::optional<EncryptionProvider> create(const std::string& home_path);
- static std::string parseMinifiHomeFromTheOptions(int argc, char* argv[]);
+ std::string encrypt(const std::string& data) const {
+ return utils::crypto::encrypt(data, encryption_key_);
+ }
- utils::crypto::Bytes getEncryptionKey() const;
- std::string hexDecodeAndValidateKey(const std::string& key) const;
- void writeEncryptionKeyToBootstrapFile(const utils::crypto::Bytes& encryption_key) const;
+ std::string decrypt(const std::string& data) const {
+ return utils::crypto::decrypt(data, encryption_key_);
+ }
- void encryptSensitiveProperties(const utils::crypto::Bytes& encryption_key) const;
-
- const std::string minifi_home_;
+ private:
+ const Bytes encryption_key_;
};
-} // namespace encrypt_config
+} // namespace crypto
+} // namespace utils
} // namespace minifi
} // namespace nifi
} // namespace apache
diff --git a/libminifi/include/utils/EncryptionUtils.h b/libminifi/include/utils/EncryptionUtils.h
index c691937..875b10b 100644
--- a/libminifi/include/utils/EncryptionUtils.h
+++ b/libminifi/include/utils/EncryptionUtils.h
@@ -19,6 +19,8 @@
#include <string>
#include <vector>
+#include "utils/OptionalUtils.h"
+
namespace org {
namespace apache {
namespace nifi {
@@ -44,6 +46,11 @@ struct EncryptionType {
static std::string separator();
};
+struct EncryptedData {
+ Bytes nonce;
+ Bytes ciphertext_plus_mac;
+};
+
/**
* Encrypt the input (raw version).
*
@@ -101,6 +108,14 @@ Bytes decryptRaw(const Bytes& input, const Bytes& key, const Bytes& nonce);
*/
std::string decrypt(const std::string& input, const Bytes& key);
+/**
+ * Checks if the input of the form nonce + EncryptionType::separator() + ciphertext_plus_MAC,
+ * indicating that it is most likely encrypted.
+ */
+bool isEncrypted(const std::string& input);
+
+EncryptedData parseEncrypted(const std::string& input);
+
} // namespace crypto
} // namespace utils
} // namespace minifi
diff --git a/libminifi/include/utils/StringUtils.h b/libminifi/include/utils/StringUtils.h
index 85d29d9..04201cb 100644
--- a/libminifi/include/utils/StringUtils.h
+++ b/libminifi/include/utils/StringUtils.h
@@ -33,6 +33,7 @@
#include <type_traits>
#include "utils/FailurePolicy.h"
#include "utils/GeneralUtils.h"
+#include "utils/OptionalUtils.h"
#if defined(WIN32) || (__cplusplus >= 201103L && (!defined(__GLIBCXX__) || (__cplusplus >= 201402L) || (defined(_GLIBCXX_RELEASE) && _GLIBCXX_RELEASE > 4)))
#define HAVE_REGEX_CPP 1
@@ -80,6 +81,8 @@ class StringUtils {
*/
static bool StringToBool(std::string input, bool &output);
+ static utils::optional<bool> toBool(const std::string& input);
+
// Trim String utils
/**
@@ -87,7 +90,7 @@ class StringUtils {
* @param s incoming string
* @returns modified string
*/
- static std::string trim(std::string s);
+ static std::string trim(const std::string& s);
/**
* Trims left most part of a string
@@ -143,10 +146,16 @@ class StringUtils {
return std::equal(endString.rbegin(), endString.rend(), value.rbegin(), [](unsigned char lc, unsigned char rc) {return tolower(lc) == tolower(rc);});
}
- inline static bool endsWith(const std::string &value, const std::string & endString) {
- if (endString.size() > value.size())
+ inline static bool startsWith(const std::string& value, const std::string& start_string) {
+ if (start_string.size() > value.size())
return false;
- return std::equal(endString.rbegin(), endString.rend(), value.rbegin());
+ return std::equal(start_string.begin(), start_string.end(), value.begin());
+ }
+
+ inline static bool endsWith(const std::string& value, const std::string& end_string) {
+ if (end_string.size() > value.size())
+ return false;
+ return std::equal(end_string.rbegin(), end_string.rend(), value.rbegin());
}
inline static std::string hex_ascii(const std::string& in) {
@@ -173,68 +182,68 @@ class StringUtils {
static size_t size(const CharT* str) noexcept { return std::char_traits<CharT>::length(str); }
struct detail {
- // add all args
- template<typename... SizeT>
- static size_t sum(SizeT... ns) {
- size_t result = 0;
- (void)(std::initializer_list<size_t>{( result += ns )...});
- return result; // (ns + ...)
- }
+ // add all args
+ template<typename... SizeT>
+ static size_t sum(SizeT... ns) {
+ size_t result = 0;
+ (void)(std::initializer_list<size_t>{( result += ns )...});
+ return result; // (ns + ...)
+ }
- #ifndef _MSC_VER
+#ifndef _MSC_VER
- // partial detection idiom impl, from cppreference.com
- struct nonesuch{};
+ // partial detection idiom impl, from cppreference.com
+ struct nonesuch{};
- template<typename Default, typename Void, template<class...> class Op, typename... Args>
- struct detector {
- using value_t = std::false_type;
- using type = Default;
- };
+ template<typename Default, typename Void, template<class...> class Op, typename... Args>
+ struct detector {
+ using value_t = std::false_type;
+ using type = Default;
+ };
- template<typename Default, template<class...> class Op, typename... Args>
- struct detector<Default, void_t<Op<Args...>>, Op, Args...> {
- using value_t = std::true_type;
- using type = Op<Args...>;
- };
+ template<typename Default, template<class...> class Op, typename... Args>
+ struct detector<Default, void_t<Op<Args...>>, Op, Args...> {
+ using value_t = std::true_type;
+ using type = Op<Args...>;
+ };
- template<template<class...> class Op, typename... Args>
- using is_detected = typename detector<nonesuch, void, Op, Args...>::value_t;
+ template<template<class...> class Op, typename... Args>
+ using is_detected = typename detector<nonesuch, void, Op, Args...>::value_t;
- // and operation for boolean template argument packs
- template<bool...>
- struct and_;
+ // and operation for boolean template argument packs
+ template<bool...>
+ struct and_;
- template<bool Head, bool... Tail>
- struct and_<Head, Tail...> : std::integral_constant<bool, Head && and_<Tail...>::value>
- {};
+ template<bool Head, bool... Tail>
+ struct and_<Head, Tail...> : std::integral_constant<bool, Head && and_<Tail...>::value>
+ {};
- template<bool B>
- struct and_<B> : std::integral_constant<bool, B>
- {};
+ template<bool B>
+ struct and_<B> : std::integral_constant<bool, B>
+ {};
- // implementation detail of join_pack
- template<typename CharT>
- struct str_detector {
- template<typename Str>
- using valid_string_t = decltype(std::declval<std::basic_string<CharT>>().append(std::declval<Str>()));
- };
-
- template<typename ResultT, typename CharT, typename... Strs>
- using valid_string_pack_t = typename std::enable_if<and_<is_detected<str_detector<CharT>::template valid_string_t, Strs>::value...>::value, ResultT>::type;
- #else
- // MSVC is broken without /permissive-
+ // implementation detail of join_pack
+ template<typename CharT>
+ struct str_detector {
+ template<typename Str>
+ using valid_string_t = decltype(std::declval<std::basic_string<CharT>>().append(std::declval<Str>()));
+ };
+
+ template<typename ResultT, typename CharT, typename... Strs>
+ using valid_string_pack_t = typename std::enable_if<and_<is_detected<str_detector<CharT>::template valid_string_t, Strs>::value...>::value, ResultT>::type;
+#else
+ // MSVC is broken without /permissive-
template<typename ResultT, typename...>
using valid_string_pack_t = ResultT;
- #endif
-
- template<typename CharT, typename... Strs, valid_string_pack_t<void, CharT, Strs...>* = nullptr>
- static std::basic_string<CharT> join_pack(const Strs&... strs) {
- std::basic_string<CharT> result;
- result.reserve(sum(size(strs)...));
- (void)(std::initializer_list<int>{( result.append(strs) , 0)...});
- return result;
- }
+#endif
+
+ template<typename CharT, typename... Strs, valid_string_pack_t<void, CharT, Strs...>* = nullptr>
+ static std::basic_string<CharT> join_pack(const Strs&... strs) {
+ std::basic_string<CharT> result;
+ result.reserve(sum(size(strs)...));
+ (void)(std::initializer_list<int>{( result.append(strs) , 0)...});
+ return result;
+ }
}; /* struct detail */
/**
diff --git a/libminifi/include/properties/Decryptor.h b/libminifi/include/utils/file/FileSystem.h
similarity index 55%
copy from libminifi/include/properties/Decryptor.h
copy to libminifi/include/utils/file/FileSystem.h
index 06bf830..d97f2e4 100644
--- a/libminifi/include/properties/Decryptor.h
+++ b/libminifi/include/utils/file/FileSystem.h
@@ -14,41 +14,43 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
#pragma once
#include <string>
-
-#include "utils/EncryptionUtils.h"
+#include <memory>
#include "utils/OptionalUtils.h"
+#include "utils/EncryptionProvider.h"
+#include "core/logging/LoggerConfiguration.h"
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
+namespace utils {
+namespace file {
-class Decryptor {
+class FileSystem {
public:
- explicit Decryptor(const utils::crypto::Bytes& encryption_key);
+ explicit FileSystem(bool should_encrypt = false, utils::optional<utils::crypto::EncryptionProvider> encryptor = {});
+
+ FileSystem(const FileSystem&) = delete;
+ FileSystem(FileSystem&&) = delete;
+ FileSystem& operator=(const FileSystem&) = delete;
+ FileSystem& operator=(FileSystem&&) = delete;
- static bool isValidEncryptionMarker(const utils::optional<std::string>& encryption_marker);
- std::string decrypt(const std::string& encrypted_text) const;
+ utils::optional<std::string> read(const std::string& file_name);
- static utils::optional<Decryptor> create(const std::string& minifi_home);
+ bool write(const std::string& file_name, const std::string& file_content);
private:
- const utils::crypto::Bytes encryption_key_;
+ bool should_encrypt_on_write_;
+ utils::optional<utils::crypto::EncryptionProvider> encryptor_;
+ std::shared_ptr<logging::Logger> logger_{logging::LoggerFactory<FileSystem>::getLogger()};
};
-inline Decryptor::Decryptor(const utils::crypto::Bytes& encryption_key) : encryption_key_(encryption_key) {}
-
-inline bool Decryptor::isValidEncryptionMarker(const utils::optional<std::string>& encryption_marker) {
- return encryption_marker && *encryption_marker == utils::crypto::EncryptionType::name();
-}
-
-inline std::string Decryptor::decrypt(const std::string& encrypted_text) const {
- return utils::crypto::decrypt(encrypted_text, encryption_key_);
-}
-
+} // namespace file
+} // namespace utils
} // namespace minifi
} // namespace nifi
} // namespace apache
diff --git a/libminifi/include/utils/file/FileUtils.h b/libminifi/include/utils/file/FileUtils.h
index 632eddb..2a043ce 100644
--- a/libminifi/include/utils/file/FileUtils.h
+++ b/libminifi/include/utils/file/FileUtils.h
@@ -82,6 +82,7 @@
#include "core/logging/LoggerConfiguration.h"
#include "utils/StringUtils.h"
+#include "utils/file/PathUtils.h"
#ifndef S_ISDIR
#define S_ISDIR(mode) (((mode) & S_IFMT) == S_IFDIR)
@@ -738,6 +739,13 @@ inline std::string get_executable_path() {
#endif
}
+inline std::string resolve(const std::string& base, const std::string& path) {
+ if (utils::file::isAbsolutePath(path.c_str())) {
+ return path;
+ }
+ return concat_path(base, path);
+}
+
/*
* Returns the absolute path to the directory containing the current executable
*/
diff --git a/libminifi/include/utils/file/PathUtils.h b/libminifi/include/utils/file/PathUtils.h
index 00db039..4960472 100644
--- a/libminifi/include/utils/file/PathUtils.h
+++ b/libminifi/include/utils/file/PathUtils.h
@@ -17,12 +17,14 @@
#ifndef LIBMINIFI_INCLUDE_UTILS_FILE_PATHUTILS_H_
#define LIBMINIFI_INCLUDE_UTILS_FILE_PATHUTILS_H_
+#include <climits>
#include <cctype>
#include <cinttypes>
#include <memory>
#include <string>
#include <system_error>
#include <utility>
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -62,6 +64,21 @@ inline bool isAbsolutePath(const char* const path) noexcept {
#endif
}
+inline utils::optional<std::string> canonicalize(const std::string &path) {
+ const char *resolved = nullptr;
+#ifndef WIN32
+ char full_path[PATH_MAX];
+ resolved = realpath(path.c_str(), full_path);
+#else
+ resolved = path.c_str();
+#endif
+
+ if (resolved == nullptr) {
+ return utils::nullopt;
+ }
+ return std::string(path);
+}
+
/**
* Represents filesystem space information in bytes
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 34718f3..dfad0ba 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -17,14 +17,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "FlowController.h"
#include <time.h>
-#include <sys/types.h>
-#include <sys/stat.h>
#include <vector>
-#include <queue>
#include <map>
-#include <set>
#include <chrono>
#include <future>
#include <thread>
@@ -32,6 +27,7 @@
#include <memory>
#include <string>
+#include "FlowController.h"
#include "core/state/nodes/AgentInformation.h"
#include "core/state/nodes/BuildInformation.h"
#include "core/state/nodes/DeviceInformation.h"
@@ -41,89 +37,53 @@
#include "core/state/nodes/RepositoryMetrics.h"
#include "core/state/nodes/SystemMetrics.h"
#include "core/state/ProcessorController.h"
-#include "yaml-cpp/yaml.h"
#include "c2/C2Agent.h"
-#include "core/ProcessContext.h"
#include "core/ProcessGroup.h"
-#include "utils/StringUtils.h"
#include "core/Core.h"
#include "core/ClassLoader.h"
#include "SchedulingAgent.h"
#include "core/controller/ControllerServiceProvider.h"
#include "core/logging/LoggerConfiguration.h"
#include "core/Connectable.h"
-#include "utils/file/FileUtils.h"
#include "utils/file/PathUtils.h"
+#include "utils/file/FileSystem.h"
#include "utils/HTTPClient.h"
#include "utils/GeneralUtils.h"
#include "io/NetworkPrioritizer.h"
#include "io/validation.h"
-#ifdef _MSC_VER
-#ifndef PATH_MAX
-#define PATH_MAX 260
-#endif
-#endif
-
namespace org {
namespace apache {
namespace nifi {
namespace minifi {
-#define DEFAULT_CONFIG_NAME "conf/config.yml"
-
-FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<Configure> configure,
- std::unique_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<core::ContentRepository> content_repo, const std::string name, bool headless_mode)
+FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<Configure> configure, std::unique_ptr<core::FlowConfiguration> flow_configuration,
+ std::shared_ptr<core::ContentRepository> content_repo, const std::string name, bool headless_mode,
+ std::shared_ptr<utils::file::FileSystem> filesystem)
: core::controller::ControllerServiceProvider(core::getClassName<FlowController>()),
- root_(nullptr),
+ c2::C2Client(std::move(configure), std::move(provenance_repo), std::move(flow_file_repo),
+ std::move(content_repo), std::move(flow_configuration), std::move(filesystem)),
running_(false),
updating_(false),
- c2_enabled_(true),
initialized_(false),
- provenance_repo_(provenance_repo),
- flow_file_repo_(flow_file_repo),
controller_service_map_(std::make_shared<core::controller::ControllerServiceMap>()),
thread_pool_(2, false, nullptr, "Flowcontroller threadpool"),
- flow_configuration_(std::move(flow_configuration)),
- configuration_(std::move(configure)),
- content_repo_(content_repo),
logger_(logging::LoggerFactory<FlowController>::getLogger()) {
- if (provenance_repo == nullptr)
+ if (provenance_repo_ == nullptr)
throw std::runtime_error("Provenance Repo should not be null");
- if (flow_file_repo == nullptr)
+ if (flow_file_repo_ == nullptr)
throw std::runtime_error("Flow Repo should not be null");
- if (IsNullOrEmpty(configuration_)) {
+ if (configuration_ == nullptr) {
throw std::runtime_error("Must supply a configuration.");
}
- flow_update_ = false;
- // Setup the default values
- if (flow_configuration_ != nullptr) {
- configuration_filename_ = flow_configuration_->getConfigurationPath();
- }
running_ = false;
initialized_ = false;
- c2_initialized_ = false;
- root_ = nullptr;
protocol_ = utils::make_unique<FlowControlProtocol>(this, configuration_);
if (!headless_mode) {
- std::string rawConfigFileString;
- configuration_->get(Configure::nifi_flow_configuration_file, rawConfigFileString);
-
- if (!rawConfigFileString.empty()) {
- configuration_filename_ = rawConfigFileString;
- }
-
- const auto adjustedFilename = [&]() -> std::string {
- if (configuration_filename_.empty()) { return {}; }
- if (utils::file::isAbsolutePath(configuration_filename_.c_str())) {
- return configuration_filename_;
- }
- return utils::file::FileUtils::concat_path(configuration_->getHome(), configuration_filename_);
- }();
initializeExternalComponents();
- initializePaths(adjustedFilename);
}
}
@@ -141,23 +101,6 @@ void FlowController::initializeExternalComponents() {
}
}
-void FlowController::initializePaths(const std::string &adjustedFilename) {
- const char *path = nullptr;
-#ifndef WIN32
- char full_path[PATH_MAX];
- path = realpath(adjustedFilename.c_str(), full_path);
-#else
- path = adjustedFilename.c_str();
-#endif
-
- if (path == nullptr) {
- throw std::runtime_error("Path is not specified. Either manually set MINIFI_HOME or ensure ../conf exists");
- }
- std::string pathString(path);
- configuration_filename_ = pathString;
- logger_->log_info("FlowController NiFi Configuration file %s", pathString);
-}
-
utils::optional<std::chrono::milliseconds> FlowController::loadShutdownTimeoutFromConfiguration() {
std::string shutdown_timeout_str;
if (configuration_->get(minifi::Configure::nifi_flowcontroller_drain_timeout, shutdown_timeout_str)) {
@@ -173,16 +116,12 @@ FlowController::~FlowController() {
stop();
stopC2();
unload();
+ // TODO(adebreceni): are these here on purpose, so they are destroyed first?
protocol_ = nullptr;
flow_file_repo_ = nullptr;
provenance_repo_ = nullptr;
}
-void FlowController::stopC2() {
- if (c2_agent_)
- c2_agent_->stop();
-}
-
bool FlowController::applyConfiguration(const std::string &source, const std::string &configurePayload) {
std::unique_ptr<core::ProcessGroup> newRoot;
try {
@@ -320,17 +259,18 @@ void FlowController::load(const std::shared_ptr<core::ProcessGroup> &root, bool
stop();
}
if (!initialized_) {
+ if (reload) {
+ io::NetworkPrioritizerFactory::getInstance()->clearPrioritizer();
+ }
+
if (root) {
logger_->log_info("Load Flow Controller from provided root");
+ this->root_ = root;
} else {
- logger_->log_info("Load Flow Controller from file %s", configuration_filename_.c_str());
+ logger_->log_info("Instantiating new flow");
+ this->root_ = std::shared_ptr<core::ProcessGroup>(flow_configuration_->getRoot());
}
- if (reload) {
- io::NetworkPrioritizerFactory::getInstance()->clearPrioritizer();
- }
-
- this->root_ = root == nullptr ? std::shared_ptr<core::ProcessGroup>(flow_configuration_->getRoot(configuration_filename_)) : root;
logger_->log_info("Loaded root processor Group");
logger_->log_info("Initializing timers");
controller_service_provider_ = flow_configuration_->getControllerServiceProvider();
@@ -408,7 +348,7 @@ int16_t FlowController::start() {
// as the thread_pool_ is started in load()
this->root_->startProcessing(timer_scheduler_, event_scheduler_, cron_scheduler_);
}
- initializeC2();
+ C2Client::initialize(this, shared_from_this());
running_ = true;
this->protocol_->start();
this->provenance_repo_->start();
@@ -420,310 +360,6 @@ int16_t FlowController::start() {
}
}
-void FlowController::initializeC2() {
- if (!c2_enabled_) {
- return;
- }
-
- std::string c2_enable_str;
- std::string class_str;
-
- // don't need to worry about the return code, only whether class_str is defined.
- configuration_->get("nifi.c2.agent.class", "c2.agent.class", class_str);
- configuration_->setAgentClass(class_str);
-
- if (configuration_->get(Configure::nifi_c2_enable, "c2.enable", c2_enable_str)) {
- bool enable_c2 = true;
- utils::StringUtils::StringToBool(c2_enable_str, enable_c2);
- c2_enabled_ = enable_c2;
- if (c2_enabled_ && class_str.empty()) {
- logger_->log_error("Class name must be defined when C2 is enabled");
- std::cerr << "Class name must be defined when C2 is enabled" << std::endl;
- stop();
- exit(1);
- }
- } else {
- /**
- * To require a C2 agent class we will disable C2 by default. If a registration process
- * is implemented we can re-enable. The reason for always enabling C2 is because this allows the controller
- * mechanism that can be used for local config/access to be used. Without this agent information cannot be
- * gathered even if a remote C2 server is enabled.
- *
- * The ticket that impacts this, MINIFICPP-664, should be reversed in the event that agent registration
- * can be performed and agent classes needn't be defined a priori.
- */
- c2_enabled_ = false;
- }
-
- if (!c2_enabled_) {
- return;
- }
-
- std::string identifier_str;
- if (!configuration_->get("nifi.c2.agent.identifier", "c2.agent.identifier", identifier_str) || identifier_str.empty()) {
- // set to the flow controller's identifier
- identifier_str = getUUIDStr();
- }
- configuration_->setAgentIdentifier(identifier_str);
-
- if (c2_initialized_ && !flow_update_) {
- return;
- }
-
- device_information_.clear();
- component_metrics_.clear();
- component_metrics_by_id_.clear();
-
- std::string class_csv;
- if (root_ != nullptr) {
- std::shared_ptr<state::response::QueueMetrics> queueMetrics = std::make_shared<state::response::QueueMetrics>();
- std::map<std::string, std::shared_ptr<Connection>> connections;
- root_->getConnections(connections);
- for (auto con : connections) {
- queueMetrics->addConnection(con.second);
- }
- device_information_[queueMetrics->getName()] = queueMetrics;
- std::shared_ptr<state::response::RepositoryMetrics> repoMetrics = std::make_shared<state::response::RepositoryMetrics>();
- repoMetrics->addRepository(provenance_repo_);
- repoMetrics->addRepository(flow_file_repo_);
- device_information_[repoMetrics->getName()] = repoMetrics;
- }
-
- if (configuration_->get("nifi.c2.root.classes", class_csv)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_csv, ",");
-
- for (std::string clazz : classes) {
- auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
- if (nullptr == ptr) {
- logger_->log_error("No metric defined for %s", clazz);
- continue;
- }
- std::shared_ptr<state::response::ResponseNode> processor = std::static_pointer_cast<state::response::ResponseNode>(ptr);
- auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(processor);
- if (identifier != nullptr) {
- identifier->setIdentifier(identifier_str);
- identifier->setAgentClass(class_str);
- }
- auto monitor = std::dynamic_pointer_cast<state::response::AgentMonitor>(processor);
- if (monitor != nullptr) {
- monitor->addRepository(provenance_repo_);
- monitor->addRepository(flow_file_repo_);
- monitor->setStateMonitor(shared_from_this());
- }
- auto flowMonitor = std::dynamic_pointer_cast<state::response::FlowMonitor>(processor);
- std::map<std::string, std::shared_ptr<Connection>> connections;
- root_->getConnections(connections);
- if (flowMonitor != nullptr) {
- for (auto con : connections) {
- flowMonitor->addConnection(con.second);
- }
- flowMonitor->setStateMonitor(shared_from_this());
- flowMonitor->setFlowVersion(flow_configuration_->getFlowVersion());
- }
- std::lock_guard<std::mutex> lock(metrics_mutex_);
- root_response_nodes_[processor->getName()] = processor;
- }
- }
-
- if (configuration_->get("nifi.flow.metrics.classes", class_csv)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_csv, ",");
- for (std::string clazz : classes) {
- auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
- if (nullptr == ptr) {
- logger_->log_error("No metric defined for %s", clazz);
- continue;
- }
- std::shared_ptr<state::response::ResponseNode> processor = std::static_pointer_cast<state::response::ResponseNode>(ptr);
- std::lock_guard<std::mutex> lock(metrics_mutex_);
- device_information_[processor->getName()] = processor;
- }
- }
-
- // first we should get all component metrics, then
- // we will build the mapping
- std::vector<std::shared_ptr<core::Processor>> processors;
- if (root_ != nullptr) {
- root_->getAllProcessors(processors);
- for (const auto &processor : processors) {
- auto rep = std::dynamic_pointer_cast<state::response::ResponseNodeSource>(processor);
- // we have a metrics source.
- if (nullptr != rep) {
- std::vector<std::shared_ptr<state::response::ResponseNode>> metric_vector;
- rep->getResponseNodes(metric_vector);
- for (auto metric : metric_vector) {
- component_metrics_[metric->getName()] = metric;
- }
- }
- }
- }
-
- std::string class_definitions;
- if (configuration_->get("nifi.flow.metrics.class.definitions", class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
-
- for (std::string metricsClass : classes) {
- try {
- int id = std::stoi(metricsClass);
- std::stringstream option;
- option << "nifi.flow.metrics.class.definitions." << metricsClass;
- if (configuration_->get(option.str(), class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
-
- for (std::string clazz : classes) {
- std::lock_guard<std::mutex> lock(metrics_mutex_);
- auto ret = component_metrics_[clazz];
- if (nullptr == ret) {
- ret = device_information_[clazz];
- }
- if (nullptr == ret) {
- logger_->log_error("No metric defined for %s", clazz);
- continue;
- }
- component_metrics_by_id_[id].push_back(ret);
- }
- }
- } catch (...) {
- logger_->log_error("Could not create metrics class %s", metricsClass);
- }
- }
- }
-
- loadC2ResponseConfiguration();
-
- if (!c2_initialized_) {
- c2_agent_ = std::unique_ptr<c2::C2Agent>(new c2::C2Agent(this,
- std::dynamic_pointer_cast<FlowController>(shared_from_this()),
- configuration_));
- c2_agent_->start();
- c2_initialized_ = true;
- }
-}
-
-void FlowController::loadC2ResponseConfiguration(const std::string &prefix) {
- std::string class_definitions;
-
- if (configuration_->get(prefix, class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
-
- for (std::string metricsClass : classes) {
- try {
- std::stringstream option;
- option << prefix << "." << metricsClass;
-
- std::stringstream classOption;
- classOption << option.str() << ".classes";
-
- std::stringstream nameOption;
- nameOption << option.str() << ".name";
- std::string name;
-
- if (configuration_->get(nameOption.str(), name)) {
- std::shared_ptr<state::response::ResponseNode> new_node = std::make_shared<state::response::ObjectNode>(name);
- if (configuration_->get(classOption.str(), class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
- for (std::string clazz : classes) {
- std::lock_guard<std::mutex> lock(metrics_mutex_);
-
- // instantiate the object
- auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
- if (nullptr == ptr) {
- auto metric = component_metrics_.find(clazz);
- if (metric != component_metrics_.end()) {
- ptr = metric->second;
- } else {
- logger_->log_error("No metric defined for %s", clazz);
- continue;
- }
- }
- auto node = std::dynamic_pointer_cast<state::response::ResponseNode>(ptr);
- std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(node);
- }
-
- } else {
- std::stringstream optionName;
- optionName << option.str() << "." << name;
- auto node = loadC2ResponseConfiguration(optionName.str(), new_node);
- }
-
- root_response_nodes_[name] = new_node;
- }
- } catch (...) {
- logger_->log_error("Could not create metrics class %s", metricsClass);
- }
- }
- }
-}
-
-std::shared_ptr<state::response::ResponseNode> FlowController::loadC2ResponseConfiguration(const std::string &prefix, std::shared_ptr<state::response::ResponseNode> prev_node) {
- std::string class_definitions;
- if (configuration_->get(prefix, class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
-
- for (std::string metricsClass : classes) {
- try {
- std::stringstream option;
- option << prefix << "." << metricsClass;
-
- std::stringstream classOption;
- classOption << option.str() << ".classes";
-
- std::stringstream nameOption;
- nameOption << option.str() << ".name";
- std::string name;
-
- if (configuration_->get(nameOption.str(), name)) {
- std::shared_ptr<state::response::ResponseNode> new_node = std::make_shared<state::response::ObjectNode>(name);
- if (name.find(",") != std::string::npos) {
- std::vector<std::string> sub_classes = utils::StringUtils::split(name, ",");
- for (std::string subClassStr : classes) {
- auto node = loadC2ResponseConfiguration(subClassStr, prev_node);
- if (node != nullptr)
- std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(node);
- }
-
- } else {
- if (configuration_->get(classOption.str(), class_definitions)) {
- std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
- for (std::string clazz : classes) {
- std::lock_guard<std::mutex> lock(metrics_mutex_);
- // instantiate the object
- auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
- if (nullptr == ptr) {
- auto metric = component_metrics_.find(clazz);
- if (metric != component_metrics_.end()) {
- ptr = metric->second;
- } else {
- logger_->log_error("No metric defined for %s", clazz);
- continue;
- }
- }
-
- auto node = std::dynamic_pointer_cast<state::response::ResponseNode>(ptr);
- std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(node);
- }
- if (!new_node->isEmpty())
- std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(new_node);
-
- } else {
- std::stringstream optionName;
- optionName << option.str() << "." << name;
- auto sub_node = loadC2ResponseConfiguration(optionName.str(), new_node);
- std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(sub_node);
- }
- }
- }
- } catch (...) {
- logger_->log_error("Could not create metrics class %s", metricsClass);
- }
- }
- }
- return prev_node;
-}
-
-void FlowController::loadC2ResponseConfiguration() {
- loadC2ResponseConfiguration("nifi.c2.root.class.definitions");
-}
-
/**
* Controller Service functions
*
@@ -884,11 +520,14 @@ void FlowController::disableAllControllerServices() {
controller_service_provider_->disableAllControllerServices();
}
-int16_t FlowController::applyUpdate(const std::string &source, const std::string &configuration) {
+int16_t FlowController::applyUpdate(const std::string &source, const std::string &configuration, bool persist) {
if (applyConfiguration(source, configuration)) {
- return 1;
- } else {
+ if (persist) {
+ flow_configuration_->persist(configuration);
+ }
return 0;
+ } else {
+ return -1;
}
}
@@ -906,38 +545,6 @@ int16_t FlowController::clearConnection(const std::string &connection) {
return -1;
}
-std::shared_ptr<state::response::ResponseNode> FlowController::getMetricsNode(const std::string& metricsClass) const {
- std::lock_guard<std::mutex> lock(metrics_mutex_);
- if (!metricsClass.empty()) {
- const auto citer = component_metrics_.find(metricsClass);
- if (citer != component_metrics_.end()) {
- return citer->second;
- }
- } else {
- const auto iter = root_response_nodes_.find("metrics");
- if (iter != root_response_nodes_.end()) {
- return iter->second;
- }
- }
- return nullptr;
-}
-
-std::vector<std::shared_ptr<state::response::ResponseNode>> FlowController::getHeartbeatNodes(bool includeManifest) const {
- std::string fullHb{"true"};
- configuration_->get("nifi.c2.full.heartbeat", fullHb);
- const bool include = includeManifest ? true : (fullHb == "true");
-
- std::vector<std::shared_ptr<state::response::ResponseNode>> nodes;
- for (const auto& entry : root_response_nodes_) {
- auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(entry.second);
- if (identifier) {
- identifier->includeAgentManifest(include);
- }
- nodes.push_back(entry.second);
- }
- return nodes;
-}
-
std::shared_ptr<state::response::ResponseNode> FlowController::getAgentManifest() const {
auto agentInfo = std::make_shared<state::response::AgentInformation>("agentInfo");
agentInfo->setIdentifier(configuration_->getAgentIdentifier());
diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp
index 09d4085..cc9c1a6 100644
--- a/libminifi/src/c2/C2Agent.cpp
+++ b/libminifi/src/c2/C2Agent.cpp
@@ -26,6 +26,7 @@
#include <map>
#include <string>
#include <memory>
+#include <cinttypes>
#include "c2/ControllerSocketProtocol.h"
#include "core/ProcessContext.h"
@@ -36,10 +37,13 @@
#include "utils/file/DiffUtils.h"
#include "utils/file/FileUtils.h"
#include "utils/file/FileManager.h"
+#include "utils/file/FileSystem.h"
#include "utils/GeneralUtils.h"
#include "utils/HTTPClient.h"
#include "utils/Environment.h"
#include "utils/Monitors.h"
+#include "utils/OptionalUtils.h"
+#include "utils/StringUtils.h"
namespace org {
namespace apache {
@@ -49,13 +53,15 @@ namespace c2 {
C2Agent::C2Agent(core::controller::ControllerServiceProvider* controller,
const std::shared_ptr<state::StateMonitor> &updateSink,
- const std::shared_ptr<Configure> &configuration)
+ const std::shared_ptr<Configure> &configuration,
+ const std::shared_ptr<utils::file::FileSystem>& filesystem)
: heart_beat_period_(3000),
max_c2_responses(5),
update_sink_(updateSink),
update_service_(nullptr),
controller_(controller),
configuration_(configuration),
+ filesystem_(filesystem),
protocol_(nullptr),
logger_(logging::LoggerFactory<C2Agent>::getLogger()),
thread_pool_(2, false, nullptr, "C2 threadpool") {
@@ -68,7 +74,7 @@ C2Agent::C2Agent(core::controller::ControllerServiceProvider* controller,
last_run_ = std::chrono::steady_clock::now();
if (nullptr != controller_) {
- update_service_ = std::static_pointer_cast<controllers::UpdatePolicyControllerService>(controller_->getControllerService(C2_AGENT_UPDATE_NAME));
+ update_service_ = std::static_pointer_cast<controllers::UpdatePolicyControllerService>(controller_->getControllerService(UPDATE_NAME));
}
if (update_service_ == nullptr) {
@@ -77,62 +83,8 @@ C2Agent::C2Agent(core::controller::ControllerServiceProvider* controller,
configure(configuration, false);
- c2_producer_ = [&]() {
- // place priority on messages to send to the c2 server
- if (protocol_.load() != nullptr) {
- std::vector<C2Payload> payload_batch;
- payload_batch.reserve(max_c2_responses);
- auto getRequestPayload = [&payload_batch] (C2Payload&& payload) { payload_batch.emplace_back(std::move(payload)); };
- for (std::size_t attempt_num = 0; attempt_num < max_c2_responses; ++attempt_num) {
- if (!requests.consume(getRequestPayload)) {
- break;
- }
- }
- std::for_each(
- std::make_move_iterator(payload_batch.begin()),
- std::make_move_iterator(payload_batch.end()),
- [&] (C2Payload&& payload) {
- try {
- C2Payload && response = protocol_.load()->consumePayload(std::move(payload));
- enqueue_c2_server_response(std::move(response));
- }
- catch(const std::exception &e) {
- logger_->log_error("Exception occurred while consuming payload. error: %s", e.what());
- }
- catch(...) {
- logger_->log_error("Unknonwn exception occurred while consuming payload.");
- }
- });
-
- try {
- performHeartBeat();
- }
- catch (const std::exception &e) {
- logger_->log_error("Exception occurred while performing heartbeat. error: %s", e.what());
- }
- catch (...) {
- logger_->log_error("Unknonwn exception occurred while performing heartbeat.");
- }
- }
-
- checkTriggers();
-
- return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(heart_beat_period_));
- };
-
- functions_.push_back(c2_producer_);
-
- c2_consumer_ = [&] {
- if (false == responses.empty()) {
- const auto call_extractPayload = [this](C2Payload&& payload) { extractPayload(std::move(payload)); };
- const auto consume_success = responses.consume(call_extractPayload);
- if (!consume_success) {
- extractPayload(C2Payload{ Operation::HEARTBEAT });
- }
- }
- return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(C2RESPONSE_POLL_MS));
- };
- functions_.push_back(c2_consumer_);
+ functions_.emplace_back([this] {return produce();});
+ functions_.emplace_back([this] {return consume();});
}
void C2Agent::start() {
@@ -255,7 +207,7 @@ void C2Agent::configure(const std::shared_ptr<Configure> &configure, bool reconf
}
}
- // if not defined we won't beable to update
+ // if not defined we won't be able to update
configure->get("nifi.c2.agent.bin.location", "c2.agent.bin.location", bin_location_);
}
std::string heartbeat_reporters;
@@ -358,12 +310,10 @@ void C2Agent::serializeMetrics(C2Payload &metric_payload, const std::string &nam
}
}
-void C2Agent::extractPayload(const C2Payload &&resp) {
+void C2Agent::extractPayload(const C2Payload &resp) {
if (resp.getStatus().getState() == state::UpdateState::NESTED) {
- const std::vector<C2Payload> &payloads = resp.getNestedPayloads();
-
- for (const auto &payload : payloads) {
- extractPayload(std::move(payload));
+ for (const C2Payload& payload : resp.getNestedPayloads()) {
+ extractPayload(payload);
}
return;
}
@@ -374,7 +324,7 @@ void C2Agent::extractPayload(const C2Payload &&resp) {
case state::UpdateState::READ_COMPLETE:
logger_->log_trace("Received Ack from Server");
// we have a heartbeat response.
- for (const auto &server_response : resp.getContent()) {
+ for (const C2ContentResponse &server_response : resp.getContent()) {
handle_c2_server_response(server_response);
}
break;
@@ -401,31 +351,12 @@ void C2Agent::extractPayload(const C2Payload &&resp) {
}
}
-void C2Agent::extractPayload(const C2Payload &resp) {
- if (resp.getStatus().getState() == state::UpdateState::NESTED) {
- const std::vector<C2Payload> &payloads = resp.getNestedPayloads();
- for (const auto &payload : payloads) {
- extractPayload(payload);
- }
- }
- switch (resp.getStatus().getState()) {
- case state::UpdateState::READ_COMPLETE:
- // we have a heartbeat response.
- for (const auto &server_response : resp.getContent()) {
- handle_c2_server_response(server_response);
- }
- break;
- default:
- break;
- }
-}
-
void C2Agent::handle_c2_server_response(const C2ContentResponse &resp) {
switch (resp.op) {
case Operation::CLEAR:
// we've been told to clear something
if (resp.name == "connection") {
- for (auto connection : resp.operation_arguments) {
+ for (const auto& connection : resp.operation_arguments) {
logger_->log_debug("Clearing connection %s", connection.second.to_string());
update_sink_->clearConnection(connection.second.to_string());
}
@@ -462,11 +393,9 @@ void C2Agent::handle_c2_server_response(const C2ContentResponse &resp) {
}
break;
- case Operation::UPDATE: {
+ case Operation::UPDATE:
handle_update(resp);
- }
break;
-
case Operation::DESCRIBE:
handle_describe(resp);
break;
@@ -624,144 +553,7 @@ void C2Agent::handle_describe(const C2ContentResponse &resp) {
void C2Agent::handle_update(const C2ContentResponse &resp) {
// we've been told to update something
if (resp.name == "configuration") {
- auto url = resp.operation_arguments.find("location");
-
- auto persist = resp.operation_arguments.find("persist");
-
- if (url != resp.operation_arguments.end()) {
- // just get the raw data.
- C2Payload payload(Operation::TRANSFER, false, true);
-
- auto urlStr = url->second.to_string();
-
- std::string file_path = urlStr;
- bool containsHttp = file_path.find("http") != std::string::npos;
- if (!containsHttp) {
- std::ifstream new_conf(file_path);
- if (!new_conf.good()) {
- containsHttp = true;
- }
- }
- if (nullptr != protocol_.load() && containsHttp) {
- std::stringstream newUrl;
- if (urlStr.find("http") == std::string::npos) {
- std::string base;
- if (configuration_->get(minifi::Configure::nifi_c2_flow_base_url, base)) {
- newUrl << base;
- if (!utils::StringUtils::endsWith(base, "/")) {
- newUrl << "/";
- }
- newUrl << urlStr;
- urlStr = newUrl.str();
- } else if (configuration_->get("c2.rest.url", base)) {
- std::string host, protocol;
- int port = -1;
- utils::parse_url(&base, &host, &port, &protocol);
- newUrl << protocol << host;
- if (port > 0) {
- newUrl << ":" << port;
- }
- newUrl << "/c2/api/" << urlStr;
- urlStr = newUrl.str();
- }
- }
-
- C2Payload &&response = protocol_.load()->consumePayload(urlStr, payload, RECEIVE, false);
-
- auto raw_data = response.getRawData();
- file_path = std::string(raw_data.data(), raw_data.size());
- }
-
- std::ifstream new_conf(file_path);
- std::string raw_data_str((std::istreambuf_iterator<char>(new_conf)), std::istreambuf_iterator<char>());
- std::remove(file_path.c_str());
- // if we can apply the update, we will acknowledge it and then backup the configuration file.
- if (update_sink_->applyUpdate(urlStr, raw_data_str)) {
- C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::FULLY_APPLIED, resp.ident, false, true);
- enqueue_c2_response(std::move(response));
-
- if (persist != resp.operation_arguments.end() && utils::StringUtils::equalsIgnoreCase(persist->second.to_string(), "true")) {
- // update nifi.flow.configuration.file=./conf/config.yml
- std::string config_file;
-
- configuration_->get(minifi::Configure::nifi_flow_configuration_file, config_file);
- std::string adjustedFilename;
- if (config_file[0] != '/') {
- adjustedFilename = adjustedFilename + configuration_->getHome() + "/" + config_file;
- } else {
- adjustedFilename += config_file;
- }
-
- config_file = adjustedFilename;
-
- std::stringstream config_file_backup;
- config_file_backup << config_file << ".bak";
- // we must be able to successfully copy the file.
- bool persist_config = true;
- bool backup_file = false;
- std::string backup_config;
-
- if (configuration_->get(minifi::Configure::nifi_flow_configuration_file_backup_update, backup_config) && utils::StringUtils::StringToBool(backup_config, backup_file)) {
- if (utils::file::FileUtils::copy_file(config_file, config_file_backup.str()) != 0) {
- logger_->log_debug("Cannot copy %s to %s", config_file, config_file_backup.str());
- persist_config = false;
- }
- }
- logger_->log_debug("Copy %s to %s %d", config_file, config_file_backup.str(), persist_config);
- if (persist_config) {
- std::ofstream writer(config_file);
- if (writer.is_open()) {
- writer.write(raw_data_str.data(), raw_data_str.size());
- }
- writer.close();
-
- // update the flow id
- configuration_->persistProperties();
- }
- }
- } else {
- logger_->log_debug("update failed.");
- C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::SET_ERROR, resp.ident, false, true);
- response.setRawData("Error while applying flow. Likely missing processors");
- enqueue_c2_response(std::move(response));
- }
- // send
- } else {
- logger_->log_debug("Did not have location within %s", resp.ident);
- auto update_text = resp.operation_arguments.find("configuration_data");
- if (update_text != resp.operation_arguments.end()) {
- if (update_sink_->applyUpdate(url->second.to_string(), update_text->second.to_string()) != 0 && persist != resp.operation_arguments.end()
- && utils::StringUtils::equalsIgnoreCase(persist->second.to_string(), "true")) {
- C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::FULLY_APPLIED, resp.ident, false, true);
- enqueue_c2_response(std::move(response));
- // update nifi.flow.configuration.file=./conf/config.yml
- std::string config_file;
- std::stringstream config_file_backup;
- config_file_backup << config_file << ".bak";
-
- bool persist_config = true;
- bool backup_file = false;
- std::string backup_config;
-
- if (configuration_->get(minifi::Configure::nifi_flow_configuration_file_backup_update, backup_config) && utils::StringUtils::StringToBool(backup_config, backup_file)) {
- if (utils::file::FileUtils::copy_file(config_file, config_file_backup.str()) != 0) {
- persist_config = false;
- }
- }
- if (persist_config) {
- std::ofstream writer(config_file);
- if (writer.is_open()) {
- auto output = update_text->second.to_string();
- writer.write(output.c_str(), output.size());
- }
- writer.close();
- }
- } else {
- C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::SET_ERROR, resp.ident, false, true);
- enqueue_c2_response(std::move(response));
- }
- }
- }
+ handleConfigurationUpdate(resp);
} else if (resp.name == "properties") {
bool update_occurred = false;
for (auto entry : resp.operation_arguments) {
@@ -873,6 +665,161 @@ void C2Agent::update_agent() {
}
}
+utils::TaskRescheduleInfo C2Agent::produce() {
+ // place priority on messages to send to the c2 server
+ if (protocol_.load() != nullptr) {
+ std::vector<C2Payload> payload_batch;
+ payload_batch.reserve(max_c2_responses);
+ auto getRequestPayload = [&payload_batch] (C2Payload&& payload) { payload_batch.emplace_back(std::move(payload)); };
+ for (std::size_t attempt_num = 0; attempt_num < max_c2_responses; ++attempt_num) {
+ if (!requests.consume(getRequestPayload)) {
+ break;
+ }
+ }
+ std::for_each(
+ std::make_move_iterator(payload_batch.begin()),
+ std::make_move_iterator(payload_batch.end()),
+ [&] (C2Payload&& payload) {
+ try {
+ C2Payload && response = protocol_.load()->consumePayload(std::move(payload));
+ enqueue_c2_server_response(std::move(response));
+ }
+ catch(const std::exception &e) {
+ logger_->log_error("Exception occurred while consuming payload. error: %s", e.what());
+ }
+ catch(...) {
+ logger_->log_error("Unknown exception occurred while consuming payload.");
+ }
+ });
+
+ try {
+ performHeartBeat();
+ }
+ catch (const std::exception &e) {
+ logger_->log_error("Exception occurred while performing heartbeat. error: %s", e.what());
+ }
+ catch (...) {
+ logger_->log_error("Unknonwn exception occurred while performing heartbeat.");
+ }
+ }
+
+ checkTriggers();
+
+ return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(heart_beat_period_));
+}
+
+utils::TaskRescheduleInfo C2Agent::consume() {
+ const auto consume_success = responses.consume([this] (C2Payload&& payload) {
+ extractPayload(std::move(payload));
+ });
+ if (!consume_success) {
+ extractPayload(C2Payload{ Operation::HEARTBEAT });
+ }
+ return utils::TaskRescheduleInfo::RetryIn(std::chrono::milliseconds(C2RESPONSE_POLL_MS));
+}
+
+utils::optional<std::string> C2Agent::fetchFlow(const std::string& uri) const {
+ if (!utils::StringUtils::startsWith(uri, "http") || protocol_.load() == nullptr) {
+ // try to open the file
+ utils::optional<std::string> content = filesystem_->read(uri);
+ if (content) {
+ return content;
+ }
+ }
+ // couldn't open as file and we have no protocol to request the file from
+ if (protocol_.load() == nullptr) {
+ return {};
+ }
+
+ std::string resolved_url = uri;
+ if (!utils::StringUtils::startsWith(uri, "http")) {
+ std::stringstream adjusted_url;
+ std::string base;
+ if (configuration_->get(minifi::Configure::nifi_c2_flow_base_url, base)) {
+ adjusted_url << base;
+ if (!utils::StringUtils::endsWith(base, "/")) {
+ adjusted_url << "/";
+ }
+ adjusted_url << uri;
+ resolved_url = adjusted_url.str();
+ } else if (configuration_->get("c2.rest.url", base)) {
+ std::string host, protocol;
+ int port = -1;
+ utils::parse_url(&base, &host, &port, &protocol);
+ adjusted_url << protocol << host;
+ if (port > 0) {
+ adjusted_url << ":" << port;
+ }
+ adjusted_url << "/c2/api/" << uri;
+ resolved_url = adjusted_url.str();
+ }
+ }
+
+ C2Payload payload(Operation::TRANSFER, false, true);
+ C2Payload &&response = protocol_.load()->consumePayload(resolved_url, payload, RECEIVE, false);
+
+ auto raw_data = response.getRawData();
+ return std::string(raw_data.data(), raw_data.size());
+}
+
+bool C2Agent::handleConfigurationUpdate(const C2ContentResponse &resp) {
+ auto url = resp.operation_arguments.find("location");
+
+ std::string file_uri;
+ std::string configuration_str;
+
+ if (url != resp.operation_arguments.end()) {
+ file_uri = url->second.to_string();
+ utils::optional<std::string> optional_configuration_str = fetchFlow(file_uri);
+ if (!optional_configuration_str) {
+ logger_->log_debug("Couldn't load new flow configuration from: \"%s\"", file_uri);
+ C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::SET_ERROR, resp.ident, false, true);
+ response.setRawData("Error while applying flow. Couldn't load flow configuration.");
+ enqueue_c2_response(std::move(response));
+ return false;
+ }
+ configuration_str = optional_configuration_str.value();
+ } else {
+ logger_->log_debug("Did not have location within %s", resp.ident);
+ auto update_text = resp.operation_arguments.find("configuration_data");
+ if (update_text == resp.operation_arguments.end()) {
+ logger_->log_debug("Neither the config file location nor the data is provided");
+ C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::SET_ERROR, resp.ident, false, true);
+ response.setRawData("Error while applying flow. Neither the config file location nor the data is provided.");
+ enqueue_c2_response(std::move(response));
+ return false;
+ }
+ configuration_str = update_text->second.to_string();
+ }
+
+ bool should_persist = [&] {
+ auto persist = resp.operation_arguments.find("persist");
+ if (persist == resp.operation_arguments.end()) {
+ return false;
+ }
+ return utils::StringUtils::equalsIgnoreCase(persist->second.to_string(), "true");
+ }();
+
+ int16_t err = {update_sink_->applyUpdate(file_uri, configuration_str, should_persist)};
+ if (err != 0) {
+ logger_->log_debug("Flow configuration update failed with error code %" PRIi16, err);
+ C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::SET_ERROR, resp.ident, false, true);
+ response.setRawData("Error while applying flow. Likely missing processors");
+ enqueue_c2_response(std::move(response));
+ return false;
+ }
+
+ C2Payload response(Operation::ACKNOWLEDGE, state::UpdateState::FULLY_APPLIED, resp.ident, false, true);
+ enqueue_c2_response(std::move(response));
+
+ if (should_persist) {
+ // update the flow id
+ configuration_->persistProperties();
+ }
+
+ return true;
+}
+
} // namespace c2
} // namespace minifi
} // namespace nifi
diff --git a/libminifi/src/c2/C2Client.cpp b/libminifi/src/c2/C2Client.cpp
new file mode 100644
index 0000000..b1d6ea9
--- /dev/null
+++ b/libminifi/src/c2/C2Client.cpp
@@ -0,0 +1,324 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <memory>
+#include <map>
+#include "c2/C2Client.h"
+#include "core/state/nodes/MetricsBase.h"
+#include "core/state/nodes/QueueMetrics.h"
+#include "core/state/nodes/AgentInformation.h"
+#include "core/state/nodes/RepositoryMetrics.h"
+#include "properties/Configure.h"
+#include "core/state/UpdateController.h"
+#include "core/controller/ControllerServiceProvider.h"
+#include "c2/C2Agent.h"
+#include "core/state/nodes/FlowInformation.h"
+#include "utils/file/FileSystem.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace c2 {
+
+C2Client::C2Client(
+ std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
+ std::shared_ptr<core::Repository> flow_file_repo, std::shared_ptr<core::ContentRepository> content_repo,
+ std::unique_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<utils::file::FileSystem> filesystem,
+ std::shared_ptr<logging::Logger> logger)
+ : core::Flow(std::move(provenance_repo), std::move(flow_file_repo), std::move(content_repo), std::move(flow_configuration)),
+ configuration_(std::move(configuration)),
+ filesystem_(std::move(filesystem)),
+ logger_(std::move(logger)) {}
+
+void C2Client::stopC2() {
+ if (c2_agent_) {
+ c2_agent_->stop();
+ }
+}
+
+bool C2Client::isC2Enabled() const {
+ std::string c2_enable_str;
+ configuration_->get(Configure::nifi_c2_enable, "c2.enable", c2_enable_str);
+ return utils::StringUtils::toBool(c2_enable_str).value_or(false);
+}
+
+void C2Client::initialize(core::controller::ControllerServiceProvider *controller, const std::shared_ptr<state::StateMonitor> &update_sink) {
+ std::string class_str;
+ configuration_->get("nifi.c2.agent.class", "c2.agent.class", class_str);
+ configuration_->setAgentClass(class_str);
+
+ if (!isC2Enabled()) {
+ return;
+ }
+
+ if (class_str.empty()) {
+ logger_->log_error("Class name must be defined when C2 is enabled");
+ throw std::runtime_error("Class name must be defined when C2 is enabled");
+ }
+
+ std::string identifier_str;
+ if (!configuration_->get("nifi.c2.agent.identifier", "c2.agent.identifier", identifier_str) || identifier_str.empty()) {
+ // set to the flow controller's identifier
+ identifier_str = getControllerUUID().to_string();
+ }
+ configuration_->setAgentIdentifier(identifier_str);
+
+ if (initialized_ && !flow_update_) {
+ return;
+ }
+
+ // root_response_nodes_ was not cleared before, it is unclear if that was intentional
+
+ std::map<std::string, std::shared_ptr<Connection>> connections;
+ if (root_ != nullptr) {
+ root_->getConnections(connections);
+ }
+
+ std::string class_csv;
+ if (configuration_->get("nifi.c2.root.classes", class_csv)) {
+ std::vector<std::string> classes = utils::StringUtils::split(class_csv, ",");
+
+ for (const std::string& clazz : classes) {
+ auto processor = std::dynamic_pointer_cast<state::response::ResponseNode>(core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz));
+ if (nullptr == processor) {
+ logger_->log_error("No metric defined for %s", clazz);
+ continue;
+ }
+ auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(processor);
+ if (identifier != nullptr) {
+ identifier->setIdentifier(identifier_str);
+ identifier->setAgentClass(class_str);
+ }
+ auto monitor = std::dynamic_pointer_cast<state::response::AgentMonitor>(processor);
+ if (monitor != nullptr) {
+ monitor->addRepository(provenance_repo_);
+ monitor->addRepository(flow_file_repo_);
+ monitor->setStateMonitor(update_sink);
+ }
+ auto flowMonitor = std::dynamic_pointer_cast<state::response::FlowMonitor>(processor);
+ if (flowMonitor != nullptr) {
+ for (auto &con : connections) {
+ flowMonitor->addConnection(con.second);
+ }
+ flowMonitor->setStateMonitor(update_sink);
+ flowMonitor->setFlowVersion(flow_configuration_->getFlowVersion());
+ }
+ std::lock_guard<std::mutex> guard(metrics_mutex_);
+ root_response_nodes_[processor->getName()] = processor;
+ }
+ }
+
+ initializeComponentMetrics();
+
+ loadC2ResponseConfiguration("nifi.c2.root.class.definitions");
+
+ if (!initialized_) {
+ c2_agent_ = std::unique_ptr<c2::C2Agent>(new c2::C2Agent(controller, update_sink, configuration_, filesystem_));
+ c2_agent_->start();
+ initialized_ = true;
+ }
+}
+
+void C2Client::initializeComponentMetrics() {
+ {
+ std::lock_guard<std::mutex> guard(metrics_mutex_);
+ component_metrics_.clear();
+ }
+
+ if (root_ == nullptr) {
+ return;
+ }
+ std::vector<std::shared_ptr<core::Processor>> processors;
+ root_->getAllProcessors(processors);
+ for (const auto &processor : processors) {
+ auto rep = std::dynamic_pointer_cast<state::response::ResponseNodeSource>(processor);
+ if (rep == nullptr) {
+ continue;
+ }
+ // we have a metrics source.
+ std::vector<std::shared_ptr<state::response::ResponseNode>> metric_vector;
+ rep->getResponseNodes(metric_vector);
+ std::lock_guard<std::mutex> guard(metrics_mutex_);
+ for (auto& metric : metric_vector) {
+ component_metrics_[metric->getName()] = metric;
+ }
+ }
+}
+
+void C2Client::loadC2ResponseConfiguration(const std::string &prefix) {
+ std::string class_definitions;
+ if (!configuration_->get(prefix, class_definitions)) {
+ return;
+ }
+
+ std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
+
+ for (const std::string& metricsClass : classes) {
+ try {
+ std::string option = prefix + "." + metricsClass;
+ std::string classOption = option + ".classes";
+ std::string nameOption = option + ".name";
+
+ std::string name;
+ if (!configuration_->get(nameOption, name)) {
+ continue;
+ }
+ std::shared_ptr<state::response::ResponseNode> new_node = std::make_shared<state::response::ObjectNode>(name);
+ if (configuration_->get(classOption, class_definitions)) {
+ std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
+ for (const std::string& clazz : classes) {
+ // instantiate the object
+ auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
+ if (nullptr == ptr) {
+ const bool found_metric = [&] {
+ std::lock_guard<std::mutex> guard{metrics_mutex_};
+ auto metric = component_metrics_.find(clazz);
+ if (metric != component_metrics_.end()) {
+ ptr = metric->second;
+ return true;
+ }
+ return false;
+ }();
+ if (!found_metric) {
+ logger_->log_error("No metric defined for %s", clazz);
+ continue;
+ }
+ }
+ auto node = std::dynamic_pointer_cast<state::response::ResponseNode>(ptr);
+ std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(node);
+ }
+
+ } else {
+ std::string optionName = option + "." + name;
+ auto node = loadC2ResponseConfiguration(optionName, new_node);
+ }
+
+ std::lock_guard<std::mutex> guard{metrics_mutex_};
+ root_response_nodes_[name] = new_node;
+ } catch (...) {
+ logger_->log_error("Could not create metrics class %s", metricsClass);
+ }
+ }
+}
+
+std::shared_ptr<state::response::ResponseNode> C2Client::loadC2ResponseConfiguration(const std::string &prefix, std::shared_ptr<state::response::ResponseNode> prev_node) {
+ std::string class_definitions;
+ if (!configuration_->get(prefix, class_definitions)) {
+ return prev_node;
+ }
+ std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
+
+ for (const std::string& metricsClass : classes) {
+ try {
+ std::string option = prefix + "." + metricsClass;
+ std::string classOption = option + ".classes";
+ std::string nameOption = option + ".name";
+
+ std::string name;
+ if (!configuration_->get(nameOption, name)) {
+ continue;
+ }
+ std::shared_ptr<state::response::ResponseNode> new_node = std::make_shared<state::response::ObjectNode>(name);
+ if (name.find(',') != std::string::npos) {
+ std::vector<std::string> sub_classes = utils::StringUtils::split(name, ",");
+ for (const std::string& subClassStr : classes) {
+ auto node = loadC2ResponseConfiguration(subClassStr, prev_node);
+ if (node != nullptr)
+ std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(node);
+ }
+ } else {
+ if (configuration_->get(classOption, class_definitions)) {
+ std::vector<std::string> classes = utils::StringUtils::split(class_definitions, ",");
+ for (const std::string& clazz : classes) {
+ // instantiate the object
+ auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
+ if (nullptr == ptr) {
+ const bool found_metric = [&] {
+ std::lock_guard<std::mutex> guard{metrics_mutex_};
+ auto metric = component_metrics_.find(clazz);
+ if (metric != component_metrics_.end()) {
+ ptr = metric->second;
+ return true;
+ }
+ return false;
+ }();
+ if (!found_metric) {
+ logger_->log_error("No metric defined for %s", clazz);
+ continue;
+ }
+ }
+
+ auto node = std::dynamic_pointer_cast<state::response::ResponseNode>(ptr);
+ std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(node);
+ }
+ if (!new_node->isEmpty())
+ std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(new_node);
+
+ } else {
+ std::string optionName = option + "." + name;
+ auto sub_node = loadC2ResponseConfiguration(optionName, new_node);
+ std::static_pointer_cast<state::response::ObjectNode>(prev_node)->add_node(sub_node);
+ }
+ }
+ } catch (...) {
+ logger_->log_error("Could not create metrics class %s", metricsClass);
+ }
+ }
+ return prev_node;
+}
+
+std::shared_ptr<state::response::ResponseNode> C2Client::getMetricsNode(const std::string& metrics_class) const {
+ if (!metrics_class.empty()) {
+ std::lock_guard<std::mutex> lock(metrics_mutex_);
+ const auto citer = component_metrics_.find(metrics_class);
+ if (citer != component_metrics_.end()) {
+ return citer->second;
+ }
+ } else {
+ std::lock_guard<std::mutex> lock(metrics_mutex_);
+ const auto iter = root_response_nodes_.find("metrics");
+ if (iter != root_response_nodes_.end()) {
+ return iter->second;
+ }
+ }
+ return nullptr;
+}
+
+std::vector<std::shared_ptr<state::response::ResponseNode>> C2Client::getHeartbeatNodes(bool include_manifest) const {
+ std::string fullHb{"true"};
+ configuration_->get("nifi.c2.full.heartbeat", fullHb);
+ const bool include = include_manifest || fullHb == "true";
+
+ std::vector<std::shared_ptr<state::response::ResponseNode>> nodes;
+ std::lock_guard<std::mutex> lock(metrics_mutex_);
+ for (const auto &entry : root_response_nodes_) {
+ auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(entry.second);
+ if (identifier) {
+ identifier->includeAgentManifest(include);
+ }
+ nodes.push_back(entry.second);
+ }
+ return nodes;
+}
+
+} // namespace c2
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
diff --git a/libminifi/src/core/ConfigurationFactory.cpp b/libminifi/src/core/ConfigurationFactory.cpp
index 0f12e92..82ec9f4 100644
--- a/libminifi/src/core/ConfigurationFactory.cpp
+++ b/libminifi/src/core/ConfigurationFactory.cpp
@@ -27,6 +27,7 @@
#include "io/StreamFactory.h"
#include "core/yaml/YamlConfiguration.h"
+#include "utils/OptionalUtils.h"
namespace org {
namespace apache {
@@ -34,32 +35,38 @@ namespace nifi {
namespace minifi {
namespace core {
-std::unique_ptr<core::FlowConfiguration> createFlowConfiguration(std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo,
- std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
- std::shared_ptr<io::StreamFactory> stream_factory, const std::string configuration_class_name, const std::string path,
- bool fail_safe) {
+std::unique_ptr<core::FlowConfiguration> createFlowConfiguration(
+ std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_file_repo,
+ std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
+ std::shared_ptr<io::StreamFactory> stream_factory, const std::string& configuration_class_name,
+ const utils::optional<std::string>& path, std::shared_ptr<utils::file::FileSystem> filesystem,
+ bool fail_safe) {
std::string class_name_lc = configuration_class_name;
std::transform(class_name_lc.begin(), class_name_lc.end(), class_name_lc.begin(), ::tolower);
try {
if (class_name_lc == "flowconfiguration") {
// load the base configuration.
- return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(repo, flow_file_repo, content_repo, stream_factory, configure, path));
+ return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(
+ repo, flow_file_repo, content_repo, stream_factory, configure, path, filesystem));
} else if (class_name_lc == "yamlconfiguration") {
// only load if the class is defined.
- return std::unique_ptr<core::FlowConfiguration>(instantiate<core::YamlConfiguration>(repo, flow_file_repo, content_repo, stream_factory, configure, path));
+ return std::unique_ptr<core::FlowConfiguration>(instantiate<core::YamlConfiguration>(
+ repo, flow_file_repo, content_repo, stream_factory, configure, path, filesystem));
} else {
if (fail_safe) {
- return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(repo, flow_file_repo, content_repo, stream_factory, configure, path));
+ return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(
+ repo, flow_file_repo, content_repo, stream_factory, configure, path, filesystem));
} else {
throw std::runtime_error("Support for the provided configuration class could not be found");
}
}
} catch (const std::runtime_error &) {
if (fail_safe) {
- return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(repo, flow_file_repo, content_repo, stream_factory, configure, path));
+ return std::unique_ptr<core::FlowConfiguration>(new core::FlowConfiguration(
+ repo, flow_file_repo, content_repo, stream_factory, configure, path, filesystem));
}
}
diff --git a/libminifi/src/core/FlowConfiguration.cpp b/libminifi/src/core/FlowConfiguration.cpp
index c9a5ae3..4c92e4c 100644
--- a/libminifi/src/core/FlowConfiguration.cpp
+++ b/libminifi/src/core/FlowConfiguration.cpp
@@ -104,6 +104,27 @@ std::unique_ptr<core::ProcessGroup> FlowConfiguration::updateFromPayload(const s
return payload;
}
+bool FlowConfiguration::persist(const std::string &configuration) {
+ if (!config_path_) {
+ logger_->log_error("No flow configuration path is specified, cannot persist changes.");
+ return false;
+ }
+
+ std::string config_file_backup = *config_path_ + ".bak";
+ bool backup_file = (configuration_->get(minifi::Configure::nifi_flow_configuration_file_backup_update)
+ | utils::flatMap(utils::StringUtils::toBool)).value_or(false);
+
+ if (backup_file) {
+ if (utils::file::FileUtils::copy_file(*config_path_, config_file_backup) != 0) {
+ logger_->log_debug("Cannot copy %s to %s", *config_path_, config_file_backup);
+ return false;
+ }
+ logger_->log_debug("Copy %s to %s", *config_path_, config_file_backup);
+ }
+
+ return filesystem_->write(*config_path_, configuration);
+}
+
std::unique_ptr<core::ProcessGroup> FlowConfiguration::createRootProcessGroup(std::string name, utils::Identifier & uuid, int version) {
return std::unique_ptr<core::ProcessGroup>(new core::ProcessGroup(core::ROOT_PROCESS_GROUP, name, uuid, version));
}
diff --git a/libminifi/src/core/yaml/YamlConfiguration.cpp b/libminifi/src/core/yaml/YamlConfiguration.cpp
index a53c0db..3d1cb9f 100644
--- a/libminifi/src/core/yaml/YamlConfiguration.cpp
+++ b/libminifi/src/core/yaml/YamlConfiguration.cpp
@@ -23,6 +23,8 @@
#include "core/yaml/YamlConfiguration.h"
#include "core/state/Value.h"
+#include "Defaults.h"
+
#ifdef YAML_CONFIGURATION_USE_REGEX
#include <regex>
#endif // YAML_CONFIGURATION_USE_REGEX
@@ -35,6 +37,15 @@ namespace core {
std::shared_ptr<utils::IdGenerator> YamlConfiguration::id_generator_ = utils::IdGenerator::getIdGenerator();
+YamlConfiguration::YamlConfiguration(const std::shared_ptr<core::Repository>& repo, const std::shared_ptr<core::Repository>& flow_file_repo,
+ const std::shared_ptr<core::ContentRepository>& content_repo, const std::shared_ptr<io::StreamFactory>& stream_factory,
+ const std::shared_ptr<Configure>& configuration, const utils::optional<std::string>& path,
+ const std::shared_ptr<utils::file::FileSystem>& filesystem)
+ : FlowConfiguration(repo, flow_file_repo, content_repo, stream_factory, configuration,
+ path.value_or(DEFAULT_NIFI_CONFIG_YML), filesystem),
+ stream_factory_(stream_factory),
+ logger_(logging::LoggerFactory<YamlConfiguration>::getLogger()) {}
+
core::ProcessGroup *YamlConfiguration::parseRootProcessGroupYaml(YAML::Node rootFlowNode) {
utils::Identifier uuid;
int version = 0;
diff --git a/libminifi/src/Decryptor.cpp b/libminifi/src/utils/EncryptionProvider.cpp
similarity index 69%
rename from libminifi/src/Decryptor.cpp
rename to libminifi/src/utils/EncryptionProvider.cpp
index 313bc61..b0ca5f1 100644
--- a/libminifi/src/Decryptor.cpp
+++ b/libminifi/src/utils/EncryptionProvider.cpp
@@ -1,5 +1,4 @@
/**
- *
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@@ -15,12 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "properties/Decryptor.h"
+#include <string>
+#include "utils/EncryptionProvider.h"
#include "properties/Properties.h"
#include "utils/OptionalUtils.h"
#include "utils/StringUtils.h"
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+namespace crypto {
+
namespace {
#ifdef WIN32
@@ -33,22 +40,19 @@ constexpr const char* CONFIG_ENCRYPTION_KEY_PROPERTY_NAME = "nifi.bootstrap.sens
} // namespace
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-
-utils::optional<minifi::Decryptor> Decryptor::create(const std::string& minifi_home) {
+utils::optional<EncryptionProvider> EncryptionProvider::create(const std::string& home_path) {
minifi::Properties bootstrap_conf;
- bootstrap_conf.setHome(minifi_home);
+ bootstrap_conf.setHome(home_path);
bootstrap_conf.loadConfigureFile(DEFAULT_NIFI_BOOTSTRAP_FILE);
return bootstrap_conf.getString(CONFIG_ENCRYPTION_KEY_PROPERTY_NAME)
- | utils::map([](const std::string& encryption_key_hex) { return utils::StringUtils::from_hex(encryption_key_hex); })
- | utils::map(&utils::crypto::stringToBytes)
- | utils::map([](const utils::crypto::Bytes& encryption_key_bytes) { return minifi::Decryptor{encryption_key_bytes}; });
+ | utils::map([](const std::string &encryption_key_hex) { return utils::StringUtils::from_hex(encryption_key_hex); })
+ | utils::map(&utils::crypto::stringToBytes)
+ | utils::map([](const utils::crypto::Bytes &encryption_key_bytes) { return EncryptionProvider{encryption_key_bytes}; });
}
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace crypto
+} // namespace utils
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
diff --git a/libminifi/src/utils/EncryptionUtils.cpp b/libminifi/src/utils/EncryptionUtils.cpp
index f102606..9dd174f 100644
--- a/libminifi/src/utils/EncryptionUtils.cpp
+++ b/libminifi/src/utils/EncryptionUtils.cpp
@@ -107,6 +107,12 @@ Bytes decryptRaw(const Bytes& input, const Bytes& key, const Bytes& nonce) {
}
std::string decrypt(const std::string& input, const Bytes& key) {
+ auto data = parseEncrypted(input);
+ Bytes plaintext = decryptRaw(data.ciphertext_plus_mac, key, data.nonce);
+ return bytesToString(plaintext);
+}
+
+EncryptedData parseEncrypted(const std::string& input) {
std::vector<std::string> nonce_and_rest = utils::StringUtils::split(input, EncryptionType::separator());
if (nonce_and_rest.size() != 2) {
throw std::invalid_argument{"Incorrect input; expected '<nonce>" + EncryptionType::separator() + "<ciphertext_plus_mac>'"};
@@ -115,8 +121,16 @@ std::string decrypt(const std::string& input, const Bytes& key) {
Bytes nonce = utils::StringUtils::from_base64(nonce_and_rest[0].data(), nonce_and_rest[0].size());
Bytes ciphertext_plus_mac = utils::StringUtils::from_base64(nonce_and_rest[1].data(), nonce_and_rest[1].size());
- Bytes plaintext = decryptRaw(ciphertext_plus_mac, key, nonce);
- return bytesToString(plaintext);
+ return EncryptedData{nonce, ciphertext_plus_mac};
+}
+
+bool isEncrypted(const std::string& input) {
+ try {
+ parseEncrypted(input);
+ return true;
+ } catch (...) {
+ return false;
+ }
}
} // namespace crypto
diff --git a/libminifi/src/utils/StringUtils.cpp b/libminifi/src/utils/StringUtils.cpp
index 7adae95..01a210e 100644
--- a/libminifi/src/utils/StringUtils.cpp
+++ b/libminifi/src/utils/StringUtils.cpp
@@ -31,7 +31,18 @@ bool StringUtils::StringToBool(std::string input, bool &output) {
return output;
}
-std::string StringUtils::trim(std::string s) {
+utils::optional<bool> StringUtils::toBool(const std::string& str) {
+ std::string trimmed = trim(str);
+ if (equalsIgnoreCase(trimmed, "true")) {
+ return true;
+ }
+ if (equalsIgnoreCase(trimmed, "false")) {
+ return false;
+ }
+ return {};
+}
+
+std::string StringUtils::trim(const std::string& s) {
return trimRight(trimLeft(s));
}
@@ -156,7 +167,7 @@ std::string StringUtils::replaceMap(std::string source_string, const std::map<st
}
std::sort(replacements.begin(), replacements.end(), [](const std::pair<size_t, std::pair<size_t, std::string>> a,
- const std::pair<size_t, std::pair<size_t, std::string>> &b) {
+ const std::pair<size_t, std::pair<size_t, std::string>> &b) {
return a.first > b.first;
});
diff --git a/libminifi/src/utils/file/FileSystem.cpp b/libminifi/src/utils/file/FileSystem.cpp
new file mode 100644
index 0000000..6776182
--- /dev/null
+++ b/libminifi/src/utils/file/FileSystem.cpp
@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <string>
+#include <fstream>
+#include "utils/file/FileSystem.h"
+#include "utils/OptionalUtils.h"
+#include "utils/EncryptionProvider.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace utils {
+namespace file {
+
+FileSystem::FileSystem(bool should_encrypt_on_write, utils::optional<utils::crypto::EncryptionProvider> encryptor)
+ : should_encrypt_on_write_(should_encrypt_on_write),
+ encryptor_(std::move(encryptor)) {
+ if (should_encrypt_on_write_ && !encryptor) {
+ std::string err_message = "Requested file encryption but no encryption utility was provided";
+ logger_->log_error(err_message.c_str());
+ throw std::invalid_argument(err_message);
+ }
+}
+
+utils::optional<std::string> FileSystem::read(const std::string& file_name) {
+ std::ifstream input{file_name, std::ios::binary};
+ if (!input) {
+ return {};
+ }
+ input.exceptions(std::ios::failbit | std::ios::badbit);
+ std::string content{std::istreambuf_iterator<char>(input), {}};
+ if (encryptor_) {
+ try {
+ logger_->log_debug("Trying to decrypt file %s", file_name);
+ content = encryptor_->decrypt(content);
+ } catch(...) {
+ // tried to decrypt file but failed, use file as-is
+ logger_->log_debug("Decrypting file %s failed, using the file as-is", file_name);
+ }
+ }
+ return content;
+}
+
+bool FileSystem::write(const std::string& file_name, const std::string& file_content) {
+ std::ofstream output{file_name, std::ios::binary};
+ if (should_encrypt_on_write_) {
+ // allow a possible exception to propagate upward
+ // if we fail to encrypt the file DON'T just write
+ // it as-is
+ logger_->log_debug("Encrypting file %s", file_name);
+ output << encryptor_->encrypt(file_content);
+ } else {
+ logger_->log_debug("No encryption is required for file %s", file_name);
+ output << file_content;
+ }
+ output.close();
+ return static_cast<bool>(output);
+}
+
+} // namespace file
+} // namespace utils
+} // namespace minifi
+} // namespace nifi
+} // namespace apache
+} // namespace org
diff --git a/libminifi/test/integration/IntegrationBase.h b/libminifi/test/integration/IntegrationBase.h
index ce22ef2..a826988 100644
--- a/libminifi/test/integration/IntegrationBase.h
+++ b/libminifi/test/integration/IntegrationBase.h
@@ -31,6 +31,7 @@
#include "core/ConfigurableComponent.h"
#include "controllers/SSLContextService.h"
#include "HTTPUtils.h"
+#include "utils/OptionalUtils.h"
class IntegrationBase {
public:
@@ -38,7 +39,7 @@ class IntegrationBase {
virtual ~IntegrationBase() = default;
- virtual void run(std::string test_file_location);
+ virtual void run(const std::string& test_file_location, const utils::optional<std::string>& bootstrap_file = {});
void setKeyDir(const std::string key_dir) {
this->key_dir = key_dir;
@@ -51,6 +52,10 @@ class IntegrationBase {
}
+ const std::shared_ptr<minifi::Configure>& getConfiguration() const {
+ return configuration;
+ }
+
virtual void cleanup() = 0;
virtual void runAssertions() = 0;
@@ -96,7 +101,7 @@ void IntegrationBase::configureSecurity() {
}
}
-void IntegrationBase::run(std::string test_file_location) {
+void IntegrationBase::run(const std::string& test_file_location, const utils::optional<std::string>& home_path) {
testSetup();
std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
@@ -110,23 +115,33 @@ void IntegrationBase::run(std::string test_file_location) {
std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>();
content_repo->initialize(configuration);
std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration);
- std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr<core::YamlConfiguration>(
- new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location));
- core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location);
+ bool should_encrypt_flow_config = (configuration->get(minifi::Configure::nifi_flow_configuration_encrypt)
+ | utils::flatMap(utils::StringUtils::toBool)).value_or(false);
+
+ std::shared_ptr<utils::file::FileSystem> filesystem;
+ if (home_path) {
+ filesystem = std::make_shared<utils::file::FileSystem>(
+ should_encrypt_flow_config,
+ utils::crypto::EncryptionProvider::create(*home_path));
+ } else {
+ filesystem = std::make_shared<utils::file::FileSystem>();
+ }
+
+ std::unique_ptr<core::FlowConfiguration> flow_config = std::unique_ptr<core::YamlConfiguration>(
+ new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location, filesystem));
- auto controller_service_provider = yaml_ptr->getControllerServiceProvider();
+ auto controller_service_provider = flow_config->getControllerServiceProvider();
char state_dir_name_template[] = "/var/tmp/integrationstate.XXXXXX";
state_dir = utils::file::FileUtils::create_temp_directory(state_dir_name_template);
core::ProcessContext::getOrCreateDefaultStateManagerProvider(controller_service_provider.get(), configuration, state_dir.c_str());
- std::shared_ptr<core::ProcessGroup> pg(yaml_config.getRoot(test_file_location));
+ std::shared_ptr<core::ProcessGroup> pg(flow_config->getRoot());
queryRootProcessGroup(pg);
std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo);
- flowController_ = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME,
- true);
+ flowController_ = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(flow_config), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
flowController_->load();
updateProperties(flowController_);
flowController_->start();
diff --git a/libminifi/test/integration/ProvenanceReportingTest.cpp b/libminifi/test/integration/ProvenanceReportingTest.cpp
index ba9d496..da07c5d 100644
--- a/libminifi/test/integration/ProvenanceReportingTest.cpp
+++ b/libminifi/test/integration/ProvenanceReportingTest.cpp
@@ -66,12 +66,12 @@ int main(int argc, char **argv) {
new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location));
std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo);
- std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME,
- true);
+ std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(
+ test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, true);
core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location);
- std::unique_ptr<core::ProcessGroup> ptr = yaml_config.getRoot(test_file_location);
+ std::unique_ptr<core::ProcessGroup> ptr = yaml_config.getRoot();
std::shared_ptr<core::ProcessGroup> pg = std::shared_ptr<core::ProcessGroup>(ptr.get());
ptr.release();
std::shared_ptr<org::apache::nifi::minifi::io::SocketContext> socket_context = std::make_shared<org::apache::nifi::minifi::io::SocketContext>(std::make_shared<minifi::Configure>());
diff --git a/libminifi/test/keyvalue-tests/PersistableKeyValueStoreServiceTest.cpp b/libminifi/test/keyvalue-tests/PersistableKeyValueStoreServiceTest.cpp
index 9a42a3e..0a8ee11 100644
--- a/libminifi/test/keyvalue-tests/PersistableKeyValueStoreServiceTest.cpp
+++ b/libminifi/test/keyvalue-tests/PersistableKeyValueStoreServiceTest.cpp
@@ -108,7 +108,7 @@ class PersistableKeyValueStoreServiceTestsFixture {
yaml_config = std::unique_ptr<core::YamlConfiguration>(new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, config_yaml));
- process_group = yaml_config->getRoot(config_yaml);
+ process_group = yaml_config->getRoot();
persistable_key_value_store_service_node = process_group->findControllerService("testcontroller");
REQUIRE(persistable_key_value_store_service_node != nullptr);
persistable_key_value_store_service_node->enable();
diff --git a/libminifi/test/nodefs/NoYamlConfiguration.cpp b/libminifi/test/nodefs/NoYamlConfiguration.cpp
index a7cabc2..bf3a241 100644
--- a/libminifi/test/nodefs/NoYamlConfiguration.cpp
+++ b/libminifi/test/nodefs/NoYamlConfiguration.cpp
@@ -27,7 +27,8 @@ TEST_CASE("NoYamlSupport1", "[NoYamlSupport1]") {
"provenancerepository", true);
REQUIRE(nullptr != prov_repo);
std::unique_ptr<core::FlowConfiguration> flow_configuration = std::move(
- core::createFlowConfiguration(prov_repo, prov_repo, std::make_shared<minifi::Configure>(), std::make_shared<minifi::io::StreamFactory>(false),
+ core::createFlowConfiguration(
+ prov_repo, prov_repo, std::make_shared<minifi::Configure>(), std::make_shared<minifi::io::StreamFactory>(false),
"yamlconfiguration"));
REQUIRE(nullptr != flow_configuration);
diff --git a/libminifi/test/persistence-tests/PersistenceTests.cpp b/libminifi/test/persistence-tests/PersistenceTests.cpp
index e81cde2..9d98e0d 100644
--- a/libminifi/test/persistence-tests/PersistenceTests.cpp
+++ b/libminifi/test/persistence-tests/PersistenceTests.cpp
@@ -166,7 +166,8 @@ TEST_CASE("Processors Can Store FlowFiles", "[TestP1]") {
content_repo->initialize(config);
auto flowConfig = std::unique_ptr<core::FlowConfiguration>{new core::FlowConfiguration(prov_repo, ff_repository, content_repo, nullptr, config, "")};
- auto flowController = std::make_shared<minifi::FlowController>(prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
+ auto flowController = std::make_shared<minifi::FlowController>(
+ prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
{
TestFlow flow(ff_repository, content_repo, prov_repo, setupMergeProcessor, MergeContent::Merge);
@@ -279,7 +280,8 @@ TEST_CASE("Persisted flowFiles are updated on modification", "[TestP1]") {
content_repo->initialize(config);
auto flowConfig = std::unique_ptr<core::FlowConfiguration>{new core::FlowConfiguration(prov_repo, ff_repository, content_repo, nullptr, config, "")};
- auto flowController = std::make_shared<minifi::FlowController>(prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
+ auto flowController = std::make_shared<minifi::FlowController>(
+ prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
{
TestFlow flow(ff_repository, content_repo, prov_repo, setupContentUpdaterProcessor, {"success", "d"});
diff --git a/libminifi/test/resources/decrypted.config.yml b/libminifi/test/resources/decrypted.config.yml
new file mode 100644
index 0000000..25a7b63
--- /dev/null
+++ b/libminifi/test/resources/decrypted.config.yml
@@ -0,0 +1,41 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+Flow Controller:
+ name: MiNiFi Flow
+ id: 2438e3c8-015a-1000-79ca-83af40ec1990
+Processors:
+ - id: 2438e3c8-015a-1000-79ca-83af40ec1991
+ name: GenerateFlowFile
+ class: org.apache.nifi.minifi.processors.GenerateFlowFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 100 ms
+ penalization period: 3 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: success
+ Properties:
+ Batch Size: '1'
+ Data Format: Text
+ File Size: 10 B
+ Unique FlowFiles: 'true'
+
+Connections:
+Remote Processing Groups:
+
diff --git a/libminifi/test/rocksdb-tests/RepoTests.cpp b/libminifi/test/rocksdb-tests/RepoTests.cpp
index d2edaa3..182de9a 100644
--- a/libminifi/test/rocksdb-tests/RepoTests.cpp
+++ b/libminifi/test/rocksdb-tests/RepoTests.cpp
@@ -278,7 +278,8 @@ TEST_CASE("Test FlowFile Restore", "[TestFFR6]") {
root->addConnection(input);
auto flowConfig = std::unique_ptr<core::FlowConfiguration>{new core::FlowConfiguration(prov_repo, ff_repository, content_repo, nullptr, config, "")};
- auto flowController = std::make_shared<minifi::FlowController>(prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
+ auto flowController = std::make_shared<minifi::FlowController>(
+ prov_repo, ff_repository, config, std::move(flowConfig), content_repo, "", true);
std::string data = "banana";
minifi::io::BufferStream content(reinterpret_cast<const uint8_t*>(data.c_str()), data.length());
diff --git a/libminifi/test/unit/CollectionUtilsTests.cpp b/libminifi/test/unit/CollectionUtilsTests.cpp
new file mode 100644
index 0000000..7aee913
--- /dev/null
+++ b/libminifi/test/unit/CollectionUtilsTests.cpp
@@ -0,0 +1,62 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <string>
+#include <vector>
+#include <set>
+#include "utils/CollectionUtils.h"
+#include "../TestBase.h"
+
+TEST_CASE("TestHaveCommonItem", "[haveCommonItem]") {
+ auto verify = [](std::initializer_list<std::string> a, std::initializer_list<std::string> b, bool result) {
+ std::vector<std::string> v1{a};
+ std::set<std::string> s1{a};
+ std::vector<std::string> v2{b};
+ std::set<std::string> s2{b};
+ REQUIRE(utils::haveCommonItem(v1, v2) == result);
+ REQUIRE(utils::haveCommonItem(v1, s2) == result);
+ REQUIRE(utils::haveCommonItem(s1, v2) == result);
+ REQUIRE(utils::haveCommonItem(s1, s2) == result);
+ };
+
+ verify({"a"}, {"a"}, true);
+ verify({"a"}, {"a", "b"}, true);
+ verify({"a"}, {}, false);
+ verify({}, {"a"}, false);
+ verify({"a"}, {"b"}, false);
+ verify({}, {}, false);
+}
+
+template<typename T>
+struct MockSet : std::set<T> {
+ using std::set<T>::set;
+ auto find(const T& item) const -> decltype(this->std::set<T>::find(item)) {
+ on_find_();
+ return std::set<T>::find(item);
+ }
+ std::function<void()> on_find_;
+};
+
+TEST_CASE("TestHaveCommonItem set::find is called", "[haveCommonItem set::find]") {
+ MockSet<std::string> s1{"a"};
+ bool member_find_called = false;
+ s1.on_find_ = [&] {member_find_called = true;};
+
+ REQUIRE(utils::haveCommonItem(std::set<std::string>{"a"}, s1));
+ REQUIRE(member_find_called);
+}
diff --git a/libminifi/test/unit/ControllerTests.cpp b/libminifi/test/unit/ControllerTests.cpp
index 504822b..04ab838 100644
--- a/libminifi/test/unit/ControllerTests.cpp
+++ b/libminifi/test/unit/ControllerTests.cpp
@@ -77,49 +77,49 @@ class TestUpdateSink : public minifi::state::StateMonitor {
controller(controller),
update_calls(0) {
}
- virtual std::vector<std::shared_ptr<StateController>> getComponents(const std::string &name) {
+ std::vector<std::shared_ptr<StateController>> getComponents(const std::string &name) override {
std::vector<std::shared_ptr<StateController>> vec;
vec.push_back(controller);
return vec;
}
- virtual std::vector<std::shared_ptr<StateController>> getAllComponents() {
+ std::vector<std::shared_ptr<StateController>> getAllComponents() override {
std::vector<std::shared_ptr<StateController>> vec;
vec.push_back(controller);
return vec;
}
- virtual std::string getComponentName() const {
+ std::string getComponentName() const override {
return "TestUpdateSink";
}
- virtual utils::Identifier getComponentUUID() const {
+ utils::Identifier getComponentUUID() const override {
static auto dummyUUID = utils::Identifier::parse("12345678-1234-1234-1234-123456789abc").value();
return dummyUUID;
}
/**
* Start the client
*/
- virtual int16_t start() {
+ int16_t start() override {
is_running = true;
return 0;
}
/**
* Stop the client
*/
- virtual int16_t stop() {
+ int16_t stop() override {
is_running = false;
return 0;
}
- virtual bool isRunning() {
+ bool isRunning() override {
return is_running;
}
- virtual int16_t pause() {
+ int16_t pause() override {
return 0;
}
- virtual std::vector<BackTrace> getTraces() {
+ std::vector<BackTrace> getTraces() override {
std::vector<BackTrace> traces;
return traces;
}
@@ -131,14 +131,14 @@ class TestUpdateSink : public minifi::state::StateMonitor {
/**
* Drain repositories
*/
- virtual int16_t drainRepositories() {
+ int16_t drainRepositories() override {
return 0;
}
/**
* Clear connection for the agent.
*/
- virtual int16_t clearConnection(const std::string &connection) {
+ int16_t clearConnection(const std::string &connection) override {
clear_calls++;
return 0;
}
@@ -149,7 +149,7 @@ class TestUpdateSink : public minifi::state::StateMonitor {
* < 0 is an error code
* 0 is success
*/
- virtual int16_t applyUpdate(const std::string &source, const std::string &configuration) {
+ int16_t applyUpdate(const std::string &source, const std::string &configuration, bool persist = false) override {
update_calls++;
return 0;
}
@@ -158,7 +158,7 @@ class TestUpdateSink : public minifi::state::StateMonitor {
* Apply an update that the agent must decode. This is useful for certain operations
* that can't be encapsulated within these definitions.
*/
- virtual int16_t applyUpdate(const std::string &source, const std::shared_ptr<minifi::state::Update> &updateController) {
+ int16_t applyUpdate(const std::string &source, const std::shared_ptr<minifi::state::Update> &updateController) override {
return 0;
}
@@ -166,7 +166,7 @@ class TestUpdateSink : public minifi::state::StateMonitor {
* Returns uptime for this module.
* @return uptime for the current state monitor.
*/
- virtual uint64_t getUptime() {
+ uint64_t getUptime() override {
return 8765309;
}
diff --git a/libminifi/test/unit/DecryptorTests.cpp b/libminifi/test/unit/DecryptorTests.cpp
index 3582be7..404c8aa 100644
--- a/libminifi/test/unit/DecryptorTests.cpp
+++ b/libminifi/test/unit/DecryptorTests.cpp
@@ -23,7 +23,7 @@ namespace utils = org::apache::nifi::minifi::utils;
TEST_CASE("Decryptor can decide whether a property is encrypted", "[isValidEncryptionMarker]") {
utils::crypto::Bytes encryption_key;
- minifi::Decryptor decryptor{encryption_key};
+ minifi::Decryptor decryptor{utils::crypto::EncryptionProvider{encryption_key}};
REQUIRE(minifi::Decryptor::isValidEncryptionMarker(utils::nullopt) == false);
REQUIRE(minifi::Decryptor::isValidEncryptionMarker(utils::optional<std::string>{""}) == false);
@@ -36,7 +36,7 @@ TEST_CASE("Decryptor can decide whether a property is encrypted", "[isValidEncry
TEST_CASE("Decryptor can decrypt a property", "[decrypt]") {
utils::crypto::Bytes encryption_key = utils::crypto::stringToBytes(utils::StringUtils::from_hex(
"4024b327fdc987ce3eb43dd1f690b9987e4072e0020e3edf4349ce1ad91a4e38"));
- minifi::Decryptor decryptor{encryption_key};
+ minifi::Decryptor decryptor{utils::crypto::EncryptionProvider{encryption_key}};
std::string encrypted_value = "l3WY1V27knTiPa6jVX0jrq4qjmKsySOu||ErntqZpHP1M+6OkA14p5sdnqJhuNHWHDVUU5EyMloTtSytKk9a5xNKo=";
REQUIRE(decryptor.decrypt(encrypted_value) == "CorrectHorseBatteryStaple");
@@ -45,7 +45,7 @@ TEST_CASE("Decryptor can decrypt a property", "[decrypt]") {
TEST_CASE("Decryptor will throw if the value is incorrect", "[decrypt]") {
utils::crypto::Bytes encryption_key = utils::crypto::stringToBytes(utils::StringUtils::from_hex(
"4024b327fdc987ce3eb43dd1f690b9987e4072e0020e3edf4349ce1ad91a4e38"));
- minifi::Decryptor decryptor{encryption_key};
+ minifi::Decryptor decryptor{utils::crypto::EncryptionProvider{encryption_key}};
// correct nonce + ciphertext and mac: "l3WY1V27knTiPa6jVX0jrq4qjmKsySOu||ErntqZpHP1M+6OkA14p5sdnqJhuNHWHDVUU5EyMloTtSytKk9a5xNKo="
@@ -87,7 +87,7 @@ TEST_CASE("Decryptor will throw if the value is incorrect", "[decrypt]") {
TEST_CASE("Decryptor can decrypt a configuration file", "[decryptSensitiveProperties]") {
utils::crypto::Bytes encryption_key = utils::crypto::stringToBytes(utils::StringUtils::from_hex(
"5506c28d0fe265299e294a4c766b723a48986764953e93d38b3c627176fd10ed"));
- minifi::Decryptor decryptor{encryption_key};
+ minifi::Decryptor decryptor{utils::crypto::EncryptionProvider{encryption_key}};
minifi::Configure configuration{decryptor};
configuration.setHome("resources");
diff --git a/libminifi/test/unit/FileSystemTests.cpp b/libminifi/test/unit/FileSystemTests.cpp
new file mode 100644
index 0000000..627bbd5
--- /dev/null
+++ b/libminifi/test/unit/FileSystemTests.cpp
@@ -0,0 +1,93 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <string>
+#include <fstream>
+#include <iterator>
+#include "../TestBase.h"
+#include "utils/file/FileSystem.h"
+
+using utils::crypto::EncryptionProvider;
+using utils::file::FileSystem;
+
+utils::crypto::Bytes encryption_key = utils::crypto::stringToBytes(utils::StringUtils::from_hex(
+ "4024b327fdc987ce3eb43dd1f690b9987e4072e0020e3edf4349ce1ad91a4e38"));
+
+struct FileSystemTest : TestController {
+ FileSystemTest() {
+ char format[] = "/var/tmp/fs.XXXXXX";
+ dir = createTempDirectory(format);
+ encrypted_file = utils::file::FileUtils::concat_path(dir, "encrypted.txt");
+ raw_file = utils::file::FileUtils::concat_path(dir, "raw.txt");
+ new_file = utils::file::FileUtils::concat_path(dir, "new.txt");
+
+ std::ofstream{encrypted_file, std::ios::binary} << crypto.encrypt("banana");
+ std::ofstream{raw_file, std::ios::binary} << "banana";
+ }
+
+ EncryptionProvider crypto{encryption_key};
+ std::string encrypted_file;
+ std::string raw_file;
+ std::string new_file;
+ std::string dir;
+};
+
+TEST_CASE_METHOD(FileSystemTest, "Can read encrypted or non-encrypted file", "[file_system]") {
+ FileSystem fs{true, crypto};
+ REQUIRE(fs.read(encrypted_file) == "banana");
+ REQUIRE(fs.read(raw_file) == "banana");
+}
+
+TEST_CASE_METHOD(FileSystemTest, "Write encrypted file", "[file_system]") {
+ FileSystem fs{true, crypto};
+
+ fs.write(new_file, "red lorry, yellow lorry");
+
+ std::ifstream file{new_file, std::ios::binary};
+ std::string file_content{std::istreambuf_iterator<char>(file), {}};
+ REQUIRE(crypto.decrypt(file_content) == "red lorry, yellow lorry");
+}
+
+TEST_CASE_METHOD(FileSystemTest, "Can read encrypted but writes non-encrypted", "[file_system]") {
+ FileSystem fs{false, crypto};
+ REQUIRE(fs.read(encrypted_file) == "banana");
+
+ fs.write(new_file, "red lorry, yellow lorry");
+
+ std::ifstream file{new_file, std::ios::binary};
+ std::string file_content{std::istreambuf_iterator<char>(file), {}};
+ REQUIRE(file_content == "red lorry, yellow lorry");
+}
+
+TEST_CASE_METHOD(FileSystemTest, "Can't read encrypted file without encryption provider", "[file_system]") {
+ FileSystem fs{false, utils::nullopt};
+ REQUIRE(fs.read(encrypted_file) != "banana");
+}
+
+TEST_CASE_METHOD(FileSystemTest, "Can read and write unencrypted", "[file_system]") {
+ FileSystem fs{false, utils::nullopt};
+ fs.write(new_file, "red lorry, yellow lorry");
+
+ std::ifstream file{new_file, std::ios::binary};
+ std::string file_content{std::istreambuf_iterator<char>(file), {}};
+ REQUIRE(file_content == "red lorry, yellow lorry");
+}
+
+TEST_CASE_METHOD(FileSystemTest, "Required to encrypt but no key was provided", "[file_system]") {
+ REQUIRE_THROWS((FileSystem{true, utils::nullopt}));
+}
diff --git a/libminifi/test/unit/ProvenanceTestHelper.h b/libminifi/test/unit/ProvenanceTestHelper.h
index 9acfa35..249b75c 100644
--- a/libminifi/test/unit/ProvenanceTestHelper.h
+++ b/libminifi/test/unit/ProvenanceTestHelper.h
@@ -289,10 +289,6 @@ class TestFlowController : public minifi::FlowController {
std::shared_ptr<minifi::Connection> createConnection(std::string name, utils::Identifier & uuid) {
return 0;
}
-
- protected:
- void initializePaths(const std::string &adjustedFilename) override {
- }
};
#if defined(__clang__)
#pragma clang diagnostic pop
diff --git a/libminifi/test/unit/StringUtilsTests.cpp b/libminifi/test/unit/StringUtilsTests.cpp
index 6a6a915..e76fa0e 100644
--- a/libminifi/test/unit/StringUtilsTests.cpp
+++ b/libminifi/test/unit/StringUtilsTests.cpp
@@ -150,6 +150,46 @@ TEST_CASE("TestStringUtils::trim", "[test trim]") {
REQUIRE("foobar\n\t " == StringUtils::trimLeft(" \n\tfoobar\n\t "));
}
+TEST_CASE("TestStringUtils::startsWith", "[test startsWith]") {
+ REQUIRE(StringUtils::startsWith("abcd", ""));
+ REQUIRE(StringUtils::startsWith("abcd", "a"));
+ REQUIRE(StringUtils::startsWith("abcd", "abcd"));
+ REQUIRE(StringUtils::startsWith("abcd", "abc"));
+ REQUIRE(!StringUtils::startsWith("abcd", "abcde"));
+
+ REQUIRE(StringUtils::startsWith("", ""));
+ REQUIRE(!StringUtils::startsWith("", "abcd"));
+ REQUIRE(!StringUtils::startsWith("abcd", "b"));
+ REQUIRE(!StringUtils::startsWith("abcd", "d"));
+}
+
+TEST_CASE("TestStringUtils::endsWith", "[test endsWith]") {
+ REQUIRE(StringUtils::endsWith("abcd", ""));
+ REQUIRE(StringUtils::endsWith("abcd", "d"));
+ REQUIRE(StringUtils::endsWith("abcd", "abcd"));
+ REQUIRE(StringUtils::endsWith("abcd", "bcd"));
+ REQUIRE(!StringUtils::endsWith("abcd", "1abcd"));
+
+ REQUIRE(StringUtils::endsWith("", ""));
+ REQUIRE(!StringUtils::endsWith("", "abcd"));
+ REQUIRE(!StringUtils::endsWith("abcd", "c"));
+ REQUIRE(!StringUtils::endsWith("abcd", "a"));
+}
+
+TEST_CASE("TestStringUtils::toBool", "[test toBool]") {
+ std::vector<std::pair<std::string, utils::optional<bool>>> cases{
+ {"", {}},
+ {"true", true},
+ {"false", false},
+ {" TrUe ", true},
+ {"\n \r FaLsE \t", false},
+ {"not false", {}}
+ };
+ for (const auto& test_case : cases) {
+ REQUIRE(StringUtils::toBool(test_case.first) == test_case.second);
+ }
+}
+
TEST_CASE("TestStringUtils::testHexEncode", "[test hex encode]") {
REQUIRE("" == StringUtils::to_hex(""));
REQUIRE("6f" == StringUtils::to_hex("o"));
diff --git a/main/MainHelper.h b/main/MainHelper.h
index 872714b..ea45833 100644
--- a/main/MainHelper.h
+++ b/main/MainHelper.h
@@ -18,10 +18,11 @@
#ifndef MAIN_MAINHELPER_H_
#define MAIN_MAINHELPER_H_
+#include <string>
+
#include "core/logging/LoggerConfiguration.h"
#include "core/logging/Logger.h"
-
-#include <string>
+#include "Defaults.h"
#ifdef WIN32
extern "C" {
@@ -41,18 +42,6 @@ extern "C" {
#define STOP_WAIT_TIME_MS 30*1000
//! Default YAML location
-#ifdef WIN32
-#define DEFAULT_NIFI_CONFIG_YML "\\conf\\config.yml"
-#define DEFAULT_NIFI_PROPERTIES_FILE "\\conf\\minifi.properties"
-#define DEFAULT_LOG_PROPERTIES_FILE "\\conf\\minifi-log.properties"
-#define DEFAULT_UID_PROPERTIES_FILE "\\conf\\minifi-uid.properties"
-#else
-#define DEFAULT_NIFI_CONFIG_YML "./conf/config.yml"
-#define DEFAULT_NIFI_PROPERTIES_FILE "./conf/minifi.properties"
-#define DEFAULT_LOG_PROPERTIES_FILE "./conf/minifi-log.properties"
-#define DEFAULT_UID_PROPERTIES_FILE "./conf/minifi-uid.properties"
-#endif
-
//! Define home environment variable
#define MINIFI_HOME_ENV_KEY "MINIFI_HOME"
diff --git a/main/MiNiFiMain.cpp b/main/MiNiFiMain.cpp
index e615ab3..6546d4d 100644
--- a/main/MiNiFiMain.cpp
+++ b/main/MiNiFiMain.cpp
@@ -297,9 +297,19 @@ int main(int argc, char **argv) {
std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configure);
- std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(prov_repo, flow_repo, content_repo, configure, stream_factory, nifi_configuration_class_name);
+ bool should_encrypt_flow_config = (configure->get(minifi::Configure::nifi_flow_configuration_encrypt)
+ | utils::flatMap(utils::StringUtils::toBool)).value_or(false);
- const auto controller = std::make_shared<minifi::FlowController>(prov_repo, flow_repo, configure, std::move(flow_configuration), content_repo);
+ auto filesystem = std::make_shared<utils::file::FileSystem>(
+ should_encrypt_flow_config,
+ utils::crypto::EncryptionProvider::create(minifiHome));
+
+ std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(
+ prov_repo, flow_repo, content_repo, configure, stream_factory, nifi_configuration_class_name,
+ configure->get(minifi::Configure::nifi_flow_configuration_file), filesystem);
+
+ const auto controller = std::make_shared<minifi::FlowController>(
+ prov_repo, flow_repo, configure, std::move(flow_configuration), content_repo, filesystem);
const bool disk_space_watchdog_enable = (configure->get(minifi::Configure::minifi_disk_space_watchdog_enable) | utils::map([](const std::string& v){ return v == "true"; })).value_or(true);
std::unique_ptr<utils::CallBackTimer> disk_space_watchdog;