Retain oversized native USB commands until they can be rejected

examples/companion_radio/main.cpp

@@ -3026,8 +3026,8 @@ bool handleCompanionWirelessCommand(const char* command, char* reply, size_t siz
 
 void setup() {
 #if MESH_ESP32_HWCDC_SESSION_GUARD
-  // HWCDC's TX ring is resized before Serial.begin() creates its mutex and
-  // enables the USB ISR. This is deliberately before the nRF52 ordering below:
+  // HWCDC's RX queue and TX ring are resized before Serial.begin() creates its
+  // mutex and enables the USB ISR. This is before the nRF52 ordering below:
   // prepareUsbLoggingPort() is otherwise a no-op outside ESP32 HWCDC.
   mesh::prepareUsbLoggingPort();
 #endif

src/helpers/UsbLogging.cpp

@@ -385,6 +385,7 @@ static portMUX_TYPE esp32_hwcdc_session_mux = portMUX_INITIALIZER_UNLOCKED;
 static uint32_t esp32_hwcdc_taken_bus_reset_generation = 0;
 static bool esp32_hwcdc_event_handler_registered = false;
 static std::atomic<size_t> esp32_hwcdc_tx_buffer_capacity{0};
+static std::atomic<bool> esp32_hwcdc_rx_queue_ready{false};
 // A busy writer or temporary allocation failure may require more than one main
 // loop to purge. Keep the original detach state/generation across retries so a
 // failed attempt never reattaches stale bytes or turns into a reboot loop.
@@ -393,7 +394,8 @@ static bool esp32_hwcdc_restore_pad_enabled = false;
 static uint32_t esp32_hwcdc_cleanup_generation = 0;
 
 static bool canAccessEsp32Hwcdc(void*) {
-  return esp32_hwcdc_allowed_generation.load(std::memory_order_acquire)
+  return esp32_hwcdc_rx_queue_ready.load(std::memory_order_acquire)
+      && esp32_hwcdc_allowed_generation.load(std::memory_order_acquire)
       == esp32_hwcdc_access_generation.load(std::memory_order_acquire);
 }
 
@@ -1053,6 +1055,20 @@ void setUsbCompanionTxBufferCapacity(size_t capacity) {
 
 void prepareUsbLoggingPort() {
 #if MESH_ESP32_HWCDC_SESSION_GUARD
+  // The SDK's 256-byte default can lose the CR of an oversized ASCII setter
+  // before the Companion's line-length check can reject it. Retain an
+  // oversized command and exceed the terminal's 542-byte line buffer so
+  // still larger single lines enter discard mode before this queue fills.
+  // RX events report only admitted bytes, with no reliable overflow signal.
+  // Like the TX ring, this queue must never be replaced under a live USB ISR.
+  static_assert(MESH_ESP32_USB_RX_BUFFER_SIZE >= 1024,
+                "HWCDC RX queue must retain oversized terminal commands");
+  // On allocation failure, begin() may fall back to the unsafe SDK default.
+  // Keep the application stream quarantined instead of parsing damaged input.
+  esp32_hwcdc_rx_queue_ready.store(
+      Serial.setRxBufferSize(MESH_ESP32_USB_RX_BUFFER_SIZE)
+          == MESH_ESP32_USB_RX_BUFFER_SIZE,
+      std::memory_order_release);
   // HWCDC::setTxBufferSize() deletes/recreates its ring without taking the TX
   // mutex or masking the USB ISR. Resize only during early setup, before
   // Serial.begin() creates that mutex and enables the interrupt handler.

src/helpers/UsbLogging.h

@@ -56,6 +56,9 @@ namespace mesh {
 #ifndef MESH_ESP32_USB_TX_BUFFER_SIZE
   #define MESH_ESP32_USB_TX_BUFFER_SIZE 4096
 #endif
+#ifndef MESH_ESP32_USB_RX_BUFFER_SIZE
+  #define MESH_ESP32_USB_RX_BUFFER_SIZE 1024
+#endif
 
 namespace detail {
 
@@ -93,8 +96,11 @@ bool saveUsbLoggingBootPreference(bool enabled);
 // instead of mistaking an allocation failure for a permanently non-empty ring.
 void setUsbCompanionTxBufferCapacity(size_t capacity);
 
-// Configure ESP32 HWCDC's bounded write timeout and TX ring before Serial.begin
-// creates its mutex and enables the USB ISR. Safe as a no-op on other ports.
+// Configure ESP32 HWCDC's RX queue, bounded write timeout and TX ring before
+// Serial.begin creates its mutex and enables the USB ISR. HWCDC silently drops
+// incoming bytes when its queue is full; hosts must wait for each command reply
+// rather than pipeline bursts larger than the bounded queue. Safe as a no-op
+// on other ports.
 void prepareUsbLoggingPort();
 
 // Start the optional dedicated USB logging interface. Ordinary and single-TTY

test/test_companion_usb_input.py

@@ -2,6 +2,7 @@
 """Run actual Companion ASCII/probe and mOTA input handlers on host streams."""
 
 from pathlib import Path
+import re
 import subprocess
 import tempfile
 import unittest
@@ -26,6 +27,7 @@ HARNESS = r'''
 #include "helpers/ota/MotaSourceSerial.h"
 #include "helpers/UsbAsciiBinarySwitch.h"
 #include "helpers/CLICommandUtils.h"
+#define MAX_TRANS_UNIT @MAX_TRANS_UNIT@
 #define COMPANION_FEATURE_USB_MOTA_SOURCE 1
 #define COMPANION_FEATURE_NETWORK_TERMINAL 0
 #define MESH_USB_LOGGING_AVAILABLE 1
@@ -37,6 +39,7 @@ struct Input : Stream {
   bool replenish=false;
   uint32_t advance_per_read=0;
   unsigned reads=0;
+  unsigned rx_dropped=0;
   int available() override { return static_cast<int>(bytes.size()); }
   int read() override {
     if(bytes.empty()) return -1;
@@ -59,6 +62,17 @@ struct Input : Stream {
     output.append(reinterpret_cast<const char*>(p),n); return n;
   }
   void push(const char* p) { while(*p) bytes.push_back(uint8_t(*p++)); }
+  // The real HWCDC ISR reads a 64-byte FIFO packet, then abandons its
+  // remaining bytes when xQueueSendFromISR finds the RX queue full.
+  void hwcdcBurst(const std::string& wire,size_t capacity) {
+    for(size_t start=0;start<wire.size();start+=64) {
+      const size_t end=std::min(start+64,wire.size());
+      for(size_t i=start;i<end;++i) {
+        if(bytes.size()==capacity) { rx_dropped+=end-i; break; }
+        bytes.push_back(uint8_t(wire[i]));
+      }
+    }
+  }
   void frame(uint8_t tail=3) {
     const uint8_t frame[]={'<',2,0,0x16,tail};
     for(uint8_t c : frame) bytes.push_back(c);
@@ -114,7 +128,8 @@ struct Mesh {
     return true;
   }
 } the_mesh;
-static char usb_terminal_line[400]={};
+@TERMINAL_LINE_DECL@
+static constexpr size_t hwcdc_rx_capacity=@HWCDC_RX_CAPACITY@;
 static size_t usb_terminal_line_len=0;
 static bool usb_terminal_discard_line=false;
 static bool usb_logging_terminal_mode=false,usb_logging_network_parked=false;
@@ -306,6 +321,52 @@ int main() {
   assert(the_mesh.commands.size()==1 && the_mesh.commands[0]=="get name");
   assert(!usb_terminal_discard_line && !usb_mota_mode && the_mesh.attaches==0);
 
+  // Reproduce the hardware failure with the SDK's old 256-byte queue: the
+  // oversized setter loses its CR before reaching the actual line limit.
+  // A later ordinary Enter would dispatch a truncated mutating command.
+  const std::string oversized_setter="set name "+std::string(600,'X')+"\r";
+  assert(oversized_setter.size()==610);
+  reset(); input.hwcdcBurst(oversized_setter,256); serviceUsbTerminal();
+  assert(input.rx_dropped==354 && input.output.size()==256+6); // Silent banner is revealed.
+  assert(the_mesh.commands.empty() && usb_terminal_line_len==256);
+  assert(!usb_terminal_discard_line && input.output.find("ERROR")==std::string::npos);
+  input.push("\r"); serviceUsbTerminal();
+  assert(the_mesh.commands==std::vector<std::string>{oversized_setter.substr(0,256)});
+
+  // The configured pre-begin RX queue preserves the complete unpaced burst.
+  // Production parsing rejects it and its delimiter produces the real prompt.
+  static_assert(hwcdc_rx_capacity>sizeof(usb_terminal_line));
+  reset(); input.hwcdcBurst(oversized_setter,hwcdc_rx_capacity); serviceUsbTerminal();
+  assert(!input.rx_dropped && input.available()==0 && the_mesh.commands.empty());
+  assert(!usb_terminal_discard_line && !usb_terminal_line_len);
+  assert(input.output.find("ERROR: command too long")!=std::string::npos);
+  assert(input.output.substr(input.output.size()-2)=="> ");
+  input.push("get name\r"); serviceUsbTerminal();
+  assert(the_mesh.commands==std::vector<std::string>{"get name"});
+
+  // Even a single line exceeding the RX queue reaches discard mode before
+  // any CR is lost. Recovery Enter cannot execute its retained setter prefix.
+  // Hosts must still use request/reply rather than overflow a multi-line batch.
+  for(size_t size : {hwcdc_rx_capacity+64,hwcdc_rx_capacity*4}) {
+    reset();
+    input.hwcdcBurst("set name "+std::string(size,'X')+"\r",hwcdc_rx_capacity);
+    serviceUsbTerminal();
+    assert(input.rx_dropped && input.available()==0 && the_mesh.commands.empty());
+    assert(usb_terminal_discard_line && !usb_terminal_line_len);
+    assert(input.output.find("ERROR: command too long")!=std::string::npos);
+    input.push("\rget name\r"); serviceUsbTerminal();
+    assert(!usb_terminal_discard_line && the_mesh.commands==std::vector<std::string>{"get name"});
+  }
+
+  // Embedded NUL rejection remains whole-line with the real HWCDC burst path.
+  reset();
+  std::string nul_setter="set name MUST-NOT-CHANGE";
+  nul_setter.push_back(0); nul_setter+="ignored\r";
+  input.hwcdcBurst(nul_setter,hwcdc_rx_capacity); serviceUsbTerminal();
+  assert(!input.rx_dropped && the_mesh.commands.empty() && !usb_terminal_discard_line);
+  assert(input.output.find("ERROR: invalid command")!=std::string::npos);
+  assert(input.output.substr(input.output.size()-2)=="> ");
+
   // Every accepted byte, UTF-8 erase, and invalid-line discard preserves the
   // terminated-length invariant; only complete valid lines prove a reader.
   reset();
@@ -377,7 +438,7 @@ int main() {
   // The ordinary ASCII handler yields too, including while discarding an
   // overlong command from a host that never stops writing.
   reset(); input.bytes={'x','x','x','x'}; input.replenish=true;
-  for(unsigned pass=0;pass<128;++pass) {
+  for(unsigned pass=0;pass<sizeof(usb_terminal_line)/4+2;++pass) {
     const unsigned before=input.reads; serviceUsbTerminal();
     assert(input.reads==before+4 && input.available()==4);
   }
@@ -535,10 +596,18 @@ class CompanionUsbInputTests(unittest.TestCase):
         functions = functions.replace("static void enterUsbTerminalMode(bool show_banner)",
                                       "static void enterUsbTerminalMode(bool show_banner = true)")
         accessor = extract_braced(CONTEXT, "static SerialMotaSource& serialFolderSource()")
+        terminal_line = re.search(r"static char usb_terminal_line\[[^\n]+;", MAIN).group()
+        rx_capacity = re.search(r"#define MESH_ESP32_USB_RX_BUFFER_SIZE (\d+)",
+                                (ROOT / "src/helpers/UsbLogging.h").read_text()).group(1)
+        max_trans_unit = re.search(r"#define MAX_TRANS_UNIT\s+(\d+)",
+                                  (ROOT / "src/MeshCore.h").read_text()).group(1)
         with tempfile.TemporaryDirectory() as directory:
             cpp = Path(directory) / "usb-input.cpp"
             cpp.write_text(HARNESS.replace("@FUNCTIONS@", functions)
-                           .replace("@SOURCE_ACCESSOR@", accessor))
+                           .replace("@SOURCE_ACCESSOR@", accessor)
+                           .replace("@TERMINAL_LINE_DECL@", terminal_line)
+                           .replace("@HWCDC_RX_CAPACITY@", rx_capacity)
+                           .replace("@MAX_TRANS_UNIT@", max_trans_unit))
             for sanitizer in (False, True):
                 with self.subTest(sanitizer=sanitizer):
                     binary = Path(directory) / ("usb-input-" + str(sanitizer))

test/test_esp32_usb_serial_hygiene.py

@@ -2,8 +2,12 @@
 
 import re
 from pathlib import Path
+import subprocess
+import tempfile
 import unittest
 
+from test_replay_reset_integration import extract_braced
+
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -13,6 +17,62 @@ def source(path: str) -> str:
 
 
 class Esp32UsbSerialHygieneTest(unittest.TestCase):
+    def test_hwcdc_rx_allocation_failure_keeps_stream_quarantined(self):
+        logging = source("src/helpers/UsbLogging.cpp")
+        functions = "\n".join(extract_braced(logging, signature) for signature in (
+            "static bool canAccessEsp32Hwcdc(void*)",
+            "void prepareUsbLoggingPort()",
+        ))
+        harness = r'''
+#include <atomic>
+#include <cassert>
+#include <cstddef>
+#define MESH_ESP32_HWCDC_SESSION_GUARD 1
+#define MESH_ESP32_USB_RX_BUFFER_SIZE 1024
+#define MESH_ESP32_USB_TX_BUFFER_SIZE 4096
+static std::atomic<bool> esp32_hwcdc_rx_queue_ready{false};
+static std::atomic<unsigned> esp32_hwcdc_allowed_generation{0};
+static std::atomic<unsigned> esp32_hwcdc_access_generation{0};
+static size_t tx_capacity=0;
+static void setUsbCompanionTxBufferCapacity(size_t capacity) { tx_capacity=capacity; }
+struct SerialMock {
+  bool live=false, fail_rx=false;
+  size_t rx=0, rx_calls=0, tx=0, timeout=0;
+  size_t setRxBufferSize(size_t size) {
+    assert(!live); ++rx_calls; rx=fail_rx?0:size; return rx;
+  }
+  size_t setTxBufferSize(size_t size) { assert(!live); return tx=size; }
+  void setTxTimeoutMs(size_t value) { timeout=value; }
+} Serial;
+@FUNCTIONS@
+int main() {
+  assert(!canAccessEsp32Hwcdc(nullptr));
+  prepareUsbLoggingPort();
+  assert(Serial.rx==1024 && Serial.rx_calls==1 && Serial.tx==4096 && Serial.timeout==5);
+  assert(tx_capacity==4096 && canAccessEsp32Hwcdc(nullptr));
+  ++esp32_hwcdc_access_generation;
+  assert(!canAccessEsp32Hwcdc(nullptr));
+  ++esp32_hwcdc_allowed_generation;
+  assert(canAccessEsp32Hwcdc(nullptr));
+  // A separate failed boot must not let the SDK's later 256-byte fallback
+  // reopen the application parser after the requested RX allocation failed.
+  Serial=SerialMock(); Serial.fail_rx=true;
+  prepareUsbLoggingPort();
+  assert(!canAccessEsp32Hwcdc(nullptr) && Serial.rx_calls==1);
+  Serial.live=true; Serial.rx=256;
+  assert(!canAccessEsp32Hwcdc(nullptr));
+}
+'''
+        with tempfile.TemporaryDirectory() as directory:
+            cpp, binary = Path(directory) / "rx-init.cpp", Path(directory) / "rx-init"
+            cpp.write_text(harness.replace("@FUNCTIONS@", functions))
+            build = subprocess.run(["g++", "-std=c++17", "-fsanitize=address,undefined",
+                                    "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                                    str(cpp), "-o", str(binary)], capture_output=True, text=True)
+            self.assertEqual(build.returncode, 0, build.stdout + build.stderr)
+            run = subprocess.run([str(binary)], capture_output=True, text=True)
+            self.assertEqual(run.returncode, 0, run.stdout + run.stderr)
+
     def test_operational_wifi_diagnostics_use_runtime_logging_port(self):
         for relative in (
             "src/helpers/ESP32Board.cpp",
@@ -163,6 +223,12 @@ class Esp32UsbSerialHygieneTest(unittest.TestCase):
             "void beginUsbLoggingPort()", prepare_start
         )]
         self.assertIn("static const size_t usb_tx_sizes[]", prepare)
+        self.assertIn("Serial.setRxBufferSize(MESH_ESP32_USB_RX_BUFFER_SIZE)", prepare)
+        self.assertIn("MESH_ESP32_USB_RX_BUFFER_SIZE >= 1024", prepare)
+        self.assertLess(prepare.index("Serial.setRxBufferSize("),
+                        prepare.index("Serial.setTxBufferSize("))
+        header = source("src/helpers/UsbLogging.h")
+        self.assertRegex(header, r"#define MESH_ESP32_USB_RX_BUFFER_SIZE 1024\b")
         self.assertIn("Serial.setTxBufferSize(candidate)", prepare)
         self.assertIn("Serial.setTxTimeoutMs(5);", prepare)
         begin_start = logging.index("void beginUsbLoggingPort()")
@@ -170,6 +236,7 @@ class Esp32UsbSerialHygieneTest(unittest.TestCase):
             "void serviceUsbLoggingPort()", begin_start
         )]
         self.assertIn("Serial.availableForWrite()", begin)
+        self.assertNotIn("Serial.setRxBufferSize(", begin)
         self.assertIn("setUsbCompanionTxBufferCapacity(", begin)
 
         debug_start = logging.index(