diff --git a/modules/meshcore_payload_decode.py b/modules/meshcore_payload_decode.py index d9c32fb..2758cc7 100644 --- a/modules/meshcore_payload_decode.py +++ b/modules/meshcore_payload_decode.py @@ -252,6 +252,26 @@ def decode_group_text(payload: bytes, key_store: Optional[ChannelKeyStore]) -> d return result +def verify_meshcore_advert_ed25519(mesh_payload: bytes) -> bool: + """Verify MeshCore ADVERT Ed25519 signature (layout from ``Mesh::createAdvert``). + + Signed message is ``pub_key (32) + timestamp (4, LE) + app_data``; signature is + ``payload[36:100]`` (64 bytes); ``app_data`` starts at byte 100. + """ + if len(mesh_payload) < 100: + return False + try: + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + pub = mesh_payload[:32] + msg = mesh_payload[:36] + mesh_payload[100:] + sig = mesh_payload[36:100] + Ed25519PublicKey.from_public_bytes(pub).verify(sig, msg) + return True + except Exception: + return False + + def parse_advert(payload: bytes) -> dict[str, Any]: """Parse an ADVERT payload (port of meshcore-packet-capture parse_advert). diff --git a/modules/packet_decode.py b/modules/packet_decode.py index dcc9c5a..b70fb54 100644 --- a/modules/packet_decode.py +++ b/modules/packet_decode.py @@ -7,10 +7,164 @@ and its logger. from __future__ import annotations -from typing import Any +import hashlib +from typing import Any, Optional from .enums import PayloadType, PayloadVersion, RouteType -from .utils import decode_path_len_byte + + +def decode_path_len_byte(path_len_byte: int, max_path_size: int = 64) -> tuple[int, int] | None: + """Decode the RF packet path_len byte per firmware ``Packet::isValidPathLen``. + + Encoding: low 6 bits = hop count, high 2 bits = size code. + ``bytes_per_hop = (path_len >> 6) + 1`` → 1, 2, 3, or 4 (4 is reserved and invalid). + + Args: + path_len_byte: The single path_len byte from the packet. + max_path_size: Max path bytes (default 64, matches ``MAX_PATH_SIZE``). + + Returns: + ``(path_byte_length, bytes_per_hop)`` if the encoding is valid on the wire. + ``None`` if reserved size class (4) or ``hop_count * bytes_per_hop > max_path_size`` + — matching MeshCore where ``readFrom`` rejects the packet (no legacy reinterpretation). + """ + hop_count = path_len_byte & 63 + size_code = path_len_byte >> 6 + bytes_per_hop = size_code + 1 # 1, 2, 3, or 4 + if bytes_per_hop == 4: + return None + path_byte_length = hop_count * bytes_per_hop + if path_byte_length > max_path_size: + return None + return (path_byte_length, bytes_per_hop) + + +def parse_trace_payload_route_hashes(payload: bytes) -> list[str]: + """Extract TRACE route hash segments from mesh payload (after tag, auth, flags). + + Matches MeshCore: ``bytes_per_hash = 1 << (flags & 3)`` for bytes at ``payload[9:]``. + If the tail length is not a multiple of ``bytes_per_hash``, falls back to 1-byte + segments (same as MessageHandler._process_packet_path). + + Args: + payload: Full mesh payload bytes (not including header/path). + + Returns: + List of uppercase hex strings, one per hop hash. + """ + if len(payload) < 9: + return [] + flags = payload[8] + path_hash_len = 1 << (flags & 3) + if path_hash_len <= 0: + path_hash_len = 1 + path_hashes_bytes = payload[9:] + if not path_hashes_bytes: + return [] + try: + if len(path_hashes_bytes) % path_hash_len == 0: + return [ + path_hashes_bytes[i : i + path_hash_len].hex().upper() + for i in range(0, len(path_hashes_bytes), path_hash_len) + ] + except Exception: + pass + return [f"{b:02X}" for b in path_hashes_bytes] + + +def encode_path_len_byte(hop_count: int, bytes_per_hop: int) -> int: + """Pack hop count and hash size into the single path_len wire byte (inverse of decode_path_len_byte). + + Firmware: low 6 bits = hop count, high 2 bits = size code with bytes_per_hop = (code + 1). + Valid bytes_per_hop are 1, 2, or 3 (size code 4 is reserved). + """ + if bytes_per_hop not in (1, 2, 3): + raise ValueError(f"bytes_per_hop must be 1, 2, or 3, got {bytes_per_hop}") + hop_count = int(hop_count) & 0x3F + size_code = (int(bytes_per_hop) - 1) & 0x03 + return (size_code << 6) | hop_count + + +def calculate_packet_hash(raw_hex: str, payload_type: Optional[int] = None) -> str: + """Calculate hash for packet identification - based on packet.cpp. + + Packet hashes are unique to the originally sent message, allowing + identification of the same message arriving via different paths. + + Args: + raw_hex: Raw packet data as hex string. + payload_type: Optional payload type as integer (if None, extracted from header). + Must be numeric value (0-15). + + Returns: + str: 16-character hex string (8 bytes) in uppercase, or "0000000000000000" on error. + """ + try: + # Parse the packet to extract payload type and payload data + byte_data = bytes.fromhex(raw_hex) + header = byte_data[0] + + # Get payload type from header (bits 2-5) + if payload_type is None: + payload_type = (header >> 2) & 0x0F + else: + # Ensure payload_type is an integer (handle enum.value if passed) + if hasattr(payload_type, 'value'): + payload_type = payload_type.value + payload_type = int(payload_type) & 0x0F # Ensure it's 0-15 + + # Check if transport codes are present + route_type = header & 0x03 + has_transport = route_type in [0x00, 0x03] # TRANSPORT_FLOOD or TRANSPORT_DIRECT + + # Calculate path length offset dynamically based on transport codes + offset = 1 # After header + if has_transport: + offset += 4 # Skip 4 bytes of transport codes + + # Validate we have enough bytes for path_len + if len(byte_data) <= offset: + return "0000000000000000" + + path_len_byte = byte_data[offset] + offset += 1 + path_parts = decode_path_len_byte(path_len_byte) + if path_parts is None: + return "0000000000000000" + path_byte_length, _ = path_parts + + # Validate we have enough bytes for the path + if len(byte_data) < offset + path_byte_length: + return "0000000000000000" + + # Skip past the path to get to payload + payload_start = offset + path_byte_length + + # Validate we have payload data + if len(byte_data) <= payload_start: + return "0000000000000000" + + payload_data = byte_data[payload_start:] + + # Calculate hash exactly like MeshCore Packet::calculatePacketHash(): + # 1. Payload type (1 byte) + # 2. Path length (2 bytes as uint16_t, little-endian) - ONLY for TRACE packets (type 9) + # 3. Payload data + hash_obj = hashlib.sha256() + hash_obj.update(bytes([payload_type])) + + if payload_type == 9: # PAYLOAD_TYPE_TRACE + # C++ does: sha.update(&path_len, sizeof(path_len)) + # path_len is the raw wire byte (uint16_t in firmware), not the decoded byte count + hash_obj.update(path_len_byte.to_bytes(2, byteorder='little')) + + hash_obj.update(payload_data) + + # Return first 16 hex characters (8 bytes) in uppercase + return hash_obj.hexdigest()[:16].upper() + except Exception: + # Return default hash on error (caller should handle logging) + return "0000000000000000" def split_path_hex(path_hex: str, hex_chars: int) -> list[str]: diff --git a/modules/utils.py b/modules/utils.py index 8d4ad50..b00e537 100644 --- a/modules/utils.py +++ b/modules/utils.py @@ -5,7 +5,6 @@ Shared helper functions used across multiple modules """ import asyncio -import hashlib import socket import urllib.error import urllib.request @@ -26,6 +25,13 @@ from .mesh_paths import ( # noqa: F401 re-exported: callers import these from parse_path_string, public_key_has_prefix, ) +from .meshcore_payload_decode import verify_meshcore_advert_ed25519 # noqa: F401 re-exported +from .packet_decode import ( # noqa: F401 re-exported: callers import these from utils + calculate_packet_hash, + decode_path_len_byte, + encode_path_len_byte, + parse_trace_payload_route_hashes, +) try: from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -478,160 +484,6 @@ def truncate_to_bytes(text: str, limit: int) -> str: return encoded[:limit].decode("utf-8", errors="ignore") -def decode_path_len_byte(path_len_byte: int, max_path_size: int = 64) -> tuple[int, int] | None: - """Decode the RF packet path_len byte per firmware ``Packet::isValidPathLen``. - - Encoding: low 6 bits = hop count, high 2 bits = size code. - ``bytes_per_hop = (path_len >> 6) + 1`` → 1, 2, 3, or 4 (4 is reserved and invalid). - - Args: - path_len_byte: The single path_len byte from the packet. - max_path_size: Max path bytes (default 64, matches ``MAX_PATH_SIZE``). - - Returns: - ``(path_byte_length, bytes_per_hop)`` if the encoding is valid on the wire. - ``None`` if reserved size class (4) or ``hop_count * bytes_per_hop > max_path_size`` - — matching MeshCore where ``readFrom`` rejects the packet (no legacy reinterpretation). - """ - hop_count = path_len_byte & 63 - size_code = path_len_byte >> 6 - bytes_per_hop = size_code + 1 # 1, 2, 3, or 4 - if bytes_per_hop == 4: - return None - path_byte_length = hop_count * bytes_per_hop - if path_byte_length > max_path_size: - return None - return (path_byte_length, bytes_per_hop) - - -def parse_trace_payload_route_hashes(payload: bytes) -> list[str]: - """Extract TRACE route hash segments from mesh payload (after tag, auth, flags). - - Matches MeshCore: ``bytes_per_hash = 1 << (flags & 3)`` for bytes at ``payload[9:]``. - If the tail length is not a multiple of ``bytes_per_hash``, falls back to 1-byte - segments (same as MessageHandler._process_packet_path). - - Args: - payload: Full mesh payload bytes (not including header/path). - - Returns: - List of uppercase hex strings, one per hop hash. - """ - if len(payload) < 9: - return [] - flags = payload[8] - path_hash_len = 1 << (flags & 3) - if path_hash_len <= 0: - path_hash_len = 1 - path_hashes_bytes = payload[9:] - if not path_hashes_bytes: - return [] - try: - if len(path_hashes_bytes) % path_hash_len == 0: - return [ - path_hashes_bytes[i : i + path_hash_len].hex().upper() - for i in range(0, len(path_hashes_bytes), path_hash_len) - ] - except Exception: - pass - return [f"{b:02X}" for b in path_hashes_bytes] - - -def encode_path_len_byte(hop_count: int, bytes_per_hop: int) -> int: - """Pack hop count and hash size into the single path_len wire byte (inverse of decode_path_len_byte). - - Firmware: low 6 bits = hop count, high 2 bits = size code with bytes_per_hop = (code + 1). - Valid bytes_per_hop are 1, 2, or 3 (size code 4 is reserved). - """ - if bytes_per_hop not in (1, 2, 3): - raise ValueError(f"bytes_per_hop must be 1, 2, or 3, got {bytes_per_hop}") - hop_count = int(hop_count) & 0x3F - size_code = (int(bytes_per_hop) - 1) & 0x03 - return (size_code << 6) | hop_count - - -def calculate_packet_hash(raw_hex: str, payload_type: Optional[int] = None) -> str: - """Calculate hash for packet identification - based on packet.cpp. - - Packet hashes are unique to the originally sent message, allowing - identification of the same message arriving via different paths. - - Args: - raw_hex: Raw packet data as hex string. - payload_type: Optional payload type as integer (if None, extracted from header). - Must be numeric value (0-15). - - Returns: - str: 16-character hex string (8 bytes) in uppercase, or "0000000000000000" on error. - """ - try: - # Parse the packet to extract payload type and payload data - byte_data = bytes.fromhex(raw_hex) - header = byte_data[0] - - # Get payload type from header (bits 2-5) - if payload_type is None: - payload_type = (header >> 2) & 0x0F - else: - # Ensure payload_type is an integer (handle enum.value if passed) - if hasattr(payload_type, 'value'): - payload_type = payload_type.value - payload_type = int(payload_type) & 0x0F # Ensure it's 0-15 - - # Check if transport codes are present - route_type = header & 0x03 - has_transport = route_type in [0x00, 0x03] # TRANSPORT_FLOOD or TRANSPORT_DIRECT - - # Calculate path length offset dynamically based on transport codes - offset = 1 # After header - if has_transport: - offset += 4 # Skip 4 bytes of transport codes - - # Validate we have enough bytes for path_len - if len(byte_data) <= offset: - return "0000000000000000" - - path_len_byte = byte_data[offset] - offset += 1 - path_parts = decode_path_len_byte(path_len_byte) - if path_parts is None: - return "0000000000000000" - path_byte_length, _ = path_parts - - # Validate we have enough bytes for the path - if len(byte_data) < offset + path_byte_length: - return "0000000000000000" - - # Skip past the path to get to payload - payload_start = offset + path_byte_length - - # Validate we have payload data - if len(byte_data) <= payload_start: - return "0000000000000000" - - payload_data = byte_data[payload_start:] - - # Calculate hash exactly like MeshCore Packet::calculatePacketHash(): - # 1. Payload type (1 byte) - # 2. Path length (2 bytes as uint16_t, little-endian) - ONLY for TRACE packets (type 9) - # 3. Payload data - hash_obj = hashlib.sha256() - hash_obj.update(bytes([payload_type])) - - if payload_type == 9: # PAYLOAD_TYPE_TRACE - # C++ does: sha.update(&path_len, sizeof(path_len)) - # path_len is the raw wire byte (uint16_t in firmware), not the decoded byte count - hash_obj.update(path_len_byte.to_bytes(2, byteorder='little')) - - hash_obj.update(payload_data) - - # Return first 16 hex characters (8 bytes) in uppercase - return hash_obj.hexdigest()[:16].upper() - except Exception: - # Return default hash on error (caller should handle logging) - return "0000000000000000" - - def get_packet_hash_placeholder(message: Any) -> str: """Return the correlated MeshCore packet hash for template placeholders. @@ -651,26 +503,6 @@ def get_packet_hash_placeholder(message: Any) -> str: return str(packet_hash) -def verify_meshcore_advert_ed25519(mesh_payload: bytes) -> bool: - """Verify MeshCore ADVERT Ed25519 signature (layout from ``Mesh::createAdvert``). - - Signed message is ``pub_key (32) + timestamp (4, LE) + app_data``; signature is - ``payload[36:100]`` (64 bytes); ``app_data`` starts at byte 100. - """ - if len(mesh_payload) < 100: - return False - try: - from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey - - pub = mesh_payload[:32] - msg = mesh_payload[:36] + mesh_payload[100:] - sig = mesh_payload[36:100] - Ed25519PublicKey.from_public_bytes(pub).verify(sig, msg) - return True - except Exception: - return False - - def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float: """Calculate haversine distance between two points in kilometers.