mirror of
https://github.com/agessaman/MeshCore.git
synced 2026-08-27 22:34:14 +00:00
Implement Let's Mesh Analyzer integration in MQTT Bridge
- Switch from PubSub to PsychicMqttClient for async operations and websockets support - Add support for US and EU Let's Mesh Analyzer servers with JWT authentication. - Introduce CLI commands to enable/disable analyzer servers. - Update NodePrefs to store analyzer server settings. - Modify MQTTBridge to publish status and packet data to analyzer servers via WebSocket MQTT. - Enhance documentation to reflect new features and configuration options.
This commit is contained in:
agessaman
+20
-1
@@ -56,6 +56,8 @@ The MQTT bridge comes with the following defaults:
|
||||
- **WiFi Password**: "password_here" (must be configured)
|
||||
- **Timezone**: "America/Los_Angeles" (Pacific Time with DST support)
|
||||
- **Timezone Offset**: -8 hours (fallback)
|
||||
- **Let's Mesh Analyzer US**: Enabled (mqtt-us-v1.letsmesh.net:443)
|
||||
- **Let's Mesh Analyzer EU**: Enabled (mqtt-eu-v1.letsmesh.net:443)
|
||||
|
||||
## CLI Commands
|
||||
|
||||
@@ -69,6 +71,8 @@ The MQTT bridge comes with the following defaults:
|
||||
- `get mqtt.raw` - Get raw message setting (on/off)
|
||||
- `get mqtt.tx` - Get TX message setting (on/off)
|
||||
- `get mqtt.interval` - Get status publish interval (ms)
|
||||
- `get mqtt.analyzer.us` - Get US Let's Mesh Analyzer server setting (on/off)
|
||||
- `get mqtt.analyzer.eu` - Get EU Let's Mesh Analyzer server setting (on/off)
|
||||
|
||||
#### Set Commands
|
||||
- `set mqtt.origin <name>` - Set device origin name
|
||||
@@ -78,6 +82,8 @@ The MQTT bridge comes with the following defaults:
|
||||
- `set mqtt.raw on|off` - Enable/disable raw messages
|
||||
- `set mqtt.tx on|off` - Enable/disable TX packet messages
|
||||
- `set mqtt.interval <ms>` - Set status publish interval (1000-3600000 ms)
|
||||
- `set mqtt.analyzer.us on|off` - Enable/disable US Let's Mesh Analyzer server
|
||||
- `set mqtt.analyzer.eu on|off` - Enable/disable EU Let's Mesh Analyzer server
|
||||
|
||||
### WiFi Commands
|
||||
|
||||
@@ -219,6 +225,16 @@ Minimal raw packet data for map integration.
|
||||
- Periodic time updates (every hour)
|
||||
- Proper UTC system time handling
|
||||
|
||||
### Let's Mesh Analyzer Integration
|
||||
- **JWT Authentication**: Ed25519-signed tokens for secure MQTT authentication
|
||||
- **WebSocket MQTT**: Support for MQTT over WebSocket connections (TLS/SSL)
|
||||
- **Dual Server Support**: Both US and EU servers enabled by default
|
||||
- **Automatic Token Generation**: Creates authentication tokens using device's Ed25519 keys
|
||||
- **Username Format**: `v1_{UPPERCASE_PUBLIC_KEY}` (e.g., `v1_7E7662676F7F0850A8A355BAAFBFC1EB7B4174C340442D7D7161C9474A2C9400`)
|
||||
- **Server Configuration**:
|
||||
- US Server: `mqtt-us-v1.letsmesh.net:443` (WebSocket with TLS)
|
||||
- EU Server: `mqtt-eu-v1.letsmesh.net:443` (WebSocket with TLS)
|
||||
|
||||
## Testing
|
||||
|
||||
1. Flash the MQTT bridge firmware to your device
|
||||
@@ -235,11 +251,14 @@ Minimal raw packet data for map integration.
|
||||
- **NTPClient**: Network time protocol client
|
||||
- **Timezone**: Timezone conversion library (JChristensen/Timezone)
|
||||
- **WiFi**: ESP32 WiFi functionality
|
||||
- **Ed25519**: Cryptographic library for JWT token signing
|
||||
- **JWTHelper**: Custom JWT token generation for Let's Mesh Analyzer authentication
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
- Device-signed JWTs for WebSocket MQTT servers
|
||||
- Full WebSocket MQTT implementation (currently JWT tokens are generated but WebSocket publishing is pending)
|
||||
- Multiple broker configuration via CLI
|
||||
- Advanced packet filtering
|
||||
- Custom topic templates
|
||||
- TLS/SSL support for secure connections
|
||||
- Real-time WebSocket MQTT publishing to Let's Mesh Analyzer servers
|
||||
|
||||
@@ -681,7 +681,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
|
||||
#elif defined(WITH_ESPNOW_BRIDGE)
|
||||
, bridge(&_prefs, _mgr, &rtc)
|
||||
#elif defined(WITH_MQTT_BRIDGE)
|
||||
, bridge(&_prefs, _mgr, &rtc)
|
||||
, bridge(&_prefs, _mgr, &rtc, &self_id)
|
||||
#endif
|
||||
{
|
||||
last_millis = 0;
|
||||
@@ -743,9 +743,13 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
|
||||
StrHelper::strncpy(_prefs.wifi_ssid, "ssid_here", sizeof(_prefs.wifi_ssid));
|
||||
StrHelper::strncpy(_prefs.wifi_password, "password_here", sizeof(_prefs.wifi_password));
|
||||
|
||||
// Timezone defaults (Pacific Time with DST support)
|
||||
StrHelper::strncpy(_prefs.timezone_string, "America/Los_Angeles", sizeof(_prefs.timezone_string));
|
||||
_prefs.timezone_offset = -8; // fallback
|
||||
// Timezone defaults (Pacific Time with DST support)
|
||||
StrHelper::strncpy(_prefs.timezone_string, "America/Los_Angeles", sizeof(_prefs.timezone_string));
|
||||
_prefs.timezone_offset = -8; // fallback
|
||||
|
||||
// Let's Mesh Analyzer defaults (both enabled by default)
|
||||
_prefs.mqtt_analyzer_us_enabled = 1; // enabled
|
||||
_prefs.mqtt_analyzer_eu_enabled = 1; // enabled
|
||||
}
|
||||
|
||||
void MyMesh::begin(FILESYSTEM *fs) {
|
||||
|
||||
@@ -0,0 +1,215 @@
|
||||
#!/usr/bin/env python
|
||||
#
|
||||
# modified ESP32 x509 certificate bundle generation utility to run with platformio
|
||||
#
|
||||
# Converts PEM and DER certificates to a custom bundle format which stores just the
|
||||
# subject name and public key to reduce space
|
||||
#
|
||||
# The bundle will have the format: number of certificates; crt 1 subject name length; crt 1 public key length;
|
||||
# crt 1 subject name; crt 1 public key; crt 2...
|
||||
#
|
||||
# SPDX-FileCopyrightText: 2018-2022 Espressif Systems (Shanghai) CO LTD
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
from __future__ import with_statement
|
||||
|
||||
from pathlib import Path
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
import requests
|
||||
from io import open
|
||||
|
||||
Import("env")
|
||||
|
||||
try:
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
except ImportError:
|
||||
env.Execute("$PYTHONEXE -m pip install cryptography")
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
|
||||
ca_bundle_bin_file = 'x509_crt_bundle.bin'
|
||||
mozilla_cacert_url = 'https://curl.se/ca/cacert.pem'
|
||||
adafruit_filtered_cacert_url = 'https://raw.githubusercontent.com/adafruit/certificates/main/data/roots-filtered.pem'
|
||||
adafruit_full_cacert_url = 'https://raw.githubusercontent.com/adafruit/certificates/main/data/roots-full.pem'
|
||||
certs_dir = Path("./ssl_certs")
|
||||
binary_dir = Path("./src/certs")
|
||||
|
||||
quiet = False
|
||||
|
||||
def download_cacert_file(source):
|
||||
if source == "mozilla":
|
||||
response = requests.get(mozilla_cacert_url)
|
||||
elif source == "adafruit":
|
||||
response = requests.get(adafruit_filtered_cacert_url)
|
||||
elif source == "adafruit-full":
|
||||
response = requests.get(adafruit_full_cacert_url)
|
||||
else:
|
||||
raise InputError('Invalid certificate source')
|
||||
|
||||
if response.status_code == 200:
|
||||
|
||||
# Ensure the directory exists, create it if necessary
|
||||
os.makedirs(certs_dir, exist_ok=True)
|
||||
|
||||
# Generate the full path to the output file
|
||||
output_file = os.path.join(certs_dir, "cacert.pem")
|
||||
|
||||
# Write the certificate bundle to the output file with utf-8 encoding
|
||||
with open(output_file, "w", encoding="utf-8") as f:
|
||||
f.write(response.text)
|
||||
|
||||
status('Certificate bundle downloaded to: %s' % output_file)
|
||||
else:
|
||||
status('Failed to fetch the certificate bundle.')
|
||||
|
||||
def status(msg):
|
||||
""" Print status message to stderr """
|
||||
if not quiet:
|
||||
critical(msg)
|
||||
|
||||
|
||||
def critical(msg):
|
||||
""" Print critical message to stderr """
|
||||
sys.stderr.write('SSL Cert Store: ')
|
||||
sys.stderr.write(msg)
|
||||
sys.stderr.write('\n')
|
||||
|
||||
|
||||
class CertificateBundle:
|
||||
def __init__(self):
|
||||
self.certificates = []
|
||||
self.compressed_crts = []
|
||||
|
||||
if os.path.isfile(ca_bundle_bin_file):
|
||||
os.remove(ca_bundle_bin_file)
|
||||
|
||||
def add_from_path(self, crts_path):
|
||||
|
||||
found = False
|
||||
for file_path in os.listdir(crts_path):
|
||||
found |= self.add_from_file(os.path.join(crts_path, file_path))
|
||||
|
||||
if found is False:
|
||||
raise InputError('No valid x509 certificates found in %s' % crts_path)
|
||||
|
||||
def add_from_file(self, file_path):
|
||||
try:
|
||||
if file_path.endswith('.pem'):
|
||||
status('Parsing certificates from %s' % file_path)
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
crt_str = f.read()
|
||||
self.add_from_pem(crt_str)
|
||||
return True
|
||||
|
||||
elif file_path.endswith('.der'):
|
||||
status('Parsing certificates from %s' % file_path)
|
||||
with open(file_path, 'rb') as f:
|
||||
crt_str = f.read()
|
||||
self.add_from_der(crt_str)
|
||||
return True
|
||||
|
||||
except ValueError:
|
||||
critical('Invalid certificate in %s' % file_path)
|
||||
raise InputError('Invalid certificate')
|
||||
|
||||
return False
|
||||
|
||||
def add_from_pem(self, crt_str):
|
||||
""" A single PEM file may have multiple certificates """
|
||||
|
||||
crt = ''
|
||||
count = 0
|
||||
start = False
|
||||
|
||||
for strg in crt_str.splitlines(True):
|
||||
if strg == '-----BEGIN CERTIFICATE-----\n' and start is False:
|
||||
crt = ''
|
||||
start = True
|
||||
elif strg == '-----END CERTIFICATE-----\n' and start is True:
|
||||
crt += strg + '\n'
|
||||
start = False
|
||||
self.certificates.append(x509.load_pem_x509_certificate(crt.encode(), default_backend()))
|
||||
count += 1
|
||||
if start is True:
|
||||
crt += strg
|
||||
|
||||
if count == 0:
|
||||
raise InputError('No certificate found')
|
||||
|
||||
status('Successfully added %d certificates' % count)
|
||||
|
||||
def add_from_der(self, crt_str):
|
||||
self.certificates.append(x509.load_der_x509_certificate(crt_str, default_backend()))
|
||||
status('Successfully added 1 certificate')
|
||||
|
||||
def create_bundle(self):
|
||||
# Sort certificates in order to do binary search when looking up certificates
|
||||
self.certificates = sorted(self.certificates, key=lambda cert: cert.subject.public_bytes(default_backend()))
|
||||
|
||||
bundle = struct.pack('>H', len(self.certificates))
|
||||
|
||||
for crt in self.certificates:
|
||||
""" Read the public key as DER format """
|
||||
pub_key = crt.public_key()
|
||||
pub_key_der = pub_key.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
|
||||
|
||||
""" Read the subject name as DER format """
|
||||
sub_name_der = crt.subject.public_bytes(default_backend())
|
||||
|
||||
name_len = len(sub_name_der)
|
||||
key_len = len(pub_key_der)
|
||||
len_data = struct.pack('>HH', name_len, key_len)
|
||||
|
||||
bundle += len_data
|
||||
bundle += sub_name_der
|
||||
bundle += pub_key_der
|
||||
|
||||
return bundle
|
||||
|
||||
class InputError(RuntimeError):
|
||||
def __init__(self, e):
|
||||
super(InputError, self).__init__(e)
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
bundle = CertificateBundle()
|
||||
|
||||
try:
|
||||
cert_source = env.GetProjectOption("board_ssl_cert_source")
|
||||
|
||||
if (cert_source == "mozilla" or cert_source == "adafruit"):
|
||||
download_cacert_file(cert_source)
|
||||
bundle.add_from_file(os.path.join(certs_dir, "cacert.pem"))
|
||||
elif (cert_source == "folder"):
|
||||
bundle.add_from_path(certs_dir)
|
||||
except ValueError:
|
||||
critical('Invalid configuration option: use \'board_ssl_cert_source\' parameter in platformio.ini' )
|
||||
raise InputError('Invalid certificate')
|
||||
|
||||
status('Successfully added %d certificates in total' % len(bundle.certificates))
|
||||
|
||||
crt_bundle = bundle.create_bundle()
|
||||
|
||||
# Ensure the directory exists, create it if necessary
|
||||
os.makedirs(binary_dir, exist_ok=True)
|
||||
|
||||
output_file = os.path.join(binary_dir, ca_bundle_bin_file)
|
||||
|
||||
with open(output_file, 'wb') as f:
|
||||
f.write(crt_bundle)
|
||||
|
||||
status('Successfully created %s' % output_file)
|
||||
|
||||
|
||||
try:
|
||||
main()
|
||||
except InputError as e:
|
||||
print(e)
|
||||
sys.exit(2)
|
||||
Binary file not shown.
@@ -70,6 +70,7 @@ void CommonCLI::loadPrefsInt(FILESYSTEM* fs, const char* filename) {
|
||||
file.read((uint8_t *)&_prefs->gps_enabled, sizeof(_prefs->gps_enabled)); // 156
|
||||
file.read((uint8_t *)&_prefs->gps_interval, sizeof(_prefs->gps_interval)); // 157
|
||||
file.read((uint8_t *)&_prefs->advert_loc_policy, sizeof (_prefs->advert_loc_policy)); // 161
|
||||
<<<<<<< HEAD
|
||||
file.read((uint8_t *)&_prefs->discovery_mod_timestamp, sizeof(_prefs->discovery_mod_timestamp)); // 162 (from upstream)
|
||||
// MQTT settings - skip reading from main prefs file (now stored separately)
|
||||
// For backward compatibility, we'll skip these bytes if they exist in old files
|
||||
@@ -98,6 +99,29 @@ void CommonCLI::loadPrefsInt(FILESYSTEM* fs, const char* filename) {
|
||||
file.read(skip_buffer, to_read);
|
||||
remaining -= to_read;
|
||||
}
|
||||
=======
|
||||
// MQTT settings
|
||||
file.read((uint8_t *)&_prefs->mqtt_origin, sizeof(_prefs->mqtt_origin)); // 162
|
||||
file.read((uint8_t *)&_prefs->mqtt_iata, sizeof(_prefs->mqtt_iata)); // 194
|
||||
file.read((uint8_t *)&_prefs->mqtt_status_enabled, sizeof(_prefs->mqtt_status_enabled)); // 202
|
||||
file.read((uint8_t *)&_prefs->mqtt_packets_enabled, sizeof(_prefs->mqtt_packets_enabled)); // 203
|
||||
file.read((uint8_t *)&_prefs->mqtt_raw_enabled, sizeof(_prefs->mqtt_raw_enabled)); // 204
|
||||
file.read((uint8_t *)&_prefs->mqtt_tx_enabled, sizeof(_prefs->mqtt_tx_enabled)); // 205
|
||||
file.read((uint8_t *)&_prefs->mqtt_status_interval, sizeof(_prefs->mqtt_status_interval)); // 206
|
||||
|
||||
// WiFi settings
|
||||
file.read((uint8_t *)&_prefs->wifi_ssid, sizeof(_prefs->wifi_ssid)); // 209
|
||||
file.read((uint8_t *)&_prefs->wifi_password, sizeof(_prefs->wifi_password)); // 241
|
||||
|
||||
// Timezone settings
|
||||
file.read((uint8_t *)&_prefs->timezone_string, sizeof(_prefs->timezone_string)); // 305
|
||||
file.read((uint8_t *)&_prefs->timezone_offset, sizeof(_prefs->timezone_offset)); // 337
|
||||
|
||||
// Let's Mesh Analyzer settings
|
||||
file.read((uint8_t *)&_prefs->mqtt_analyzer_us_enabled, sizeof(_prefs->mqtt_analyzer_us_enabled)); // 338
|
||||
file.read((uint8_t *)&_prefs->mqtt_analyzer_eu_enabled, sizeof(_prefs->mqtt_analyzer_eu_enabled)); // 339
|
||||
// 209
|
||||
>>>>>>> 6f42dc3 (Implement Let's Mesh Analyzer integration in MQTT Bridge)
|
||||
|
||||
// sanitise bad pref values
|
||||
_prefs->rx_delay_base = constrain(_prefs->rx_delay_base, 0, 20.0f);
|
||||
@@ -177,6 +201,7 @@ void CommonCLI::savePrefs(FILESYSTEM* fs) {
|
||||
file.write((uint8_t *)&_prefs->gps_enabled, sizeof(_prefs->gps_enabled)); // 156
|
||||
file.write((uint8_t *)&_prefs->gps_interval, sizeof(_prefs->gps_interval)); // 157
|
||||
file.write((uint8_t *)&_prefs->advert_loc_policy, sizeof(_prefs->advert_loc_policy)); // 161
|
||||
<<<<<<< HEAD
|
||||
file.write((uint8_t *)&_prefs->discovery_mod_timestamp, sizeof(_prefs->discovery_mod_timestamp)); // 162 (from upstream)
|
||||
// MQTT settings - no longer saved here (stored in separate /mqtt_prefs file)
|
||||
// Write zeros/padding to maintain file format compatibility
|
||||
@@ -203,6 +228,29 @@ void CommonCLI::savePrefs(FILESYSTEM* fs) {
|
||||
file.write(pad, to_write);
|
||||
remaining -= to_write;
|
||||
}
|
||||
=======
|
||||
// MQTT settings
|
||||
file.write((uint8_t *)&_prefs->mqtt_origin, sizeof(_prefs->mqtt_origin)); // 162
|
||||
file.write((uint8_t *)&_prefs->mqtt_iata, sizeof(_prefs->mqtt_iata)); // 194
|
||||
file.write((uint8_t *)&_prefs->mqtt_status_enabled, sizeof(_prefs->mqtt_status_enabled)); // 202
|
||||
file.write((uint8_t *)&_prefs->mqtt_packets_enabled, sizeof(_prefs->mqtt_packets_enabled)); // 203
|
||||
file.write((uint8_t *)&_prefs->mqtt_raw_enabled, sizeof(_prefs->mqtt_raw_enabled)); // 204
|
||||
file.write((uint8_t *)&_prefs->mqtt_tx_enabled, sizeof(_prefs->mqtt_tx_enabled)); // 205
|
||||
file.write((uint8_t *)&_prefs->mqtt_status_interval, sizeof(_prefs->mqtt_status_interval)); // 206
|
||||
|
||||
// WiFi settings
|
||||
file.write((uint8_t *)&_prefs->wifi_ssid, sizeof(_prefs->wifi_ssid)); // 209
|
||||
file.write((uint8_t *)&_prefs->wifi_password, sizeof(_prefs->wifi_password)); // 241
|
||||
|
||||
// Timezone settings
|
||||
file.write((uint8_t *)&_prefs->timezone_string, sizeof(_prefs->timezone_string)); // 305
|
||||
file.write((uint8_t *)&_prefs->timezone_offset, sizeof(_prefs->timezone_offset)); // 337
|
||||
|
||||
// Let's Mesh Analyzer settings
|
||||
file.write((uint8_t *)&_prefs->mqtt_analyzer_us_enabled, sizeof(_prefs->mqtt_analyzer_us_enabled)); // 338
|
||||
file.write((uint8_t *)&_prefs->mqtt_analyzer_eu_enabled, sizeof(_prefs->mqtt_analyzer_eu_enabled)); // 339
|
||||
// 209
|
||||
>>>>>>> 6f42dc3 (Implement Let's Mesh Analyzer integration in MQTT Bridge)
|
||||
|
||||
file.close();
|
||||
}
|
||||
@@ -413,6 +461,10 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch
|
||||
sprintf(reply, "> %s", _prefs->timezone_string);
|
||||
} else if (memcmp(config, "timezone.offset", 15) == 0) {
|
||||
sprintf(reply, "> %d", _prefs->timezone_offset);
|
||||
} else if (memcmp(config, "mqtt.analyzer.us", 16) == 0) {
|
||||
sprintf(reply, "> %s", _prefs->mqtt_analyzer_us_enabled ? "on" : "off");
|
||||
} else if (memcmp(config, "mqtt.analyzer.eu", 16) == 0) {
|
||||
sprintf(reply, "> %s", _prefs->mqtt_analyzer_eu_enabled ? "on" : "off");
|
||||
#endif
|
||||
} else {
|
||||
sprintf(reply, "??: %s", config);
|
||||
@@ -661,6 +713,14 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch
|
||||
} else {
|
||||
strcpy(reply, "Error: timezone offset must be between -12 and +14");
|
||||
}
|
||||
} else if (memcmp(config, "mqtt.analyzer.us ", 17) == 0) {
|
||||
_prefs->mqtt_analyzer_us_enabled = memcmp(&config[17], "on", 2) == 0;
|
||||
savePrefs();
|
||||
strcpy(reply, "OK");
|
||||
} else if (memcmp(config, "mqtt.analyzer.eu ", 17) == 0) {
|
||||
_prefs->mqtt_analyzer_eu_enabled = memcmp(&config[17], "on", 2) == 0;
|
||||
savePrefs();
|
||||
strcpy(reply, "OK");
|
||||
#endif
|
||||
} else {
|
||||
sprintf(reply, "unknown config: %s", config);
|
||||
|
||||
@@ -98,6 +98,7 @@ struct MQTTPrefs {
|
||||
// Timezone settings
|
||||
char timezone_string[32]; // Timezone string (e.g., "America/Los_Angeles")
|
||||
int8_t timezone_offset; // Timezone offset in hours (-12 to +14) - fallback
|
||||
<<<<<<< HEAD
|
||||
<<<<<<< HEAD
|
||||
|
||||
// MQTT server settings
|
||||
@@ -105,12 +106,17 @@ struct MQTTPrefs {
|
||||
uint16_t mqtt_port; // MQTT server port
|
||||
char mqtt_username[32]; // MQTT username
|
||||
char mqtt_password[64]; // MQTT password
|
||||
=======
|
||||
>>>>>>> 6f42dc3 (Implement Let's Mesh Analyzer integration in MQTT Bridge)
|
||||
|
||||
// Let's Mesh Analyzer settings
|
||||
uint8_t mqtt_analyzer_us_enabled; // Enable US analyzer server
|
||||
uint8_t mqtt_analyzer_eu_enabled; // Enable EU analyzer server
|
||||
<<<<<<< HEAD
|
||||
char mqtt_owner_public_key[65]; // Owner public key (hex string, same length as repeater public key)
|
||||
char mqtt_email[64]; // Owner email address for matching nodes with owners
|
||||
=======
|
||||
>>>>>>> 6f42dc3 (Implement Let's Mesh Analyzer integration in MQTT Bridge)
|
||||
};
|
||||
#endif
|
||||
|
||||
|
||||
@@ -0,0 +1,252 @@
|
||||
#include "JWTHelper.h"
|
||||
#include <ArduinoJson.h>
|
||||
#include <SHA256.h>
|
||||
#include <string.h>
|
||||
#include "ed_25519.h"
|
||||
#include "mbedtls/base64.h"
|
||||
|
||||
// Base64 URL encoding table (without padding)
|
||||
static const char base64url_chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
|
||||
|
||||
bool JWTHelper::createAuthToken(
|
||||
const mesh::LocalIdentity& identity,
|
||||
const char* audience,
|
||||
unsigned long issuedAt,
|
||||
unsigned long expiresIn,
|
||||
char* token,
|
||||
size_t tokenSize
|
||||
) {
|
||||
Serial.printf("JWTHelper: Starting JWT creation for audience: %s\n", audience);
|
||||
|
||||
if (!audience || !token || tokenSize == 0) {
|
||||
Serial.printf("JWTHelper: Invalid parameters - audience: %p, token: %p, tokenSize: %d\n", audience, token, (int)tokenSize);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Use current time if not specified
|
||||
if (issuedAt == 0) {
|
||||
issuedAt = time(nullptr);
|
||||
}
|
||||
Serial.printf("JWTHelper: Using issuedAt: %lu\n", issuedAt);
|
||||
|
||||
// Create header
|
||||
char header[256];
|
||||
size_t headerLen = createHeader(header, sizeof(header));
|
||||
if (headerLen == 0) {
|
||||
Serial.printf("JWTHelper: Failed to create header\n");
|
||||
return false;
|
||||
}
|
||||
Serial.printf("JWTHelper: Header created, length: %d\n", (int)headerLen);
|
||||
Serial.printf("JWTHelper: Header: %s\n", header);
|
||||
|
||||
// Get public key as UPPERCASE HEX string (not base64!)
|
||||
char publicKeyHex[65]; // 32 bytes * 2 + null terminator
|
||||
mesh::Utils::toHex(publicKeyHex, identity.pub_key, PUB_KEY_SIZE);
|
||||
|
||||
// Convert to uppercase
|
||||
for (int i = 0; publicKeyHex[i]; i++) {
|
||||
publicKeyHex[i] = toupper(publicKeyHex[i]);
|
||||
}
|
||||
|
||||
Serial.printf("JWTHelper: Public key hex: %s (length: %d)\n", publicKeyHex, (int)strlen(publicKeyHex));
|
||||
|
||||
// Create payload with HEX public key (not base64!)
|
||||
char payload[512];
|
||||
size_t payloadLen = createPayload(publicKeyHex, audience, issuedAt, expiresIn, payload, sizeof(payload));
|
||||
if (payloadLen == 0) {
|
||||
Serial.printf("JWTHelper: Failed to create payload\n");
|
||||
return false;
|
||||
}
|
||||
Serial.printf("JWTHelper: Payload created, length: %d\n", (int)payloadLen);
|
||||
Serial.printf("JWTHelper: Payload: %s\n", payload);
|
||||
|
||||
// Create signing input: header.payload
|
||||
char signingInput[768];
|
||||
size_t signingInputLen = headerLen + 1 + payloadLen;
|
||||
if (signingInputLen >= sizeof(signingInput)) {
|
||||
Serial.printf("JWTHelper: Signing input too large: %d >= %d\n", (int)signingInputLen, (int)sizeof(signingInput));
|
||||
return false;
|
||||
}
|
||||
|
||||
memcpy(signingInput, header, headerLen);
|
||||
signingInput[headerLen] = '.';
|
||||
memcpy(signingInput + headerLen + 1, payload, payloadLen);
|
||||
Serial.printf("JWTHelper: Signing input created, length: %d\n", (int)signingInputLen);
|
||||
|
||||
// Sign the data using direct Ed25519 signing
|
||||
uint8_t signature[64];
|
||||
|
||||
// Create a non-const copy of the identity to access writeTo method
|
||||
mesh::LocalIdentity identity_copy = identity;
|
||||
|
||||
// Export the private and public keys using the writeTo method
|
||||
uint8_t export_buffer[96]; // PRV_KEY_SIZE + PUB_KEY_SIZE = 64 + 32 = 96 bytes
|
||||
size_t exported_size = identity_copy.writeTo(export_buffer, sizeof(export_buffer));
|
||||
|
||||
if (exported_size != 96) {
|
||||
Serial.printf("JWTHelper: Failed to export keys, got %d bytes instead of 96\n", (int)exported_size);
|
||||
return false;
|
||||
}
|
||||
|
||||
// The first 64 bytes are the private key, next 32 bytes are the public key
|
||||
uint8_t* private_key = export_buffer; // First 64 bytes
|
||||
uint8_t* public_key = export_buffer + 64; // Next 32 bytes
|
||||
|
||||
Serial.printf("JWTHelper: Using direct Ed25519 signing\n");
|
||||
Serial.printf("JWTHelper: Private key length: %d, Public key length: %d\n", 64, 32);
|
||||
|
||||
// Use direct Ed25519 signing
|
||||
ed25519_sign(signature, (const unsigned char*)signingInput, signingInputLen, public_key, private_key);
|
||||
Serial.printf("JWTHelper: Signature created using direct Ed25519\n");
|
||||
|
||||
// Verify the signature locally
|
||||
int verify_result = ed25519_verify(signature, (const unsigned char*)signingInput, signingInputLen, public_key);
|
||||
Serial.printf("JWTHelper: Signature verification result: %d (should be 1 for valid)\n", verify_result);
|
||||
|
||||
if (verify_result != 1) {
|
||||
Serial.println("JWTHelper: ERROR - Signature verification failed!");
|
||||
return false;
|
||||
}
|
||||
|
||||
// Log the exact signing input
|
||||
Serial.printf("JWTHelper: Signing input: %s\n", signingInput);
|
||||
Serial.printf("JWTHelper: Signing input hex: ");
|
||||
for (size_t i = 0; i < signingInputLen; i++) {
|
||||
Serial.printf("%02x", signingInput[i]);
|
||||
}
|
||||
Serial.println();
|
||||
|
||||
// Log the signature
|
||||
Serial.printf("JWTHelper: Signature hex: ");
|
||||
for (int i = 0; i < 64; i++) {
|
||||
Serial.printf("%02x", signature[i]);
|
||||
}
|
||||
Serial.println();
|
||||
|
||||
// Convert signature to hex (MeshCore Decoder expects hex, not base64url)
|
||||
char signatureHex[129]; // 64 bytes * 2 + null terminator
|
||||
for (int i = 0; i < 64; i++) {
|
||||
sprintf(signatureHex + (i * 2), "%02X", signature[i]);
|
||||
}
|
||||
signatureHex[128] = '\0';
|
||||
|
||||
Serial.printf("JWTHelper: Signature converted to hex, length: %d\n", (int)strlen(signatureHex));
|
||||
Serial.printf("JWTHelper: Signature Hex: %s\n", signatureHex);
|
||||
|
||||
// Create final token: header.payload.signatureHex (MeshCore Decoder format)
|
||||
size_t sigHexLen = strlen(signatureHex);
|
||||
size_t totalLen = headerLen + 1 + payloadLen + 1 + sigHexLen;
|
||||
if (totalLen >= tokenSize) {
|
||||
Serial.printf("JWTHelper: Token too large: %d >= %d\n", (int)totalLen, (int)tokenSize);
|
||||
return false;
|
||||
}
|
||||
|
||||
memcpy(token, header, headerLen);
|
||||
token[headerLen] = '.';
|
||||
memcpy(token + headerLen + 1, payload, payloadLen);
|
||||
token[headerLen + 1 + payloadLen] = '.';
|
||||
memcpy(token + headerLen + 1 + payloadLen + 1, signatureHex, sigHexLen);
|
||||
token[totalLen] = '\0';
|
||||
|
||||
Serial.printf("JWTHelper: JWT token created successfully, total length: %d\n", (int)totalLen);
|
||||
Serial.printf("JWTHelper: JWT Token: %s\n", token);
|
||||
return true;
|
||||
}
|
||||
|
||||
size_t JWTHelper::base64UrlEncode(const uint8_t* input, size_t inputLen, char* output, size_t outputSize) {
|
||||
Serial.printf("JWTHelper: base64UrlEncode called with inputLen: %d, outputSize: %d\n", (int)inputLen, (int)outputSize);
|
||||
|
||||
if (!input || !output || outputSize == 0) {
|
||||
Serial.printf("JWTHelper: base64UrlEncode invalid parameters\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Use ESP32's built-in mbedTLS base64 encoding
|
||||
size_t outlen = 0;
|
||||
int ret = mbedtls_base64_encode((unsigned char*)output, outputSize - 1, &outlen, input, inputLen);
|
||||
|
||||
if (ret != 0) {
|
||||
Serial.printf("JWTHelper: mbedtls_base64_encode failed with error: %d\n", ret);
|
||||
return 0;
|
||||
}
|
||||
|
||||
Serial.printf("JWTHelper: mbedtls_base64_encode result: %s (outlen: %d)\n", output, (int)outlen);
|
||||
|
||||
// Convert to base64 URL format (replace + with -, / with _, remove padding =)
|
||||
String encoded(output);
|
||||
encoded.replace('+', '-');
|
||||
encoded.replace('/', '_');
|
||||
encoded.replace("=", "");
|
||||
|
||||
// Copy back to output buffer
|
||||
size_t len = encoded.length();
|
||||
if (len >= outputSize) {
|
||||
Serial.printf("JWTHelper: base64UrlEncode output too large: %d >= %d\n", (int)len, (int)outputSize);
|
||||
return 0;
|
||||
}
|
||||
|
||||
strcpy(output, encoded.c_str());
|
||||
Serial.printf("JWTHelper: base64UrlEncode completed, outputLen: %d\n", (int)len);
|
||||
return len;
|
||||
}
|
||||
|
||||
size_t JWTHelper::createHeader(char* output, size_t outputSize) {
|
||||
Serial.printf("JWTHelper: createHeader called with outputSize: %d\n", (int)outputSize);
|
||||
|
||||
// Create JWT header: {"alg":"Ed25519","typ":"JWT"}
|
||||
DynamicJsonDocument doc(256);
|
||||
doc["alg"] = "Ed25519";
|
||||
doc["typ"] = "JWT";
|
||||
|
||||
// Use temporary buffer for JSON
|
||||
char jsonBuffer[256];
|
||||
size_t len = serializeJson(doc, jsonBuffer, sizeof(jsonBuffer));
|
||||
Serial.printf("JWTHelper: JSON serialized, length: %d\n", (int)len);
|
||||
if (len == 0 || len >= sizeof(jsonBuffer)) {
|
||||
Serial.printf("JWTHelper: JSON serialization failed or too large\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Base64 URL encode from temporary buffer to output
|
||||
size_t encodedLen = base64UrlEncode((uint8_t*)jsonBuffer, len, output, outputSize);
|
||||
Serial.printf("JWTHelper: Header base64 encoded, length: %d\n", (int)encodedLen);
|
||||
return encodedLen;
|
||||
}
|
||||
|
||||
size_t JWTHelper::createPayload(
|
||||
const char* publicKey,
|
||||
const char* audience,
|
||||
unsigned long issuedAt,
|
||||
unsigned long expiresIn,
|
||||
char* output,
|
||||
size_t outputSize
|
||||
) {
|
||||
Serial.printf("JWTHelper: createPayload called with outputSize: %d\n", (int)outputSize);
|
||||
Serial.printf("JWTHelper: publicKey: %s, audience: %s, issuedAt: %lu, expiresIn: %lu\n",
|
||||
publicKey, audience, issuedAt, expiresIn);
|
||||
|
||||
// Create JWT payload
|
||||
DynamicJsonDocument doc(512);
|
||||
doc["publicKey"] = publicKey;
|
||||
doc["aud"] = audience;
|
||||
doc["iat"] = issuedAt;
|
||||
|
||||
if (expiresIn > 0) {
|
||||
doc["exp"] = issuedAt + expiresIn;
|
||||
}
|
||||
|
||||
// Use temporary buffer for JSON
|
||||
char jsonBuffer[512];
|
||||
size_t len = serializeJson(doc, jsonBuffer, sizeof(jsonBuffer));
|
||||
Serial.printf("JWTHelper: Payload JSON serialized, length: %d\n", (int)len);
|
||||
if (len == 0 || len >= sizeof(jsonBuffer)) {
|
||||
Serial.printf("JWTHelper: Payload JSON serialization failed or too large\n");
|
||||
return 0;
|
||||
}
|
||||
|
||||
// Base64 URL encode from temporary buffer to output
|
||||
size_t encodedLen = base64UrlEncode((uint8_t*)jsonBuffer, len, output, outputSize);
|
||||
Serial.printf("JWTHelper: Payload base64 encoded, length: %d\n", (int)encodedLen);
|
||||
return encodedLen;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
#pragma once
|
||||
|
||||
#include "MeshCore.h"
|
||||
#include "Identity.h"
|
||||
|
||||
/**
|
||||
* JWT Helper for creating authentication tokens
|
||||
*
|
||||
* This class provides functionality to create JWT-style authentication tokens
|
||||
* signed with Ed25519 private keys for MQTT authentication.
|
||||
*/
|
||||
class JWTHelper {
|
||||
public:
|
||||
/**
|
||||
* Create an authentication token for MQTT authentication
|
||||
*
|
||||
* @param identity LocalIdentity instance for signing
|
||||
* @param audience Audience string (e.g., "mqtt-us-v1.letsmesh.net")
|
||||
* @param issuedAt Unix timestamp (0 for current time)
|
||||
* @param expiresIn Expiration time in seconds (0 for no expiration)
|
||||
* @param token Buffer to store the resulting token
|
||||
* @param tokenSize Size of the token buffer
|
||||
* @return true if token was created successfully
|
||||
*/
|
||||
static bool createAuthToken(
|
||||
const mesh::LocalIdentity& identity,
|
||||
const char* audience,
|
||||
unsigned long issuedAt = 0,
|
||||
unsigned long expiresIn = 0,
|
||||
char* token = nullptr,
|
||||
size_t tokenSize = 0
|
||||
);
|
||||
|
||||
private:
|
||||
/**
|
||||
* Base64 URL encode data
|
||||
*
|
||||
* @param input Input data
|
||||
* @param inputLen Length of input data
|
||||
* @param output Output buffer
|
||||
* @param outputSize Size of output buffer
|
||||
* @return Length of encoded data, or 0 on error
|
||||
*/
|
||||
static size_t base64UrlEncode(const uint8_t* input, size_t inputLen, char* output, size_t outputSize);
|
||||
|
||||
/**
|
||||
* Create JWT header
|
||||
*
|
||||
* @param output Output buffer
|
||||
* @param outputSize Size of output buffer
|
||||
* @return Length of header, or 0 on error
|
||||
*/
|
||||
static size_t createHeader(char* output, size_t outputSize);
|
||||
|
||||
/**
|
||||
* Create JWT payload
|
||||
*
|
||||
* @param publicKey Public key in hex format
|
||||
* @param audience Audience string
|
||||
* @param issuedAt Issued at timestamp
|
||||
* @param expiresIn Expiration time in seconds (0 for no expiration)
|
||||
* @param output Output buffer
|
||||
* @param outputSize Size of output buffer
|
||||
* @return Length of payload, or 0 on error
|
||||
*/
|
||||
static size_t createPayload(
|
||||
const char* publicKey,
|
||||
const char* audience,
|
||||
unsigned long issuedAt,
|
||||
unsigned long expiresIn,
|
||||
char* output,
|
||||
size_t outputSize
|
||||
);
|
||||
|
||||
};
|
||||
@@ -4,14 +4,18 @@
|
||||
#include <WiFiUdp.h>
|
||||
#include <Timezone.h>
|
||||
|
||||
// Using ESP32's built-in certificate bundle
|
||||
|
||||
#ifdef WITH_MQTT_BRIDGE
|
||||
|
||||
MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc)
|
||||
: BridgeBase(prefs, mgr, rtc), _mqtt_client(nullptr), _wifi_client(nullptr),
|
||||
MQTTBridge::MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity)
|
||||
: BridgeBase(prefs, mgr, rtc), _mqtt_client(nullptr),
|
||||
_active_brokers(0), _queue_head(0), _queue_tail(0), _queue_count(0),
|
||||
_last_status_publish(0), _status_interval(300000), // 5 minutes default
|
||||
_ntp_client(_ntp_udp, "pool.ntp.org", 0, 60000), _last_ntp_sync(0), _ntp_synced(false),
|
||||
_timezone(nullptr), _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0) {
|
||||
_ntp_client(_ntp_udp, "pool.ntp.org", 0, 60000), _last_ntp_sync(0), _ntp_synced(false),
|
||||
_timezone(nullptr), _last_raw_len(0), _last_snr(0), _last_rssi(0), _last_raw_timestamp(0),
|
||||
_analyzer_us_enabled(false), _analyzer_eu_enabled(false), _identity(identity),
|
||||
_analyzer_us_client(nullptr), _analyzer_eu_client(nullptr) {
|
||||
|
||||
// Initialize default values
|
||||
strncpy(_origin, "MeshCore-Repeater", sizeof(_origin) - 1);
|
||||
@@ -102,13 +106,45 @@ void MQTTBridge::begin() {
|
||||
return;
|
||||
}
|
||||
|
||||
// Initialize WiFi client
|
||||
_wifi_client = new WiFiClient();
|
||||
_mqtt_client = new PubSubClient(*_wifi_client);
|
||||
// Initialize PsychicMqttClient
|
||||
_mqtt_client = new PsychicMqttClient();
|
||||
|
||||
// Set up event callbacks for the main MQTT client
|
||||
_mqtt_client->onConnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT client connected, session present: %s", sessionPresent ? "true" : "false");
|
||||
// Update broker connection status
|
||||
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
|
||||
if (_brokers[i].enabled && !_brokers[i].connected) {
|
||||
_brokers[i].connected = true;
|
||||
_active_brokers++;
|
||||
MQTT_DEBUG_PRINTLN("Broker %d marked as connected", i);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
_mqtt_client->onDisconnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT client disconnected, session present: %s", sessionPresent ? "true" : "false");
|
||||
// Update broker connection status
|
||||
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
|
||||
if (_brokers[i].connected) {
|
||||
_brokers[i].connected = false;
|
||||
_active_brokers--;
|
||||
MQTT_DEBUG_PRINTLN("Broker %d marked as disconnected", i);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Set default broker (meshtastic.pugetmesh.org)
|
||||
setBroker(0, "meshtastic.pugetmesh.org", 1883, "meshdev", "large4cats", true);
|
||||
|
||||
// Setup Let's Mesh Analyzer servers
|
||||
setupAnalyzerServers();
|
||||
|
||||
// Setup PsychicMqttClient WebSocket clients for analyzer servers
|
||||
setupAnalyzerClients();
|
||||
|
||||
// Connect to brokers
|
||||
connectToBrokers();
|
||||
|
||||
@@ -127,6 +163,18 @@ void MQTTBridge::end() {
|
||||
}
|
||||
}
|
||||
|
||||
// Disconnect analyzer clients
|
||||
if (_analyzer_us_client) {
|
||||
_analyzer_us_client->disconnect();
|
||||
delete _analyzer_us_client;
|
||||
_analyzer_us_client = nullptr;
|
||||
}
|
||||
if (_analyzer_eu_client) {
|
||||
_analyzer_eu_client->disconnect();
|
||||
delete _analyzer_eu_client;
|
||||
_analyzer_eu_client = nullptr;
|
||||
}
|
||||
|
||||
// Clear packet queue
|
||||
_queue_count = 0;
|
||||
_queue_head = 0;
|
||||
@@ -137,10 +185,6 @@ void MQTTBridge::end() {
|
||||
delete _mqtt_client;
|
||||
_mqtt_client = nullptr;
|
||||
}
|
||||
if (_wifi_client) {
|
||||
delete _wifi_client;
|
||||
_wifi_client = nullptr;
|
||||
}
|
||||
|
||||
_initialized = false;
|
||||
MQTT_DEBUG_PRINTLN("MQTT Bridge stopped");
|
||||
@@ -152,6 +196,9 @@ void MQTTBridge::loop() {
|
||||
// Maintain broker connections
|
||||
connectToBrokers();
|
||||
|
||||
// Maintain analyzer server connections
|
||||
maintainAnalyzerConnections();
|
||||
|
||||
// Process packet queue
|
||||
processPacketQueue();
|
||||
|
||||
@@ -168,8 +215,13 @@ void MQTTBridge::loop() {
|
||||
}
|
||||
|
||||
void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
|
||||
if (!_initialized || !_packets_enabled) return;
|
||||
if (!_initialized || !_packets_enabled) {
|
||||
MQTT_DEBUG_PRINTLN("Packet received but not processing - initialized: %s, packets_enabled: %s",
|
||||
_initialized ? "true" : "false", _packets_enabled ? "true" : "false");
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Packet received, queuing for transmission");
|
||||
// Queue packet for transmission
|
||||
queuePacket(packet, false);
|
||||
}
|
||||
@@ -182,6 +234,8 @@ void MQTTBridge::sendPacket(mesh::Packet *packet) {
|
||||
}
|
||||
|
||||
void MQTTBridge::connectToBrokers() {
|
||||
// For now, connect to the first enabled broker
|
||||
// TODO: Implement multi-broker support with PsychicMqttClient
|
||||
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
|
||||
if (!_brokers[i].enabled) continue;
|
||||
|
||||
@@ -191,42 +245,32 @@ void MQTTBridge::connectToBrokers() {
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Connecting to broker %d: %s:%d", i, _brokers[i].host, _brokers[i].port);
|
||||
|
||||
// Set broker for this connection
|
||||
_mqtt_client->setServer(_brokers[i].host, _brokers[i].port);
|
||||
|
||||
// Generate unique client ID
|
||||
char client_id[32];
|
||||
snprintf(client_id, sizeof(client_id), "%s_%d_%lu", _origin, i, millis());
|
||||
|
||||
// Attempt connection
|
||||
bool connected = _mqtt_client->connect(
|
||||
client_id,
|
||||
_brokers[i].username,
|
||||
_brokers[i].password
|
||||
);
|
||||
// Set broker URI and connect using PsychicMqttClient API
|
||||
char broker_uri[128];
|
||||
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->setServer(broker_uri);
|
||||
|
||||
if (connected) {
|
||||
_brokers[i].connected = true;
|
||||
_brokers[i].reconnect_interval = 5000; // Reset to 5 seconds
|
||||
_active_brokers++;
|
||||
MQTT_DEBUG_PRINTLN("Connected to broker %d", i);
|
||||
|
||||
// Publish initial status
|
||||
if (_status_enabled) {
|
||||
publishStatus();
|
||||
}
|
||||
} else {
|
||||
_brokers[i].connected = false;
|
||||
_brokers[i].last_attempt = millis();
|
||||
// Exponential backoff: 5s, 10s, 20s, 30s max
|
||||
_brokers[i].reconnect_interval = min(30000UL, _brokers[i].reconnect_interval * 2);
|
||||
MQTT_DEBUG_PRINTLN("Failed to connect to broker %d", i);
|
||||
// Set credentials if provided
|
||||
if (strlen(_brokers[i].username) > 0) {
|
||||
_mqtt_client->setCredentials(_brokers[i].username, _brokers[i].password);
|
||||
}
|
||||
|
||||
// Connect to the broker (PsychicMqttClient uses async connection)
|
||||
_mqtt_client->connect();
|
||||
|
||||
// Update attempt timestamp
|
||||
_brokers[i].last_attempt = millis();
|
||||
MQTT_DEBUG_PRINTLN("Initiating connection to broker %d", i);
|
||||
}
|
||||
|
||||
// Maintain connection
|
||||
if (_brokers[i].connected) {
|
||||
_mqtt_client->loop();
|
||||
// PsychicMqttClient handles connection maintenance internally
|
||||
// TODO: Implement proper connection state checking with callbacks
|
||||
if (!_mqtt_client->connected()) {
|
||||
_brokers[i].connected = false;
|
||||
_active_brokers--;
|
||||
@@ -237,13 +281,22 @@ void MQTTBridge::connectToBrokers() {
|
||||
}
|
||||
|
||||
void MQTTBridge::processPacketQueue() {
|
||||
if (_queue_count == 0 || !isAnyBrokerConnected()) return;
|
||||
if (_queue_count == 0 || !isAnyBrokerConnected()) {
|
||||
if (_queue_count > 0) {
|
||||
MQTT_DEBUG_PRINTLN("Queue has %d packets but no brokers connected", _queue_count);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Processing packet queue - count: %d", _queue_count);
|
||||
|
||||
// Process up to 5 packets per loop to avoid blocking
|
||||
int processed = 0;
|
||||
while (_queue_count > 0 && processed < 5) {
|
||||
QueuedPacket& queued = _packet_queue[_queue_head];
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Processing queued packet (is_tx: %s)", queued.is_tx ? "true" : "false");
|
||||
|
||||
// Publish packet
|
||||
publishPacket(queued.packet, queued.is_tx);
|
||||
|
||||
@@ -296,19 +349,27 @@ void MQTTBridge::publishStatus() {
|
||||
sizeof(json_buffer)
|
||||
);
|
||||
|
||||
if (len > 0) {
|
||||
// Publish to all connected brokers
|
||||
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
|
||||
if (_brokers[i].enabled && _brokers[i].connected) {
|
||||
char topic[128];
|
||||
snprintf(topic, sizeof(topic), "meshcore/%s/%s/status", _iata, _device_id);
|
||||
MQTT_DEBUG_PRINTLN("Publishing status to topic: %s", topic);
|
||||
|
||||
_mqtt_client->setServer(_brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->publish(topic, json_buffer, true); // retained
|
||||
}
|
||||
}
|
||||
}
|
||||
if (len > 0) {
|
||||
// Publish to all connected brokers
|
||||
for (int i = 0; i < MAX_MQTT_BROKERS_COUNT; i++) {
|
||||
if (_brokers[i].enabled && _brokers[i].connected) {
|
||||
char topic[128];
|
||||
snprintf(topic, sizeof(topic), "meshcore/%s/%s/status", _iata, _device_id);
|
||||
MQTT_DEBUG_PRINTLN("Publishing status to topic: %s", topic);
|
||||
|
||||
// Set broker for this connection (PsychicMqttClient uses URI format)
|
||||
char broker_uri[128];
|
||||
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->setServer(broker_uri);
|
||||
_mqtt_client->publish(topic, 1, true, json_buffer, strlen(json_buffer)); // qos=1, retained=true
|
||||
}
|
||||
}
|
||||
|
||||
// Also publish to Let's Mesh Analyzer servers
|
||||
char analyzer_topic[128];
|
||||
snprintf(analyzer_topic, sizeof(analyzer_topic), "meshcore/%s/%s/status", _iata, _device_id);
|
||||
publishToAnalyzerServers(analyzer_topic, json_buffer, true);
|
||||
}
|
||||
}
|
||||
|
||||
void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx) {
|
||||
@@ -344,10 +405,18 @@ void MQTTBridge::publishPacket(mesh::Packet* packet, bool is_tx) {
|
||||
snprintf(topic, sizeof(topic), "meshcore/%s/%s/packets", _iata, _device_id);
|
||||
MQTT_DEBUG_PRINTLN("Publishing packet to topic: %s", topic);
|
||||
|
||||
_mqtt_client->setServer(_brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->publish(topic, json_buffer);
|
||||
// Set broker for this connection (PsychicMqttClient uses URI format)
|
||||
char broker_uri[128];
|
||||
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->setServer(broker_uri);
|
||||
_mqtt_client->publish(topic, 1, false, json_buffer, strlen(json_buffer)); // qos=1, retained=false
|
||||
}
|
||||
}
|
||||
|
||||
// Also publish to Let's Mesh Analyzer servers
|
||||
char analyzer_topic[128];
|
||||
snprintf(analyzer_topic, sizeof(analyzer_topic), "meshcore/%s/%s/packets", _iata, _device_id);
|
||||
publishToAnalyzerServers(analyzer_topic, json_buffer, false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -373,10 +442,18 @@ void MQTTBridge::publishRaw(mesh::Packet* packet) {
|
||||
char topic[128];
|
||||
snprintf(topic, sizeof(topic), "meshcore/%s/%s/raw", _iata, _device_id);
|
||||
|
||||
_mqtt_client->setServer(_brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->publish(topic, json_buffer);
|
||||
// Set broker for this connection (PsychicMqttClient uses URI format)
|
||||
char broker_uri[128];
|
||||
snprintf(broker_uri, sizeof(broker_uri), "mqtt://%s:%d", _brokers[i].host, _brokers[i].port);
|
||||
_mqtt_client->setServer(broker_uri);
|
||||
_mqtt_client->publish(topic, 1, false, json_buffer, strlen(json_buffer)); // qos=1, retained=false
|
||||
}
|
||||
}
|
||||
|
||||
// Also publish to Let's Mesh Analyzer servers
|
||||
char analyzer_topic[128];
|
||||
snprintf(analyzer_topic, sizeof(analyzer_topic), "meshcore/%s/%s/raw", _iata, _device_id);
|
||||
publishToAnalyzerServers(analyzer_topic, json_buffer, false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -473,6 +550,279 @@ void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr,
|
||||
}
|
||||
}
|
||||
|
||||
void MQTTBridge::setupAnalyzerServers() {
|
||||
// Update analyzer server settings from preferences
|
||||
_analyzer_us_enabled = _prefs->mqtt_analyzer_us_enabled;
|
||||
_analyzer_eu_enabled = _prefs->mqtt_analyzer_eu_enabled;
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Analyzer servers - US: %s, EU: %s",
|
||||
_analyzer_us_enabled ? "enabled" : "disabled",
|
||||
_analyzer_eu_enabled ? "enabled" : "disabled");
|
||||
|
||||
// Create authentication token if any analyzer servers are enabled
|
||||
if (_analyzer_us_enabled || _analyzer_eu_enabled) {
|
||||
if (createAuthToken()) {
|
||||
MQTT_DEBUG_PRINTLN("Created authentication token for analyzer servers");
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Failed to create authentication token");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
bool MQTTBridge::createAuthToken() {
|
||||
if (!_identity) {
|
||||
MQTT_DEBUG_PRINTLN("No identity available for creating auth token");
|
||||
return false;
|
||||
}
|
||||
|
||||
// Create username in the format: v1_{UPPERCASE_PUBLIC_KEY}
|
||||
char public_key_hex[65];
|
||||
mesh::Utils::toHex(public_key_hex, _identity->pub_key, PUB_KEY_SIZE);
|
||||
|
||||
snprintf(_analyzer_username, sizeof(_analyzer_username), "v1_%s", public_key_hex);
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Creating auth token for username: %s", _analyzer_username);
|
||||
|
||||
bool us_token_created = false;
|
||||
bool eu_token_created = false;
|
||||
|
||||
// Create JWT token for US server
|
||||
if (_analyzer_us_enabled) {
|
||||
MQTT_DEBUG_PRINTLN("Creating JWT token for US server...");
|
||||
if (JWTHelper::createAuthToken(
|
||||
*_identity, "mqtt-us-v1.letsmesh.net",
|
||||
0, 86400, _auth_token_us, sizeof(_auth_token_us))) {
|
||||
MQTT_DEBUG_PRINTLN("Created auth token for US server");
|
||||
us_token_created = true;
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Failed to create auth token for US server");
|
||||
}
|
||||
}
|
||||
|
||||
// Create JWT token for EU server
|
||||
if (_analyzer_eu_enabled) {
|
||||
MQTT_DEBUG_PRINTLN("Creating JWT token for EU server...");
|
||||
if (JWTHelper::createAuthToken(
|
||||
*_identity, "mqtt-eu-v1.letsmesh.net",
|
||||
0, 86400, _auth_token_eu, sizeof(_auth_token_eu))) {
|
||||
MQTT_DEBUG_PRINTLN("Created auth token for EU server");
|
||||
eu_token_created = true;
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Failed to create auth token for EU server");
|
||||
}
|
||||
}
|
||||
|
||||
return us_token_created || eu_token_created;
|
||||
}
|
||||
|
||||
void MQTTBridge::publishToAnalyzerServers(const char* topic, const char* payload, bool retained) {
|
||||
if (!_analyzer_us_enabled && !_analyzer_eu_enabled) {
|
||||
MQTT_DEBUG_PRINTLN("No analyzer servers enabled, skipping publish to topic: %s", topic);
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Publishing to analyzer servers via WebSocket MQTT");
|
||||
MQTT_DEBUG_PRINTLN("Topic: %s", topic);
|
||||
MQTT_DEBUG_PRINTLN("Payload length: %d", strlen(payload));
|
||||
MQTT_DEBUG_PRINTLN("US enabled: %s, EU enabled: %s", _analyzer_us_enabled ? "true" : "false", _analyzer_eu_enabled ? "true" : "false");
|
||||
|
||||
// Publish to US server if enabled
|
||||
if (_analyzer_us_enabled && _analyzer_us_client) {
|
||||
MQTT_DEBUG_PRINTLN("Publishing to US analyzer server");
|
||||
publishToAnalyzerClient(_analyzer_us_client, topic, payload, retained);
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("US analyzer server not available (enabled: %s, client: %s)",
|
||||
_analyzer_us_enabled ? "true" : "false", _analyzer_us_client ? "exists" : "null");
|
||||
}
|
||||
|
||||
// Publish to EU server if enabled
|
||||
if (_analyzer_eu_enabled && _analyzer_eu_client) {
|
||||
MQTT_DEBUG_PRINTLN("Publishing to EU analyzer server");
|
||||
publishToAnalyzerClient(_analyzer_eu_client, topic, payload, retained);
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("EU analyzer server not available (enabled: %s, client: %s)",
|
||||
_analyzer_eu_enabled ? "true" : "false", _analyzer_eu_client ? "exists" : "null");
|
||||
}
|
||||
}
|
||||
|
||||
// Google Trust Services - GTS Root R4
|
||||
const char* GTS_ROOT_R4 =
|
||||
"-----BEGIN CERTIFICATE-----\n"
|
||||
"MIIDejCCAmKgAwIBAgIQf+UwvzMTQ77dghYQST2KGzANBgkqhkiG9w0BAQsFADBX\n"
|
||||
"MQswCQYDVQQGEwJCRTEZMBcGA1UEChMQR2xvYmFsU2lnbiBudi1zYTEQMA4GA1UE\n"
|
||||
"CxMHUm9vdCBDQTEbMBkGA1UEAxMSR2xvYmFsU2lnbiBSb290IENBMB4XDTIzMTEx\n"
|
||||
"NTAzNDMyMVoXDTI4MDEyODAwMDA0MlowRzELMAkGA1UEBhMCVVMxIjAgBgNVBAoT\n"
|
||||
"GUdvb2dsZSBUcnVzdCBTZXJ2aWNlcyBMTEMxFDASBgNVBAMTC0dUUyBSb290IFI0\n"
|
||||
"MHYwEAYHKoZIzj0CAQYFK4EEACIDYgAE83Rzp2iLYK5DuDXFgTB7S0md+8Fhzube\n"
|
||||
"Rr1r1WEYNa5A3XP3iZEwWus87oV8okB2O6nGuEfYKueSkWpz6bFyOZ8pn6KY019e\n"
|
||||
"WIZlD6GEZQbR3IvJx3PIjGov5cSr0R2Ko4H/MIH8MA4GA1UdDwEB/wQEAwIBhjAd\n"
|
||||
"BgNVHSUEFjAUBggrBgEFBQcDAQYIKwYBBQUHAwIwDwYDVR0TAQH/BAUwAwEB/zAd\n"
|
||||
"BgNVHQ4EFgQUgEzW63T/STaj1dj8tT7FavCUHYwwHwYDVR0jBBgwFoAUYHtmGkUN\n"
|
||||
"l8qJUC99BM00qP/8/UswNgYIKwYBBQUHAQEEKjAoMCYGCCsGAQUFBzAChhpodHRw\n"
|
||||
"Oi8vaS5wa2kuZ29vZy9nc3IxLmNydDAtBgNVHR8EJjAkMCKgIKAehhxodHRwOi8v\n"
|
||||
"Yy5wa2kuZ29vZy9yL2dzcjEuY3JsMBMGA1UdIAQMMAowCAYGZ4EMAQIBMA0GCSqG\n"
|
||||
"SIb3DQEBCwUAA4IBAQAYQrsPBtYDh5bjP2OBDwmkoWhIDDkic574y04tfzHpn+cJ\n"
|
||||
"odI2D4SseesQ6bDrarZ7C30ddLibZatoKiws3UL9xnELz4ct92vID24FfVbiI1hY\n"
|
||||
"+SW6FoVHkNeWIP0GCbaM4C6uVdF5dTUsMVs/ZbzNnIdCp5Gxmx5ejvEau8otR/Cs\n"
|
||||
"kGN+hr/W5GvT1tMBjgWKZ1i4//emhA1JG1BbPzoLJQvyEotc03lXjTaCzv8mEbep\n"
|
||||
"8RqZ7a2CPsgRbuvTPBwcOMBBmuFeU88+FSBX6+7iP0il8b4Z0QFqIwwMHfs/L6K1\n"
|
||||
"vepuoxtGzi4CZ68zJpiq1UvSqTbFJjtbD4seiMHl\n"
|
||||
"-----END CERTIFICATE-----\n";
|
||||
|
||||
void MQTTBridge::setupAnalyzerClients() {
|
||||
if (!_analyzer_us_enabled && !_analyzer_eu_enabled) {
|
||||
MQTT_DEBUG_PRINTLN("No analyzer servers enabled, skipping PsychicMqttClient setup");
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Setting up PsychicMqttClient WebSocket clients...");
|
||||
|
||||
// Setup US server client
|
||||
if (_analyzer_us_enabled) {
|
||||
_analyzer_us_client = new PsychicMqttClient();
|
||||
|
||||
// Set up event callbacks for US server
|
||||
_analyzer_us_client->onConnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("Connected to Let's Mesh US server, session present: %s", sessionPresent ? "true" : "false");
|
||||
// Publish status message when connected
|
||||
publishStatusToAnalyzerClient(_analyzer_us_client, "mqtt-us-v1.letsmesh.net");
|
||||
});
|
||||
|
||||
_analyzer_us_client->onDisconnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("Disconnected from Let's Mesh US server, session present: %s", sessionPresent ? "true" : "false");
|
||||
});
|
||||
|
||||
_analyzer_us_client->onError([this](esp_mqtt_error_codes error) {
|
||||
MQTT_DEBUG_PRINTLN("Let's Mesh US server error - error_type: %d, connect_return_code: %d",
|
||||
error.error_type, error.connect_return_code);
|
||||
});
|
||||
|
||||
// Set up WebSocket MQTT over TLS connection to US server
|
||||
_analyzer_us_client->setServer("wss://mqtt-us-v1.letsmesh.net:443/mqtt");
|
||||
MQTT_DEBUG_PRINTLN("US Server - Username: %s", _analyzer_username);
|
||||
MQTT_DEBUG_PRINTLN("US Server - Auth token length: %d", strlen(_auth_token_us));
|
||||
MQTT_DEBUG_PRINTLN("US Server - Auth token (first 50 chars): %.50s...", _auth_token_us);
|
||||
_analyzer_us_client->setCredentials(_analyzer_username, _auth_token_us);
|
||||
|
||||
// Configure TLS - use specific GTS Root R4 certificate
|
||||
MQTT_DEBUG_PRINTLN("Using GTS Root R4 certificate for US server");
|
||||
_analyzer_us_client->setCACert(GTS_ROOT_R4);
|
||||
|
||||
// Connect to US server (async connection)
|
||||
_analyzer_us_client->connect();
|
||||
MQTT_DEBUG_PRINTLN("Initiating connection to Let's Mesh US server");
|
||||
}
|
||||
|
||||
// Setup EU server client
|
||||
if (_analyzer_eu_enabled) {
|
||||
_analyzer_eu_client = new PsychicMqttClient();
|
||||
|
||||
// Set up event callbacks for EU server
|
||||
_analyzer_eu_client->onConnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("Connected to Let's Mesh EU server, session present: %s", sessionPresent ? "true" : "false");
|
||||
// Publish status message when connected
|
||||
publishStatusToAnalyzerClient(_analyzer_eu_client, "mqtt-eu-v1.letsmesh.net");
|
||||
});
|
||||
|
||||
_analyzer_eu_client->onDisconnect([this](bool sessionPresent) {
|
||||
MQTT_DEBUG_PRINTLN("Disconnected from Let's Mesh EU server, session present: %s", sessionPresent ? "true" : "false");
|
||||
});
|
||||
|
||||
_analyzer_eu_client->onError([this](esp_mqtt_error_codes error) {
|
||||
MQTT_DEBUG_PRINTLN("Let's Mesh EU server error - error_type: %d, connect_return_code: %d",
|
||||
error.error_type, error.connect_return_code);
|
||||
});
|
||||
|
||||
// Set up WebSocket MQTT over TLS connection to EU server
|
||||
_analyzer_eu_client->setServer("wss://mqtt-eu-v1.letsmesh.net:443/mqtt");
|
||||
MQTT_DEBUG_PRINTLN("EU Server - Username: %s", _analyzer_username);
|
||||
MQTT_DEBUG_PRINTLN("EU Server - Auth token length: %d", strlen(_auth_token_eu));
|
||||
MQTT_DEBUG_PRINTLN("EU Server - Auth token (first 50 chars): %.50s...", _auth_token_eu);
|
||||
_analyzer_eu_client->setCredentials(_analyzer_username, _auth_token_eu);
|
||||
|
||||
// Configure TLS - use specific GTS Root R4 certificate
|
||||
MQTT_DEBUG_PRINTLN("Using GTS Root R4 certificate for EU server");
|
||||
_analyzer_eu_client->setCACert(GTS_ROOT_R4);
|
||||
|
||||
// Connect to EU server (async connection)
|
||||
_analyzer_eu_client->connect();
|
||||
MQTT_DEBUG_PRINTLN("Initiating connection to Let's Mesh EU server");
|
||||
}
|
||||
}
|
||||
|
||||
void MQTTBridge::publishToAnalyzerClient(PsychicMqttClient* client, const char* topic, const char* payload, bool retained) {
|
||||
if (!client) {
|
||||
MQTT_DEBUG_PRINTLN("PsychicMqttClient is null");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!client->connected()) {
|
||||
MQTT_DEBUG_PRINTLN("PsychicMqttClient not connected - skipping publish to topic: %s", topic);
|
||||
return;
|
||||
}
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Publishing to analyzer client - topic: %s, payload length: %d, retained: %s",
|
||||
topic, strlen(payload), retained ? "true" : "false");
|
||||
|
||||
// Publish message using PsychicMqttClient API
|
||||
int result = client->publish(topic, 1, retained, payload, strlen(payload));
|
||||
if (result > 0) {
|
||||
MQTT_DEBUG_PRINTLN("PsychicMqttClient message published successfully, result=%d", result);
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("PsychicMqttClient publish failed, result=%d", result);
|
||||
}
|
||||
}
|
||||
|
||||
void MQTTBridge::publishStatusToAnalyzerClient(PsychicMqttClient* client, const char* server_name) {
|
||||
if (!client || !client->connected()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Create status message
|
||||
char status_topic[128];
|
||||
snprintf(status_topic, sizeof(status_topic), "meshcore/%s/%s/status", _iata, _device_id);
|
||||
|
||||
// Build status JSON
|
||||
char status_payload[512];
|
||||
snprintf(status_payload, sizeof(status_payload),
|
||||
"{"
|
||||
"\"device_id\":\"%s\","
|
||||
"\"origin_id\":\"%s\","
|
||||
"\"origin\":\"%s\","
|
||||
"\"iata\":\"%s\","
|
||||
"\"server\":\"%s\","
|
||||
"\"status\":\"connected\","
|
||||
"\"timestamp\":%lu,"
|
||||
"\"uptime\":%lu"
|
||||
"}",
|
||||
_device_id,
|
||||
_device_id, // origin_id should be the device_id (public key)
|
||||
_origin,
|
||||
_iata,
|
||||
server_name,
|
||||
time(nullptr),
|
||||
millis() / 1000
|
||||
);
|
||||
|
||||
MQTT_DEBUG_PRINTLN("Publishing status to %s server", server_name);
|
||||
MQTT_DEBUG_PRINTLN("Status topic: %s", status_topic);
|
||||
MQTT_DEBUG_PRINTLN("Status payload: %s", status_payload);
|
||||
|
||||
// Publish status message (retained)
|
||||
int result = client->publish(status_topic, 1, true, status_payload, strlen(status_payload));
|
||||
if (result > 0) {
|
||||
MQTT_DEBUG_PRINTLN("Status published to %s server successfully, result=%d", server_name, result);
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("Status publish to %s server failed, result=%d", server_name, result);
|
||||
}
|
||||
}
|
||||
|
||||
void MQTTBridge::maintainAnalyzerConnections() {
|
||||
// PsychicMqttClient handles connection maintenance and reconnection automatically
|
||||
// No manual maintenance needed - the library manages this internally
|
||||
// Connection state changes are handled via the onConnect/onDisconnect callbacks
|
||||
}
|
||||
|
||||
void MQTTBridge::setMessageTypes(bool status, bool packets, bool raw) {
|
||||
_status_enabled = status;
|
||||
_packets_enabled = packets;
|
||||
|
||||
@@ -2,11 +2,12 @@
|
||||
|
||||
#include "MeshCore.h"
|
||||
#include "helpers/bridges/BridgeBase.h"
|
||||
#include <PubSubClient.h>
|
||||
#include <PsychicMqttClient.h>
|
||||
#include <WiFi.h>
|
||||
#include <NTPClient.h>
|
||||
#include <WiFiUdp.h>
|
||||
#include <Timezone.h>
|
||||
#include "helpers/JWTHelper.h"
|
||||
|
||||
#if defined(MQTT_DEBUG) && defined(ARDUINO)
|
||||
#include <Arduino.h>
|
||||
@@ -44,8 +45,7 @@
|
||||
*/
|
||||
class MQTTBridge : public BridgeBase {
|
||||
private:
|
||||
PubSubClient* _mqtt_client;
|
||||
WiFiClient* _wifi_client;
|
||||
PsychicMqttClient* _mqtt_client;
|
||||
|
||||
// MQTT broker configuration
|
||||
struct MQTTBroker {
|
||||
@@ -107,6 +107,20 @@ private:
|
||||
float _last_rssi;
|
||||
unsigned long _last_raw_timestamp;
|
||||
|
||||
// Let's Mesh Analyzer support
|
||||
bool _analyzer_us_enabled;
|
||||
bool _analyzer_eu_enabled;
|
||||
char _auth_token_us[1024]; // JWT token for US server authentication
|
||||
char _auth_token_eu[1024]; // JWT token for EU server authentication
|
||||
char _analyzer_username[70]; // Username in format v1_{UPPERCASE_PUBLIC_KEY}
|
||||
|
||||
// Device identity for JWT token creation
|
||||
mesh::LocalIdentity *_identity;
|
||||
|
||||
// PsychicMqttClient instances for different brokers
|
||||
PsychicMqttClient* _analyzer_us_client;
|
||||
PsychicMqttClient* _analyzer_eu_client;
|
||||
|
||||
// Internal methods
|
||||
void connectToBrokers();
|
||||
void processPacketQueue();
|
||||
@@ -127,8 +141,9 @@ public:
|
||||
* @param prefs Node preferences for configuration settings
|
||||
* @param mgr PacketManager for allocating and queuing packets
|
||||
* @param rtc RTCClock for timestamping debug messages
|
||||
* @param identity Device identity for JWT token creation
|
||||
*/
|
||||
MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc);
|
||||
MQTTBridge(NodePrefs *prefs, mesh::PacketManager *mgr, mesh::RTCClock *rtc, mesh::LocalIdentity *identity);
|
||||
|
||||
/**
|
||||
* Initializes the MQTT bridge
|
||||
@@ -229,6 +244,17 @@ public:
|
||||
* @param rssi Received signal strength indicator
|
||||
*/
|
||||
void storeRawRadioData(const uint8_t* raw_data, int len, float snr, float rssi);
|
||||
|
||||
// Let's Mesh Analyzer methods
|
||||
void setupAnalyzerServers();
|
||||
bool createAuthToken();
|
||||
void publishToAnalyzerServers(const char* topic, const char* payload, bool retained = false);
|
||||
|
||||
// PsychicMqttClient WebSocket methods
|
||||
void setupAnalyzerClients();
|
||||
void maintainAnalyzerConnections();
|
||||
void publishToAnalyzerClient(PsychicMqttClient* client, const char* topic, const char* payload, bool retained = false);
|
||||
void publishStatusToAnalyzerClient(PsychicMqttClient* client, const char* server_name);
|
||||
|
||||
/**
|
||||
* Enable/disable message types
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -119,17 +119,22 @@ build_flags =
|
||||
-D MQTT_DEBUG=1
|
||||
-D MESH_PACKET_LOGGING=1
|
||||
-D MESH_DEBUG=1
|
||||
-D CONFIG_MBEDTLS_CERTIFICATE_BUNDLE=y
|
||||
# -D WIFI_SSID='"ssid"'
|
||||
# -D WIFI_PWD='"password"'
|
||||
|
||||
; Use ESP32's built-in certificate bundle
|
||||
build_src_filter = ${Heltec_lora32_v3.build_src_filter}
|
||||
+<helpers/bridges/MQTTBridge.cpp>
|
||||
+<helpers/MQTTMessageBuilder.cpp>
|
||||
+<helpers/JWTHelper.cpp>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
+<../examples/simple_repeater>
|
||||
lib_deps =
|
||||
${Heltec_lora32_v3.lib_deps}
|
||||
${esp32_ota.lib_deps}
|
||||
knolleary/PubSubClient @ ^2.8
|
||||
; PsychicMqttClient - unified MQTT client with WebSocket support
|
||||
elims/PsychicMqttClient@^0.2.4
|
||||
bblanchon/ArduinoJson @ 6.17.3
|
||||
arduino-libraries/NTPClient
|
||||
JChristensen/Timezone
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
[Station_G2]
|
||||
extends = esp32_base
|
||||
board = station-g2
|
||||
board_build.partitions = huge_app.csv
|
||||
build_flags =
|
||||
${esp32_base.build_flags}
|
||||
${sensor_base.build_flags}
|
||||
@@ -254,12 +255,14 @@ build_flags =
|
||||
build_src_filter = ${Station_G2.build_src_filter}
|
||||
+<helpers/bridges/MQTTBridge.cpp>
|
||||
+<helpers/MQTTMessageBuilder.cpp>
|
||||
+<helpers/JWTHelper.cpp>
|
||||
+<helpers/ui/SH1106Display.cpp>
|
||||
+<../examples/simple_repeater>
|
||||
lib_deps =
|
||||
${Station_G2.lib_deps}
|
||||
${esp32_ota.lib_deps}
|
||||
knolleary/PubSubClient @ ^2.8
|
||||
; PsychicMqttClient - unified MQTT client with WebSocket support
|
||||
elims/PsychicMqttClient@^0.2.4
|
||||
bblanchon/ArduinoJson @ 6.17.3
|
||||
arduino-libraries/NTPClient
|
||||
JChristensen/Timezone
|
||||
|
||||
Reference in New Issue
Block a user