diff --git a/src/meshcore/meshcore_parser.py b/src/meshcore/meshcore_parser.py index 046709a..d46d998 100644 --- a/src/meshcore/meshcore_parser.py +++ b/src/meshcore/meshcore_parser.py @@ -160,11 +160,15 @@ class MeshcorePacketParser: log_data["txt_type"] = txt_type else: # found: copy - log_data["message"] = logged["message"] - log_data["msg_hash"] = logged["msg_hash"] - log_data["sender_timestamp"] = logged["sender_timestamp"] - log_data["attempt"] = logged["attempt"] - log_data["txt_type"] = logged["txt_type"] + # A prior channels_log entry matched by pkt_hash may have been logged while the + # channel could not be decrypted, so it has no "message"/"msg_hash"/etc. Copy with + # .get() so a duplicate of an unreadable frame does not raise KeyError and abort + # handle_rx for the whole packet. + log_data["message"] = logged.get("message") + log_data["msg_hash"] = logged.get("msg_hash") + log_data["sender_timestamp"] = logged.get("sender_timestamp") + log_data["attempt"] = logged.get("attempt") + log_data["txt_type"] = logged.get("txt_type") self.channels_log.append(log_data) if len(self.channels_log) > 100: diff --git a/tests/unit/test_rxlog_keyerror_regression.py b/tests/unit/test_rxlog_keyerror_regression.py new file mode 100644 index 0000000..810a1df --- /dev/null +++ b/tests/unit/test_rxlog_keyerror_regression.py @@ -0,0 +1,50 @@ +import pytest +from Crypto.Cipher import AES +from Crypto.Hash import HMAC, SHA256 + +from meshcore.meshcore_parser import MeshcorePacketParser + + +@pytest.mark.asyncio +async def test_duplicate_of_undecryptable_channel_frame_does_not_keyerror(): + """A channel frame first heard without the key is logged with no "message". + + When a duplicate of it arrives after the key becomes known (a channel-table refresh + landing between two copies of a flooded transmission), parsePacketPayload matches the + prior, keyless channels_log entry by pkt_hash and copies its fields. That entry has no + "message"/"msg_hash"/etc., so `logged["message"]` raised KeyError and aborted handle_rx + for the whole packet — dropping its RX_LOG_DATA. It must not raise. + """ + parser = MeshcorePacketParser() + parser.decrypt_channels = True + + channel_secret = b"\x01" * 16 + parser.channels[0] = { + "channel_idx": 0, + "channel_name": "test", + "channel_hash": "ab", + "channel_secret": channel_secret, + } + + # A valid GRP_TXT (channel) frame: header (route_type=1, payload_type=5), path_byte=0, + # then pkt_payload = chan_hash(1) + cipher_mac(2) + AES-ECB ciphertext(16). + plaintext = b"\x00\x00\x00\x00\x15hello\x00\x00\x00\x00\x00\x00" + assert len(plaintext) == 16 + encrypted = AES.new(channel_secret, AES.MODE_ECB).encrypt(plaintext) + h = HMAC.new(channel_secret, digestmod=SHA256) + h.update(encrypted) + cipher_mac = h.digest()[:2] + pkt_payload = bytes([0xAB]) + cipher_mac + encrypted + payload = bytes([0x15, 0x00]) + pkt_payload + + # Simulate the earlier reception, logged while the channel could not be decrypted: + # a channels_log entry for this pkt_hash with NO "message"/"msg_hash"/etc. + pkt_hash = int.from_bytes(SHA256.new(pkt_payload).digest()[0:4], "little", signed=False) + parser.channels_log.append({"pkt_hash": pkt_hash, "chan_hash": "ab"}) + + # The duplicate now decrypts and reaches the "found: copy" branch. + log_data = await parser.parsePacketPayload(payload, log_data={}) + + assert log_data["payload_type"] == 0x05 + # Copied (absent) from the keyless entry instead of raising KeyError. + assert log_data["message"] is None