845 lines
32 KiB
C++
845 lines
32 KiB
C++
#include "MQTT.h"
|
|
#include "MeshService.h"
|
|
#include "NodeDB.h"
|
|
#include "PowerFSM.h"
|
|
#include "ServiceEnvelope.h"
|
|
#include "configuration.h"
|
|
#include "main.h"
|
|
#include "mesh/Channels.h"
|
|
#include "mesh/CryptoEngine.h"
|
|
#include "mesh/Router.h"
|
|
#include "mesh/generated/meshtastic/mqtt.pb.h"
|
|
#include "mesh/generated/meshtastic/telemetry.pb.h"
|
|
#include "modules/RoutingModule.h"
|
|
#if defined(ARCH_ESP32)
|
|
#include "../mesh/generated/meshtastic/paxcount.pb.h"
|
|
#endif
|
|
#include "mesh/generated/meshtastic/remote_hardware.pb.h"
|
|
#include "sleep.h"
|
|
#if HAS_WIFI
|
|
#include "mesh/wifi/WiFiAPClient.h"
|
|
#include <WiFi.h>
|
|
#endif
|
|
#if HAS_ETHERNET && defined(ARCH_ESP32)
|
|
#include <ETH.h>
|
|
#endif // HAS_ETHERNET
|
|
#if HAS_ETHERNET && defined(USE_CH390D)
|
|
#include "ESP32_CH390.h"
|
|
#endif // USE_CH390D
|
|
#include "Default.h"
|
|
#include <Throttle.h>
|
|
#include <assert.h>
|
|
#include <utility>
|
|
|
|
#include "IPAddress.h"
|
|
#if defined(ARCH_PORTDUINO)
|
|
#if defined(_WIN32)
|
|
#include <winsock2.h> // ntohl()
|
|
#else
|
|
#include <netinet/in.h>
|
|
#endif
|
|
#elif !defined(ntohl)
|
|
#include <machine/endian.h>
|
|
#define ntohl __ntohl
|
|
#endif
|
|
#include <gps/RTC.h>
|
|
|
|
MQTT *mqtt;
|
|
|
|
namespace
|
|
{
|
|
constexpr int reconnectMax = 5;
|
|
|
|
// FIXME - this size calculation is super sloppy, but it will go away once we dynamically alloc meshpackets
|
|
static uint8_t bytes[meshtastic_MqttClientProxyMessage_size + 30]; // 12 for channel name and 16 for nodeid
|
|
|
|
static bool isMqttServerAddressPrivate = false;
|
|
static bool isConnected = false;
|
|
|
|
static uint32_t lastPositionUnavailableWarning = 0;
|
|
static const uint32_t POSITION_UNAVAILABLE_WARNING_INTERVAL_MS = 15000; // 15 seconds
|
|
|
|
inline bool shouldDropMqttDownlink(const meshtastic_MeshPacket &packet)
|
|
{
|
|
if (is_in_repeated(config.lora.ignore_incoming, packet.from)) {
|
|
LOG_INFO("Drop MQTT ignored 0x%08x", packet.from);
|
|
return true;
|
|
}
|
|
|
|
const meshtastic_NodeInfoLite *node = nodeDB->getMeshNode(packet.from);
|
|
if (nodeInfoLiteIsIgnored(node)) {
|
|
LOG_INFO("Drop MQTT node 0x%08x", packet.from);
|
|
return true;
|
|
}
|
|
|
|
if (packet.from == NODENUM_BROADCAST) {
|
|
LOG_INFO("Drop MQTT broadcast src");
|
|
return true;
|
|
}
|
|
|
|
if (config.lora.ignore_mqtt) {
|
|
LOG_INFO("Drop MQTT ignore_mqtt");
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
inline void onReceiveProto(char *topic, byte *payload, size_t length)
|
|
{
|
|
const DecodedServiceEnvelope e(payload, length);
|
|
if (!e.validDecode || e.channel_id == NULL || e.gateway_id == NULL || e.packet == NULL) {
|
|
LOG_ERROR("Invalid MQTT service envelope, topic %s, len %u!", topic, length);
|
|
return;
|
|
}
|
|
|
|
const meshtastic_Channel &ch = channels.getByName(e.channel_id);
|
|
// Find channel by channel_id and check downlink_enabled
|
|
if (!(strcmp(e.channel_id, "PKI") == 0 ||
|
|
(strcmp(e.channel_id, channels.getGlobalId(ch.index)) == 0 && ch.settings.downlink_enabled))) {
|
|
return;
|
|
}
|
|
|
|
bool anyChannelHasDownlink = false;
|
|
size_t numChan = channels.getNumChannels();
|
|
for (size_t i = 0; i < numChan; ++i) {
|
|
const auto &c = channels.getByIndex(i);
|
|
if (c.settings.downlink_enabled) {
|
|
anyChannelHasDownlink = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (strcmp(e.channel_id, "PKI") == 0 && !anyChannelHasDownlink) {
|
|
return;
|
|
}
|
|
// Generate node ID from nodenum for comparison
|
|
std::string nodeId = nodeDB->getNodeId();
|
|
if (strcmp(e.gateway_id, nodeId.c_str()) == 0) {
|
|
// Generate an implicit ACK towards ourselves (handled and processed only locally!) for this message.
|
|
// We do this because packets are not rebroadcasted back into MQTT anymore and we assume that at least one node
|
|
// receives it when we get our own packet back. Then we'll stop our retransmissions.
|
|
if (isFromUs(e.packet)) {
|
|
auto pAck = routingModule->allocAckNak(meshtastic_Routing_Error_NONE, getFrom(e.packet), e.packet->id, ch.index);
|
|
pAck->transport_mechanism = meshtastic_MeshPacket_TransportMechanism_TRANSPORT_MQTT;
|
|
router->sendLocal(pAck);
|
|
} else {
|
|
LOG_INFO("Ignore downlink message we originally sent");
|
|
}
|
|
return;
|
|
}
|
|
if (isFromUs(e.packet)) {
|
|
LOG_INFO("Ignore downlink message we originally sent");
|
|
return;
|
|
}
|
|
|
|
LOG_INFO("Received MQTT topic %s, len=%u", topic, length);
|
|
if (e.packet->hop_limit > HOP_MAX || e.packet->hop_start > HOP_MAX) {
|
|
LOG_INFO("Invalid hop_limit(%u) or hop_start(%u)", e.packet->hop_limit, e.packet->hop_start);
|
|
return;
|
|
}
|
|
|
|
UniquePacketPoolPacket p = packetPool.allocUniqueZeroed();
|
|
p->from = e.packet->from;
|
|
p->to = e.packet->to;
|
|
p->id = e.packet->id;
|
|
p->channel = e.packet->channel;
|
|
p->hop_limit = e.packet->hop_limit;
|
|
p->hop_start = e.packet->hop_start;
|
|
p->want_ack = e.packet->want_ack;
|
|
p->via_mqtt = true; // Mark that the packet was received via MQTT
|
|
p->transport_mechanism = meshtastic_MeshPacket_TransportMechanism_TRANSPORT_MQTT;
|
|
p->which_payload_variant = e.packet->which_payload_variant;
|
|
memcpy(&p->decoded, &e.packet->decoded, std::max(sizeof(p->decoded), sizeof(p->encrypted)));
|
|
|
|
if (shouldDropMqttDownlink(*p))
|
|
return;
|
|
|
|
if (p->which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
|
|
if (moduleConfig.mqtt.encryption_enabled) {
|
|
LOG_INFO("Ignore decoded message on MQTT, encryption is enabled");
|
|
return;
|
|
}
|
|
if (p->decoded.portnum == meshtastic_PortNum_ADMIN_APP) {
|
|
LOG_INFO("Ignore decoded admin packet");
|
|
return;
|
|
}
|
|
p->channel = ch.index;
|
|
#if !(MESHTASTIC_EXCLUDE_PKI) && !(MESHTASTIC_EXCLUDE_XEDDSA)
|
|
// Already-decoded downlink skips perhapsDecode's crypto path entirely, so enforce the
|
|
// signature policy here: verify a carried signature and apply unsigned-downgrade
|
|
// protection for known signers. Without this, a peer on a plaintext broker could
|
|
// impersonate a signing node with unsigned broadcasts. Hold cryptLock like the RF path
|
|
// (perhapsDecode) does - checkXeddsaReceivePolicy -> xeddsa_verify mutates shared
|
|
// CryptoEngine cache state, and MQTT ingress can run on a different task.
|
|
{
|
|
concurrency::LockGuard g(cryptLock);
|
|
if (!checkXeddsaReceivePolicy(p.get())) {
|
|
LOG_INFO("Ignore decoded message failing XEdDSA policy");
|
|
return;
|
|
}
|
|
}
|
|
#endif
|
|
}
|
|
|
|
// PKI messages get accepted even if we can't decrypt
|
|
if (router && p->which_payload_variant == meshtastic_MeshPacket_encrypted_tag && strcmp(e.channel_id, "PKI") == 0) {
|
|
const meshtastic_NodeInfoLite *tx = nodeDB->getMeshNode(getFrom(p.get()));
|
|
const meshtastic_NodeInfoLite *rx = nodeDB->getMeshNode(p->to);
|
|
// Only accept PKI messages to us, or if we have both the sender and receiver in our nodeDB, as then it's
|
|
// likely they discovered each other via a channel we have downlink enabled for
|
|
if (isToUs(p.get()) || (nodeInfoLiteHasUser(tx) && nodeInfoLiteHasUser(rx)))
|
|
router->enqueueReceivedMessage(p.release());
|
|
} else if (router &&
|
|
perhapsDecode(p.get()) == DecodeState::DECODE_SUCCESS) // ignore messages if we don't have the channel key
|
|
router->enqueueReceivedMessage(p.release());
|
|
}
|
|
|
|
/// Determines if the given IPAddress is a private IPv4 address, i.e. not routable on the public internet.
|
|
bool isPrivateIpAddress(const IPAddress &ip)
|
|
{
|
|
constexpr struct {
|
|
uint32_t network;
|
|
uint32_t mask;
|
|
} privateCidrRanges[] = {
|
|
{.network = 192u << 24 | 168 << 16, .mask = 0xffff0000}, // 192.168.0.0/16
|
|
{.network = 172u << 24 | 16 << 16, .mask = 0xfff00000}, // 172.16.0.0/12
|
|
{.network = 169u << 24 | 254 << 16, .mask = 0xffff0000}, // 169.254.0.0/16
|
|
{.network = 10u << 24, .mask = 0xff000000}, // 10.0.0.0/8
|
|
{.network = 127u << 24 | 1, .mask = 0xffffffff}, // 127.0.0.1/32
|
|
{.network = 100u << 24 | 64 << 16, .mask = 0xffc00000}, // 100.64.0.0/10
|
|
};
|
|
const uint32_t addr = ntohl(ip);
|
|
for (const auto &cidrRange : privateCidrRanges) {
|
|
if (cidrRange.network == (addr & cidrRange.mask)) {
|
|
LOG_INFO("MQTT server on a private IP");
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// Separate a <host>[:<port>] string. Returns a pair containing the parsed host and port. If the port is
|
|
// not present in the input string, or is invalid, the value of the `port` argument will be returned.
|
|
std::pair<String, uint16_t> parseHostAndPort(String server, uint16_t port = 0)
|
|
{
|
|
const int delimIndex = server.indexOf(':');
|
|
if (delimIndex > 0) {
|
|
const long parsedPort = server.substring(delimIndex + 1, server.length()).toInt();
|
|
if (parsedPort < 1 || parsedPort > UINT16_MAX) {
|
|
LOG_WARN("Invalid MQTT port %d: %s", parsedPort, server.c_str());
|
|
} else {
|
|
port = parsedPort;
|
|
}
|
|
server[delimIndex] = 0;
|
|
}
|
|
return std::make_pair(std::move(server), port);
|
|
}
|
|
|
|
bool isDefaultServer(const String &host)
|
|
{
|
|
return host.length() == 0 || host == default_mqtt_address;
|
|
}
|
|
|
|
bool isDefaultRootTopic(const String &root)
|
|
{
|
|
return root.length() == 0 || root == default_mqtt_root;
|
|
}
|
|
|
|
struct PubSubConfig {
|
|
explicit PubSubConfig(const meshtastic_ModuleConfig_MQTTConfig &config)
|
|
{
|
|
if (*config.address) {
|
|
serverAddr = config.address;
|
|
mqttUsername = config.username;
|
|
mqttPassword = config.password;
|
|
}
|
|
if (config.tls_enabled) {
|
|
serverPort = 8883;
|
|
}
|
|
auto [parsedServerAddr, parsedServerPort] = parseHostAndPort(serverAddr.c_str(), serverPort);
|
|
serverAddr = std::move(parsedServerAddr);
|
|
serverPort = parsedServerPort;
|
|
}
|
|
|
|
// Defaults
|
|
static constexpr uint16_t defaultPort = 1883;
|
|
static constexpr uint16_t defaultPortTls = 8883;
|
|
|
|
uint16_t serverPort = defaultPort;
|
|
String serverAddr = default_mqtt_address;
|
|
const char *mqttUsername = default_mqtt_username;
|
|
const char *mqttPassword = default_mqtt_password;
|
|
};
|
|
|
|
#if HAS_NETWORKING
|
|
bool connectPubSub(const PubSubConfig &config, PubSubClient &pubSub, Client &client)
|
|
{
|
|
pubSub.setBufferSize(1024, 1024);
|
|
pubSub.setClient(client);
|
|
pubSub.setServer(config.serverAddr.c_str(), config.serverPort);
|
|
|
|
LOG_INFO("Connecting directly to MQTT server %s, port: %d, username: %s, password: ***", config.serverAddr.c_str(),
|
|
config.serverPort, config.mqttUsername);
|
|
|
|
// Generate node ID from nodenum for client identification
|
|
std::string nodeId = nodeDB->getNodeId();
|
|
const bool connected = pubSub.connect(nodeId.c_str(), config.mqttUsername, config.mqttPassword);
|
|
if (connected) {
|
|
isConnected = true;
|
|
LOG_INFO("MQTT connected");
|
|
} else {
|
|
isConnected = false;
|
|
LOG_WARN("Failed to connect to MQTT server");
|
|
}
|
|
return connected;
|
|
}
|
|
#endif
|
|
|
|
inline bool isConnectedToNetwork()
|
|
{
|
|
#ifdef USE_WS5500
|
|
if (ETH.connected())
|
|
return true;
|
|
#elif defined(USE_CH390D)
|
|
if (CH390.isConnected())
|
|
return true;
|
|
#endif
|
|
|
|
#if HAS_WIFI
|
|
return WiFi.isConnected();
|
|
#elif HAS_ETHERNET
|
|
return Ethernet.linkStatus() == LinkON;
|
|
#elif defined(ARCH_PORTDUINO)
|
|
return true;
|
|
#else
|
|
return false;
|
|
#endif
|
|
}
|
|
|
|
/** return true if we have a channel that wants uplink/downlink or map reporting is enabled
|
|
*/
|
|
bool wantsLink()
|
|
{
|
|
const bool hasChannelorMapReport =
|
|
moduleConfig.mqtt.enabled && (moduleConfig.mqtt.map_reporting_enabled || channels.anyMqttEnabled());
|
|
return hasChannelorMapReport && (moduleConfig.mqtt.proxy_to_client_enabled || isConnectedToNetwork());
|
|
}
|
|
} // namespace
|
|
|
|
void MQTT::mqttCallback(char *topic, byte *payload, unsigned int length)
|
|
{
|
|
mqtt->onReceive(topic, payload, length);
|
|
}
|
|
|
|
void MQTT::onClientProxyReceive(meshtastic_MqttClientProxyMessage msg)
|
|
{
|
|
onReceive(msg.topic, msg.payload_variant.data.bytes, msg.payload_variant.data.size);
|
|
}
|
|
|
|
void MQTT::onReceive(char *topic, byte *payload, size_t length)
|
|
{
|
|
if (length == 0) {
|
|
LOG_WARN("Empty MQTT payload received, topic %s!", topic);
|
|
return;
|
|
}
|
|
|
|
onReceiveProto(topic, payload, length);
|
|
}
|
|
|
|
void mqttInit()
|
|
{
|
|
if (!moduleConfig.mqtt.enabled)
|
|
return;
|
|
|
|
new MQTT();
|
|
}
|
|
|
|
#if HAS_NETWORKING
|
|
MQTT::MQTT() : MQTT(std::unique_ptr<MQTTClient>(new MQTTClient())) {}
|
|
MQTT::MQTT(std::unique_ptr<MQTTClient> _mqttClient)
|
|
: concurrency::OSThread("mqtt"), mqttQueue(MAX_MQTT_QUEUE), mqttClient(std::move(_mqttClient)), pubSub(*mqttClient)
|
|
#else
|
|
MQTT::MQTT() : concurrency::OSThread("mqtt"), mqttQueue(MAX_MQTT_QUEUE)
|
|
#endif
|
|
{
|
|
if (moduleConfig.mqtt.enabled) {
|
|
LOG_DEBUG("Init MQTT");
|
|
|
|
assert(!mqtt);
|
|
mqtt = this;
|
|
|
|
if (*moduleConfig.mqtt.root) {
|
|
cryptTopic = moduleConfig.mqtt.root + cryptTopic;
|
|
mapTopic = moduleConfig.mqtt.root + mapTopic;
|
|
isConfiguredForDefaultRootTopic = isDefaultRootTopic(moduleConfig.mqtt.root);
|
|
} else {
|
|
cryptTopic = "msh" + cryptTopic;
|
|
mapTopic = "msh" + mapTopic;
|
|
isConfiguredForDefaultRootTopic = true;
|
|
}
|
|
|
|
if (moduleConfig.mqtt.map_reporting_enabled && moduleConfig.mqtt.has_map_report_settings) {
|
|
map_position_precision = Default::getConfiguredOrDefault(moduleConfig.mqtt.map_report_settings.position_precision,
|
|
default_map_position_precision);
|
|
map_publish_interval_msecs = Default::getConfiguredOrDefaultMs(
|
|
moduleConfig.mqtt.map_report_settings.publish_interval_secs, default_map_publish_interval_secs);
|
|
}
|
|
|
|
auto [host, parsedPort] = parseHostAndPort(moduleConfig.mqtt.address);
|
|
(void)parsedPort;
|
|
isConfiguredForDefaultServer = isDefaultServer(host);
|
|
IPAddress ip;
|
|
isMqttServerAddressPrivate = ip.fromString(host.c_str()) && isPrivateIpAddress(ip);
|
|
|
|
#if HAS_NETWORKING
|
|
if (!moduleConfig.mqtt.proxy_to_client_enabled)
|
|
pubSub.setCallback(mqttCallback);
|
|
#endif
|
|
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled) {
|
|
LOG_INFO("MQTT configured to use client proxy");
|
|
enabled = true;
|
|
runASAP = true;
|
|
reconnectCount = 0;
|
|
#ifndef PIO_UNIT_TESTING
|
|
publishNodeInfo();
|
|
#endif
|
|
}
|
|
// preflightSleepObserver.observe(&preflightSleep);
|
|
} else {
|
|
disable();
|
|
}
|
|
}
|
|
|
|
bool MQTT::isConnectedDirectly()
|
|
{
|
|
#if HAS_NETWORKING
|
|
return pubSub.connected();
|
|
#else
|
|
return false;
|
|
#endif
|
|
}
|
|
|
|
bool MQTT::publish(const char *topic, const char *payload, bool retained)
|
|
{
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled) {
|
|
meshtastic_MqttClientProxyMessage *msg = mqttClientProxyMessagePool.allocZeroed();
|
|
if (!msg)
|
|
return false;
|
|
msg->which_payload_variant = meshtastic_MqttClientProxyMessage_text_tag;
|
|
strncpy(msg->topic, topic, sizeof(msg->topic));
|
|
msg->topic[sizeof(msg->topic) - 1] = '\0';
|
|
strncpy(msg->payload_variant.text, payload, sizeof(msg->payload_variant.text));
|
|
msg->payload_variant.text[sizeof(msg->payload_variant.text) - 1] = '\0';
|
|
msg->retained = retained;
|
|
service->sendMqttMessageToClientProxy(msg);
|
|
return true;
|
|
}
|
|
#if HAS_NETWORKING
|
|
else if (isConnectedDirectly()) {
|
|
return pubSub.publish(topic, payload, retained);
|
|
}
|
|
#endif
|
|
return false;
|
|
}
|
|
|
|
bool MQTT::publish(const char *topic, const uint8_t *payload, size_t length, bool retained)
|
|
{
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled) {
|
|
meshtastic_MqttClientProxyMessage *msg = mqttClientProxyMessagePool.allocZeroed();
|
|
if (!msg)
|
|
return false;
|
|
msg->which_payload_variant = meshtastic_MqttClientProxyMessage_data_tag;
|
|
strncpy(msg->topic, topic, sizeof(msg->topic));
|
|
msg->topic[sizeof(msg->topic) - 1] = '\0'; // Ensure null termination
|
|
if (length > sizeof(msg->payload_variant.data.bytes))
|
|
length = sizeof(msg->payload_variant.data.bytes);
|
|
msg->payload_variant.data.size = length;
|
|
memcpy(msg->payload_variant.data.bytes, payload, length);
|
|
msg->retained = retained;
|
|
service->sendMqttMessageToClientProxy(msg);
|
|
return true;
|
|
}
|
|
#if HAS_NETWORKING
|
|
else if (isConnectedDirectly()) {
|
|
return pubSub.publish(topic, payload, length, retained);
|
|
}
|
|
#endif
|
|
return false;
|
|
}
|
|
|
|
void MQTT::reconnect()
|
|
{
|
|
isConnected = false;
|
|
if (wantsLink()) {
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled) {
|
|
LOG_INFO("MQTT connect via client proxy instead");
|
|
enabled = true;
|
|
runASAP = true;
|
|
reconnectCount = 0;
|
|
|
|
publishNodeInfo();
|
|
return; // Don't try to connect directly to the server
|
|
}
|
|
#if HAS_NETWORKING
|
|
const PubSubConfig ps_config(moduleConfig.mqtt);
|
|
MQTTClient *clientConnection = mqttClient.get();
|
|
#if MQTT_SUPPORTS_TLS
|
|
if (moduleConfig.mqtt.tls_enabled) {
|
|
mqttClientTLS.setInsecure();
|
|
LOG_INFO("Use TLS-encrypted session");
|
|
clientConnection = &mqttClientTLS;
|
|
} else {
|
|
LOG_INFO("Use non-TLS-encrypted session");
|
|
}
|
|
#endif
|
|
if (connectPubSub(ps_config, pubSub, *clientConnection)) {
|
|
enabled = true; // Start running background process again
|
|
runASAP = true;
|
|
reconnectCount = 0;
|
|
isMqttServerAddressPrivate = isPrivateIpAddress(clientConnection->remoteIP());
|
|
isConnected = true;
|
|
publishNodeInfo();
|
|
sendSubscriptions();
|
|
} else {
|
|
#if HAS_WIFI && !defined(ARCH_PORTDUINO)
|
|
reconnectCount++;
|
|
LOG_ERROR("Failed to contact MQTT server directly (%d/%d)", reconnectCount, reconnectMax);
|
|
if (reconnectCount >= reconnectMax) {
|
|
#if defined(USE_WS5500) || defined(USE_CH390D)
|
|
LOG_WARN("MQTT connect failed repeatedly; waiting for Ethernet reconnect");
|
|
#else
|
|
needReconnect = true;
|
|
if (wifiReconnect) {
|
|
wifiReconnect->setIntervalFromNow(0);
|
|
} else {
|
|
LOG_WARN("MQTT connect failed repeatedly, but WiFi reconnect is unavailable");
|
|
}
|
|
#endif
|
|
reconnectCount = 0;
|
|
}
|
|
#endif
|
|
}
|
|
#endif
|
|
}
|
|
}
|
|
|
|
void MQTT::sendSubscriptions()
|
|
{
|
|
#if HAS_NETWORKING
|
|
bool hasDownlink = false;
|
|
size_t numChan = channels.getNumChannels();
|
|
for (size_t i = 0; i < numChan; i++) {
|
|
const auto &ch = channels.getByIndex(i);
|
|
if (ch.settings.downlink_enabled) {
|
|
hasDownlink = true;
|
|
std::string topic = cryptTopic + channels.getGlobalId(i) + "/+";
|
|
LOG_INFO("Subscribe to %s", topic.c_str());
|
|
pubSub.subscribe(topic.c_str(), 1); // FIXME, is QOS 1 right?
|
|
}
|
|
}
|
|
#if !MESHTASTIC_EXCLUDE_PKI
|
|
if (hasDownlink) {
|
|
std::string topic = cryptTopic + "PKI/+";
|
|
LOG_INFO("Subscribe to %s", topic.c_str());
|
|
pubSub.subscribe(topic.c_str(), 1);
|
|
}
|
|
#endif
|
|
#endif
|
|
}
|
|
|
|
int32_t MQTT::runOnce()
|
|
{
|
|
if (!moduleConfig.mqtt.enabled || !(moduleConfig.mqtt.map_reporting_enabled || channels.anyMqttEnabled()))
|
|
return disable();
|
|
bool wantConnection = wantsLink();
|
|
|
|
perhapsReportToMap();
|
|
|
|
// If connected poll rapidly, otherwise only occasionally check for a wifi connection change and ability to contact server
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled) {
|
|
publishQueuedMessages();
|
|
return 200;
|
|
}
|
|
#if HAS_NETWORKING
|
|
else if (!pubSub.loop()) {
|
|
if (!wantConnection)
|
|
return 5000; // If we don't want connection now, check again in 5 secs
|
|
else {
|
|
reconnect();
|
|
// If we succeeded, empty the queue one by one and start reading rapidly, else try again in 30 seconds (TCP
|
|
// connections are EXPENSIVE so try rarely)
|
|
if (isConnectedDirectly()) {
|
|
publishQueuedMessages();
|
|
return 200;
|
|
} else
|
|
return 30000;
|
|
}
|
|
} else {
|
|
// we are connected to server, check often for new requests on the TCP port
|
|
if (!wantConnection) {
|
|
LOG_INFO("MQTT link not needed, drop");
|
|
pubSub.disconnect();
|
|
}
|
|
|
|
powerFSM.trigger(EVENT_CONTACT_FROM_PHONE); // Suppress entering light sleep (because that would turn off bluetooth)
|
|
return 20;
|
|
}
|
|
#else
|
|
// No networking available, return default interval
|
|
return 30000;
|
|
#endif
|
|
}
|
|
|
|
bool MQTT::isValidConfig(const meshtastic_ModuleConfig_MQTTConfig &config, MQTTClient *client)
|
|
{
|
|
const PubSubConfig parsed(config);
|
|
|
|
if (config.enabled && !config.proxy_to_client_enabled) {
|
|
#if HAS_NETWORKING
|
|
if (config.tls_enabled) {
|
|
#if !MQTT_SUPPORTS_TLS
|
|
LOG_ERROR("Invalid MQTT config: tls_enabled is not supported on this node");
|
|
return false;
|
|
#endif
|
|
}
|
|
// Perform a lightweight TCP connectivity check without using connectPubSub(),
|
|
// which mutates the module's isConnected state. This only checks if the server
|
|
// is reachable - it does not establish an MQTT session.
|
|
// Settings are always saved regardless of the result.
|
|
if (isConnectedToNetwork()) {
|
|
MQTTClient testClient;
|
|
if (!testClient.connect(parsed.serverAddr.c_str(), parsed.serverPort)) {
|
|
const char *warning = "Could not reach the MQTT server. Settings will be saved, but please verify the server "
|
|
"address and credentials.";
|
|
LOG_WARN(warning);
|
|
#ifndef PIO_UNIT_TESTING
|
|
meshtastic_ClientNotification *cn = clientNotificationPool.allocZeroed();
|
|
if (cn) {
|
|
cn->level = meshtastic_LogRecord_Level_WARNING;
|
|
cn->time = getValidTime(RTCQualityFromNet);
|
|
strncpy(cn->message, warning, sizeof(cn->message) - 1);
|
|
cn->message[sizeof(cn->message) - 1] = '\0';
|
|
service->sendClientNotification(cn);
|
|
}
|
|
#endif
|
|
}
|
|
testClient.stop();
|
|
}
|
|
#else
|
|
const char *warning = "Invalid MQTT config: proxy_to_client_enabled must be enabled on nodes that do not have a network";
|
|
LOG_ERROR(warning);
|
|
#ifndef PIO_UNIT_TESTING
|
|
meshtastic_ClientNotification *cn = clientNotificationPool.allocZeroed();
|
|
if (cn) {
|
|
cn->level = meshtastic_LogRecord_Level_ERROR;
|
|
cn->time = getValidTime(RTCQualityFromNet);
|
|
strncpy(cn->message, warning, sizeof(cn->message) - 1);
|
|
cn->message[sizeof(cn->message) - 1] = '\0'; // Ensure null termination
|
|
service->sendClientNotification(cn);
|
|
}
|
|
#endif
|
|
return false;
|
|
#endif
|
|
}
|
|
|
|
const bool defaultServer = isDefaultServer(parsed.serverAddr);
|
|
if (defaultServer && !IS_ONE_OF(parsed.serverPort, PubSubConfig::defaultPort, PubSubConfig::defaultPortTls)) {
|
|
const char *warning = "Invalid MQTT config: default server address must not have a port specified";
|
|
LOG_ERROR(warning);
|
|
#ifndef PIO_UNIT_TESTING
|
|
meshtastic_ClientNotification *cn = clientNotificationPool.allocZeroed();
|
|
if (cn) {
|
|
cn->level = meshtastic_LogRecord_Level_ERROR;
|
|
cn->time = getValidTime(RTCQualityFromNet);
|
|
strncpy(cn->message, warning, sizeof(cn->message) - 1);
|
|
cn->message[sizeof(cn->message) - 1] = '\0'; // Ensure null termination
|
|
service->sendClientNotification(cn);
|
|
}
|
|
#endif
|
|
return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
void MQTT::publishNodeInfo()
|
|
{
|
|
// TODO: NodeInfo broadcast over MQTT only (NODENUM_BROADCAST_NO_LORA)
|
|
}
|
|
void MQTT::publishQueuedMessages()
|
|
{
|
|
if (mqttQueue.isEmpty())
|
|
return;
|
|
|
|
if (!moduleConfig.mqtt.proxy_to_client_enabled && !isConnected)
|
|
return;
|
|
|
|
LOG_DEBUG("Publish enqueued MQTT message");
|
|
const std::unique_ptr<QueueEntry> entry(mqttQueue.dequeuePtr(0));
|
|
LOG_INFO("publish %s, %u bytes from queue", entry->topic.c_str(), entry->envBytes.size());
|
|
publish(entry->topic.c_str(), entry->envBytes.data(), entry->envBytes.size(), false);
|
|
}
|
|
|
|
void MQTT::onSend(const meshtastic_MeshPacket &mp_encrypted, const meshtastic_MeshPacket &mp_decoded, ChannelIndex chIndex)
|
|
{
|
|
if (mp_encrypted.via_mqtt)
|
|
return; // Don't send messages that came from MQTT back into MQTT
|
|
bool uplinkEnabled = false;
|
|
for (int i = 0; i <= 7; i++) {
|
|
if (channels.getByIndex(i).settings.uplink_enabled)
|
|
uplinkEnabled = true;
|
|
}
|
|
if (!uplinkEnabled)
|
|
return; // no channels have an uplink enabled
|
|
auto &ch = channels.getByIndex(chIndex);
|
|
|
|
// mp_decoded will not be decoded when it's PKI encrypted and not directed to us
|
|
if (mp_decoded.which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
|
|
// For uplinking other's packets, check if it's not OK to MQTT or if it's an older packet without the bitfield
|
|
bool dontUplink = !mp_decoded.decoded.has_bitfield || !(mp_decoded.decoded.bitfield & BITFIELD_OK_TO_MQTT_MASK);
|
|
// Respect the DontMqttMeBro flag for other nodes' packets on public MQTT servers
|
|
if (!isFromUs(&mp_decoded) && !isMqttServerAddressPrivate && dontUplink) {
|
|
LOG_INFO("MQTT onSend - Not forwarding packet due to DontMqttMeBro flag");
|
|
return;
|
|
}
|
|
|
|
if (isConfiguredForDefaultServer && (mp_decoded.decoded.portnum == meshtastic_PortNum_RANGE_TEST_APP ||
|
|
mp_decoded.decoded.portnum == meshtastic_PortNum_DETECTION_SENSOR_APP)) {
|
|
LOG_DEBUG("MQTT onSend - Ignoring range test or detection sensor message on public mqtt");
|
|
return;
|
|
}
|
|
}
|
|
// Either encrypted packet (we couldn't decrypt) is marked as pki_encrypted, or we could decode the PKI encrypted packet
|
|
bool isPKIEncrypted = mp_encrypted.pki_encrypted || mp_decoded.pki_encrypted;
|
|
// If it was to a channel, check uplink enabled, else must be pki_encrypted
|
|
if (!(ch.settings.uplink_enabled || isPKIEncrypted))
|
|
return;
|
|
const char *channelId = isPKIEncrypted ? "PKI" : channels.getGlobalId(chIndex);
|
|
|
|
LOG_DEBUG("MQTT onSend - Publish ");
|
|
const meshtastic_MeshPacket *p;
|
|
if (moduleConfig.mqtt.encryption_enabled) {
|
|
p = &mp_encrypted;
|
|
LOG_DEBUG("encrypted message");
|
|
} else if (mp_decoded.which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
|
|
p = &mp_decoded;
|
|
LOG_DEBUG("portnum %i message", mp_decoded.decoded.portnum);
|
|
} else {
|
|
LOG_DEBUG("nothing, pkt not decrypted");
|
|
return; // Don't upload a still-encrypted PKI packet if not encryption_enabled
|
|
}
|
|
|
|
// Generate node ID from nodenum for service envelope
|
|
std::string nodeId = nodeDB->getNodeId();
|
|
|
|
const meshtastic_ServiceEnvelope env = {.packet = const_cast<meshtastic_MeshPacket *>(p),
|
|
.channel_id = const_cast<char *>(channelId),
|
|
.gateway_id = const_cast<char *>(nodeId.c_str())};
|
|
size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, &env);
|
|
std::string topic = cryptTopic + channelId + "/" + nodeId;
|
|
|
|
if (moduleConfig.mqtt.proxy_to_client_enabled || this->isConnectedDirectly()) {
|
|
LOG_DEBUG("MQTT Publish %s, %u bytes", topic.c_str(), numBytes);
|
|
publish(topic.c_str(), bytes, numBytes, false);
|
|
} else {
|
|
LOG_INFO("MQTT not connected, queue packet");
|
|
QueueEntry *entry;
|
|
if (mqttQueue.numFree() == 0) {
|
|
LOG_WARN("MQTT queue is full, discard oldest");
|
|
entry = mqttQueue.dequeuePtr(0);
|
|
} else {
|
|
entry = new QueueEntry;
|
|
}
|
|
entry->topic = std::move(topic);
|
|
entry->envBytes.assign(bytes, numBytes);
|
|
if (mqttQueue.enqueue(entry, 0) == false) {
|
|
LOG_CRIT("Failed to add a message to mqttQueue!");
|
|
abort();
|
|
}
|
|
}
|
|
}
|
|
|
|
void MQTT::perhapsReportToMap()
|
|
{
|
|
if (!moduleConfig.mqtt.map_reporting_enabled || !moduleConfig.mqtt.map_report_settings.should_report_location ||
|
|
!(moduleConfig.mqtt.proxy_to_client_enabled || isConnectedDirectly()))
|
|
return;
|
|
|
|
// Coerce the map position precision to be within the valid range
|
|
// This removes obtusely large radius and privacy problematic ones from the map
|
|
if (map_position_precision < 12 || map_position_precision > 15) {
|
|
LOG_WARN("MQTT Map report position precision %u is out of range, using default %u", map_position_precision,
|
|
default_map_position_precision);
|
|
map_position_precision = default_map_position_precision;
|
|
}
|
|
|
|
if (Throttle::isWithinTimespanMs(last_report_to_map, map_publish_interval_msecs) && last_report_to_map != 0)
|
|
return;
|
|
|
|
if (localPosition.latitude_i == 0 && localPosition.longitude_i == 0) {
|
|
if (Throttle::isWithinTimespanMs(lastPositionUnavailableWarning, POSITION_UNAVAILABLE_WARNING_INTERVAL_MS) == false) {
|
|
LOG_WARN("MQTT Map report enabled, but no position available");
|
|
lastPositionUnavailableWarning = millis();
|
|
}
|
|
return;
|
|
}
|
|
|
|
// Allocate MeshPacket and fill it
|
|
meshtastic_MeshPacket *mp = packetPool.allocZeroed();
|
|
if (!mp)
|
|
return;
|
|
mp->which_payload_variant = meshtastic_MeshPacket_decoded_tag;
|
|
mp->from = nodeDB->getNodeNum();
|
|
mp->to = NODENUM_BROADCAST;
|
|
mp->decoded.portnum = meshtastic_PortNum_MAP_REPORT_APP;
|
|
|
|
// Fill MapReport message
|
|
meshtastic_MapReport mapReport = meshtastic_MapReport_init_default;
|
|
// owner.long_name (40) is wider than mapReport.long_name (25); bound by the destination
|
|
strncpy(mapReport.long_name, owner.long_name, sizeof(mapReport.long_name));
|
|
mapReport.long_name[sizeof(mapReport.long_name) - 1] = '\0';
|
|
memcpy(mapReport.short_name, owner.short_name, sizeof(owner.short_name));
|
|
mapReport.role = config.device.role;
|
|
mapReport.hw_model = owner.hw_model;
|
|
strncpy(mapReport.firmware_version, optstr(APP_VERSION), sizeof(mapReport.firmware_version));
|
|
mapReport.region = config.lora.region;
|
|
mapReport.modem_preset = config.lora.modem_preset;
|
|
mapReport.has_default_channel = channels.hasDefaultChannel();
|
|
mapReport.has_opted_report_location = true;
|
|
|
|
// Set position with precision (same as in PositionModule)
|
|
mapReport.latitude_i = localPosition.latitude_i & (UINT32_MAX << (32 - map_position_precision));
|
|
mapReport.longitude_i = localPosition.longitude_i & (UINT32_MAX << (32 - map_position_precision));
|
|
mapReport.latitude_i += (1 << (31 - map_position_precision));
|
|
mapReport.longitude_i += (1 << (31 - map_position_precision));
|
|
|
|
mapReport.altitude = localPosition.altitude;
|
|
mapReport.position_precision = map_position_precision;
|
|
|
|
mapReport.num_online_local_nodes = nodeDB->getNumOnlineMeshNodes(true);
|
|
|
|
// Encode MapReport message into the MeshPacket
|
|
mp->decoded.payload.size =
|
|
pb_encode_to_bytes(mp->decoded.payload.bytes, sizeof(mp->decoded.payload.bytes), &meshtastic_MapReport_msg, &mapReport);
|
|
|
|
// Generate node ID from nodenum for service envelope
|
|
std::string nodeId = nodeDB->getNodeId();
|
|
|
|
// Encode the MeshPacket into a binary ServiceEnvelope and publish
|
|
const meshtastic_ServiceEnvelope se = {
|
|
.packet = mp,
|
|
.channel_id = (char *)channels.getGlobalId(channels.getPrimaryIndex()), // Use primary channel as the channel_id
|
|
.gateway_id = const_cast<char *>(nodeId.c_str())};
|
|
size_t numBytes = pb_encode_to_bytes(bytes, sizeof(bytes), &meshtastic_ServiceEnvelope_msg, &se);
|
|
|
|
LOG_INFO("MQTT Publish map report to %s", mapTopic.c_str());
|
|
publish(mapTopic.c_str(), bytes, numBytes, false);
|
|
|
|
// Release the allocated memory for MeshPacket
|
|
packetPool.release(mp);
|
|
|
|
// Update the last report time
|
|
last_report_to_map = millis();
|
|
}
|