Merge branch 'feature/execute_cli_command' into 'master'

peterdragun · peterdragun · commit 823b6822ebd6 · 2025-12-19T15:38:31.000+08:00
feat(serial_handler): Support executing CLI commands from monitor logs

See merge request espressif/esp-idf-monitor!100
diff --git a/README.md b/README.md
@@ -19,6 +19,7 @@ Other advanced topics like configuration file will be described in the following
     - [Custom Reset Sequence](#custom-reset-sequence)
       - [Share Configuration Across Tools](#share-configuration-across-tools)
   - [Syntax](#syntax)
+- [Embedded Command Execution](#embedded-command-execution)
 
 ## Installation
 
@@ -169,6 +170,54 @@ exit_menu_key = X
 skip_menu_key = False
 ```
 
+## Embedded Command Execution
+
+`esp-idf-monitor` includes an advanced feature that automatically executes host-side tools when the target device outputs specific markers in its logs. This is particularly useful for workflows where device firmware needs to perform operations that are better handled on the host computer—such as decoding or analyzing chip data (for example, reading and interpreting eFuse dump).
+
+### How It Works
+
+When the monitor detects one of the predefined markers in the device output, it automatically executes the corresponding command template. The command substitutes data from the device output (such as eFuse tokens) into the template, allowing seamless data analysis without manual intervention.
+
+### Supported Markers
+
+The following markers are currently supported:
+
+| Marker                                 | Command Template                           | Use Case                           |
+|----------------------------------------|--------------------------------------------|------------------------------------|
+| `IDF_MONITOR_EXECUTE_ESPEFUSE_SUMMARY` | `espefuse --token {ARGS} summary --active` | Display active eFuse summary       |
+| `IDF_MONITOR_EXECUTE_ESPEFUSE_DUMP`    | `espefuse --token {ARGS} dump`             | Display eFuse dump                 |
+
+For both commands, `{ARGS}` must include:
+- A token eFuse dump (format: `EFSR:chiptype:size:hexdata...`)
+- Optionally, additional flags such as `--extend-efuse-table main/esp_efuse_custom_table.csv` to extend eFuse field definitions
+
+### Usage Example
+
+When your firmware outputs a line containing `IDF_MONITOR_EXECUTE_ESPEFUSE_DUMP`:
+
+```text
+I (481) example: IDF_MONITOR_EXECUTE_ESPEFUSE_DUMP EFSR:esp32c3:100:AAAAAAAAAAAAAAAAAAAAAAAAAIAAAAAA:zIH3-VVgAAAAAAAAAAAAS8kmEVKwQgYB:ZSd8yloMSAJssOWmfZQw8lFbphuTZH574QcV3ggAAAA:AAAAAAAAAAEayAcAAAAAAAAAAAAAAAAAAAAAAAAAAAA:::::::::ydrNkQ
+--- Executing monitor command: espefuse --token EFSR:esp32c3:100:... dump
+espefuse v5.1.0
+=== Run "dump" command ===
+BLOCK0          (                ) [0 ] dump: 00000000 00000000 00000000 00000000 80000000 00000000
+MAC_SPI_8M_0    (BLOCK1          ) [1 ] dump: f9f781cc 00006055 00000000 4b000000 521126c9 010642b0
+BLOCK_SYS_DATA  (BLOCK2          ) [2 ] dump: ca7c2765 02480c5a a6e5b06c f230947d 1ba65b51 7b7e6493 de1507e1 00000008
+...
+I (331) example: read efuse fields
+```
+
+### Security and Limitations
+
+For your security and to ensure predictable behavior, IDF Monitor:
+
+- Does not execute arbitrary commands printed by the device
+- Supports only a small, predefined set of markers mapped to fixed command templates
+- Accepts only `<ARGS>` from the device—the eFuse token and optional flags—which are substituted into the template
+- Executes all commands with `shell=False`, preventing shell metacharacters (`&&`, `;`, `|`, `>`) from being interpreted
+
+This intentional limitation ensures that only specific, safe espefuse operations are available. Any future extensions would require careful review for security implications.
+
 ## Contributing
 
 ### Code Style & Static Analysis
diff --git a/esp_idf_monitor/base/monitor_secure_exec.py b/esp_idf_monitor/base/monitor_secure_exec.py
@@ -0,0 +1,130 @@
+# SPDX-FileCopyrightText: 2025 Espressif Systems (Shanghai) CO LTD
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Secure command executor for monitor-triggered CLI commands.
+
+This module provides a controlled interface for executing pre-approved commands
+found in monitor log lines, while preventing arbitrary code execution.
+
+Expected format in the log line:
+
+    ... IDF_MONITOR_EXECUTE_<TYPE> <ARGS>
+
+For example:
+
+    I (311) example: IDF_MONITOR_EXECUTE_ESPEFUSE_SUMMARY EFSR:esp32:300:AAAA...
+"""
+
+import os
+import shlex
+import subprocess
+
+from .output_helpers import note_print
+from .output_helpers import warning_print
+
+# Prefix used in log lines to indicate an executable monitor command
+MONITOR_EXECUTE_PREFIX = 'IDF_MONITOR_EXECUTE_'
+
+
+class SecureMonitorCommandExecutor:
+    """
+    Secure executor for commands embedded in monitor log lines.
+
+    The executor:
+      - parses log lines for MONITOR_EXECUTE_PREFIX,
+      - extracts the command type and arguments that follow the marker,
+      - maps the type to a fixed command template,
+      - executes the command if it is allowed.
+
+    This prevents arbitrary commands from being run based solely on device output.
+    """
+
+    def __init__(self, logger) -> None:
+        """
+        Initialize the command executor.
+
+        Args:
+            logger: Logger instance used to print command output.
+        """
+        self._logger = logger
+        self._incomplete_line = ''
+        self.enable = True
+        # Allowed commands keyed by TYPE after IDF_MONITOR_EXECUTE_
+        # The placeholder '{}' is filled with the argument string from the log line
+        self._allowed_cmds = {
+            'ESPEFUSE_SUMMARY': 'espefuse --token {} summary --active',
+            'ESPEFUSE_DUMP': 'espefuse --token {} dump',
+        }
+
+    def execute_from_log_line(self, chunk: bytes) -> None:
+        """
+        Consume a chunk of bytes from the monitor and execute an embedded
+        command if a *complete* line with MONITOR_EXECUTE_PREFIX is present.
+
+        This function is robust to partial lines: it accumulates data until
+        a newline is seen, then processes whole lines one by one.
+        """
+        if not self.enable:
+            return
+
+        # log output is ASCII-like; ignore any problematic bytes
+        text = chunk.decode('ascii', errors='ignore')
+
+        if '\n' not in text:
+            # No complete line yet; accumulate
+            self._incomplete_line += text
+            return
+
+        # Complete line(s) present
+        if self._incomplete_line:
+            text = self._incomplete_line + text
+            self._incomplete_line = ''
+
+        self._process_complete_line(text)
+
+    def _process_complete_line(self, line_text: str) -> None:
+        """
+        Handle a single *complete* log line (without trailing newline).
+        If it contains a valid IDF_MONITOR_EXECUTE_<TYPE> command, execute it.
+        """
+
+        if MONITOR_EXECUTE_PREFIX not in line_text:
+            return
+
+        # Extract everything after the first occurrence of "IDF_MONITOR_EXECUTE_"
+        parts = line_text.split(MONITOR_EXECUTE_PREFIX, 1)
+        if len(parts) < 2:
+            return
+
+        # Split into TYPE and the rest (arguments)
+        try:
+            cmd_type, cmd_args = parts[1].strip().split(maxsplit=1)
+        except ValueError:
+            cmd_type, cmd_args = parts[1].strip(), ''
+
+        template = self._allowed_cmds.get(cmd_type)
+        if template is None:  # Unknown cmd type
+            warning_print(f'Ignoring unknown monitor command type: "IDF_MONITOR_EXECUTE_{cmd_type}"')
+            return
+
+        if '{}' in template:
+            cmd_args = cmd_args.strip()
+            if not cmd_args:
+                warning_print(f'Ignoring monitor command: no arguments provided for command "{cmd_type}": "{template}"')
+                return
+            formatted = template.format(cmd_args)
+        else:
+            formatted = template
+
+        cmd_argv = shlex.split(formatted)  # Convert to argv list
+
+        note_print(f'Executing monitor command: {" ".join(cmd_argv)}')
+
+        try:
+            output = subprocess.check_output(cmd_argv, stderr=subprocess.STDOUT, env=os.environ, shell=False)
+            self._logger.print(output)
+        except subprocess.CalledProcessError as e:
+            warning_print(f'Command failed: {e}\n{e.output.decode(errors="ignore")}')
+        except OSError as e:
+            warning_print(f'Failed to execute command: {e}')
diff --git a/esp_idf_monitor/base/serial_handler.py b/esp_idf_monitor/base/serial_handler.py
@@ -44,6 +44,7 @@
 from .key_config import MENU_KEY
 from .line_matcher import LineMatcher  # noqa: F401
 from .logger import Logger  # noqa: F401
+from .monitor_secure_exec import SecureMonitorCommandExecutor
 from .output_helpers import ANSI_GREEN_B
 from .output_helpers import ANSI_NORMAL_B
 from .output_helpers import ANSI_RED_B
@@ -123,6 +124,7 @@ def __init__(
         self.disable_auto_color = disable_auto_color
         self.binlog = BinaryLog(elf_files)
         self.binary_log_detected = False
+        self.monitor_cmd_executor = SecureMonitorCommandExecutor(self.logger)
 
     def splitdata(self, data):  # type: (bytes) -> List[bytes]
         """
@@ -207,6 +209,7 @@ def handle_serial_input(
                 for line in text_lines:
                     self.print_colored(line)
                     self.logger.handle_possible_pc_address_in_line(line)
+                    self.monitor_cmd_executor.execute_from_log_line(line)
                 return
             except ValueError:
                 # If no valid binary log frames were found, or if we have too much accumulated data
@@ -241,6 +244,7 @@ def handle_serial_input(
                     self.print_colored(line)
                     self.compare_elf_sha256(decoded_line)
                     self.logger.handle_possible_pc_address_in_line(line_strip)
+                    self.monitor_cmd_executor.execute_from_log_line(line)
             check_gdb_stub_and_run(line_strip)
             self._force_line_print = False
 
@@ -262,6 +266,7 @@ def handle_serial_input(
             self._force_line_print = True
             self.print_colored(self._last_line_part)
             self.logger.handle_possible_pc_address_in_line(self._last_line_part, insert_new_line=True)
+            self.monitor_cmd_executor.execute_from_log_line(self._last_line_part)
             check_gdb_stub_and_run(self._last_line_part)
             # It is possible that the incomplete line cuts in half the PC
             # address. A small buffer is kept and will be used the next time
@@ -393,6 +398,7 @@ def handle_serial_input(
 
             if self._force_line_print or line_matcher.match(line.decode(errors='ignore')):
                 self.print_colored(line)
+                self.monitor_cmd_executor.execute_from_log_line(line)
                 self._force_line_print = False
 
         if self._last_line_part.startswith(CONSOLE_STATUS_QUERY):
@@ -409,4 +415,5 @@ def handle_serial_input(
         if self._last_line_part != b'' and force_print_or_matched:
             self._force_line_print = True
             self.print_colored(self._last_line_part)
+            self.monitor_cmd_executor.execute_from_log_line(self._last_line_part)
             self._last_line_part = b''
diff --git a/test/host_test/test_monitor.py b/test/host_test/test_monitor.py
@@ -667,3 +667,145 @@ def test_c_format(self, c_fmt, arg, pythonic_fmt, output):
         assert converted_format == pythonic_fmt, f"Expected Pythonic format '{pythonic_fmt}', got '{converted_format}'"
         formatted_output = formatter.c_format(c_fmt, [arg])
         assert formatted_output == output, f"Expected '{output}', got '{formatted_output}'"
+
+
+class TestEmbeddedMonitorCommands:
+    """Tests for SecureMonitorCommandExecutor handling embedded monitor commands."""
+
+    class DummyLogger:
+        def __init__(self) -> None:
+            self.outputs: List[bytes] = []
+
+        def print(self, data: bytes) -> None:
+            self.outputs.append(data)
+
+    @pytest.mark.parametrize(
+        'input_line, expect_called, expected_argv',
+        [
+            # Unknown marker type after IDF_MONITOR_EXECUTE_: should be ignored
+            (
+                'I (20) test: IDF_MONITOR_EXECUTE_UNKNOWN EFSR:esp32c3:100:AAA\n',
+                False,
+                None,
+            ),
+            # Marker without any arguments: should be ignored
+            (
+                'I (20) test: IDF_MONITOR_EXECUTE_ESPEFUSE_SUMMARY\n',
+                False,
+                None,
+            ),
+            # Valid ESPEFUSE_SUMMARY with token
+            (
+                'I (311) example: IDF_MONITOR_EXECUTE_ESPEFUSE_SUMMARY EFSR:esp32c3:100:AAA\n',
+                True,
+                ['espefuse', '--token', 'EFSR:esp32c3:100:AAA', 'summary', '--active'],
+            ),
+            # Valid ESPEFUSE_DUMP with token
+            (
+                'I (331) example: IDF_MONITOR_EXECUTE_ESPEFUSE_DUMP EFSR:esp32c3:100:AAA\n',
+                True,
+                ['espefuse', '--token', 'EFSR:esp32c3:100:AAA', 'dump'],
+            ),
+        ],
+    )
+    def test_monitor_embedded_command_execution(
+        self,
+        input_line: str,
+        expect_called: bool,
+        expected_argv: Optional[List[str]],
+        monkeypatch,
+    ):
+        """
+        Verify that SecureMonitorCommandExecutor:
+        """
+        from esp_idf_monitor.base.monitor_secure_exec import SecureMonitorCommandExecutor
+
+        calls = []
+
+        def fake_check_output(argv, stderr=None, env=None, shell=None):
+            calls.append(
+                {
+                    'argv': argv,
+                    'stderr': stderr,
+                    'env': env,
+                    'shell': shell,
+                }
+            )
+            return b'OK\n'
+
+        # Patch subprocess.check_output used inside monitor_secure_exec
+        monkeypatch.setattr(
+            'esp_idf_monitor.base.monitor_secure_exec.subprocess.check_output',
+            fake_check_output,
+        )
+
+        logger = self.DummyLogger()
+        executor = SecureMonitorCommandExecutor(logger)
+
+        # Run executor for this test case (single full line, with '\n')
+        executor.execute_from_log_line(input_line.encode('ascii'))
+
+        if not expect_called:
+            assert calls == []
+            assert logger.outputs == []
+            return
+
+        # Exactly one subprocess call expected
+        assert len(calls) == 1
+        call = calls[0]
+        argv = call['argv']
+
+        # Full argv must match the expected expansion of the template
+        assert argv == expected_argv
+
+        # Explicitly verify that shell=False was used
+        assert call['shell'] is False
+
+        # Logger should receive the output from the subprocess
+        assert logger.outputs == [b'OK\n']
+
+    def test_monitor_embedded_command_streaming_chunks(self, monkeypatch):
+        """
+        Verify that execute_from_log_line handles partial lines and only
+        executes once a complete line (with '\n') is received.
+        """
+        from esp_idf_monitor.base.monitor_secure_exec import SecureMonitorCommandExecutor
+
+        calls = []
+
+        def fake_check_output(argv, stderr=None, env=None, shell=None):
+            calls.append(
+                {
+                    'argv': argv,
+                    'stderr': stderr,
+                    'env': env,
+                    'shell': shell,
+                }
+            )
+            return b'OK\n'
+
+        monkeypatch.setattr(
+            'esp_idf_monitor.base.monitor_secure_exec.subprocess.check_output',
+            fake_check_output,
+        )
+
+        logger = self.DummyLogger()
+        executor = SecureMonitorCommandExecutor(logger)
+
+        # First chunk: no newline yet, should not trigger execution
+        chunk1 = b'I (311) example: IDF_MONITOR_EXECUTE_ESPEFUSE_SUMMARY EFSR:esp32c3:100:AAA'
+        executor.execute_from_log_line(chunk1)
+        assert calls == []
+        assert logger.outputs == []
+
+        # Second chunk: completes the line with '\n'
+        chunk2 = b'BBB\n'
+        executor.execute_from_log_line(chunk2)
+
+        # Now we expect exactly one call, with the combined token "AAABBB"
+        assert len(calls) == 1
+        call = calls[0]
+        argv = call['argv']
+        assert argv == ['espefuse', '--token', 'EFSR:esp32c3:100:AAABBB', 'summary', '--active']
+        assert call['shell'] is False
+        assert logger.outputs == [b'OK\n']
