diff --git a/CMakeLists.txt b/CMakeLists.txt index f0af9e0d..94b72b93 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,9 +1,6 @@ # The version number. -set(AGENT_VERSION_MAJOR 2) -set(AGENT_VERSION_MINOR 8) -set(AGENT_VERSION_PATCH 0) -set(AGENT_VERSION_BUILD 4) -set(AGENT_VERSION_RC "") +include(cmake/get_version_number.cmake) + # This minimum version is to support Visual Studio 2019 and C++ feature checking and FetchContent cmake_minimum_required(VERSION 3.23 FATAL_ERROR) @@ -28,7 +25,9 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_CXX_EXTENSIONS OFF) set(CXX_COMPILE_FEATURES cxx_std_20) -project(cppagent LANGUAGES C CXX) +project(cppagent + VERSION ${AGENT_VERSION_MAJOR}.${AGENT_VERSION_MINOR}.${AGENT_VERSION_PATCH}.${AGENT_VERSION_BUILD} + LANGUAGES C CXX) set(WITH_RUBY ON CACHE STRING "With Ruby Support") diff --git a/README.md b/README.md index c687c4e8..5daa04c6 100755 --- a/README.md +++ b/README.md @@ -270,6 +270,44 @@ MqttService { Additional message mapping rules may be needed depending on your topic structure and payload format. +### Embedded MQTT Broker Adapter + +The agent can also act as the MQTT broker itself (`Protocol = mqtt_broker`). Devices connect and publish directly to the agent; every publication is forwarded to any other subscribers, and publications matching `Topics` are ingested with the same topic and payload mapping as the MQTT client adapter (`Protocol = mqtt`). + +``` +Adapters { + Cell { + Protocol = mqtt_broker + Device = Mill + Topics = Mill/# + MqttHost = 0.0.0.0 # bind address, default 0.0.0.0 + MqttPort = 8883 # default 1883, or 8883 with TLS + MqttTls = true + TlsCertificateChain = server.crt + TlsPrivateKey = server.key + MqttUserName = device + MqttPassword = secret + } +} +``` + +* Listener, transport, and credential settings (`MqttHost`, `MqttPort`, `MqttTls`, `MqttWs`, `MqttUserName`, `MqttPassword`, `MqttBrokerSecurity`) are only read from the adapter block, so they do not conflict with the MQTT sink settings. TLS certificate settings fall back to the top level values. +* `TlsVerifyClientCertificate = true` with `TlsClientCAs` rejects TLS clients without a valid client certificate. `TlsCertificatePassword` and `TlsDHKey` are optional. +* `MqttBrokerSecurity` is the path to an mqtt_cpp broker security JSON file for multiple users, salted `sha256` passwords, groups, and per-topic publish/subscribe ACLs. It replaces `MqttUserName`/`MqttPassword`. For example: + + ``` + { + "authentication": [ + { "name": "device", "method": "sha256", "salt": "abc", "digest": "" }, + { "name": "dashboard", "method": "plain_password", "password": "viewer" } + ], + "authorization": [ + { "topic": "Mill/#", "allow": { "pub": ["device"], "sub": ["device", "dashboard"] } } + ] + } + ``` +* With neither credentials nor a security file, clients connect anonymously and a warning is logged. + --- ## Ruby Extensions @@ -405,6 +443,12 @@ This section provides an overview of all the necessary configurable parameters. _Default_: 1000 +* `DisconnectRemovesAllAssets` - When an adapter disconnects, remove all assets for the + device it is the primary adapter for (the adapter's `Device`). Assets of devices listed in + `AdditionalDevices` are not removed. This can be overridden on a per adapter basis. + + _Default_: false + * `IgnoreTimestamps` - Overwrite timestamps with the agent time. This will correct clock drift but will not give as accurate relative time since it will not take into consideration network latencies. This can be overridden on a per adapter basis. diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 00000000..c55dc643 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,43 @@ +# Security Policy + +## Supported Versions + +Security fixes are made for every 2.x release line of the MTConnect C++ Agent, from 2.0 on, when the line contains the affected code. A fix ships as a new patch release of each affected line. If you run a 1.x version, upgrade to a 2.x release to get them. + +| Version | Supported | +| ------- | --------- | +| 2.x | Yes | +| < 2.0 | No | + +## Reporting a Vulnerability + +Please do not report security vulnerabilities in public issues, pull requests or discussions. + +Report them privately through GitHub instead: + +1. Go to the [Security tab](https://github.com/mtconnect/cppagent/security) of this repository. +2. Select **Report a vulnerability**. +3. Fill in the form. + +Please include: + +- The agent version, and whether you built it yourself or used a published binary or Docker image. +- The relevant parts of your `agent.cfg`, such as `BrowserView`, `AllowPut`, the sinks and the sources. +- Steps to reproduce the problem, with a request, adapter data or file that shows it if you have one. +- What an attacker can do with it, and what they need first (for example network access to the agent, `AllowPut`, or control of an adapter). +- Any mitigation you have found. + +## What to Expect + +We will acknowledge your report, look into it, and keep you updated through the advisory on GitHub. When a fix is ready we will publish a release and a security advisory. We are glad to credit you in the advisory; tell us if you would rather not be named. + +## Scope + +The agent is usually deployed on a plant network and often has no authentication in front of it. Reports are in scope when they affect the agent's own code, for example: + +- The REST sink (HTTP and WebSocket), including the browser viewer in `styles/`. +- The MQTT sinks and sources. +- Parsing of adapter (SHDR and MQTT) data, assets, `Devices.xml` and `agent.cfg`. +- The mruby scripting integration. + +Problems in a dependency (Boost, libxml2, mruby, and others) should be reported to that project. Let us know as well if the agent needs to update the dependency or work around it. diff --git a/agent_lib/CMakeLists.txt b/agent_lib/CMakeLists.txt index e0934526..e3c74ad9 100644 --- a/agent_lib/CMakeLists.txt +++ b/agent_lib/CMakeLists.txt @@ -238,6 +238,7 @@ set(AGENT_SOURCES "${SOURCE_DIR}/source/adapter/agent_adapter/session.hpp" "${SOURCE_DIR}/source/adapter/agent_adapter/session_impl.hpp" "${SOURCE_DIR}/source/adapter/mqtt/mqtt_adapter.hpp" + "${SOURCE_DIR}/source/adapter/mqtt/mqtt_broker_adapter.hpp" "${SOURCE_DIR}/source/adapter/shdr/connector.hpp" "${SOURCE_DIR}/source/adapter/shdr/shdr_adapter.hpp" "${SOURCE_DIR}/source/adapter/shdr/shdr_pipeline.hpp" @@ -249,6 +250,7 @@ set(AGENT_SOURCES "${SOURCE_DIR}/source/adapter/adapter_pipeline.cpp" "${SOURCE_DIR}/source/adapter/mqtt/mqtt_adapter.cpp" + "${SOURCE_DIR}/source/adapter/mqtt/mqtt_broker_adapter.cpp" "${SOURCE_DIR}/source/adapter/shdr/connector.cpp" "${SOURCE_DIR}/source/adapter/shdr/shdr_adapter.cpp" "${SOURCE_DIR}/source/adapter/shdr/shdr_pipeline.cpp" @@ -368,6 +370,7 @@ if(MSVC) "${SOURCE_DIR}/sink/mqtt_entity_sink/mqtt_entity_sink.cpp" "${SOURCE_DIR}/sink/rest_sink/session_impl.cpp" "${SOURCE_DIR}/source/adapter/mqtt/mqtt_adapter.cpp" + "${SOURCE_DIR}/source/adapter/mqtt/mqtt_broker_adapter.cpp" "${SOURCE_DIR}/source/adapter/agent_adapter/agent_adapter.cpp" "${SOURCE_DIR}/source/adapter/shdr/shdr_pipeline.cpp" "${SOURCE_DIR}/source/adapter/adapter_pipeline.cpp" diff --git a/agent_version.txt b/agent_version.txt new file mode 100644 index 00000000..342f75f9 --- /dev/null +++ b/agent_version.txt @@ -0,0 +1 @@ +2.8.0.5 diff --git a/cmake/get_version_number.cmake b/cmake/get_version_number.cmake new file mode 100644 index 00000000..7f6de365 --- /dev/null +++ b/cmake/get_version_number.cmake @@ -0,0 +1,22 @@ +set(AGENT_VERSION_FILE "${CMAKE_CURRENT_SOURCE_DIR}/agent_version.txt") + +# Read the first line and trim whitespace / stray \r from CRLF files +file(STRINGS "${AGENT_VERSION_FILE}" _agent_version_line LIMIT_COUNT 1) +string(STRIP "${_agent_version_line}" _agent_version_line) + +# MAJOR.MINOR.PATCH.BUILD with an optional RC suffix: 2.8.0.4, 2.8.0.4-RC1, 2.8.0.4_RC1, 2.8.0.4RC1 +if(NOT _agent_version_line MATCHES + "^([0-9]+)\\.([0-9]+)\\.([0-9]+)\\.([0-9]+)([-_]?(RC[0-9]+))?$") + message(FATAL_ERROR + "Invalid version '${_agent_version_line}' in ${AGENT_VERSION_FILE}; " + "expected MAJOR.MINOR.PATCH.BUILD[-RCn]") +endif() + +set(AGENT_VERSION_MAJOR ${CMAKE_MATCH_1}) +set(AGENT_VERSION_MINOR ${CMAKE_MATCH_2}) +set(AGENT_VERSION_PATCH ${CMAKE_MATCH_3}) +set(AGENT_VERSION_BUILD ${CMAKE_MATCH_4}) +set(AGENT_VERSION_RC "${CMAKE_MATCH_6}") # "" when there's no RC + +# Re-run configure automatically when the version file changes +set_property(DIRECTORY APPEND PROPERTY CMAKE_CONFIGURE_DEPENDS "${AGENT_VERSION_FILE}") diff --git a/conanfile.py b/conanfile.py index f9ff3215..a0ae1857 100644 --- a/conanfile.py +++ b/conanfile.py @@ -1,7 +1,7 @@ from conan import ConanFile from conan.tools.microsoft import is_msvc, is_msvc_static_runtime from conan.tools.cmake import CMake, CMakeToolchain, CMakeDeps, cmake_layout -from conan.tools.files import copy +from conan.tools.files import copy, load import os import io @@ -9,7 +9,6 @@ class MTConnectAgentConan(ConanFile): name = "mtconnect_agent" - version = "2.7" url = "https://github.com/mtconnect/cppagent.git" license = "Apache License 2.0" settings = "os", "compiler", "arch", "build_type" @@ -27,7 +26,6 @@ class MTConnectAgentConan(ConanFile): "cpack_name": [None, "ANY"], "cpack_generator": [None, "ANY"] } - description = "MTConnect reference C++ agent copyright Association for Manufacturing Technology" build_policy = "missing" @@ -74,7 +72,7 @@ class MTConnectAgentConan(ConanFile): } exports_sources = "*", "!build", "!test_package/build", "!*~" - exports = "conan/mqtt_cpp/*", "conan/mruby/*", "conan/oniguruma/*" + exports = "conan/mqtt_cpp/*", "conan/mruby/*", "conan/oniguruma/*", "agent_version.txt" def validate(self): if is_msvc(self) and self.options.shared and self.settings.compiler.runtime != 'dynamic': @@ -90,6 +88,11 @@ def config_options(self): if is_msvc(self): self.options.rm_safe("fPIC") + def set_version(self): + if not self.version: + self.version = load(self, os.path.join(self.recipe_folder, "agent_version.txt")).splitlines()[0].strip() + print(f"Building MTConnect Version {self.version}") + def tool_requires_version(self, package, version): self.output.info(f"Checking version of {package} > {version}") buf = io.StringIO() diff --git a/demo/styles/viewer.html b/demo/styles/viewer.html index ba076c78..3a0bc0e0 100644 --- a/demo/styles/viewer.html +++ b/demo/styles/viewer.html @@ -18,6 +18,7 @@ + MTConnect Agent diff --git a/src/mtconnect/agent.cpp b/src/mtconnect/agent.cpp index 47dd3a64..117ccaee 100644 --- a/src/mtconnect/agent.cpp +++ b/src/mtconnect/agent.cpp @@ -697,6 +697,23 @@ namespace mtconnect { LOG(warning) << "Free data item found in device data items"; } } + + // Remove the assets of the device this adapter is the primary adapter for + auto source = std::dynamic_pointer_cast(findSource(adapter)); + if (source && IsOptionSet(source->getOptions(), config::DisconnectRemovesAllAssets)) + { + auto name = GetOption(source->getOptions(), config::Device); + DevicePtr device; + if (name) + device = findDeviceByUUIDorName(*name); + if (device && device->getUuid()) + { + asset::AssetList list; + removeAllAssets(*device->getUuid(), nullopt, nullopt, list); + LOG(info) << "Adapter " << adapter << " disconnected, removed " << list.size() + << " assets for device " << *device->getComponentName(); + } + } } void Agent::connected(const std::string& adapter, const StringList& devices, bool autoAvailable) diff --git a/src/mtconnect/configuration/agent_config.cpp b/src/mtconnect/configuration/agent_config.cpp index e638a491..2949d1dc 100644 --- a/src/mtconnect/configuration/agent_config.cpp +++ b/src/mtconnect/configuration/agent_config.cpp @@ -65,6 +65,7 @@ #include "mtconnect/sink/rest_sink/rest_service.hpp" #include "mtconnect/source/adapter/agent_adapter/agent_adapter.hpp" #include "mtconnect/source/adapter/mqtt/mqtt_adapter.hpp" +#include "mtconnect/source/adapter/mqtt/mqtt_broker_adapter.hpp" #include "mtconnect/source/adapter/shdr/shdr_adapter.hpp" #include "mtconnect/version.h" @@ -115,6 +116,7 @@ namespace mtconnect::configuration { sink::mqtt_entity_sink::MqttEntitySink::registerFactory(m_sinkFactory); adapter::shdr::ShdrAdapter::registerFactory(m_sourceFactory); adapter::mqtt_adapter::MqttAdapter::registerFactory(m_sourceFactory); + adapter::mqtt_adapter::MqttBrokerAdapter::registerFactory(m_sourceFactory); adapter::agent_adapter::AgentAdapter::registerFactory(m_sourceFactory); #if _WINDOWS @@ -791,6 +793,7 @@ namespace mtconnect::configuration { ConfigOptions options; GetOptions(config, options, {{configuration::PreserveUUID, true}, + {configuration::DisconnectRemovesAllAssets, false}, {configuration::DisableAgentDevice, false}, {configuration::MonitorResources, false}, {configuration::WorkingDirectory, m_working.string()}, diff --git a/src/mtconnect/configuration/config_options.hpp b/src/mtconnect/configuration/config_options.hpp index 336ea887..51435fb9 100644 --- a/src/mtconnect/configuration/config_options.hpp +++ b/src/mtconnect/configuration/config_options.hpp @@ -110,6 +110,7 @@ namespace mtconnect { DECLARE_CONFIGURATION(MqttMaxTopicDepth); DECLARE_CONFIGURATION(MqttLastWillTopic); DECLARE_CONFIGURATION(MqttXPath); + DECLARE_CONFIGURATION(MqttBrokerSecurity); DECLARE_CONFIGURATION(ObservationTopicPrefix); DECLARE_CONFIGURATION(DeviceTopicPrefix); DECLARE_CONFIGURATION(AssetTopicPrefix); @@ -132,6 +133,7 @@ namespace mtconnect { DECLARE_CONFIGURATION(Path); DECLARE_CONFIGURATION(PollingInterval); DECLARE_CONFIGURATION(PreserveUUID); + DECLARE_CONFIGURATION(DisconnectRemovesAllAssets); DECLARE_CONFIGURATION(Protocol); DECLARE_CONFIGURATION(RealTime); DECLARE_CONFIGURATION(ReconnectInterval); diff --git a/src/mtconnect/mqtt/mqtt_client_impl.hpp b/src/mtconnect/mqtt/mqtt_client_impl.hpp index d7d0a4d5..10af4d29 100644 --- a/src/mtconnect/mqtt/mqtt_client_impl.hpp +++ b/src/mtconnect/mqtt/mqtt_client_impl.hpp @@ -188,8 +188,8 @@ namespace mtconnect { mqtt::buffer contents) { if (packet_id) LOG(debug) << "packet_id: " << *packet_id; - LOG(debug) << "topic_name: " << topic_name; - LOG(debug) << "contents: " << contents; + LOG(trace) << "topic_name: " << topic_name; + LOG(trace) << "contents: " << contents; if (m_running) { diff --git a/src/mtconnect/mqtt/mqtt_server_impl.hpp b/src/mtconnect/mqtt/mqtt_server_impl.hpp index 63f44fbe..931cd3f1 100644 --- a/src/mtconnect/mqtt/mqtt_server_impl.hpp +++ b/src/mtconnect/mqtt/mqtt_server_impl.hpp @@ -212,8 +212,8 @@ namespace mtconnect { if (packet_id) LOG(debug) << "Server packet_id: " << *packet_id; - LOG(debug) << "Server topic_name: " << topic_name; - LOG(debug) << "Server contents: " << contents; + LOG(trace) << "Server topic_name: " << topic_name; + LOG(trace) << "Server contents: " << contents; for (const auto& sub : m_subs) { diff --git a/src/mtconnect/printer/html_printer.cpp b/src/mtconnect/printer/html_printer.cpp index 4150d888..e8f8edae 100644 --- a/src/mtconnect/printer/html_printer.cpp +++ b/src/mtconnect/printer/html_printer.cpp @@ -17,6 +17,8 @@ #include "mtconnect/printer/html_printer.hpp" +#include + #include #include @@ -65,10 +67,20 @@ namespace mtconnect::printer { if (!m_loaded) return content; - // Single allocation and copy: prefix + content + suffix + // The document is base64 encoded so nothing in it can end the script element it is placed in. + // Extended assets keep their raw XML, including CDATA sections, so the document can contain + // any text, such as . The base64 alphabet has no markup characters. + namespace base64 = boost::beast::detail::base64; + auto encodedSize = base64::encoded_size(content.size()); + + // Single allocation: prefix + encoded content + suffix string result; - result.reserve(m_prefix.size() + content.size() + m_suffix.size()); - result.append(m_prefix).append(content).append(m_suffix); + result.reserve(m_prefix.size() + encodedSize + m_suffix.size()); + result.append(m_prefix); + auto pos = result.size(); + result.resize(pos + encodedSize); + result.resize(pos + base64::encode(result.data() + pos, content.data(), content.size())); + result.append(m_suffix); return result; } diff --git a/src/mtconnect/printer/html_printer.hpp b/src/mtconnect/printer/html_printer.hpp index 4e8d8cd3..d006e8a4 100644 --- a/src/mtconnect/printer/html_printer.hpp +++ b/src/mtconnect/printer/html_printer.hpp @@ -67,7 +67,7 @@ namespace mtconnect::printer { /// @param in function to be called to resolve the file name. void resolveBrowserView(FileResolver&& resolver) { loadFile(std::move(resolver)); } - /// @brief Inject the content into the browser view at the placeholder. + /// @brief Inject the base64 encoded content into the browser view at the placeholder. /// @param content the document generated by the delegate printer /// @return the html document, or the content if the browser view is not available std::string formatResult(const std::string& content) const; diff --git a/src/mtconnect/source/adapter/mqtt/mqtt_adapter.hpp b/src/mtconnect/source/adapter/mqtt/mqtt_adapter.hpp index 8efe9144..a8943e16 100644 --- a/src/mtconnect/source/adapter/mqtt/mqtt_adapter.hpp +++ b/src/mtconnect/source/adapter/mqtt/mqtt_adapter.hpp @@ -80,11 +80,10 @@ namespace mtconnect::source::adapter::mqtt_adapter { /// @brief subcribe to topics void subscribeToTopics(); - protected: /// @brief load all topics /// @param ptree the property tree coming from configuration parser /// @param options configation options - void loadTopics(const boost::property_tree::ptree& tree, ConfigOptions& options); + static void loadTopics(const boost::property_tree::ptree& tree, ConfigOptions& options); protected: boost::asio::io_context& m_ioContext; diff --git a/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.cpp b/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.cpp new file mode 100644 index 00000000..03c846b9 --- /dev/null +++ b/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.cpp @@ -0,0 +1,548 @@ +// +// Copyright 2009-2026, AMT – The Association For Manufacturing Technology (“AMT”) +// All rights reserved. +// +// Licensed 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 "mqtt_broker_adapter.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "mtconnect/configuration/config_options.hpp" +#include "mtconnect/mqtt/mqtt_client_impl.hpp" + +using namespace std; +namespace asio = boost::asio; +namespace pt = boost::property_tree; + +namespace mtconnect { + using namespace pipeline; + using namespace mqtt_client; + + namespace source::adapter::mqtt_adapter { + namespace mb = MQTT_NS::broker; + + /// @brief Embedded MQTT broker with an external listener for devices and an internal + /// loopback listener used by the adapter to subscribe to the ingest topics. + class MqttBroker : public std::enable_shared_from_this + { + public: + MqttBroker(asio::io_context& io, const ConfigOptions& options) + : m_ioContext(io), m_options(options), m_broker(io) + { + // The ingest user only exists inside this process and can only subscribe to the + // configured Topics. + boost::uuids::random_generator gen; + m_ingestUser = "mtconnect-ingest-" + boost::uuids::to_string(gen()); + m_ingestPassword = boost::uuids::to_string(gen()) + boost::uuids::to_string(gen()); + m_broker.set_security(makeSecurity()); + } + + ~MqttBroker() { close(); } + + /// @brief Start the external and internal listeners + /// @return the port the external listener is bound to + uint16_t start(const asio::ip::address& address, uint16_t port) + { + asio::ip::tcp::endpoint endpoint(address, port); + bool tls = IsOptionSet(m_options, configuration::MqttTls); + bool ws = IsOptionSet(m_options, configuration::MqttWs); + + if (tls && ws) + m_port = listen(m_tlsWsServer.emplace(endpoint, makeTlsContext(), m_ioContext), true); + else if (tls) + m_port = listen(m_tlsServer.emplace(endpoint, makeTlsContext(), m_ioContext), true); + else if (ws) + m_port = listen(m_wsServer.emplace(endpoint, m_ioContext), false); + else + m_port = listen(m_tcpServer.emplace(endpoint, m_ioContext), false); + + m_internalPort = listen( + m_internalServer.emplace(asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), 0), + m_ioContext), + false); + + LOG(info) << "MQTT Broker: listening on " << address << ':' << m_port + << (tls ? " (TLS)" : "") << (ws ? " (WebSocket)" : ""); + + return m_port; + } + + /// @brief Close the listeners and disconnect all clients + void stop() + { + if (!close()) + return; + + // The broker's connection handlers capture the broker, keep it alive until the aborted + // operations from the closed connections have completed. + if (!m_ioContext.stopped()) + { + auto timer = make_shared(m_ioContext, 2s); + timer->async_wait([self = shared_from_this(), timer](boost::system::error_code) {}); + } + } + + auto getPort() const { return m_port; } + auto getInternalPort() const { return m_internalPort; } + const auto& getIngestUser() const { return m_ingestUser; } + const auto& getIngestPassword() const { return m_ingestPassword; } + + protected: + bool close() + { + if (!m_running) + return false; + m_running = false; + + if (m_tcpServer) + m_tcpServer->close(); + if (m_tlsServer) + m_tlsServer->close(); + if (m_wsServer) + m_wsServer->close(); + if (m_tlsWsServer) + m_tlsWsServer->close(); + if (m_internalServer) + m_internalServer->close(); + + { + std::lock_guard lock(m_mutex); + for (auto& wp : m_connections) + { + if (auto con = wp.lock()) + con->force_disconnect(); + } + m_connections.clear(); + } + + m_broker.clear_all_sessions(); + m_broker.clear_all_retained_topics(); + + return true; + } + + template + uint16_t listen(Server& server, bool tls) + { + if constexpr (std::is_same_v> || + std::is_same_v>) + { + // mqtt_cpp always requests a peer certificate, so accept any presented certificate + // unless client verification is required. + bool verify = IsOptionSet(m_options, configuration::TlsVerifyClientCertificate); + server.set_verify_callback( + [verify](bool preverified, asio::ssl::verify_context&, + std::shared_ptr> const&) { + return !verify || preverified; + }); + } + + server.set_accept_handler([this, tls](mb::con_sp_t spep) { accept(std::move(spep), tls); }); + server.set_error_handler([this](MQTT_NS::error_code ec) { + if (m_running && ec != asio::error::operation_aborted) + LOG(error) << "MQTT Broker: accept failed: " << ec.message(); + }); + server.listen(); + m_running = true; + + return server.port(); + } + + void accept(mb::con_sp_t spep, bool tls) + { + if (tls && IsOptionSet(m_options, configuration::TlsVerifyClientCertificate) && + !hasVerifiedPeerCertificate(*spep)) + { + LOG(warning) << "MQTT Broker: rejecting connection without a valid client certificate"; + boost::system::error_code ec; + spep->socket().lowest_layer().close(ec); + return; + } + + { + std::lock_guard lock(m_mutex); + m_connections.remove_if([](const auto& wp) { return wp.expired(); }); + m_connections.emplace_back(spep); + } + + m_broker.handle_accept(std::move(spep)); + } + + static bool hasVerifiedPeerCertificate(mb::endpoint_t& ep) + { + try + { + auto ssl = MQTT_NS::any_cast(ep.socket().native_handle()); + if (auto cert = SSL_get1_peer_certificate(ssl)) + { + X509_free(cert); + return SSL_get_verify_result(ssl) == X509_V_OK; + } + } + catch (std::exception& e) + { + LOG(error) << "MQTT Broker: cannot get TLS session: " << e.what(); + } + return false; + } + + asio::ssl::context makeTlsContext() + { + using namespace configuration; + using ctx_t = asio::ssl::context; + + auto cert = GetOption(m_options, TlsCertificateChain); + auto key = GetOption(m_options, TlsPrivateKey); + if (!cert || !key) + { + string msg("MQTT Broker: MqttTls requires TlsCertificateChain and TlsPrivateKey"); + LOG(fatal) << msg; + throw FatalException(msg); + } + + ctx_t ctx(ctx_t::tls_server); + ctx.set_options(ctx_t::default_workarounds | ctx_t::no_sslv2 | ctx_t::no_sslv3 | + ctx_t::no_tlsv1 | ctx_t::no_tlsv1_1 | ctx_t::single_dh_use); + + if (auto password = GetOption(m_options, TlsCertificatePassword)) + { + ctx.set_password_callback( + [password = *password](size_t, ctx_t::password_purpose) { return password; }); + } + + ctx.use_certificate_chain_file(*cert); + ctx.use_private_key_file(*key, ctx_t::pem); + if (auto dh = GetOption(m_options, TlsDHKey)) + ctx.use_tmp_dh_file(*dh); + + if (IsOptionSet(m_options, TlsVerifyClientCertificate)) + { + LOG(info) << "MQTT Broker: Will only accept client connections with valid certificates"; + if (auto cas = GetOption(m_options, TlsClientCAs)) + ctx.load_verify_file(*cas); + else + LOG(warning) << "MQTT Broker: TlsVerifyClientCertificate without TlsClientCAs"; + } + + return ctx; + } + + /// @brief Build the broker security from either MqttBrokerSecurity or + /// MqttUserName/MqttPassword, and add the internal ingest user. + mb::security makeSecurity() + { + using namespace configuration; + + auto user = [](const string& name, const string& method) { + pt::ptree u; + u.put("name", name); + u.put("method", method); + return u; + }; + auto names = [](const StringList& list) { + pt::ptree a; + for (const auto& n : list) + a.push_back({"", pt::ptree(n)}); + return a; + }; + auto rule = [&names](const string& topic, const StringList& pub, const StringList& sub) { + pt::ptree r, allow; + r.put("topic", topic); + if (!pub.empty()) + allow.add_child("pub", names(pub)); + if (!sub.empty()) + allow.add_child("sub", names(sub)); + r.add_child("allow", allow); + return r; + }; + + pt::ptree root; + if (auto file = GetOption(m_options, MqttBrokerSecurity)) + { + ifstream in(*file); + if (!in) + { + string msg("MQTT Broker: Cannot open MqttBrokerSecurity file: " + *file); + LOG(fatal) << msg; + throw FatalException(msg); + } + istringstream json(mb::json_remove_comments(in)); + pt::read_json(json, root); + LOG(info) << "MQTT Broker: Loaded security configuration from " << *file; + } + else + { + string name; + pt::ptree authentication; + if (auto username = GetOption(m_options, MqttUserName)) + { + auto password = GetOption(m_options, MqttPassword); + if (!password) + { + string msg("MQTT Broker: MqttPassword is required with MqttUserName"); + LOG(fatal) << msg; + throw FatalException(msg); + } + name = *username; + auto u = user(name, "plain_password"); + u.put("password", *password); + authentication.push_back({"", u}); + } + else + { + LOG(warning) << "MQTT Broker: No MqttUserName or MqttBrokerSecurity given, clients " + "can connect anonymously"; + name = "anonymous"; + authentication.push_back({"", user(name, "anonymous")}); + } + + pt::ptree authorization; + authorization.push_back({"", rule("#", {name}, {name})}); + root.add_child("authentication", authentication); + root.add_child("authorization", authorization); + } + + // Add the internal ingest user, it can only subscribe to the ingest topics + auto ingest = user(m_ingestUser, "plain_password"); + ingest.put("password", m_ingestPassword); + if (!root.get_child_optional("authentication")) + root.add_child("authentication", pt::ptree()); + root.get_child("authentication").push_back({"", ingest}); + + if (!root.get_child_optional("authorization")) + root.add_child("authorization", pt::ptree()); + auto& authorization = root.get_child("authorization"); + if (auto topics = GetOption(m_options, Topics)) + { + for (const auto& topic : *topics) + authorization.push_back({"", rule(topic, {}, {m_ingestUser})}); + } + + stringstream json; + pt::write_json(json, root); + mb::security security; + security.load_json(json); + return security; + } + + protected: + asio::io_context& m_ioContext; + ConfigOptions m_options; + mb::broker_t m_broker; + bool m_running {false}; + + std::optional> m_tcpServer; + std::optional> m_tlsServer; + std::optional> m_wsServer; + std::optional> m_tlsWsServer; + std::optional> m_internalServer; + + uint16_t m_port {0}; + uint16_t m_internalPort {0}; + + std::string m_ingestUser; + std::string m_ingestPassword; + + std::mutex m_mutex; + std::list> m_connections; + }; + + MqttBrokerAdapter::MqttBrokerAdapter(boost::asio::io_context& io, + pipeline::PipelineContextPtr pipelineContext, + const ConfigOptions& options, + const boost::property_tree::ptree& block) + : Adapter("MQTT Broker", io, options), + m_ioContext(io), + m_pipeline(pipelineContext, Source::m_strand) + { + using namespace configuration; + + GetOptions(block, m_options, options); + + // Listener, transport, and credentials must come from the adapter block, the top level + // values of these options configure the MQTT sinks. + for (auto o : {MqttHost, MqttPort, MqttTls, MqttWs, MqttUserName, MqttPassword, + MqttBrokerSecurity, MqttClientId}) + m_options.erase(o); + + AddOptions(block, m_options, + {{UUID, string()}, + {Manufacturer, string()}, + {AdapterIdentity, string()}, + {Station, string()}, + {Url, string()}, + {Topics, StringList()}, + {TlsCertificateChain, string()}, + {TlsPrivateKey, string()}, + {TlsCertificatePassword, string()}, + {TlsDHKey, string()}, + {TlsClientCAs, string()}, + {MqttUserName, string()}, + {MqttPassword, string()}, + {MqttBrokerSecurity, string()}, + {MqttHost, string()}, + {MqttPort, int()}}); + + AddDefaultedOptions(block, m_options, + {{MqttTls, false}, + {MqttWs, false}, + {TlsVerifyClientCertificate, false}, + {AutoAvailable, false}, + {RealTime, false}, + {RelativeTime, false}}); + MqttAdapter::loadTopics(block, m_options); + + bool tls = IsOptionSet(m_options, MqttTls); + m_host = GetOption(m_options, MqttHost) + .value_or(GetOption(m_options, Host).value_or("0.0.0.0")); + m_port = GetOption(m_options, MqttPort) + .value_or(GetOption(m_options, Port).value_or(tls ? 8883 : 1883)); + + stringstream url; + url << (tls ? "mqtts://" : "mqtt://") << m_host << ':' << m_port << '/'; + m_name = url.str(); + + if (auto ident = GetOption(m_options, AdapterIdentity)) + { + m_identity = *ident; + } + else + { + stringstream identity; + identity << m_name; + if (auto topics = GetOption(m_options, Topics)) + { + for (const auto& s : *topics) + identity << s; + } + + m_identity = CreateIdentityHash(identity.str()); + m_options[AdapterIdentity] = m_identity; + } + + m_broker = make_shared(m_ioContext, m_options); + + m_handler = m_pipeline.makeHandler(); + m_pipeline.m_handler = m_handler.get(); + m_pipeline.build(m_options); + } + + MqttBrokerAdapter::~MqttBrokerAdapter() + { + if (m_client) + m_client->stop(); + if (m_broker) + m_broker->stop(); + } + + void MqttBrokerAdapter::registerFactory(SourceFactory& factory) + { + factory.registerFactory( + "mqtt_broker", + [](const std::string& name, boost::asio::io_context& io, + pipeline::PipelineContextPtr context, const ConfigOptions& options, + const boost::property_tree::ptree& block) -> source::SourcePtr { + return std::make_shared(io, context, options, block); + }); + } + + unsigned int MqttBrokerAdapter::getPort() const + { + if (m_broker && m_broker->getPort() != 0) + return m_broker->getPort(); + return m_port; + } + + bool MqttBrokerAdapter::start() + { + using namespace configuration; + + m_pipeline.start(); + + boost::system::error_code ec; + auto address = asio::ip::make_address(m_host, ec); + if (ec) + { + LOG(error) << "MQTT Broker: Invalid bind address '" << m_host << "': " << ec.message(); + return false; + } + + try + { + m_broker->start(address, static_cast(m_port)); + } + catch (std::exception& e) + { + LOG(error) << "MQTT Broker: Cannot start broker on " << m_host << ':' << m_port << ": " + << e.what(); + return false; + } + + // The ingest subscriber connects to the broker on the loopback listener + ConfigOptions clientOptions {{MqttHost, "127.0.0.1"s}, + {MqttPort, int(m_broker->getInternalPort())}, + {MqttUserName, m_broker->getIngestUser()}, + {MqttPassword, m_broker->getIngestPassword()}, + {MqttClientId, m_broker->getIngestUser()}, + {MqttConnectInterval, Seconds(1)}}; + + auto clientHandler = make_unique(); + clientHandler->m_connecting = [this](shared_ptr client) { + m_handler->m_connecting(m_identity); + }; + clientHandler->m_connected = [this](shared_ptr client) { + client->connectComplete(); + m_handler->m_connected(m_identity); + subscribeToTopics(); + }; + clientHandler->m_disconnected = [this](shared_ptr client) { + m_handler->m_disconnected(m_identity); + }; + clientHandler->m_receive = [this](shared_ptr client, const std::string& topic, + const std::string& payload) { + m_handler->m_processMessage(topic, payload, m_identity); + }; + + m_client = make_shared(m_ioContext, clientOptions, std::move(clientHandler)); + return m_client->start(); + } + + void MqttBrokerAdapter::stop() + { + if (m_client) + m_client->stop(); + if (m_broker) + m_broker->stop(); + m_pipeline.clear(); + } + + void MqttBrokerAdapter::subscribeToTopics() + { + if (auto topics = GetOption(m_options, configuration::Topics)) + { + for (const auto& topic : *topics) + m_client->subscribe(topic); + } + } + } // namespace source::adapter::mqtt_adapter +} // namespace mtconnect diff --git a/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.hpp b/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.hpp new file mode 100644 index 00000000..e2f95c9e --- /dev/null +++ b/src/mtconnect/source/adapter/mqtt/mqtt_broker_adapter.hpp @@ -0,0 +1,95 @@ +// +// Copyright 2009-2026, AMT – The Association For Manufacturing Technology (“AMT”) +// All rights reserved. +// +// Licensed 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 "mtconnect/config.hpp" +#include "mtconnect/mqtt/mqtt_client.hpp" +#include "mtconnect/source/adapter/adapter.hpp" +#include "mtconnect/source/adapter/mqtt/mqtt_adapter.hpp" + +namespace mtconnect::source::adapter::mqtt_adapter { + class MqttBroker; + + /// @brief An adapter that hosts an embedded MQTT broker. + /// + /// Devices publish to the agent directly. Every publication is routed to any other + /// subscribers of the broker, and publications matching `Topics` are ingested into the + /// agent using the same pipeline as the MQTT client adapter (`Protocol = mqtt`). + /// + /// Configuration: + /// - `Protocol = mqtt_broker` + /// - `Topics`: topic filters to ingest (required) + /// - `MqttHost`/`Host`: bind address, defaults to `0.0.0.0` + /// - `MqttPort`/`Port`: listen port, defaults to 1883 (8883 with TLS) + /// - `MqttTls`, `MqttWs`: transport selection + /// - `TlsCertificateChain`, `TlsPrivateKey`, `TlsCertificatePassword`, `TlsDHKey`, + /// `TlsVerifyClientCertificate`, `TlsClientCAs`: TLS server settings + /// - `MqttUserName`, `MqttPassword`: single user with full access + /// - `MqttBrokerSecurity`: path to an mqtt_cpp broker security JSON file for multiple users, + /// salted SHA256 passwords, groups and topic ACLs + class AGENT_LIB_API MqttBrokerAdapter : public Adapter + { + public: + /// @brief Create an Mqtt broker adapter + /// @param io boost asio io context + /// @param context pipeline context + /// @param options configation options + /// @param block additional configuration options + MqttBrokerAdapter(boost::asio::io_context& io, pipeline::PipelineContextPtr pipelineContext, + const ConfigOptions& options, const boost::property_tree::ptree& block); + + ~MqttBrokerAdapter() override; + + /// @brief Register the Mqtt broker adapter with the factory as `mqtt_broker` + /// @param factory source factory + static void registerFactory(SourceFactory& factory); + + /// @name Agent Device methods + ///@{ + const std::string& getHost() const override { return m_host; } + + unsigned int getPort() const override; + ///@} + + /// @name Source interface + ///@{ + bool start() override; + + void stop() override; + + pipeline::Pipeline* getPipeline() override { return &m_pipeline; } + ///@} + + /// @brief is the internal ingest subscriber connected to the broker + bool isIngesting() const { return m_client && m_client->isConnected(); } + + protected: + void subscribeToTopics(); + + protected: + boost::asio::io_context& m_ioContext; + + std::string m_host; + unsigned int m_port; + + MqttPipeline m_pipeline; + + std::shared_ptr m_broker; + std::shared_ptr m_client; + }; +} // namespace mtconnect::source::adapter::mqtt_adapter diff --git a/styles/README.md b/styles/README.md index 310fd8f7..167cdb81 100644 --- a/styles/README.md +++ b/styles/README.md @@ -78,7 +78,7 @@ The Agent returns the viewer for a GET of `/probe`, `/current`, `/sample`, `/ass | `application/json` or `application/mtconnect+json` | JSON | | Any of the above plus `?format=xml` or `?format=json` | The raw document | -The Agent puts the document it would have returned in the page, in ` in an asset's CDATA, +// can end the script element. atob gives one character per byte, decoded here as UTF-8. +function decodeBase64(text) { + let binary + try { + binary = atob(text) + } catch { + throw new Error('The embedded MTConnect document is not base64 encoded') + } + const bytes = Uint8Array.from(binary, (c) => c.charCodeAt(0)) + return new TextDecoder().decode(bytes) +} + // Read the document embedded in a page: this one, or one returned by fetchPage export function embeddedDocument(page = document) { const script = page.getElementById(DATA_ID) if (!script) throw new Error('The page does not contain an MTConnect document') - return { type: script.type, text: script.textContent.trim() } + return { type: script.type, text: decodeBase64(script.textContent.trim()) } } // Request a page from the agent and read the document embedded in it. MTConnectError documents diff --git a/styles/viewer.html b/styles/viewer.html index 9e079167..09905146 100644 --- a/styles/viewer.html +++ b/styles/viewer.html @@ -18,6 +18,7 @@ + MTConnect Agent diff --git a/test_package/CMakeLists.txt b/test_package/CMakeLists.txt index eb79d380..ffce8c28 100644 --- a/test_package/CMakeLists.txt +++ b/test_package/CMakeLists.txt @@ -273,6 +273,7 @@ add_agent_test(html_printer TRUE html) add_agent_test(adapter FALSE adapter) add_agent_test(connector FALSE adapter) add_agent_test(mqtt_adapter FALSE adapter) +add_agent_test(mqtt_broker_adapter FALSE adapter TRUE) add_agent_test(url_parser FALSE adapter) add_agent_test(agent_adapter FALSE adapter) diff --git a/test_package/agent_asset_test.cpp b/test_package/agent_asset_test.cpp index 8772ab3d..b89cd761 100644 --- a/test_package/agent_asset_test.cpp +++ b/test_package/agent_asset_test.cpp @@ -1164,3 +1164,46 @@ TEST_F(AgentAssetTest, should_return_many_assets_by_id_with_removed_query) ASSERT_XML_PATH_COUNT(doc, "//m:Assets/*", 50); } } + +TEST_F(AgentAssetTest, should_remove_all_device_assets_when_primary_adapter_disconnects) +{ + addAdapter({{configuration::DisconnectRemovesAllAssets, true}}); + auto agent = m_agentTestHelper->getAgent(); + const auto& storage = agent->getAssetStorage(); + + m_agentTestHelper->m_adapter->processData( + "2021-02-01T12:00:00Z|@ASSET@|P1|FakeAsset|TEST 1"); + m_agentTestHelper->m_adapter->processData( + "2021-02-01T12:00:00Z|@ASSET@|P2|FakeAsset|TEST 2"); + ASSERT_EQ((unsigned int)2, storage->getCount()); + + m_agentTestHelper->m_adapter->disconnected(); + + ASSERT_EQ((unsigned int)0, storage->getCount()); + ASSERT_EQ((unsigned int)2, storage->getCount(false)); + + { + PARSE_XML_RESPONSE("/asset"); + ASSERT_XML_PATH_EQUAL(doc, "//m:Header@assetCount", "0"); + } + + { + PARSE_XML_RESPONSE("/current"); + ASSERT_XML_PATH_EQUAL(doc, "//m:AssetRemoved", "P2"); + } +} + +TEST_F(AgentAssetTest, should_keep_assets_when_adapter_disconnects_by_default) +{ + addAdapter(); + auto agent = m_agentTestHelper->getAgent(); + const auto& storage = agent->getAssetStorage(); + + m_agentTestHelper->m_adapter->processData( + "2021-02-01T12:00:00Z|@ASSET@|P1|FakeAsset|TEST 1"); + ASSERT_EQ((unsigned int)1, storage->getCount()); + + m_agentTestHelper->m_adapter->disconnected(); + + ASSERT_EQ((unsigned int)1, storage->getCount()); +} diff --git a/test_package/html_printer_test.cpp b/test_package/html_printer_test.cpp index e50eda9c..6146e2ee 100644 --- a/test_package/html_printer_test.cpp +++ b/test_package/html_printer_test.cpp @@ -24,6 +24,8 @@ #include #include +#include + #include "agent_test_helper.hpp" #include "mtconnect/agent.hpp" #include "mtconnect/printer/html_printer.hpp" @@ -83,9 +85,9 @@ class HtmlPrinterTest : public testing::Test return m_agentTestHelper->session()->m_body; } - /// @brief Check the body is the browser view with a document in place of the placeholder and - /// return that document. - string embeddedDocument(const string& body) + /// @brief Check the body is the browser view with base64 text in place of the placeholder and + /// return that text. + string embeddedText(const string& body) { EXPECT_EQ("text/html", m_agentTestHelper->session()->m_mimeType); if (body.size() < m_prefix.size() + m_suffix.size() || !body.starts_with(m_prefix) || @@ -95,14 +97,32 @@ class HtmlPrinterTest : public testing::Test return ""; } - return body.substr(m_prefix.size(), body.size() - m_prefix.size() - m_suffix.size()); + auto text = body.substr(m_prefix.size(), body.size() - m_prefix.size() - m_suffix.size()); + EXPECT_EQ(string::npos, text.find_first_not_of("ABCDEFGHIJKLMNOPQRSTUVWXYZ" + "abcdefghijklmnopqrstuvwxyz0123456789+/=")) + << "Embedded content is not base64, it could close the script element:\n" + << text; + return text; + } + + /// @brief Decode the document embedded in the browser view + string embeddedDocument(const string& body) + { + namespace base64 = boost::beast::detail::base64; + auto text = embeddedText(body); + string doc(base64::decoded_size(text.size()), '\0'); + auto [written, read] = base64::decode(doc.data(), text.data(), text.size()); + // decode stops at the padding, so the padding is not counted as read + auto unpadded = text.find_last_not_of('=') + 1; + EXPECT_EQ(unpadded, read) << "Embedded content is not valid base64"; + doc.resize(written); + return doc; } /// @brief Parse the document embedded in the browser view xmlDocPtr parseEmbedded(const string& body) { auto content = embeddedDocument(body); - EXPECT_EQ(string::npos, content.find(""; + string body = ""; + QueryMap queries {{"type", "Widget"}, {"device", "LinuxCNC"}}; + m_agentTestHelper->makeRequest(__FILE__, __LINE__, boost::beast::http::verb::put, body, queries, + "/asset/W1", "text/xml"); + ASSERT_EQ(status::ok, m_agentTestHelper->session()->m_code); + + auto& page = request("/assets"); + ASSERT_EQ(status::ok, m_agentTestHelper->session()->m_code); + + // The only script end tags are the ones in the browser view + auto count = [](const string& s) { + size_t n = 0; + for (auto pos = s.find(" +// Keep this comment to keep gtest.h above. (clang-format off/on is not working here!) + +#include +#include +#include +#include +#include + +#include + +#include "agent_test_helper.hpp" +#include "mtconnect/configuration/config_options.hpp" +#include "mtconnect/source/adapter/mqtt/mqtt_broker_adapter.hpp" + +using namespace std; +using namespace mtconnect; +using namespace mtconnect::configuration; +using namespace mtconnect::source::adapter::mqtt_adapter; + +using namespace std::literals; + +const string ServerCertFile(TEST_RESOURCE_DIR "/user.crt"); +const string ServerKeyFile(TEST_RESOURCE_DIR "/user.key"); +const string ServerCAFile(TEST_RESOURCE_DIR "/rootca.crt"); +const string ClientCAFile(TEST_RESOURCE_DIR "/clientca.crt"); +const string ClientCertFile(TEST_RESOURCE_DIR "/client.crt"); +const string ClientKeyFile(TEST_RESOURCE_DIR "/client.key"); + +// main +int main(int argc, char* argv[]) +{ + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} + +class MqttBrokerAdapterTest : public testing::Test +{ +protected: + void SetUp() override + { + m_agentTestHelper = make_unique(); + m_agentTestHelper->createAgent("/samples/test_config.xml", 8, 4, "2.0", 25); + } + + void TearDown() override + { + m_clients.clear(); + m_agentTestHelper->getAgent()->stop(); + m_agentTestHelper->m_ioContext.run_for(100ms); + m_adapter.reset(); + m_agentTestHelper.reset(); + } + + void startBroker(const map& settings) + { + boost::property_tree::ptree block; + block.put(MqttPort, "0"); + block.put(MqttHost, "127.0.0.1"); + block.put(Topics, "LinuxCNC/#"); + for (const auto& [k, v] : settings) + block.put(k, v); + + ConfigOptions options {{Device, "LinuxCNC"s}}; + m_adapter = make_shared(m_agentTestHelper->m_ioContext.get(), + m_agentTestHelper->m_context, options, block); + m_agentTestHelper->getAgent()->addSource(m_adapter, false); + m_agentTestHelper->getAgent()->start(); + + ASSERT_TRUE(waitFor(5s, [this]() { return m_adapter->isIngesting(); })); + ASSERT_NE(0, m_adapter->getPort()); + } + + template + optional connect(Client& client, + const optional& user = nullopt, + const optional& password = nullopt) + { + optional result; + bool closed = false; + client->set_client_id("test_"s + to_string(m_clients.size())); + client->set_clean_session(true); + if (user) + client->set_user_name(*user); + if (password) + client->set_password(*password); + client->set_connack_handler([&result](bool, MQTT_NS::connect_return_code rc) { + result = rc; + return true; + }); + client->set_close_handler([&closed]() { closed = true; }); + client->set_error_handler([&closed](MQTT_NS::error_code) { closed = true; }); + client->async_connect([&closed](MQTT_NS::error_code ec) { + if (ec) + closed = true; + }); + waitFor(5s, [&]() { return result.has_value() || closed; }); + + client->set_connack_handler([](bool, MQTT_NS::connect_return_code) { return true; }); + client->set_close_handler([]() {}); + client->set_error_handler([](MQTT_NS::error_code) {}); + m_clients.push_back(client); + + return result; + } + + template + void publish(Client& client, const string& topic, const string& payload) + { + client->async_publish(topic, payload, MQTT_NS::qos::at_least_once); + } + + observation::ObservationPtr latest(const string& id) + { + return m_agentTestHelper->getAgent()->getCircularBuffer().getLatest().getObservation(id); + } + + bool waitForValue(const string& id, const string& value) + { + return waitFor(5s, [&]() { + auto obs = latest(id); + return obs && obs->getValue() == value; + }); + } + + template + bool waitFor(const chrono::duration& time, function pred) + { + auto end = chrono::steady_clock::now() + time; + while (!pred() && chrono::steady_clock::now() < end) + m_agentTestHelper->m_ioContext.run_for(50ms); + return pred(); + } + + auto& ioc() { return m_agentTestHelper->m_ioContext.get(); } + + std::unique_ptr m_agentTestHelper; + std::shared_ptr m_adapter; + std::list> m_clients; +}; + +TEST_F(MqttBrokerAdapterTest, should_ingest_published_observations) +{ + startBroker({}); + + auto client = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(client)); + + publish(client, "LinuxCNC/line", "204"); + ASSERT_TRUE(waitForValue("p3", "204")); +} + +TEST_F(MqttBrokerAdapterTest, should_only_ingest_matching_topics) +{ + startBroker({{Topics, "LinuxCNC/program"}}); + + auto client = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(client)); + + publish(client, "LinuxCNC/line", "204"); + publish(client, "LinuxCNC/program", "main.ngc"); + ASSERT_TRUE(waitForValue("p4", "main.ngc")); + + auto line = latest("p3"); + ASSERT_TRUE(line); + ASSERT_TRUE(line->isUnavailable()); +} + +TEST_F(MqttBrokerAdapterTest, should_mirror_publications_to_subscribers) +{ + startBroker({}); + + auto subscriber = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(subscriber)); + + string received; + subscriber->set_publish_handler([&received](MQTT_NS::optional, + MQTT_NS::publish_options, MQTT_NS::buffer topic, + MQTT_NS::buffer contents) { + received = string(topic) + "=" + string(contents); + return true; + }); + bool subscribed = false; + subscriber->set_suback_handler( + [&subscribed](std::uint16_t, std::vector) { + subscribed = true; + return true; + }); + subscriber->async_subscribe("LinuxCNC/#", MQTT_NS::qos::at_least_once); + ASSERT_TRUE(waitFor(5s, [&]() { return subscribed; })); + + auto publisher = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(publisher)); + publish(publisher, "LinuxCNC/line", "100"); + + ASSERT_TRUE(waitFor(5s, [&]() { return !received.empty(); })); + ASSERT_EQ("LinuxCNC/line=100", received); + ASSERT_TRUE(waitForValue("p3", "100")); +} + +TEST_F(MqttBrokerAdapterTest, should_require_user_name_and_password) +{ + startBroker({{MqttUserName, "device"}, {MqttPassword, "secret"}}); + + auto anonymous = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + auto rc = connect(anonymous); + ASSERT_NE(MQTT_NS::connect_return_code::accepted, rc.value_or(MQTT_NS::connect_return_code::accepted)); + + auto wrong = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + rc = connect(wrong, "device"s, "wrong"s); + ASSERT_NE(MQTT_NS::connect_return_code::accepted, rc.value_or(MQTT_NS::connect_return_code::accepted)); + + auto client = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(client, "device"s, "secret"s)); + + publish(client, "LinuxCNC/line", "301"); + ASSERT_TRUE(waitForValue("p3", "301")); +} + +TEST_F(MqttBrokerAdapterTest, should_load_users_and_acls_from_security_file) +{ + auto file = filesystem::temp_directory_path() / "mqtt_broker_security_test.json"; + { + ofstream out(file); + out << R"({ + # Comments are allowed + "authentication": [ + { "name": "writer", "method": "plain_password", "password": "w" }, + { "name": "reader", "method": "plain_password", "password": "r" } + ], + "authorization": [ + { "topic": "#", "allow": { "pub": ["writer"], "sub": ["writer", "reader"] } } + ] +})"; + } + + startBroker({{MqttBrokerSecurity, file.string()}}); + + auto reader = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(reader, "reader"s, "r"s)); + publish(reader, "LinuxCNC/line", "1"); + + auto writer = MQTT_NS::make_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(writer, "writer"s, "w"s)); + publish(writer, "LinuxCNC/program", "writer.ngc"); + + ASSERT_TRUE(waitForValue("p4", "writer.ngc")); + ASSERT_TRUE(latest("p3")->isUnavailable()); + + filesystem::remove(file); +} + +TEST_F(MqttBrokerAdapterTest, should_accept_tls_connections) +{ + startBroker({{MqttTls, "true"}, + {TlsCertificateChain, ServerCertFile}, + {TlsPrivateKey, ServerKeyFile}, + {MqttUserName, "device"}, + {MqttPassword, "secret"}}); + + auto client = MQTT_NS::make_tls_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + client->get_ssl_context().load_verify_file(ServerCAFile); + client->get_ssl_context().set_verify_mode(boost::asio::ssl::verify_peer); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(client, "device"s, "secret"s)); + + publish(client, "LinuxCNC/line", "42"); + ASSERT_TRUE(waitForValue("p3", "42")); +} + +TEST_F(MqttBrokerAdapterTest, should_require_client_certificate_when_verifying) +{ + startBroker({{MqttTls, "true"}, + {TlsCertificateChain, ServerCertFile}, + {TlsPrivateKey, ServerKeyFile}, + {TlsVerifyClientCertificate, "true"}, + {TlsClientCAs, ClientCAFile}}); + + auto noCert = MQTT_NS::make_tls_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + noCert->get_ssl_context().load_verify_file(ServerCAFile); + ASSERT_FALSE(connect(noCert)); + + auto client = MQTT_NS::make_tls_async_client(ioc(), "127.0.0.1", m_adapter->getPort()); + client->get_ssl_context().load_verify_file(ServerCAFile); + client->get_ssl_context().use_certificate_chain_file(ClientCertFile); + client->get_ssl_context().use_private_key_file(ClientKeyFile, boost::asio::ssl::context::pem); + ASSERT_EQ(MQTT_NS::connect_return_code::accepted, connect(client)); + + publish(client, "LinuxCNC/line", "7"); + ASSERT_TRUE(waitForValue("p3", "7")); +}