swaits
8a5d8c419e
MeshCore is a protocol implementation ON TOP of DongLoRa, not part of the dongle itself. Moving it to examples/ makes the separation clear while keeping it accessible as a real-world usage example.
328 lines
11 KiB
Python
Executable File
328 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Generate MeshCore test vectors using Hypothesis (property-based testing)."""
|
|
# /// script
|
|
# requires-python = ">=3.10"
|
|
# dependencies = ["hypothesis", "jsonschema"]
|
|
# ///
|
|
|
|
import argparse
|
|
import json
|
|
import struct
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from hypothesis import given, settings, HealthCheck, Phase
|
|
from hypothesis import strategies as st
|
|
|
|
CORPUS_DIR = Path(__file__).parent
|
|
OUT = CORPUS_DIR / "generated"
|
|
SCHEMA_PATH = CORPUS_DIR / "schema.json"
|
|
|
|
# MeshCore constants
|
|
MAX_PAYLOAD = 184
|
|
MAX_PATH_SIZE = 64
|
|
MAX_ADVERT_DATA_SIZE = 32
|
|
PUB_KEY_SIZE = 32
|
|
SIGNATURE_SIZE = 64
|
|
|
|
ROUTE_TYPES = {0: "tflood", 1: "flood", 2: "direct", 3: "tdirect"}
|
|
PAYLOAD_TYPES = {
|
|
0: "REQ", 1: "RESPONSE", 2: "TXT_MSG", 3: "ACK", 4: "ADVERT",
|
|
5: "GRP_TXT", 6: "GRP_DATA", 7: "ANON_REQ", 8: "PATH",
|
|
9: "TRACE", 10: "MULTIPART", 11: "CONTROL", 15: "RAW_CUSTOM",
|
|
}
|
|
NODE_TYPES = {1: "chat", 2: "repeater", 3: "room", 4: "sensor"}
|
|
|
|
|
|
def mc_header(payload_type: int, route_type: int, ver: int = 0) -> int:
|
|
return (ver << 6) | (payload_type << 2) | route_type
|
|
|
|
|
|
def path_len_byte(hop_count: int, hash_size: int) -> int:
|
|
return ((hash_size - 1) << 6) | (hop_count & 0x3F)
|
|
|
|
|
|
# ── Hypothesis strategies ─────────────────────────────────────────
|
|
|
|
hash_sizes = st.sampled_from([1, 2, 3])
|
|
route_types_no_tc = st.sampled_from([1, 2]) # flood, direct (no transport codes)
|
|
route_types_all = st.sampled_from([0, 1, 2, 3])
|
|
payload_vers = st.sampled_from([0]) # only v0 is well-defined
|
|
|
|
|
|
@st.composite
|
|
def path_strategy(draw, hash_size=None):
|
|
hs = hash_size or draw(hash_sizes)
|
|
max_hops = min(63, MAX_PATH_SIZE // hs)
|
|
hop_count = draw(st.integers(min_value=0, max_value=min(max_hops, 10)))
|
|
hops = [draw(st.binary(min_size=hs, max_size=hs)) for _ in range(hop_count)]
|
|
return hs, hops
|
|
|
|
|
|
@st.composite
|
|
def advert_appdata(draw):
|
|
"""Generate valid ADVERT app_data."""
|
|
node_type = draw(st.sampled_from([1, 2, 3, 4]))
|
|
has_loc = draw(st.booleans())
|
|
has_name = draw(st.booleans())
|
|
flags = node_type
|
|
data = b""
|
|
loc = None
|
|
name = None
|
|
|
|
if has_loc:
|
|
flags |= 0x10
|
|
lat = draw(st.integers(min_value=-90_000_000, max_value=90_000_000))
|
|
lon = draw(st.integers(min_value=-180_000_000, max_value=180_000_000))
|
|
data += struct.pack("<ii", lat, lon)
|
|
loc = {"lat": round(lat / 1e6, 4), "lon": round(lon / 1e6, 4)}
|
|
|
|
if has_name:
|
|
flags |= 0x80
|
|
# Generate ASCII names (to avoid UTF-8 scoring ambiguity in generated tests)
|
|
name = draw(st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "P", "Z"),
|
|
max_codepoint=127),
|
|
min_size=1, max_size=20
|
|
))
|
|
name_bytes = name.encode("ascii")
|
|
data += name_bytes
|
|
|
|
if len(data) + 1 > MAX_ADVERT_DATA_SIZE:
|
|
data = data[:MAX_ADVERT_DATA_SIZE - 1]
|
|
if name:
|
|
name = data[8 if has_loc else 0:].decode("ascii", errors="replace")
|
|
|
|
app_data = bytes([flags]) + data
|
|
node_label = NODE_TYPES.get(node_type, f"type={node_type}")
|
|
return app_data, {"flags": flags, "node_type": node_label, "location": loc, "name": name}
|
|
|
|
|
|
@st.composite
|
|
def valid_advert_packet(draw):
|
|
"""Generate a complete valid ADVERT packet with expected decode."""
|
|
route_type = draw(route_types_no_tc)
|
|
ver = draw(payload_vers)
|
|
hs, hops = draw(path_strategy())
|
|
pubkey = draw(st.binary(min_size=PUB_KEY_SIZE, max_size=PUB_KEY_SIZE))
|
|
timestamp = draw(st.integers(min_value=0, max_value=0xFFFFFFFF))
|
|
signature = draw(st.binary(min_size=SIGNATURE_SIZE, max_size=SIGNATURE_SIZE))
|
|
app_data, advert_expected = draw(advert_appdata())
|
|
|
|
hdr = bytes([mc_header(0x04, route_type, ver)])
|
|
pl = bytes([path_len_byte(len(hops), hs)])
|
|
path = b"".join(hops)
|
|
payload = pubkey + struct.pack("<I", timestamp) + signature + app_data
|
|
packet = hdr + pl + path + payload
|
|
|
|
return packet, {
|
|
"route_type": ROUTE_TYPES[route_type],
|
|
"payload_type": "ADVERT",
|
|
"payload_ver": ver,
|
|
"transport_codes": None,
|
|
"path": {"hash_size": hs, "hops": [h.hex() for h in hops]},
|
|
"payload": {
|
|
"pubkey_prefix": pubkey[:4].hex(),
|
|
"timestamp": timestamp,
|
|
"advert": advert_expected,
|
|
}
|
|
}
|
|
|
|
|
|
@st.composite
|
|
def valid_ack_packet(draw):
|
|
route_type = draw(route_types_no_tc)
|
|
hs, hops = draw(path_strategy())
|
|
crc = draw(st.integers(min_value=0, max_value=0xFFFFFFFF))
|
|
|
|
hdr = bytes([mc_header(0x03, route_type)])
|
|
pl = bytes([path_len_byte(len(hops), hs)])
|
|
path = b"".join(hops)
|
|
packet = hdr + pl + path + struct.pack("<I", crc)
|
|
|
|
return packet, {
|
|
"route_type": ROUTE_TYPES[route_type],
|
|
"payload_type": "ACK",
|
|
"payload_ver": 0,
|
|
"transport_codes": None,
|
|
"path": {"hash_size": hs, "hops": [h.hex() for h in hops]},
|
|
"payload": {"ack_crc": crc}
|
|
}
|
|
|
|
|
|
@st.composite
|
|
def valid_encrypted_peer_packet(draw):
|
|
ptype = draw(st.sampled_from([0, 1, 2, 8])) # REQ/RESPONSE/TXT_MSG/PATH
|
|
route_type = draw(route_types_no_tc)
|
|
hs, hops = draw(path_strategy())
|
|
dst = draw(st.integers(min_value=0, max_value=255))
|
|
src = draw(st.integers(min_value=0, max_value=255))
|
|
mac = draw(st.binary(min_size=2, max_size=2))
|
|
ct_len = draw(st.integers(min_value=0, max_value=64))
|
|
ct = draw(st.binary(min_size=ct_len, max_size=ct_len))
|
|
|
|
hdr = bytes([mc_header(ptype, route_type)])
|
|
pl = bytes([path_len_byte(len(hops), hs)])
|
|
path = b"".join(hops)
|
|
packet = hdr + pl + path + bytes([dst, src]) + mac + ct
|
|
|
|
return packet, {
|
|
"route_type": ROUTE_TYPES[route_type],
|
|
"payload_type": PAYLOAD_TYPES[ptype],
|
|
"payload_ver": 0,
|
|
"transport_codes": None,
|
|
"path": {"hash_size": hs, "hops": [h.hex() for h in hops]},
|
|
"payload": {
|
|
"dst": f"{dst:02x}",
|
|
"src": f"{src:02x}",
|
|
"mac": mac.hex(),
|
|
"ciphertext_len": ct_len
|
|
}
|
|
}
|
|
|
|
|
|
@st.composite
|
|
def valid_group_packet(draw):
|
|
ptype = draw(st.sampled_from([5, 6])) # GRP_TXT/GRP_DATA
|
|
route_type = draw(route_types_no_tc)
|
|
hs, hops = draw(path_strategy())
|
|
ch = draw(st.integers(min_value=0, max_value=255))
|
|
mac = draw(st.binary(min_size=2, max_size=2))
|
|
ct_len = draw(st.integers(min_value=0, max_value=64))
|
|
ct = draw(st.binary(min_size=ct_len, max_size=ct_len))
|
|
|
|
hdr = bytes([mc_header(ptype, route_type)])
|
|
pl = bytes([path_len_byte(len(hops), hs)])
|
|
path = b"".join(hops)
|
|
packet = hdr + pl + path + bytes([ch]) + mac + ct
|
|
|
|
return packet, {
|
|
"route_type": ROUTE_TYPES[route_type],
|
|
"payload_type": PAYLOAD_TYPES[ptype],
|
|
"payload_ver": 0,
|
|
"transport_codes": None,
|
|
"path": {"hash_size": hs, "hops": [h.hex() for h in hops]},
|
|
"payload": {
|
|
"channel_hash": f"{ch:02x}",
|
|
"mac": mac.hex(),
|
|
"ciphertext_len": ct_len
|
|
}
|
|
}
|
|
|
|
|
|
@st.composite
|
|
def invalid_packet(draw):
|
|
"""Generate an invalid MeshCore packet via mutation."""
|
|
mutation = draw(st.sampled_from(["truncate", "bad_path_len", "random"]))
|
|
|
|
if mutation == "truncate":
|
|
# Valid packet truncated to random length
|
|
pkt, _ = draw(valid_ack_packet())
|
|
cut = draw(st.integers(min_value=0, max_value=max(0, len(pkt) - 2)))
|
|
packet = pkt[:cut]
|
|
if len(packet) < 2:
|
|
error = "too_short"
|
|
else:
|
|
error = "bad_framing"
|
|
elif mutation == "bad_path_len":
|
|
# Header + impossible path_len
|
|
route_type = draw(route_types_no_tc)
|
|
ptype = draw(st.sampled_from(list(PAYLOAD_TYPES.keys())))
|
|
hdr = bytes([mc_header(ptype, route_type)])
|
|
# path_len claiming huge path
|
|
bad_pl = draw(st.integers(min_value=0xC0, max_value=0xFF))
|
|
tail = draw(st.binary(min_size=1, max_size=20))
|
|
packet = hdr + bytes([bad_pl]) + tail
|
|
error = "bad_framing"
|
|
else: # random
|
|
packet = draw(st.binary(min_size=2, max_size=50))
|
|
error = "not_meshcore"
|
|
|
|
header = packet[0] if packet else 0
|
|
route_type = header & 0x03
|
|
payload_type = (header >> 2) & 0x0F
|
|
|
|
return packet, {
|
|
"route_type": ROUTE_TYPES.get(route_type, f"rt{route_type}"),
|
|
"payload_type": PAYLOAD_TYPES.get(payload_type, f"0x{payload_type:02x}"),
|
|
"error": error if len(packet) < 2 else error
|
|
}
|
|
|
|
|
|
# ── Generation ────────────────────────────────────────────────────
|
|
|
|
_counter = 0
|
|
|
|
|
|
def save_case(packet: bytes, expected: dict, valid: bool, prefix: str):
|
|
global _counter
|
|
_counter += 1
|
|
ptype = expected["payload_type"].lower()
|
|
name = f"{prefix}_{ptype}_{_counter:04d}"
|
|
|
|
tags = [ptype, expected["route_type"]]
|
|
if not valid:
|
|
tags.append("invalid")
|
|
if "error" in expected:
|
|
tags.append(expected["error"])
|
|
|
|
case = {
|
|
"description": f"Generated {prefix}: {expected['payload_type']} {expected['route_type']}",
|
|
"tags": tags,
|
|
"packet_hex": packet.hex(),
|
|
"valid": valid,
|
|
"expected": expected,
|
|
}
|
|
|
|
# Validate against schema
|
|
import jsonschema
|
|
schema = json.loads(SCHEMA_PATH.read_text())
|
|
try:
|
|
jsonschema.validate(case, schema)
|
|
except jsonschema.ValidationError as e:
|
|
print(f" SKIP {name}: schema error: {e.message}")
|
|
return
|
|
|
|
(OUT / f"{name}.json").write_text(json.dumps(case, indent=2, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def generate(count: int):
|
|
OUT.mkdir(exist_ok=True)
|
|
|
|
# Clear previous generated files
|
|
for f in OUT.glob("*.json"):
|
|
f.unlink()
|
|
|
|
global _counter
|
|
_counter = 0
|
|
|
|
strategies = [
|
|
("valid", valid_advert_packet(), True),
|
|
("valid", valid_ack_packet(), True),
|
|
("valid", valid_encrypted_peer_packet(), True),
|
|
("valid", valid_group_packet(), True),
|
|
("invalid", invalid_packet(), False),
|
|
]
|
|
|
|
per_strategy = max(1, count // len(strategies))
|
|
|
|
for prefix, strategy, valid in strategies:
|
|
for i in range(per_strategy):
|
|
example = strategy.example()
|
|
packet, expected = example
|
|
save_case(packet, expected, valid, prefix)
|
|
|
|
total = len(list(OUT.glob("*.json")))
|
|
print(f"Generated {total} test vectors in {OUT}/")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Generate MeshCore test vectors")
|
|
parser.add_argument("--count", type=int, default=50, help="Total test cases to generate")
|
|
args = parser.parse_args()
|
|
generate(args.count)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|