Merge branch 'fix/binary_log_invalid_data' into 'master'

fix(binary_log): Fix monitor being stuck on decoding invalid binary log data

Closes IDFGH-16338

See merge request espressif/esp-idf-monitor!95

Author: peterdragun

esp_idf_monitor/base/binlog.py

@@ -222,25 +222,35 @@ def find_frames(self, data: bytes) -> Tuple[List[bytes], bytes]:
         frames: List[bytes] = []
         i = 0
         idx_of_last_found_pkg = 0
+
         while i < len(data):
             if len(data[i:]) < 15:  # Minimal frame len
                 break
             if self.detected(int(data[i])):
                 start_idx = i
-                control = Control(data[start_idx + 1:])
-                if control.pkg_len > len(data[i:]):
-                    break
-                frame = data[start_idx:start_idx + control.pkg_len]
-                if control.pkg_len != 0 and self.crc8(frame) == 0:
-                    frames.append(frame)
-                    idx_of_last_found_pkg = start_idx + control.pkg_len
-                    i += control.pkg_len - 1
+                try:
+                    control = Control(data[start_idx + 1:])
+                    if control.pkg_len > len(data[i:]):
+                        break
+                    frame = data[start_idx:start_idx + control.pkg_len]
+                    if control.pkg_len != 0 and self.crc8(frame) == 0:
+                        frames.append(frame)
+                        idx_of_last_found_pkg = start_idx + control.pkg_len
+                        i += control.pkg_len - 1
+                    else:
+                        raise ValueError('Invalid binary log; invalid CRC')
+                except (struct.error, IndexError):
+                    # Invalid control structure
+                    raise ValueError('Invalid binary log; invalid control structure')
+            else:
+                raise ValueError('Invalid binary log; invalid start of frame')
+
             i += 1
         # Return recognized frames and any remaining unprocessed data
         return frames, data[idx_of_last_found_pkg:]
 
     def convert_to_text(self, data: bytes) -> Tuple[List[bytes], bytes]:
-        messages: List[bytes] = [b'']
+        messages: List[bytes] = []
         frames, incomplete_fragment = self.find_frames(data)
         for pkg_msg in frames:
             elf_path = self.source_of_message(pkg_msg[0])
@@ -257,7 +267,7 @@ def format_message(self, message: Message) -> bytes:
         return f'{level_name} ({message.timestamp}) {message.tag}: {text_msg}\n'.encode('ascii')
 
     def format_buffer_message(self, message) -> List[bytes]:
-        text_msg: List[bytes] = [b'']
+        text_msg: List[bytes] = []
         BYTES_PER_LINE = 16
         buff_len = message.args[0]
         buffer = message.args[1]

esp_idf_monitor/base/serial_handler.py

@@ -164,11 +164,21 @@ def handle_serial_input(self, data, console_parser, coredump, gdb_helper, line_m
 
         sp = self.splitdata(data)
         if self.binary_log_detected:
-            text_lines, self._last_line_part = self.binlog.convert_to_text(sp[0])
-            for line in text_lines:
-                self.print_colored(line)
-                self.logger.handle_possible_pc_address_in_line(line)
-            return
+            try:
+                text_lines, self._last_line_part = self.binlog.convert_to_text(sp[0])
+                for line in text_lines:
+                    self.print_colored(line)
+                    self.logger.handle_possible_pc_address_in_line(line)
+                return
+            except ValueError:
+                # If no valid binary log frames were found, or if we have too much accumulated data
+                # without valid frames, exit binary log mode
+                self.binary_log_detected = False
+                # Process the accumulated data as regular text instead
+                accumulated_data = sp[0] if sp else b''
+                if accumulated_data:
+                    self._last_line_part = accumulated_data
+                    sp = self.splitdata(b'')  # Re-split the data normally
 
         for line in sp:
             line_strip = line.strip()

test/host_test/test_monitor.py

@@ -6,6 +6,7 @@
 import errno
 import filecmp
 import os
+import random
 import re
 import socket
 import subprocess
@@ -267,7 +268,7 @@ def test_auto_color_advanced(self):
     @pytest.mark.skipif(os.name == 'nt', reason='Linux/MacOS only')
     def test_rfc2217(self, rfc2217: str):
         """Run monitor with RFC2217 port"""
-        # run with no reset because it is not supported for socker ports
+        # run with no reset because it is not supported for socket ports
         input_file = 'in1.txt'
         out, err = self.run_monitor(['--no-reset'], input_file, custom_port=rfc2217)
         with open(err, 'r') as f:
@@ -375,6 +376,42 @@ def test_binary_logging(self):
                 log_clean
             ), "Expected address, 'app_main', and 'main.c:<line>' in the output"
 
+    @pytest.fixture
+    def invalid_binary_log(self):
+        with NamedTemporaryFile(delete=False) as f:
+            f.write(b'I (1) main: Starting\r\n')
+            # Binary log detection trigger
+            f.write(b'\x01')
+            # Corrupted/invalid binary log data that would cause stuck behavior
+            # The max length of the binary log frame is 1023 bytes so just to be sure we write more
+            f.write(b'\x01' + random.randbytes(1024))
+            # Text after binary log (should be processed normally)
+            f.write(b'I (1000) main: Application started\r\n')
+            # Add some valid binary log data frame from inputs/binlog
+            # Should be decoded as "I (259) example: >>> String Formatting Tests <<<"
+            f.write(b'\x02\x0c\x10\x00\x00\x85\x9c\x3f\x40\x09\x9c\x00\x00\x01\x03\xd6')
+
+        yield f.name
+        os.unlink(f.name)
+
+    def test_binary_log_invalid_data(self, invalid_binary_log: str):
+        """Test the binary log with invalid data to make sure it is processed normally and not stuck"""
+        args = [os.path.join(IN_DIR, 'log.elf')]
+        out, err = self.run_monitor(args, invalid_binary_log, timeout=15)
+        print('Using binary log file: ', invalid_binary_log)
+        with open(err, 'r') as f_err:
+            stderr = f_err.read()
+            assert 'Stopping condition has been received' in stderr
+
+        # Verify that monitor didn't get stuck and processed all data; ignore errors because we are using random data
+        with open(out, 'r', errors='ignore') as f_out:
+            output = f_out.read()
+            # Should contain messages from both before and after invalid binary log processing
+            assert 'I (1) main: Starting' in output
+            assert 'I (1000) main: Application started' in output
+            # Valid binary log data frame should be decoded
+            assert 'I (259) example: >>> String Formatting Tests <<<' in output
+
 
 @pytest.mark.skipif(os.name == 'nt', reason='Linux/MacOS only')
 class TestConfig(TestBaseClass):