#!/usr/bin/env python3 """DongLoRa MeshCore receiver: decode and display MeshCore packets.""" # /// script # requires-python = ">=3.10" # dependencies = ["cobs", "pyserial", "pycryptodome"] # /// import base64 import csv import glob import hashlib import hmac import json import struct import serial import sys import tempfile import time from cobs import cobs from Crypto.Cipher import AES from datetime import datetime, timezone from pathlib import Path # ── ANSI colors ─────────────────────────────────────────────────── RST = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" RED = "\033[31m" GRN = "\033[32m" YEL = "\033[33m" BLU = "\033[34m" MAG = "\033[35m" CYN = "\033[36m" WHT = "\033[37m" BRED = "\033[1;31m" BGRN = "\033[1;32m" BYEL = "\033[1;33m" BBLU = "\033[1;34m" BMAG = "\033[1;35m" BCYN = "\033[1;36m" BWHT = "\033[1;37m" TYPE_COLORS = { "ADVERT": BGRN, "ACK": DIM, "TXT_MSG": BCYN, "REQ": BYEL, "RESPONSE": BYEL, "GRP_TXT": BMAG, "GRP_DATA": BMAG, "ANON_REQ": YEL, "PATH": BBLU, "TRACE": BLU, "MULTIPART": YEL, "CONTROL": BLU, "RAW_CUSTOM": DIM, } def _rssi_color(rssi: int) -> str: if rssi >= -70: return GRN if rssi >= -100: return YEL return RED def _snr_color(snr: int) -> str: if snr >= 0: return GRN if snr >= -10: return YEL return RED # ── MeshCore channel crypto ──────────────────────────────────────── def _channel_secret_from_hashtag(name: str) -> bytes: """Derive 32-byte secret from a hashtag channel name (e.g., '#test').""" h = hashlib.sha256(name.encode()).digest() return h[:16] + b"\x00" * 16 def _channel_secret_from_psk(psk_b64: str) -> bytes: """Decode a base64 PSK into a 32-byte secret (16-byte key + 16 zero bytes).""" key = base64.b64decode(psk_b64) return key[:16] + b"\x00" * 16 def _channel_hash(secret: bytes) -> int: """Compute the 1-byte channel hash from a 32-byte secret.""" # Hash over the key portion (16 bytes for hashtag/PSK channels) key_len = 16 if secret[16:] == b"\x00" * 16 else 32 return hashlib.sha256(secret[:key_len]).digest()[0] def _grp_verify_and_decrypt(secret: bytes, mac_bytes: bytes, ciphertext: bytes) -> bytes | None: """Verify MAC then decrypt a GRP_TXT/GRP_DATA payload. Returns plaintext or None.""" if not ciphertext or len(ciphertext) % 16 != 0: return None # HMAC-SHA256 over ciphertext, truncated to 2 bytes key_len = 16 if secret[16:] == b"\x00" * 16 else 32 computed_mac = hmac.new(secret[:key_len], ciphertext, hashlib.sha256).digest()[:2] if computed_mac != mac_bytes: return None # AES-128-ECB decrypt cipher = AES.new(secret[:16], AES.MODE_ECB) plaintext = b"" for i in range(0, len(ciphertext), 16): plaintext += cipher.decrypt(ciphertext[i:i + 16]) return plaintext def _parse_grp_plaintext(plaintext: bytes) -> tuple[int, str] | None: """Parse decrypted GRP_TXT: timestamp(4) + flags(1) + text. Returns (timestamp, text).""" if len(plaintext) < 5: return None timestamp = struct.unpack_from(" bytes: """Build an encrypted GRP_TXT payload (channel_hash + mac + ciphertext).""" ch = _channel_hash(secret) # Plaintext: timestamp(4) + flags(1) + "sender: text\0" + zero-pad plaintext = struct.pack(" bytes: """Wrap encrypted GRP_TXT payload into a full MeshCore packet.""" header = bytes([0x15]) # GRP_TXT flood: (5 << 2) | 1 path_len = bytes([0x40]) # 0 hops, 2-byte hash mode (hash_size_code=1 << 6) return header + path_len + channel_payload def _grp_transmit(ser, channel_name: str, sender: str, text: str): """Encrypt and transmit a GRP_TXT message, then resume RX.""" secret = _KNOWN_CHANNELS.get(channel_name) if secret is None: return payload = _grp_encrypt(secret, sender, text) packet = _grp_build_packet(payload) try: send_cmd(ser, {"type": "Transmit", "payload": packet}, f"TX GRP_TXT → {channel_name}") except Exception as e: print(f" {RED}[TX failed: {e}]{RST}") # Always try to resume RX, even if TX failed try: send_cmd(ser, {"type": "StartRx"}, "StartRx (resume)") except Exception as e: print(f" {RED}[StartRx failed: {e}]{RST}") # ── Loop/collision aggregator ───────────────────────────────────── REPORT_CHANNEL = "#watchman" REPORT_SENDER = "Watchman" REPORT_WINDOW = 60 # seconds — collect events before reporting REPORT_COOLDOWN = 3600 # seconds — suppress repeat reports for same issue def _nodes_for_hash(hash_hex: str, hash_size: int) -> int: """Count how many known nodes share this path hash.""" return sum(1 for pk in _known_nodes if pk[:hash_size * 2] == hash_hex) class LoopAggregator: def __init__(self): self._pending: list[dict] = [] self._cooldown: dict[str, float] = {} self._window_start = time.monotonic() def record(self, dupes: dict[str, int], hash_size: int, all_hops: list[str]): """Record duplicate hops from one packet.""" first_hop = all_hops[0] if all_hops else None self._pending.append({ "dupes": dupes, "hash_size": hash_size, "hops": all_hops, "first_hop": first_hop, }) def maybe_report(self, ser) -> bool: if not self._pending: self._window_start = time.monotonic() return False if time.monotonic() - self._window_start < REPORT_WINDOW: return False self._send_report(ser) return True def _send_report(self, ser): now = time.monotonic() messages: list[str] = [] # Aggregate: (hash, hash_size) → {packets, relayers, senders} agg: dict[tuple[str, int], dict] = {} for ev in self._pending: for h, count in ev["dupes"].items(): key = (h, ev["hash_size"]) if key not in agg: agg[key] = {"packets": 0, "relayers": set(), "senders": set()} agg[key]["packets"] += 1 if ev["first_hop"]: agg[key]["senders"].add(ev["first_hop"]) for hop in ev["hops"]: if hop != h: agg[key]["relayers"].add(hop) for (h, hs), info in agg.items(): issue_key = f"{h}:{hs}" if issue_key in self._cooldown and now - self._cooldown[issue_key] < REPORT_COOLDOWN: continue n_known = _nodes_for_hash(h, hs) pkts = info["packets"] senders = sorted(info["senders"]) sender_str = ",".join(senders[:3]) if senders else "?" if len(senders) > 3: sender_str += f"+{len(senders) - 3}" relayer_str = ",".join(sorted(info["relayers"])[:3]) if len(info["relayers"]) > 3: relayer_str += f"+{len(info['relayers']) - 3}" if n_known == 1: # Exactly one node matches → confirmed loop msg = f"Loop: {h} re-sent its own traffic ({pkts} pkt)." if relayer_str: msg += f" Relayed by {relayer_str}." messages.append(msg) elif hs >= 2: if n_known > 1: # 2B+ hash collision — rare, definitely a routing issue msg = (f"Routing conflict: {n_known} nodes share {hs}B " f"hash {h} ({pkts} pkt).") messages.append(msg) else: # 2B+ duplicate, unknown node → almost certainly a loop msg = f"Probable loop: {h} appeared {pkts}x." if relayer_str: msg += f" Relayed by {relayer_str}." messages.append(msg) else: # 1B hash — can't distinguish loop from collision if n_known > 1: msg = (f"1B hash {h} duplicated in {pkts} pkt " f"({n_known} known nodes share it). " f"Sender {sender_str}: use 2B routing!") messages.append(msg) else: msg = (f"1B hash {h} duplicated in {pkts} pkt. " f"Sender {sender_str}: use 2B routing to clarify!") messages.append(msg) self._cooldown[issue_key] = now if not messages: self._reset() return full = " | ".join(messages) max_len = MAX_GRP_TEXT - len(REPORT_SENDER) - 2 if len(full) > max_len: full = full[:max_len - 5] + "+more" print(f" {BYEL}>>> Reporting to {REPORT_CHANNEL}: {full}{RST}") _grp_transmit(ser, REPORT_CHANNEL, REPORT_SENDER, full) self._reset() def _reset(self): self._pending.clear() self._window_start = time.monotonic() _aggregator = LoopAggregator() # Open-source VID 1209, PID "WA" (0x5741) USB_VID_PID = "1209:5741" def find_serial_port() -> str | None: """Find the serial port for our USB device.""" import subprocess # Try to find by USB VID:PID for path in sorted(glob.glob("/dev/ttyACM*")) + sorted(glob.glob("/dev/ttyUSB*")): try: result = subprocess.run( ["udevadm", "info", "--query=property", f"--name={path}"], capture_output=True, text=True, timeout=2, ) vid = "" pid = "" for line in result.stdout.splitlines(): if line.startswith("ID_VENDOR_ID="): vid = line.split("=", 1)[1].lower() elif line.startswith("ID_MODEL_ID="): pid = line.split("=", 1)[1].lower() if f"{vid}:{pid}" == USB_VID_PID: return path except Exception: continue # Fallback: first ttyACM device ports = sorted(glob.glob("/dev/ttyACM*")) return ports[0] if ports else None def wait_for_device() -> str: """Poll until the USB device appears.""" print("Waiting for DongLoRa...", end="", flush=True) while True: port = find_serial_port() if port: print(f" found {port}") time.sleep(0.3) # let the device settle return port print(".", end="", flush=True) time.sleep(0.5) def open_serial(port: str) -> serial.Serial: return serial.Serial(port, timeout=2) # ── COBS framing ─────────────────────────────────────────────────── def cobs_frame(payload: bytes) -> bytes: return cobs.encode(payload) + b"\x00" def read_frame(ser: serial.Serial) -> bytes | None: buf = b"" while True: b = ser.read(1) if not b: return None if b == b"\x00": break buf += b if not buf: return None try: return cobs.decode(buf) except cobs.DecodeError: return None # ── Fixed-size LE serialization ──────────────────────────────────── def encode_radio_config(cfg: dict) -> bytes: """Encode RadioConfig to 10 fixed-size LE bytes.""" return struct.pack(" bytes: """Encode a command dict to fixed-size LE bytes.""" kind = cmd["type"] tags = { "Ping": 0, "GetConfig": 1, "SetConfig": 2, "StartRx": 3, "StopRx": 4, "Transmit": 5, "DisplayOn": 6, "DisplayOff": 7, } out = bytes([tags[kind]]) if kind == "SetConfig": out += encode_radio_config(cmd["config"]) elif kind == "Transmit": config = cmd.get("config") if config is None: out += b"\x00" else: out += b"\x01" + encode_radio_config(config) payload = cmd["payload"] out += struct.pack(" dict: """Decode a fixed-size LE response.""" if not data: return {"type": "Empty"} tag = data[0] rest = data[1:] if tag == 0: return {"type": "Pong"} elif tag == 1: return {"type": "Config", "raw": rest.hex()} elif tag == 2: rssi = struct.unpack_from(" cutoff} except (FileNotFoundError, json.JSONDecodeError, ValueError): _known_nodes = {} def _save_adverts(): """Persist current adverts to disk (atomic write).""" try: tmp = _ADVERT_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(_known_nodes)) tmp.rename(_ADVERT_FILE) except OSError: pass def _lookup_hash(hash_hex: str, hash_size: int) -> list[str]: """Find node names whose pubkey prefix matches a path hash.""" return [ v["name"] for pk_hex, v in _known_nodes.items() if pk_hex[:hash_size * 2] == hash_hex ] def _register_node(pubkey: bytes, name: str): """Register or update a node from an ADVERT.""" pk_hex = pubkey.hex() _known_nodes[pk_hex] = {"name": name, "seen": time.time()} _save_adverts() def _utf8_score(data: bytes) -> int: """Score how cleanly data decodes as UTF-8. Higher = more replacement chars.""" return data.decode("utf-8", errors="replace").count("\ufffd") def _best_name(candidates: list[bytes]) -> bytes: """Pick the candidate with the fewest UTF-8 replacement chars; longest wins ties.""" return min(candidates, key=lambda d: (_utf8_score(d), -len(d))) def _decode_advert_appdata(app_data: bytes) -> tuple[str, str]: """Decode advert app_data. Returns (display_string, node_name).""" if not app_data: return "", "" flags = app_data[0] node_type = flags & 0x0F node_label = NODE_TYPES.get(node_type, f"type={node_type}") parts = [f"{DIM}flags={RST}{flags:02x} {DIM}node={RST}{node_label}"] # Parse location if flagged and coords are valid off = 1 loc_str = "" if flags & 0x10 and off + 8 <= len(app_data): lat_raw = struct.unpack_from(" str | None: """Check for duplicate hops in a path. Returns a warning line or None.""" if len(hops) < 2: return None counts: dict[str, int] = {} for h in hops: counts[h] = counts.get(h, 0) + 1 dupes = {h: c for h, c in counts.items() if c > 1} if not dupes: return None # Feed the aggregator for deferred reporting _aggregator.record(dupes, hash_size, hops) parts = [] for h, c in dupes.items(): n_known = _nodes_for_hash(h, hash_size) if n_known == 1: # Exactly one node matches → confirmed loop parts.append(f"{BRED}{h} x{c} LOOP{RST}") elif hash_size >= 2: if n_known > 1: # 2B+ collision is rare and real parts.append(f"{BCYN}{h} x{c} COLLISION {DIM}({n_known} nodes){RST}") else: # 2B+ duplicate, unknown node → almost certainly a loop parts.append(f"{BRED}{h} x{c} LOOP{RST}") else: # 1B hash — ambiguous: could be loop or collision if n_known > 1: parts.append(f"{YEL}{h} x{c} {DIM}(loop or collision? {n_known} nodes share this 1B hash){RST}") else: parts.append(f"{YEL}{h} x{c} {DIM}(loop or collision? 1B hash){RST}") return f"{LOOP_INDENT}\u26a0 {', '.join(parts)}" def _try_decode_body(data: bytes, pos: int, type_name: str, type_color: str, payload_type: int, payload_ver: int, route_name: str, tc_str: str) -> str | None: """Try to parse path + payload starting at pos. Returns decoded string or None.""" if pos >= len(data): return None path_len_byte = data[pos] pos += 1 hash_size = (path_len_byte >> 6) + 1 hop_count = path_len_byte & 0x3F path_bytes = hop_count * hash_size if path_bytes > MAX_PATH_SIZE or pos + path_bytes > len(data): return None hops_str = "" loop_warn = "" hops: list[str] = [] if hop_count > 0: for i in range(hop_count): h = data[pos + i * hash_size : pos + (i + 1) * hash_size] hops.append(h.hex()) hops_str = f" {DIM}path=[{RST}{'>'.join(hops)}{DIM}]{RST}" warn = _detect_loops(hops, hash_size) if warn: loop_warn = f"\n{warn}" pos += path_bytes payload = data[pos:] prefix = f"{type_color}{type_name}{RST} {route_name}{tc_str}{hops_str}" if payload_type == 0x04: result = _decode_advert(prefix, payload) elif payload_type == 0x03: result = _decode_ack(prefix, payload) elif payload_type in (0x00, 0x01, 0x02, 0x08): result = _decode_encrypted_peer(prefix, type_name, payload) elif payload_type in (0x05, 0x06): result = _decode_encrypted_group(prefix, payload) elif payload_type == 0x07: result = _decode_anon_req(prefix, payload) elif payload_type == 0x09: result = _decode_trace(prefix, payload) elif payload_type == 0x0A: result = _decode_multipart(prefix, payload) elif payload_type == 0x0B: result = _decode_control(prefix, payload) else: result = f"{prefix} {DIM}{payload.hex()}{RST}" if payload else prefix return result + loop_warn def decode_meshcore_packet(data: bytes) -> str: """Decode a MeshCore packet into a human-readable string.""" if len(data) < 2: return f"{RED}{RST} {DIM}{data.hex()}{RST}" header = data[0] if header == 0xFF: return f"{DIM} {data[1:].hex()}{RST}" route_type = header & 0x03 payload_type = (header >> 2) & 0x0F payload_ver = (header >> 6) & 0x03 # MeshCore only implements PAYLOAD_VER_1 (0x00) — versions 1-3 are reserved # and rejected by real nodes. Packets with ver > 0 are noise or non-MeshCore. if payload_ver != 0: route_name = ROUTE_NAMES.get(route_type, f"rt{route_type}") type_name = PAYLOAD_NAMES.get(payload_type, f"0x{payload_type:02x}") return (f"{RED}{RST} {DIM}hdr=0x{header:02x} " f"(ver={payload_ver} — only v0 exists) {data[1:].hex()}{RST}") route_name = ROUTE_NAMES.get(route_type, f"rt{route_type}") type_name = PAYLOAD_NAMES.get(payload_type, f"0x{payload_type:02x}") type_color = TYPE_COLORS.get(type_name, WHT) has_tc = route_type in (0, 3) # Try with transport codes first if indicated, fallback without if has_tc and len(data) >= 6: tc1, tc2 = struct.unpack_from("{RST} {type_color}{type_name}{RST} " f"{route_name} {DIM}{data[1:].hex()}{RST}") def _decode_advert(prefix: str, payload: bytes) -> str: if len(payload) < 100: return f"{prefix} {RED}=100>{RST} {DIM}{payload.hex()}{RST}" pubkey = payload[0:32] timestamp = struct.unpack_from(" str: if len(payload) < 4: return f"{prefix} {RED}{RST} {DIM}{payload.hex()}{RST}" crc = struct.unpack_from(" str: if len(payload) < 4: return f"{prefix} {RED}{RST} {DIM}{payload.hex()}{RST}" dst = payload[0] src = payload[1] mac = payload[2:4] ct_len = len(payload) - 4 return f"{prefix} {DIM}dst={RST}{dst:02x} {DIM}src={RST}{src:02x} {DIM}mac={RST}{mac.hex()} {DIM}[{ct_len}B]{RST}" def _decode_encrypted_group(prefix: str, payload: bytes) -> str: if len(payload) < 3: return f"{prefix} {RED}{RST} {DIM}{payload.hex()}{RST}" ch = payload[0] mac_bytes = payload[1:3] ciphertext = payload[3:] # Try to decrypt with known channels matching this hash candidates = _CHANNEL_BY_HASH.get(ch, []) for chan_name, secret in candidates: plaintext = _grp_verify_and_decrypt(secret, mac_bytes, ciphertext) if plaintext is not None: parsed = _parse_grp_plaintext(plaintext) if parsed: ts, text = parsed try: ts_str = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%H:%M:%S") except (OSError, OverflowError, ValueError): ts_str = f"epoch={ts}" return (f"{prefix} {BGRN}{chan_name}{RST} " f"{DIM}ts={RST}{ts_str}Z {BWHT}{text}{RST}") # MAC matched but plaintext unparseable return (f"{prefix} {BGRN}{chan_name}{RST} " f"{DIM}[decrypted but unparseable]{RST}") # No known channel matched — show raw info return (f"{prefix} {DIM}ch={RST}{ch:02x} {DIM}mac={RST}{mac_bytes.hex()} " f"{DIM}[{len(ciphertext)}B]{RST}") def _decode_anon_req(prefix: str, payload: bytes) -> str: if len(payload) < 35: return f"{prefix} {RED}{RST} {DIM}{payload.hex()}{RST}" dst = payload[0] ephem_pub = payload[1:33] mac = payload[33:35] ct_len = len(payload) - 35 return f"{prefix} {DIM}dst={RST}{dst:02x} {DIM}ephem={RST}{ephem_pub[:4].hex()}.. {DIM}mac={RST}{mac.hex()} {DIM}[{ct_len}B]{RST}" def _decode_trace(prefix: str, payload: bytes) -> str: if len(payload) < 9: return f"{prefix} {RED}{RST} {DIM}{payload.hex()}{RST}" tag = struct.unpack_from("'.join(hops)}{DIM}]{RST}" if hops else "" return f"{prefix} {DIM}tag={RST}0x{tag:08x}{hops_str}" def _decode_multipart(prefix: str, payload: bytes) -> str: if len(payload) < 1: return f"{prefix} {RED}{RST}" remaining = (payload[0] >> 4) & 0x0F inner_type = payload[0] & 0x0F inner_name = PAYLOAD_NAMES.get(inner_type, f"0x{inner_type:02x}") return f"{prefix} {DIM}remaining={RST}{remaining} {DIM}inner={RST}{inner_name} {DIM}[{len(payload) - 1}B]{RST}" def _decode_control(prefix: str, payload: bytes) -> str: if len(payload) < 1: return f"{prefix} {RED}{RST}" sub_type = (payload[0] >> 4) & 0x0F sub_names = {0x8: "DISCOVER_REQ", 0x9: "DISCOVER_RESP"} sub_name = sub_names.get(sub_type, f"sub=0x{sub_type:x}") if sub_type == 0x8 and len(payload) >= 6: type_filter = payload[1] tag = struct.unpack_from("= 6: node_type = payload[0] & 0x0F snr_raw = struct.unpack_from(" dict | None: payload = encode_command(cmd) frame = cobs_frame(payload) print(f"{DIM}>>>{RST} {label}") ser.write(frame) ser.flush() resp_data = read_frame(ser) if resp_data is None: print(f" {YEL}timeout{RST}") return None resp = decode_response(resp_data) print(f"{DIM}<<<{RST} {resp}") return resp RADIO_CONFIG = { "freq_hz": 910_525_000, "bw": 6, # Khz62 = variant index 6 "sf": 7, "cr": 5, # CR 4/5 — denominator value "sync_word": 0x3444, "tx_power_dbm": -128, # TX_POWER_MAX: use board's maximum } def _format_rssi_snr(rssi: int, snr: int) -> str: rc = _rssi_color(rssi) sc = _snr_color(snr) line = f"{rc}RSSI:{rssi:4d}dBm{RST} {sc}SNR:{snr:3d}dB{RST}" # Flag bogus dongle responses (SNR outside SX126x range -32..+32) if not (-32 <= snr <= 32): line += f" {BRED}[bad SNR]{RST}" return line def configure_and_listen(ser: serial.Serial): send_cmd(ser, {"type": "Ping"}, "Ping") send_cmd(ser, {"type": "SetConfig", "config": RADIO_CONFIG}, "SetConfig 910.525/62.5k/SF7/CR4/5") send_cmd(ser, {"type": "StartRx"}, "StartRx") print(f"\n{BWHT}Listening for packets{RST} {DIM}(Ctrl+C to stop){RST}\n") ser.timeout = 1 while True: try: _aggregator.maybe_report(ser) except Exception as e: print(f" {RED}[report error: {e}]{RST}") data = read_frame(ser) if data is None: continue try: resp = decode_response(data) if resp["type"] == "RxPacket": payload = resp["payload"] decoded = decode_meshcore_packet(payload) rssi_snr = _format_rssi_snr(resp["rssi"], resp["snr"]) print( f" {rssi_snr} " f"{DIM}len:{len(payload):3d}{RST} " f"{decoded}" ) else: print(f" {DIM}{resp}{RST}") except Exception as e: print(f" {RED}[decode error: {e}] raw={data.hex()}{RST}") def main(): _load_adverts() n = len(_known_nodes) if n: print(f"{DIM}Loaded {n} known node(s) from {_ADVERT_FILE}{RST}") port = sys.argv[1] if len(sys.argv) > 1 else None while True: if port is None: port = wait_for_device() try: print(f"Opening {port}") ser = open_serial(port) ser.reset_input_buffer() configure_and_listen(ser) except (serial.SerialException, ConnectionError, OSError) as e: print(f"\n{RED}Disconnected: {e}{RST}") print("Will reconnect when device reappears...") port = None time.sleep(1) except KeyboardInterrupt: print(f"\n{DIM}Stopping...{RST}") try: ser.timeout = 2 send_cmd(ser, {"type": "StopRx"}, "StopRx") except Exception: pass break if __name__ == "__main__": main()