From de4995468dc6279431d0ccfd2098335d09bc491f Mon Sep 17 00:00:00 2001
From: Dorian <dorian@Dorians-MacBook-Air.local>
Date: Wed, 1 Jul 2026 16:23:44 +0100
Subject: [PATCH] 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>
---
 firmware/tools/meshtastic_decode.py | 114 ++++++++++++++++++++++++++++
 1 file changed, 114 insertions(+)
 create mode 100644 firmware/tools/meshtastic_decode.py

diff --git a/firmware/tools/meshtastic_decode.py b/firmware/tools/meshtastic_decode.py
new file mode 100644
index 0000000..0e3f5c0
--- /dev/null
+++ b/firmware/tools/meshtastic_decode.py
@@ -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))