Verify XMODEM downloads against advertised MD5

CHANGELOG.md

@@ -1,3 +1,4 @@
+- Fixed: Reject incomplete downloads that do not match the machine-provided MD5 checksum
 - Enhancement: Add multi-select to the remote file browser
 - Enhancement: Display error message in halt popup. Requires halt errors to start with "ERROR: " in the firmware
 - Enhancement: Add popup notice when using stock firmware instead of the Community Firmware

carveracontroller/XMODEM.py

@@ -92,6 +92,7 @@
 __license__ = "MIT"
 __version__ = "0.4.5"
 
+import hashlib
 import logging
 import platform
 import sys
@@ -717,6 +718,8 @@ def callback(success_count, error_count)
         cancel = 0
         retrans = retry + 1
         md5_received = False
+        expected_md5 = None
+        received_md5 = hashlib.md5()
 
         while True:
             if self.canceled:
@@ -732,9 +735,17 @@ def callback(success_count, error_count)
                 if char in (SOH, STX):
                     break
                 if char == EOT:
-                    # We received an EOT, so send an ACK and return the
-                    # received data length.
+                    # ACK the transport-level completion before validating the
+                    # whole-file digest advertised in block zero.
                     self.putc(ACK)
+                    actual_md5 = received_md5.hexdigest().encode()
+                    if expected_md5 is None or actual_md5 != expected_md5:
+                        self.log.error(
+                            "Download error: MD5 mismatch (expected=%r, actual=%r)",
+                            expected_md5,
+                            actual_md5,
+                        )
+                        return None
                     self.log.info("Transmission complete, %d bytes", income_size)
                     return income_size
                 if char == CAN:
@@ -813,7 +824,9 @@ def callback(success_count, error_count)
                     retrans = retry + 1
                     if sequence == 0 and not md5_received:
                         md5_received = True
-                        if md5.encode() == data[1 + is_stx : 33 + is_stx]:
+                        data_len = data[0] << 8 | data[1] if is_stx else data[0]
+                        expected_md5 = data[1 + is_stx : (data_len + 1 + is_stx)].lower()
+                        if md5.encode().lower() == expected_md5:
                             self.putc(CAN)
                             self.putc(CAN)
                             self.putc(CAN)
@@ -823,7 +836,9 @@ def callback(success_count, error_count)
                     else:
                         income_size += len(data) - 1 - is_stx
                         data_len = data[0] << 8 | data[1] if is_stx else data[0]
-                        stream.write(data[1 + is_stx : (data_len + 1 + is_stx)])
+                        payload = data[1 + is_stx : (data_len + 1 + is_stx)]
+                        stream.write(payload)
+                        received_md5.update(payload)
                         success_count = success_count + 1
                         if callable(callback):
                             callback(packet_size, success_count, error_count)

tests/unit/test_xmodem_download_integrity.py

@@ -0,0 +1,73 @@
+"""Tests for whole-file integrity checks in XMODEM downloads."""
+
+import hashlib
+from io import BytesIO
+
+from carveracontroller.XMODEM import ACK, CAN, CRC, EOT, XMODEM
+
+
+def _packet(modem, sequence, payload):
+    packet_size = 8192
+    header = modem._make_send_header(packet_size, sequence)
+    data = bytes([len(payload) >> 8, len(payload) & 0xFF]) + payload.ljust(packet_size, modem.pad)
+    checksum = modem._make_send_checksum(1, data)
+    return bytes(header + data + checksum)
+
+
+def _receive(expected_payload, received_payload, local_md5=""):
+    transport = BytesIO()
+    writes = []
+
+    def getc(size, timeout=0.5):
+        return transport.read(size) or None
+
+    def putc(data, timeout=0.5):
+        writes.append(data)
+        return len(data)
+
+    modem = XMODEM(getc, putc, "xmodem8k")
+    expected_md5 = hashlib.md5(expected_payload).hexdigest().encode()
+    transport.write(_packet(modem, 0, expected_md5))
+    if received_payload:
+        transport.write(_packet(modem, 1, received_payload))
+    transport.write(EOT)
+    transport.seek(0)
+
+    output = BytesIO()
+    result = modem.recv(output, md5=local_md5)
+    return result, output.getvalue(), writes
+
+
+def test_recv_accepts_download_matching_block_zero_md5():
+    payload = b"G0 X1 Y2\nM2\n"
+
+    result, output, writes = _receive(payload, payload)
+
+    assert result is not None
+    assert result > 0
+    assert output == payload
+    assert writes == [CRC, ACK, ACK, ACK]
+
+
+def test_recv_short_circuits_when_uppercase_local_md5_matches_block_zero():
+    payload = b"G0 X1 Y2\nM2\n"
+    local_md5 = hashlib.md5(payload).hexdigest().upper()
+
+    result, output, writes = _receive(payload, payload, local_md5=local_md5)
+
+    assert result == 0
+    assert output == b""
+    assert writes == [CRC, CAN, CAN, CAN]
+
+
+def test_recv_rejects_eot_when_download_does_not_match_block_zero_md5():
+    expected_payload = b"G0 X1 Y2\nG1 X3 Y4\nM2\n"
+    truncated_payload = b"G0 X1 Y2\n"
+
+    result, output, writes = _receive(expected_payload, truncated_payload)
+
+    assert result is None
+    assert output == truncated_payload
+    # Transport completion remains wire-compatible even though the local
+    # artifact is rejected after its whole-file digest is checked.
+    assert writes == [CRC, ACK, ACK, ACK]