archy/core/archipelago/src/mesh/meshtastic.rs

731 lines
25 KiB
Rust
Raw Normal View History

//! Async serial driver for Meshtastic devices.
//!
//! Meshtastic uses protobuf payloads over a SLIP-like serial stream. This
//! module implements only the small subset Archipelago needs: connect,
//! discover the local node, send/receive text packets, and provide synthetic
//! contacts to the existing mesh listener.
use super::protocol::{InboundFrame, ParsedContact};
use super::types::{DeviceInfo, DeviceType};
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::time::Duration;
use tracing::{debug, info, warn};
const BAUD_RATE: u32 = 115200;
const READ_TIMEOUT: Duration = Duration::from_secs(5);
const WRITE_TIMEOUT: Duration = Duration::from_secs(2);
const READ_BUF_SIZE: usize = 512;
const START1: u8 = 0x94;
const START2: u8 = 0xc3;
const TO_RADIO_MAX: usize = 512;
const BROADCAST_NUM: u32 = 0xffff_ffff;
const TEXT_MESSAGE_APP: u32 = 1;
const TO_RADIO_PACKET: u64 = 1;
const TO_RADIO_WANT_CONFIG_ID: u64 = 3;
const TO_RADIO_HEARTBEAT: u64 = 7;
const FROM_RADIO_PACKET: u64 = 2;
const FROM_RADIO_MY_INFO: u64 = 3;
const FROM_RADIO_NODE_INFO: u64 = 4;
const FROM_RADIO_CONFIG_COMPLETE_ID: u64 = 7;
const FROM_RADIO_REBOOTED: u64 = 8;
/// Async Meshtastic device handle.
pub struct MeshtasticDevice {
port: serial2_tokio::SerialPort,
read_buf: Vec<u8>,
node_num: Option<u32>,
user_id: Option<String>,
long_name: Option<String>,
short_name: Option<String>,
contacts: HashMap<u32, ParsedContact>,
/// Real Curve25519 public keys, keyed by node-num, as learned from NodeInfo
/// (`User.public_key`) or PKC-encrypted inbound packets (`MeshPacket
/// .public_key`). Kept SEPARATE from `contacts[*].public_key_hex`, which is
/// the synthetic node-num-derived routing key that `send_text_msg` relies
/// on — we must not overwrite that or unicast routing breaks. This map only
/// records which peers are PKC-capable, so we can tell a true end-to-end
/// (PKI) DM from a channel-PSK fallback.
peer_pubkeys: HashMap<u32, Vec<u8>>,
device_path: String,
}
impl MeshtasticDevice {
pub async fn open(path: &str) -> Result<Self> {
match tokio::fs::metadata(path).await {
Ok(meta) => {
debug!(path = %path, permissions = ?meta.permissions(), "Device node exists")
}
Err(e) => anyhow::bail!("Serial device {} not accessible: {}", path, e),
}
let port = serial2_tokio::SerialPort::open(path, BAUD_RATE).context(format!(
"Failed to open serial port {} (permission denied? device busy?)",
path
))?;
info!(path = %path, baud = BAUD_RATE, "Opened Meshtastic serial port");
Ok(Self {
port,
read_buf: Vec::with_capacity(READ_BUF_SIZE),
node_num: None,
user_id: None,
long_name: None,
short_name: None,
contacts: HashMap::new(),
peer_pubkeys: HashMap::new(),
device_path: path.to_string(),
})
}
pub async fn initialize(&mut self) -> Result<DeviceInfo> {
info!(path = %self.device_path, "Starting Meshtastic handshake");
self.send_to_radio(&encode_want_config()).await?;
let deadline = tokio::time::Instant::now() + READ_TIMEOUT;
let mut saw_meshtastic_frame = false;
2026-05-17 19:44:54 -04:00
let mut saw_config_complete = false;
loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
break;
}
match tokio::time::timeout(
remaining.min(Duration::from_millis(250)),
self.read_from_radio(),
)
.await
{
Ok(Ok(Some(frame))) => {
saw_meshtastic_frame = true;
2026-05-17 19:44:54 -04:00
if matches!(
decode_top_level_variant(&frame),
Some((FROM_RADIO_CONFIG_COMPLETE_ID, _))
) {
saw_config_complete = true;
}
self.handle_from_radio(&frame);
2026-05-17 19:44:54 -04:00
if saw_config_complete && self.node_num.is_some() {
break;
}
}
Ok(Ok(None)) | Err(_) => {}
Ok(Err(e)) => return Err(e),
}
}
if !saw_meshtastic_frame {
anyhow::bail!("No Meshtastic serial API response");
}
2026-05-17 20:45:56 -04:00
let node_id = self
.node_num
.ok_or_else(|| anyhow::anyhow!("Meshtastic serial API did not provide MyInfo"))?;
if self.user_id.is_none() && self.long_name.is_none() && self.short_name.is_none() {
anyhow::bail!("Meshtastic serial API did not provide node identity");
}
let firmware_version = self
.long_name
.clone()
.or_else(|| self.user_id.clone())
.unwrap_or_else(|| "Meshtastic".to_string());
info!(node_id, name = %firmware_version, "Meshtastic identity");
Ok(DeviceInfo {
firmware_version,
node_id,
max_contacts: 200,
device_type: DeviceType::Meshtastic,
})
}
pub async fn set_advert_name(&mut self, name: &str) -> Result<()> {
self.long_name = Some(name.to_string());
Ok(())
}
pub async fn send_self_advert(&mut self) -> Result<()> {
self.send_to_radio(&encode_heartbeat()).await
}
pub async fn send_channel_text(&mut self, _channel: u8, msg: &[u8]) -> Result<()> {
let text = String::from_utf8_lossy(msg);
let packet = encode_mesh_packet(BROADCAST_NUM, TEXT_MESSAGE_APP, text.as_bytes());
self.send_to_radio(&encode_to_radio_variant(TO_RADIO_PACKET, &packet))
.await
}
/// Native Meshtastic unicast DM. Our synthetic Meshtastic pubkeys carry the
/// numeric node-id in their first 4 bytes (little-endian, see
/// `synthetic_pubkey`), so `dest_pubkey_prefix` directly yields the
/// destination node number. We send a directed MeshPacket (`to` = node num)
/// rather than a `BROADCAST_NUM` channel blast — this is the Meshtastic
/// analog of the meshcore `CMD_SEND_TXT_MSG` fix: the message is delivered
/// as a real DM (only the recipient's client surfaces it) instead of
/// polluting the shared primary channel where every node would see it.
///
/// If the prefix decodes to node 0 / broadcast (e.g. a non-Meshtastic
/// synthetic key routed here by mistake), fall back to a channel send so the
/// device interface stays uniform and the message still goes out.
pub async fn send_text_msg(&mut self, dest_pubkey_prefix: &[u8; 6], msg: &[u8]) -> Result<()> {
let node_num = u32::from_le_bytes([
dest_pubkey_prefix[0],
dest_pubkey_prefix[1],
dest_pubkey_prefix[2],
dest_pubkey_prefix[3],
]);
if node_num == 0 || node_num == BROADCAST_NUM {
return self.send_channel_text(0, msg).await;
}
let text = String::from_utf8_lossy(msg);
let packet = encode_mesh_packet(node_num, TEXT_MESSAGE_APP, text.as_bytes());
self.send_to_radio(&encode_to_radio_variant(TO_RADIO_PACKET, &packet))
.await
}
/// Meshtastic has no meshcore-style contact table; these are no-ops so the
/// device interface stays uniform.
pub async fn remove_contact(&mut self, _pubkey: &[u8; 32]) -> Result<()> {
Ok(())
}
pub async fn add_contact(
&mut self,
_pubkey: &[u8; 32],
_contact_type: u8,
_flags: u8,
_out_path_len: u8,
_name: &str,
_last_advert: u32,
) -> Result<()> {
Ok(())
}
pub async fn get_contacts(&mut self) -> Result<Vec<ParsedContact>> {
2026-05-17 19:44:54 -04:00
if self.contacts.is_empty() {
self.send_to_radio(&encode_want_config()).await?;
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
while tokio::time::Instant::now() < deadline {
match self.read_from_radio().await? {
Some(frame) => {
let config_complete = matches!(
decode_top_level_variant(&frame),
Some((FROM_RADIO_CONFIG_COMPLETE_ID, _))
);
self.handle_from_radio(&frame);
if config_complete || !self.contacts.is_empty() {
break;
}
}
None => tokio::time::sleep(Duration::from_millis(50)).await,
}
}
}
Ok(self.contacts.values().cloned().collect())
}
pub async fn reset_contact_path(&mut self, _pubkey: &[u8; 32]) -> Result<()> {
Ok(())
}
pub async fn sync_messages(&mut self) -> Result<Vec<InboundFrame>> {
Ok(Vec::new())
}
pub async fn try_recv_frame(&mut self) -> Result<Option<InboundFrame>> {
let Some(frame) = self.read_from_radio().await? else {
return Ok(None);
};
Ok(self.handle_from_radio(&frame))
}
/// Whether we've learned `node_num`'s real PKI (Curve25519) key — from a
/// NodeInfo `public_key` or an inbound PKC DM — meaning the firmware can
/// deliver DMs to/from it end-to-end encrypted instead of falling back to
/// the channel PSK. Driver-internal for now; lets a future mesh-tab badge
/// distinguish a true E2E DM from a channel-encrypted one without changing
/// the shared device interface (which would break meshcore hot-swap).
#[allow(dead_code)] // seam: consumed when the mesh-tab E2E badge lands
pub fn peer_is_pkc_capable(&self, node_num: u32) -> bool {
self.peer_pubkeys
.get(&node_num)
.is_some_and(|k| !k.is_empty())
}
pub fn advert_name(&self) -> Option<String> {
self.long_name
.clone()
.or_else(|| self.short_name.clone())
.or_else(|| self.user_id.clone())
}
async fn send_to_radio(&mut self, payload: &[u8]) -> Result<()> {
if payload.len() > TO_RADIO_MAX {
anyhow::bail!("Meshtastic payload too large: {} bytes", payload.len());
}
let mut frame = Vec::with_capacity(4 + payload.len());
frame.push(START1);
frame.push(START2);
frame.extend_from_slice(&(payload.len() as u16).to_be_bytes());
frame.extend_from_slice(payload);
tokio::time::timeout(WRITE_TIMEOUT, self.port.write_all(&frame))
.await
.context("Meshtastic serial write timed out")?
.context("Meshtastic serial write failed")?;
Ok(())
}
async fn read_from_radio(&mut self) -> Result<Option<Vec<u8>>> {
if let Some(frame) = decode_serial_frame(&mut self.read_buf) {
return Ok(Some(frame));
}
let mut tmp = [0u8; READ_BUF_SIZE];
match tokio::time::timeout(Duration::from_millis(50), self.port.read(&mut tmp)).await {
Ok(Ok(0)) => anyhow::bail!("Meshtastic serial port closed"),
Ok(Ok(n)) => self.read_buf.extend_from_slice(&tmp[..n]),
Ok(Err(e)) => return Err(e).context("Meshtastic serial read error"),
Err(_) => return Ok(None),
}
Ok(decode_serial_frame(&mut self.read_buf))
}
fn handle_from_radio(&mut self, frame: &[u8]) -> Option<InboundFrame> {
let Some((field, value)) = decode_top_level_variant(frame) else {
return None;
};
match field {
FROM_RADIO_MY_INFO => {
if let Some((node_num, user_id)) = parse_my_info(value) {
self.node_num = Some(node_num);
if let Some(user_id) = user_id {
self.user_id = Some(user_id);
}
}
None
}
FROM_RADIO_NODE_INFO => {
self.update_node_info(value);
None
}
FROM_RADIO_PACKET => self.packet_to_inbound_frame(value),
FROM_RADIO_CONFIG_COMPLETE_ID | FROM_RADIO_REBOOTED => None,
other => {
debug!(
field = other,
len = value.len(),
"Unhandled Meshtastic FromRadio field"
);
None
}
}
}
fn update_node_info(&mut self, data: &[u8]) {
if let Some(node) = parse_node_info(data) {
if let Some(pk) = node.public_key.as_ref() {
if self.peer_pubkeys.insert(node.num, pk.clone()).is_none() {
debug!(
node = node.num,
key_len = pk.len(),
"Meshtastic peer is PKC-capable (NodeInfo public_key)"
);
}
}
let key = synthetic_pubkey(node.num);
let name = node
.long_name
.or(node.short_name)
.or(node.id)
.unwrap_or_else(|| format!("Meshtastic !{:08x}", node.num));
if Some(node.num) == self.node_num {
self.long_name = Some(name.clone());
}
self.contacts.insert(
node.num,
ParsedContact {
public_key_hex: hex::encode(key),
advert_name: name,
last_advert: node.last_heard.unwrap_or_default(),
contact_type: 1,
path_len: 0xff,
flags: 0,
},
);
}
}
fn packet_to_inbound_frame(&mut self, data: &[u8]) -> Option<InboundFrame> {
let packet = parse_mesh_packet(data)?;
if packet.portnum != TEXT_MESSAGE_APP || packet.payload.is_empty() {
return None;
}
let from = packet.from.unwrap_or(0);
if Some(from) == self.node_num {
return None;
}
// Record E2E status: a `pki_encrypted` packet (or one carrying the
// sender's `public_key`) proves this DM arrived end-to-end encrypted via
// the PKI, not the shared channel PSK. We learn the sender's key here too
// — but keep it OUT of the routing `public_key_hex` (synthetic) so the
// device interface stays identical to meshcore's and the two remain
// hot-swappable behind the mesh listener.
if let Some(pk) = packet.public_key.as_ref() {
self.peer_pubkeys.entry(from).or_insert_with(|| pk.clone());
}
if packet.pki_encrypted {
debug!(node = from, "Meshtastic DM received end-to-end encrypted (PKI)");
}
let from_key = synthetic_pubkey(from);
self.contacts.entry(from).or_insert_with(|| ParsedContact {
public_key_hex: hex::encode(synthetic_pubkey(from)),
advert_name: format!("Meshtastic !{:08x}", from),
last_advert: 0,
contact_type: 1,
path_len: 0xff,
flags: 0,
});
let mut payload = Vec::with_capacity(15 + packet.payload.len());
payload.push(0); // SNR unknown
payload.extend_from_slice(&[0, 0]); // reserved
payload.extend_from_slice(&from_key[..6]);
payload.push(0xff); // unknown/flood path
payload.push(0); // text type
payload.extend_from_slice(&0u32.to_le_bytes());
payload.extend_from_slice(&packet.payload);
Some(InboundFrame {
code: super::protocol::RESP_CONTACT_MSG_V3,
data: payload,
bytes_consumed: 0,
})
}
}
fn decode_serial_frame(buf: &mut Vec<u8>) -> Option<Vec<u8>> {
let start = buf.windows(2).position(|w| w == [START1, START2])?;
if start > 0 {
buf.drain(..start);
}
if buf.len() < 4 {
return None;
}
let len = u16::from_be_bytes([buf[2], buf[3]]) as usize;
if buf.len() < 4 + len {
return None;
}
let payload = buf[4..4 + len].to_vec();
buf.drain(..4 + len);
Some(payload)
}
fn encode_want_config() -> Vec<u8> {
2026-05-17 20:45:56 -04:00
encode_varint_field(TO_RADIO_WANT_CONFIG_ID, 1)
}
fn encode_heartbeat() -> Vec<u8> {
encode_to_radio_variant(TO_RADIO_HEARTBEAT, &[])
}
fn encode_to_radio_variant(field: u64, bytes: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
encode_len_field(field, bytes, &mut out);
out
}
fn encode_mesh_packet(to: u32, portnum: u32, payload: &[u8]) -> Vec<u8> {
let mut decoded = Vec::new();
encode_varint_field_into(1, portnum as u64, &mut decoded);
encode_len_field(2, payload, &mut decoded);
let mut packet = Vec::new();
encode_fixed32_field(2, to, &mut packet);
encode_len_field(4, &decoded, &mut packet);
packet
}
fn decode_top_level_variant(buf: &[u8]) -> Option<(u64, &[u8])> {
let mut idx = 0;
while idx < buf.len() {
let (key, n) = read_varint(&buf[idx..])?;
idx += n;
let field = key >> 3;
match key & 0x07 {
0 => {
let (_, n) = read_varint(&buf[idx..])?;
idx += n;
2026-05-17 19:44:54 -04:00
if matches!(field, FROM_RADIO_CONFIG_COMPLETE_ID | FROM_RADIO_REBOOTED) {
return Some((field, &[]));
}
}
2 => {
let (len, n) = read_varint(&buf[idx..])?;
idx += n;
let end = idx.checked_add(len as usize)?;
if end > buf.len() {
return None;
}
if matches!(
field,
FROM_RADIO_PACKET | FROM_RADIO_MY_INFO | FROM_RADIO_NODE_INFO
) {
return Some((field, &buf[idx..end]));
}
idx = end;
}
_ => return None,
}
}
None
}
fn parse_my_info(data: &[u8]) -> Option<(u32, Option<String>)> {
let mut idx = 0;
let mut node_num = None;
let mut user_id = None;
while idx < data.len() {
let (field, value, next) = next_field(data, idx)?;
idx = next;
match (field, value) {
(1, FieldValue::Varint(v)) => node_num = Some(v as u32),
(1, FieldValue::Fixed32(v)) => node_num = Some(v),
(3, FieldValue::Bytes(b)) => user_id = parse_user(b).and_then(|u| u.id),
_ => {}
}
}
node_num.map(|n| (n, user_id))
}
struct ParsedNode {
num: u32,
id: Option<String>,
long_name: Option<String>,
short_name: Option<String>,
last_heard: Option<u32>,
public_key: Option<Vec<u8>>,
}
fn parse_node_info(data: &[u8]) -> Option<ParsedNode> {
let mut idx = 0;
let mut node = ParsedNode {
num: 0,
id: None,
long_name: None,
short_name: None,
last_heard: None,
public_key: None,
};
while idx < data.len() {
let (field, value, next) = next_field(data, idx)?;
idx = next;
match (field, value) {
(1, FieldValue::Varint(v)) => node.num = v as u32,
(1, FieldValue::Fixed32(v)) => node.num = v,
(2, FieldValue::Bytes(b)) => {
if let Some(user) = parse_user(b) {
node.id = user.id;
node.long_name = user.long_name;
node.short_name = user.short_name;
node.public_key = user.public_key;
}
}
(5, FieldValue::Fixed32(v)) => node.last_heard = Some(v),
_ => {}
}
}
if node.num == 0 {
None
} else {
Some(node)
}
}
struct ParsedUser {
id: Option<String>,
long_name: Option<String>,
short_name: Option<String>,
public_key: Option<Vec<u8>>,
}
fn parse_user(data: &[u8]) -> Option<ParsedUser> {
let mut idx = 0;
let mut user = ParsedUser {
id: None,
long_name: None,
short_name: None,
public_key: None,
};
while idx < data.len() {
let (field, value, next) = next_field(data, idx)?;
idx = next;
match (field, value) {
(1, FieldValue::Bytes(b)) => user.id = string_field(b),
(2, FieldValue::Bytes(b)) => user.long_name = string_field(b),
(3, FieldValue::Bytes(b)) => user.short_name = string_field(b),
// User.public_key (field 8): the peer's Curve25519 key. Its presence
// means the radio can PKC-encrypt DMs to this node end-to-end.
(8, FieldValue::Bytes(b)) if !b.is_empty() => user.public_key = Some(b.to_vec()),
_ => {}
}
}
Some(user)
}
struct ParsedPacket {
from: Option<u32>,
portnum: u32,
payload: Vec<u8>,
/// MeshPacket.pki_encrypted (field 17): the firmware decrypted this packet
/// with the PKI (Curve25519) key, i.e. it arrived end-to-end encrypted
/// rather than via the shared channel PSK.
pki_encrypted: bool,
/// MeshPacket.public_key (field 16): the sender's key, carried on PKC DMs.
public_key: Option<Vec<u8>>,
}
fn parse_mesh_packet(data: &[u8]) -> Option<ParsedPacket> {
let mut idx = 0;
let mut from = None;
let mut decoded = None;
let mut pki_encrypted = false;
let mut public_key = None;
while idx < data.len() {
let (field, value, next) = next_field(data, idx)?;
idx = next;
match (field, value) {
(1, FieldValue::Fixed32(v)) => from = Some(v),
(4, FieldValue::Bytes(b)) => decoded = Some(b),
(16, FieldValue::Bytes(b)) if !b.is_empty() => public_key = Some(b.to_vec()),
(17, FieldValue::Varint(v)) => pki_encrypted = v != 0,
_ => {}
}
}
let decoded = decoded?;
let mut didx = 0;
let mut portnum = 0;
let mut payload = Vec::new();
while didx < decoded.len() {
let (field, value, next) = next_field(decoded, didx)?;
didx = next;
match (field, value) {
(1, FieldValue::Varint(v)) => portnum = v as u32,
(2, FieldValue::Bytes(b)) => payload = b.to_vec(),
_ => {}
}
}
Some(ParsedPacket {
from,
portnum,
payload,
pki_encrypted,
public_key,
})
}
enum FieldValue<'a> {
Varint(u64),
Fixed32(u32),
Bytes(&'a [u8]),
}
fn next_field(buf: &[u8], idx: usize) -> Option<(u64, FieldValue<'_>, usize)> {
let (key, n) = read_varint(&buf[idx..])?;
let field = key >> 3;
let mut pos = idx + n;
match key & 0x07 {
0 => {
let (v, n) = read_varint(&buf[pos..])?;
pos += n;
Some((field, FieldValue::Varint(v), pos))
}
2 => {
let (len, n) = read_varint(&buf[pos..])?;
pos += n;
let end = pos.checked_add(len as usize)?;
if end > buf.len() {
return None;
}
Some((field, FieldValue::Bytes(&buf[pos..end]), end))
}
5 => {
let end = pos.checked_add(4)?;
if end > buf.len() {
None
} else {
let value =
u32::from_le_bytes([buf[pos], buf[pos + 1], buf[pos + 2], buf[pos + 3]]);
Some((field, FieldValue::Fixed32(value), end))
}
}
1 => {
let end = pos.checked_add(8)?;
if end > buf.len() {
None
} else {
Some((field, FieldValue::Bytes(&buf[pos..end]), end))
}
}
wire => {
warn!(wire, "Unsupported Meshtastic protobuf wire type");
None
}
}
}
fn read_varint(buf: &[u8]) -> Option<(u64, usize)> {
let mut out = 0u64;
for (i, b) in buf.iter().copied().enumerate().take(10) {
out |= ((b & 0x7f) as u64) << (7 * i);
if b & 0x80 == 0 {
return Some((out, i + 1));
}
}
None
}
fn encode_varint_field(field: u64, value: u64) -> Vec<u8> {
let mut out = Vec::new();
encode_varint_field_into(field, value, &mut out);
out
}
fn encode_varint_field_into(field: u64, value: u64, out: &mut Vec<u8>) {
write_varint((field << 3) | 0, out);
write_varint(value, out);
}
fn encode_len_field(field: u64, bytes: &[u8], out: &mut Vec<u8>) {
write_varint((field << 3) | 2, out);
write_varint(bytes.len() as u64, out);
out.extend_from_slice(bytes);
}
fn encode_fixed32_field(field: u64, value: u32, out: &mut Vec<u8>) {
write_varint((field << 3) | 5, out);
out.extend_from_slice(&value.to_le_bytes());
}
fn write_varint(mut value: u64, out: &mut Vec<u8>) {
while value >= 0x80 {
out.push((value as u8 & 0x7f) | 0x80);
value >>= 7;
}
out.push(value as u8);
}
fn string_field(bytes: &[u8]) -> Option<String> {
std::str::from_utf8(bytes).ok().map(|s| s.to_string())
}
fn synthetic_pubkey(node_num: u32) -> [u8; 32] {
let mut out = [0u8; 32];
out[..4].copy_from_slice(&node_num.to_le_bytes());
out[4..15].copy_from_slice(b"meshtastic:");
out
}