feat: reference Meshtastic decoder, validated on real captures

AES-128-CTR (public LongFast key) + minimal protobuf walk. Verified against
packets our scanner captured on-air: decoded a live NodeInfo (longName
'Arch Optiplex', shortName 'ARCH') and POSITION packets. Serves as the
reference for the on-chip C++ port in src/main.cpp.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dorian 2026-07-01 16:23:44 +01:00
parent 1fcb0f4c59
commit de4995468d

View File

@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""Reference Meshtastic packet decoder — validates the crypto/parse logic that
the firmware (src/main.cpp) reimplements in C++, and is handy for decoding
captured hex offline.
Meshtastic default-channel ("LongFast") packets are AES-128-CTR encrypted with
a PUBLIC, hardcoded key (that's why default-channel traffic is not private).
The wire format is a 16-byte header + encrypted `Data` protobuf.
Validated 2026-07-01 against real packets our Phase 1 scanner captured on-air:
decoded a live NodeInfo broadcast (longName "Arch Optiplex", shortName "ARCH")
and POSITION packets from node !4358d230.
Usage:
python3 meshtastic_decode.py <hexpacket> [<hexpacket> ...]
"""
import struct
import sys
from Crypto.Cipher import AES
from Crypto.Util import Counter
# Meshtastic default "LongFast" PSK (public — key index 1, "AQ==").
DEFAULT_KEY = bytes.fromhex("d4f1bb3a20290759f0bcffabcf4e6901")
PORTNUMS = {1: "TEXT_MESSAGE", 3: "POSITION", 4: "NODEINFO", 67: "TELEMETRY"}
def parse_header(b: bytes):
dest, sender, pid = struct.unpack_from("<III", b, 0)
flags, chan_hash, next_hop, relay = b[12], b[13], b[14], b[15]
return {
"dest": dest, "sender": sender, "id": pid,
"flags": flags, "chan_hash": chan_hash,
"hop_limit": flags & 0x07, "next_hop": next_hop, "relay": relay,
}
def ctr_decrypt(sender: int, pid: int, ciphertext: bytes, key: bytes = DEFAULT_KEY) -> bytes:
# CTR nonce block = packetId (8 bytes LE) + fromNode (4 LE) + extraNonce (4 LE = 0)
nonce = struct.pack("<QII", pid, sender, 0)
ctr = Counter.new(128, initial_value=int.from_bytes(nonce, "big"))
return AES.new(key, AES.MODE_CTR, counter=ctr).decrypt(ciphertext)
def _read_varint(b: bytes, i: int):
v, shift = 0, 0
while True:
v |= (b[i] & 0x7F) << shift
shift += 7
if not (b[i] & 0x80):
return v, i + 1
i += 1
def parse_fields(b: bytes) -> dict:
"""Minimal protobuf field walk -> {field_number: value}."""
out, i = {}, 0
while i < len(b):
tag = b[i]; i += 1
field, wt = tag >> 3, tag & 7
if wt == 0:
v, i = _read_varint(b, i)
out[field] = v
elif wt == 2:
ln, i = _read_varint(b, i)
out[field] = b[i:i + ln]; i += ln
else: # unsupported wire type for this minimal parser
break
return out
def decode(hexpkt: str) -> dict:
b = bytes.fromhex(hexpkt)
h = parse_header(b)
plain = ctr_decrypt(h["sender"], h["id"], b[16:])
data = parse_fields(plain) # Meshtastic Data message
portnum = data.get(1)
payload = data.get(2, b"")
result = {"header": h, "portnum": portnum, "payload": payload}
if portnum == 1:
result["text"] = payload.decode("utf-8", "replace")
elif portnum == 4: # NodeInfo -> User protobuf
user = parse_fields(payload)
result["user"] = {
"id": user.get(1, b"").decode("utf-8", "replace") if isinstance(user.get(1), bytes) else user.get(1),
"long_name": user.get(2, b"").decode("utf-8", "replace") if isinstance(user.get(2), bytes) else None,
"short_name": user.get(3, b"").decode("utf-8", "replace") if isinstance(user.get(3), bytes) else None,
}
return result
def main(argv) -> int:
if len(argv) < 2:
print(__doc__)
return 1
for hexpkt in argv[1:]:
r = decode(hexpkt)
h = r["header"]
port = PORTNUMS.get(r["portnum"], r["portnum"])
print(f"from=0x{h['sender']:08x} id=0x{h['id']:08x} chan=0x{h['chan_hash']:02x} port={port}")
if "text" in r:
print(f" TEXT: {r['text']!r}")
elif "user" in r:
u = r["user"]
print(f" NODEINFO: {u['long_name']!r} / {u['short_name']!r} ({u['id']})")
else:
print(f" payload({len(r['payload'])}B): {r['payload'].hex()}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))