From f073c1cabe180f10fc9af0b345a2b03cc4ec164e Mon Sep 17 00:00:00 2001 From: Oliver Kleinecke Date: Fri, 1 May 2026 12:43:13 +0200 Subject: [PATCH 1/5] [usb_host][usb_uart] Add configurable max packet size (#14584) --- esphome/components/usb_host/__init__.py | 25 ++++++++++++++++--- esphome/components/usb_host/usb_host.h | 2 ++ .../components/usb_host/usb_host_client.cpp | 2 +- esphome/components/usb_uart/__init__.py | 12 ++++++--- esphome/components/usb_uart/usb_uart.cpp | 14 +++++------ esphome/components/usb_uart/usb_uart.h | 11 ++++---- 6 files changed, 44 insertions(+), 22 deletions(-) diff --git a/esphome/components/usb_host/__init__.py b/esphome/components/usb_host/__init__.py index 338bd8d572..8e591bd80c 100644 --- a/esphome/components/usb_host/__init__.py +++ b/esphome/components/usb_host/__init__.py @@ -10,6 +10,7 @@ from esphome.components.esp32 import ( ) import esphome.config_validation as cv from esphome.const import CONF_DEVICES, CONF_ID +from esphome.core import CORE from esphome.cpp_types import Component from esphome.types import ConfigType @@ -19,14 +20,15 @@ DEPENDENCIES = ["esp32"] usb_host_ns = cg.esphome_ns.namespace("usb_host") USBHost = usb_host_ns.class_("USBHost", Component) USBClient = usb_host_ns.class_("USBClient", Component) - +DOMAIN = "usb_host" CONF_VID = "vid" CONF_PID = "pid" CONF_ENABLE_HUBS = "enable_hubs" CONF_MAX_TRANSFER_REQUESTS = "max_transfer_requests" +CONF_MAX_PACKET_SIZE = "max_packet_size" -def usb_device_schema(cls=USBClient, vid: int = None, pid: [int] = None) -> cv.Schema: +def usb_device_schema(cls=USBClient, vid: int = None, pid: int = None) -> cv.Schema: schema = cv.COMPONENT_SCHEMA.extend( { cv.GenerateID(): cv.declare_id(cls), @@ -43,6 +45,17 @@ def usb_device_schema(cls=USBClient, vid: int = None, pid: [int] = None) -> cv.S return schema +def _set_max_packet_size(config: dict) -> dict: + CORE.data.setdefault(DOMAIN, {})[CONF_MAX_PACKET_SIZE] = config[ + CONF_MAX_PACKET_SIZE + ] + return config + + +def get_max_packet_size() -> int: + return CORE.data.get(DOMAIN, {}).get(CONF_MAX_PACKET_SIZE, 64) + + CONFIG_SCHEMA = cv.All( cv.COMPONENT_SCHEMA.extend( { @@ -51,10 +64,14 @@ CONFIG_SCHEMA = cv.All( cv.Optional(CONF_MAX_TRANSFER_REQUESTS, default=16): cv.int_range( min=1, max=32 ), + cv.Optional(CONF_MAX_PACKET_SIZE, default=64): cv.one_of( + 64, 128, 256, 512, 1024, int=True + ), cv.Optional(CONF_DEVICES): cv.ensure_list(usb_device_schema()), } ), only_on_variant(supported=[VARIANT_ESP32P4, VARIANT_ESP32S2, VARIANT_ESP32S3]), + _set_max_packet_size, ) @@ -72,8 +89,8 @@ async def to_code(config: ConfigType) -> None: if config.get(CONF_ENABLE_HUBS): add_idf_sdkconfig_option("CONFIG_USB_HOST_HUBS_SUPPORTED", True) - max_requests = config[CONF_MAX_TRANSFER_REQUESTS] - cg.add_define("USB_HOST_MAX_REQUESTS", max_requests) + cg.add_define("USB_HOST_MAX_REQUESTS", config[CONF_MAX_TRANSFER_REQUESTS]) + cg.add_define("USB_HOST_MAX_PACKET_SIZE", config[CONF_MAX_PACKET_SIZE]) var = cg.new_Pvariable(config[CONF_ID]) await cg.register_component(var, config) diff --git a/esphome/components/usb_host/usb_host.h b/esphome/components/usb_host/usb_host.h index dcb76a3a3b..480fd86750 100644 --- a/esphome/components/usb_host/usb_host.h +++ b/esphome/components/usb_host/usb_host.h @@ -66,6 +66,8 @@ static_assert(MAX_REQUESTS >= 1 && MAX_REQUESTS <= 32, "MAX_REQUESTS must be bet using trq_bitmask_t = std::conditional<(MAX_REQUESTS <= 16), uint16_t, uint32_t>::type; static constexpr trq_bitmask_t ALL_REQUESTS_IN_USE = MAX_REQUESTS == 32 ? ~0 : (1 << MAX_REQUESTS) - 1; +static constexpr size_t USB_MAX_PACKET_SIZE = + USB_HOST_MAX_PACKET_SIZE; // Max USB packet size (64 for FS, 512 for P4 HS) static constexpr size_t USB_EVENT_QUEUE_SIZE = 32; // Size of event queue between USB task and main loop static constexpr size_t USB_TASK_STACK_SIZE = 4096; // Stack size for USB task (same as ESP-IDF USB examples) static constexpr UBaseType_t USB_TASK_PRIORITY = 5; // Higher priority than main loop (tskIDLE_PRIORITY + 5) diff --git a/esphome/components/usb_host/usb_host_client.cpp b/esphome/components/usb_host/usb_host_client.cpp index c34c7ef67d..4ee8e2ac5e 100644 --- a/esphome/components/usb_host/usb_host_client.cpp +++ b/esphome/components/usb_host/usb_host_client.cpp @@ -217,7 +217,7 @@ void USBClient::setup() { // Pre-allocate USB transfer buffers for all slots at startup // This avoids any dynamic allocation during runtime for (auto &request : this->requests_) { - usb_host_transfer_alloc(64, 0, &request.transfer); + usb_host_transfer_alloc(USB_MAX_PACKET_SIZE, 0, &request.transfer); request.client = this; // Set once, never changes } diff --git a/esphome/components/usb_uart/__init__.py b/esphome/components/usb_uart/__init__.py index d542788fb9..1cf78fdbd5 100644 --- a/esphome/components/usb_uart/__init__.py +++ b/esphome/components/usb_uart/__init__.py @@ -1,7 +1,11 @@ import esphome.codegen as cg from esphome.components.const import CONF_DATA_BITS, CONF_PARITY, CONF_STOP_BITS from esphome.components.uart import CONF_DEBUG_PREFIX, CONF_FLUSH_TIMEOUT, UARTComponent -from esphome.components.usb_host import register_usb_client, usb_device_schema +from esphome.components.usb_host import ( + get_max_packet_size, + register_usb_client, + usb_device_schema, +) import esphome.config_validation as cv from esphome.const import ( CONF_BAUD_RATE, @@ -118,14 +122,14 @@ CONFIG_SCHEMA = cv.ensure_list( async def to_code(config): # The output chunk pool/queue are compile-time-sized templates shared by all # USBUartChannel instances, so use the largest buffer_size across every channel - # of every device. Each chunk is 64 bytes (USB FS MPS); add one extra slot - # because LockFreeQueue is a ring buffer that wastes one entry. + # of every device. Add one extra slot because LockFreeQueue is a ring + # buffer that wastes one entry. max_buffer_size = max( channel[CONF_BUFFER_SIZE] for device in config for channel in device[CONF_CHANNELS] ) - output_chunk_count = max_buffer_size // 64 + 1 + output_chunk_count = max(max_buffer_size // get_max_packet_size(), 2) + 1 cg.add_define("USB_UART_OUTPUT_CHUNK_COUNT", output_chunk_count) for device in config: diff --git a/esphome/components/usb_uart/usb_uart.cpp b/esphome/components/usb_uart/usb_uart.cpp index 30ec61fdc4..e3bf5e40bc 100644 --- a/esphome/components/usb_uart/usb_uart.cpp +++ b/esphome/components/usb_uart/usb_uart.cpp @@ -157,7 +157,7 @@ void USBUartChannel::write_array(const uint8_t *data, size_t len) { ESP_LOGE(TAG, "Output pool full - lost %zu bytes", len); break; } - size_t chunk_len = std::min(len, UsbOutputChunk::MAX_CHUNK_SIZE); + uint16_t chunk_len = std::min(len, UsbOutputChunk::MAX_CHUNK_SIZE); memcpy(chunk->data, data, chunk_len); chunk->length = static_cast(chunk_len); // Push always succeeds: pool is sized to queue capacity (SIZE-1), so if @@ -222,7 +222,7 @@ void USBUartComponent::loop() { #ifdef USE_UART_DEBUGGER if (channel->debug_) { - char buf[4 + format_hex_pretty_size(UsbDataChunk::MAX_CHUNK_SIZE)]; // "<<< " + hex + char buf[4 + format_hex_pretty_size(usb_host::USB_MAX_PACKET_SIZE)]; // "<<< " + hex memcpy(buf, "<<< ", 4); format_hex_pretty_to(buf + 4, sizeof(buf) - 4, chunk->data, chunk->length, ','); ESP_LOGD(TAG, "%s%s", channel->debug_prefix_.c_str(), buf); @@ -377,7 +377,7 @@ void USBUartComponent::start_output(USBUartChannel *channel) { this->start_output(channel); }; - const uint8_t len = chunk->length; + const auto len = chunk->length; if (!this->transfer_out(ep->bEndpointAddress, callback, chunk->data, len)) { // Transfer submission failed — return chunk and release flag so callers can retry. channel->output_pool_.release(chunk); @@ -394,10 +394,10 @@ void USBUartComponent::start_output(USBUartChannel *channel) { static void fix_mps(const usb_ep_desc_t *ep) { if (ep != nullptr) { auto *ep_mutable = const_cast(ep); - if (ep->wMaxPacketSize > 64) { - ESP_LOGW(TAG, "Corrected MPS of EP 0x%02X from %u to 64", static_cast(ep->bEndpointAddress & 0xFF), - ep->wMaxPacketSize); - ep_mutable->wMaxPacketSize = 64; + if (ep->wMaxPacketSize > usb_host::USB_MAX_PACKET_SIZE) { + ESP_LOGW(TAG, "Corrected MPS of EP 0x%02X from %u to %u", static_cast(ep->bEndpointAddress & 0xFF), + ep->wMaxPacketSize, usb_host::USB_MAX_PACKET_SIZE); + ep_mutable->wMaxPacketSize = usb_host::USB_MAX_PACKET_SIZE; } } } diff --git a/esphome/components/usb_uart/usb_uart.h b/esphome/components/usb_uart/usb_uart.h index f9648b795b..e88c41c0cb 100644 --- a/esphome/components/usb_uart/usb_uart.h +++ b/esphome/components/usb_uart/usb_uart.h @@ -106,20 +106,19 @@ class RingBuffer { // Structure for queuing received USB data chunks struct UsbDataChunk { - static constexpr size_t MAX_CHUNK_SIZE = 64; // USB packet size - uint8_t data[MAX_CHUNK_SIZE]; - uint8_t length; // Max 64 bytes, so uint8_t is sufficient + uint8_t data[usb_host::USB_MAX_PACKET_SIZE]; + uint16_t length; USBUartChannel *channel; // Required for EventPool - no cleanup needed for POD types void release() {} }; -// Structure for queuing outgoing USB data chunks (one per USB FS packet) +// Structure for queuing outgoing USB data chunks (one per USB packet) struct UsbOutputChunk { - static constexpr size_t MAX_CHUNK_SIZE = 64; // USB FS MPS + static constexpr size_t MAX_CHUNK_SIZE = usb_host::USB_MAX_PACKET_SIZE; uint8_t data[MAX_CHUNK_SIZE]; - uint8_t length; + uint16_t length; // Required for EventPool - no cleanup needed for POD types void release() {} From 3dd60c57134eea138ed0d78d84ff5d2465670e29 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Fri, 1 May 2026 08:55:08 -0400 Subject: [PATCH 2/5] [core] Support allocating ring buffer in internal memory (#16187) --- esphome/core/ring_buffer.cpp | 13 +++++++------ esphome/core/ring_buffer.h | 7 ++++++- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/esphome/core/ring_buffer.cpp b/esphome/core/ring_buffer.cpp index 6a2232599f..2e0802eceb 100644 --- a/esphome/core/ring_buffer.cpp +++ b/esphome/core/ring_buffer.cpp @@ -1,11 +1,9 @@ #include "ring_buffer.h" -#include "esphome/core/helpers.h" -#include "esphome/core/log.h" - #ifdef USE_ESP32 -#include "helpers.h" +#include "esphome/core/helpers.h" +#include "esphome/core/log.h" namespace esphome { @@ -19,12 +17,15 @@ RingBuffer::~RingBuffer() { } } -std::unique_ptr RingBuffer::create(size_t len) { +std::unique_ptr RingBuffer::create(size_t len, MemoryPreference preference) { std::unique_ptr rb = make_unique(); rb->size_ = len; - RAMAllocator allocator; + const uint8_t type = (preference == MemoryPreference::INTERNAL_FIRST) ? RAMAllocator::PREFER_INTERNAL + : RAMAllocator::NONE; + + RAMAllocator allocator(type); rb->storage_ = allocator.allocate(rb->size_); if (rb->storage_ == nullptr) { return nullptr; diff --git a/esphome/core/ring_buffer.h b/esphome/core/ring_buffer.h index 98a273781f..4acd07d5b0 100644 --- a/esphome/core/ring_buffer.h +++ b/esphome/core/ring_buffer.h @@ -80,7 +80,12 @@ class RingBuffer { */ BaseType_t reset(); - static std::unique_ptr create(size_t len); + enum class MemoryPreference { + EXTERNAL_FIRST, // External RAM preferred, fall back to internal (default) + INTERNAL_FIRST, // Internal RAM preferred, fall back to external + }; + + static std::unique_ptr create(size_t len, MemoryPreference preference = MemoryPreference::EXTERNAL_FIRST); protected: /// @brief Discards data from the ring buffer. From 58cb7effd45323c28c679036b9574d52d0d85e95 Mon Sep 17 00:00:00 2001 From: Mat931 <49403702+Mat931@users.noreply.github.com> Date: Fri, 1 May 2026 15:40:14 +0000 Subject: [PATCH 3/5] [ota] Add extended OTA protocol (#16164) Co-authored-by: J. Nick Koston --- esphome/__main__.py | 11 +- .../components/esphome/ota/ota_esphome.cpp | 50 +++- esphome/components/esphome/ota/ota_esphome.h | 4 +- esphome/components/ota/ota_backend.h | 8 + esphome/espota2.py | 220 ++++++++++------ tests/unit_tests/test_espota2.py | 244 +++++++++++++++++- tests/unit_tests/test_main.py | 23 +- 7 files changed, 456 insertions(+), 104 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 781bcd6288..e7ce36ae2d 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1125,15 +1125,16 @@ def upload_program( remote_port = int(ota_conf[CONF_PORT]) password = ota_conf.get(CONF_PASSWORD) - if getattr(args, "file", None) is not None: - binary = Path(args.file) - else: - binary = CORE.firmware_bin # Resolve MQTT magic strings to actual IP addresses network_devices = _resolve_network_devices(devices, config, args) - return espota2.run_ota(network_devices, remote_port, password, binary) + binary = CORE.firmware_bin + ota_type = espota2.OTA_TYPE_UPDATE_APP + if getattr(args, "file", None) is not None: + binary = Path(args.file) + + return espota2.run_ota(network_devices, remote_port, password, binary, ota_type) def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int | None: diff --git a/esphome/components/esphome/ota/ota_esphome.cpp b/esphome/components/esphome/ota/ota_esphome.cpp index be771eb689..955b4dc96f 100644 --- a/esphome/components/esphome/ota/ota_esphome.cpp +++ b/esphome/components/esphome/ota/ota_esphome.cpp @@ -114,8 +114,10 @@ void ESPHomeOTAComponent::loop() { this->handle_handshake_(); } -static const uint8_t FEATURE_SUPPORTS_COMPRESSION = 0x01; -static const uint8_t FEATURE_SUPPORTS_SHA256_AUTH = 0x02; +static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01; +static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02; +static constexpr uint8_t CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04; +static constexpr uint8_t SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01; void ESPHomeOTAComponent::handle_handshake_() { /// Handle the OTA handshake and authentication. @@ -201,16 +203,30 @@ void ESPHomeOTAComponent::handle_handshake_() { this->ota_features_ = this->handshake_buf_[0]; ESP_LOGV(TAG, "Features: 0x%02X", this->ota_features_); this->transition_ota_state_(OTAState::FEATURE_ACK); - this->handshake_buf_[0] = - ((this->ota_features_ & FEATURE_SUPPORTS_COMPRESSION) != 0 && this->backend_->supports_compression()) - ? ota::OTA_RESPONSE_SUPPORTS_COMPRESSION - : ota::OTA_RESPONSE_HEADER_OK; + + const bool supports_compression = + (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_COMPRESSION) != 0 && this->backend_->supports_compression(); + + // Compose the feature-ack response. When the client negotiates the extended protocol we emit + // a 2-byte response (marker + server feature flags); otherwise we emit the single-byte + // legacy response. + this->extended_proto_ = (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL) != 0; + if (this->extended_proto_) { + static_assert(HANDSHAKE_BUF_SIZE >= 2, "handshake_buf_ must hold the 2-byte extended-protocol feature ack"); + this->handshake_buf_[0] = ota::OTA_RESPONSE_FEATURE_FLAGS; + this->handshake_buf_[1] = (supports_compression ? SERVER_FEATURE_SUPPORTS_COMPRESSION : 0); + } else { + this->handshake_buf_[0] = + supports_compression ? ota::OTA_RESPONSE_SUPPORTS_COMPRESSION : ota::OTA_RESPONSE_HEADER_OK; + } [[fallthrough]]; } case OTAState::FEATURE_ACK: { - // Acknowledge header - 1 byte - if (!this->try_write_(1, LOG_STR("ack feature"))) { + static constexpr size_t STANDARD_PROTO_ACK_SIZE = 1; + static constexpr size_t EXTENDED_PROTO_ACK_SIZE = 2; + const size_t ack_size = this->extended_proto_ ? EXTENDED_PROTO_ACK_SIZE : STANDARD_PROTO_ACK_SIZE; + if (!this->try_write_(ack_size, LOG_STR("ack feature"))) { return; } #ifdef USE_OTA_PASSWORD @@ -296,6 +312,7 @@ void ESPHomeOTAComponent::handle_data_() { uint8_t buf[OTA_BUFFER_SIZE]; char *sbuf = reinterpret_cast(buf); size_t ota_size; + ota::OTAType ota_type = ota::OTA_TYPE_UPDATE_APP; #if USE_OTA_VERSION == 2 size_t size_acknowledged = 0; #endif @@ -311,6 +328,16 @@ void ESPHomeOTAComponent::handle_data_() { // Acknowledge auth OK - 1 byte this->write_byte_(ota::OTA_RESPONSE_AUTH_OK); + if (this->extended_proto_) { + // Read ota type, 1 byte + if (!this->readall_(buf, 1)) { + this->log_read_error_(LOG_STR("OTA type")); + goto error; // NOLINT(cppcoreguidelines-avoid-goto) + } + ota_type = static_cast(buf[0]); + } + ESP_LOGV(TAG, "OTA type is 0x%02x", ota_type); + // Read size, 4 bytes MSB first if (!this->readall_(buf, 4)) { this->log_read_error_(LOG_STR("size")); @@ -320,6 +347,11 @@ void ESPHomeOTAComponent::handle_data_() { (static_cast(buf[2]) << 8) | buf[3]; ESP_LOGV(TAG, "Size is %u bytes", ota_size); + if (ota_type != ota::OTA_TYPE_UPDATE_APP) { + error_code = ota::OTA_RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE; + goto error; // NOLINT(cppcoreguidelines-avoid-goto) + } + // Now that we've passed authentication and are actually // starting the update, set the warning status and notify // listeners. This ensures that port scanners do not @@ -616,7 +648,7 @@ void ESPHomeOTAComponent::yield_and_feed_watchdog_() { void ESPHomeOTAComponent::log_auth_warning_(const LogString *msg) { ESP_LOGW(TAG, "Auth: %s", LOG_STR_ARG(msg)); } bool ESPHomeOTAComponent::select_auth_type_() { - bool client_supports_sha256 = (this->ota_features_ & FEATURE_SUPPORTS_SHA256_AUTH) != 0; + bool client_supports_sha256 = (this->ota_features_ & CLIENT_FEATURE_SUPPORTS_SHA256_AUTH) != 0; // Require SHA256 if (!client_supports_sha256) { diff --git a/esphome/components/esphome/ota/ota_esphome.h b/esphome/components/esphome/ota/ota_esphome.h index 53288fc000..5043bc33ef 100644 --- a/esphome/components/esphome/ota/ota_esphome.h +++ b/esphome/components/esphome/ota/ota_esphome.h @@ -97,8 +97,9 @@ class ESPHomeOTAComponent final : public ota::OTAComponent { ota::OTABackendPtr backend_; uint32_t client_connect_time_{0}; + static constexpr size_t HANDSHAKE_BUF_SIZE = 5; uint16_t port_; - uint8_t handshake_buf_[5]; + uint8_t handshake_buf_[HANDSHAKE_BUF_SIZE]; OTAState ota_state_{OTAState::IDLE}; uint8_t handshake_buf_pos_{0}; uint8_t ota_features_{0}; @@ -106,6 +107,7 @@ class ESPHomeOTAComponent final : public ota::OTAComponent { uint8_t auth_buf_pos_{0}; uint8_t auth_type_{0}; // Store auth type to know which hasher to use #endif // USE_OTA_PASSWORD + bool extended_proto_{false}; }; } // namespace esphome diff --git a/esphome/components/ota/ota_backend.h b/esphome/components/ota/ota_backend.h index bd9c481901..7e7b0f6523 100644 --- a/esphome/components/ota/ota_backend.h +++ b/esphome/components/ota/ota_backend.h @@ -4,6 +4,8 @@ #include "esphome/core/defines.h" #include "esphome/core/helpers.h" +#include + #ifdef USE_OTA_STATE_LISTENER #include #endif @@ -23,6 +25,7 @@ enum OTAResponseTypes { OTA_RESPONSE_UPDATE_END_OK = 0x45, OTA_RESPONSE_SUPPORTS_COMPRESSION = 0x46, OTA_RESPONSE_CHUNK_OK = 0x47, + OTA_RESPONSE_FEATURE_FLAGS = 0x48, OTA_RESPONSE_ERROR_MAGIC = 0x80, OTA_RESPONSE_ERROR_UPDATE_PREPARE = 0x81, @@ -38,6 +41,7 @@ enum OTAResponseTypes { OTA_RESPONSE_ERROR_MD5_MISMATCH = 0x8B, OTA_RESPONSE_ERROR_RP2040_NOT_ENOUGH_SPACE = 0x8C, OTA_RESPONSE_ERROR_SIGNATURE_INVALID = 0x8D, + OTA_RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE = 0x8E, OTA_RESPONSE_ERROR_UNKNOWN = 0xFF, }; @@ -49,6 +53,10 @@ enum OTAState { OTA_ERROR, }; +enum OTAType : uint8_t { + OTA_TYPE_UPDATE_APP = 0x00, +}; + /** Listener interface for OTA state changes. * * Components can implement this interface to receive OTA state updates diff --git a/esphome/espota2.py b/esphome/espota2.py index 39f51e02e9..f4c0c73589 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -15,6 +15,8 @@ from typing import Any from esphome.core import EsphomeError from esphome.helpers import ProgressBar, resolve_ip_address +OTA_TYPE_UPDATE_APP = 0x00 + RESPONSE_OK = 0x00 RESPONSE_REQUEST_AUTH = 0x01 RESPONSE_REQUEST_SHA256_AUTH = 0x02 @@ -27,6 +29,7 @@ RESPONSE_RECEIVE_OK = 0x44 RESPONSE_UPDATE_END_OK = 0x45 RESPONSE_SUPPORTS_COMPRESSION = 0x46 RESPONSE_CHUNK_OK = 0x47 +RESPONSE_FEATURE_FLAGS = 0x48 RESPONSE_ERROR_MAGIC = 0x80 RESPONSE_ERROR_UPDATE_PREPARE = 0x81 @@ -42,6 +45,7 @@ RESPONSE_ERROR_NO_UPDATE_PARTITION = 0x8A RESPONSE_ERROR_MD5_MISMATCH = 0x8B RESPONSE_ERROR_RP2040_NOT_ENOUGH_SPACE = 0x8C RESPONSE_ERROR_SIGNATURE_INVALID = 0x8D +RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE = 0x8E RESPONSE_ERROR_UNKNOWN = 0xFF OTA_VERSION_1_0 = 1 @@ -49,9 +53,16 @@ OTA_VERSION_2_0 = 2 MAGIC_BYTES = [0x6C, 0x26, 0xF7, 0x5C, 0x45] -FEATURE_SUPPORTS_COMPRESSION = 0x01 -FEATURE_SUPPORTS_SHA256_AUTH = 0x02 +CLIENT_FEATURE_SUPPORTS_COMPRESSION = 0x01 +CLIENT_FEATURE_SUPPORTS_SHA256_AUTH = 0x02 +CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL = 0x04 +SERVER_FEATURE_SUPPORTS_COMPRESSION = 0x01 +SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS = 0x02 +# OTA types this client knows how to send. Future PRs that add bootloader/partition +# updates extend this set. Anything outside the set is rejected up front so callers +# of perform_ota/run_ota get a clear error instead of a post-auth 0x8E from the device. +_SUPPORTED_OTA_TYPES: frozenset[int] = frozenset({OTA_TYPE_UPDATE_APP}) UPLOAD_BLOCK_SIZE = 8192 UPLOAD_BUFFER_SIZE = UPLOAD_BLOCK_SIZE * 8 @@ -64,6 +75,62 @@ _AUTH_METHODS: dict[int, tuple[Callable[..., Any], int, str]] = { RESPONSE_REQUEST_AUTH: (hashlib.md5, 32, "MD5"), } +# Error response code -> human-readable message (without the "Error: " prefix; check_error() +# prepends it uniformly). Looked up by check_error() to translate a single byte from the device +# into an OTAError. Add new error codes here rather than extending the if-chain in check_error(). +_ERROR_MESSAGES: dict[int, str] = { + RESPONSE_ERROR_MAGIC: "Invalid magic byte", + RESPONSE_ERROR_UPDATE_PREPARE: ( + "Couldn't prepare flash memory for update. Is the binary too big? " + "Please try restarting the ESP." + ), + RESPONSE_ERROR_AUTH_INVALID: "Authentication invalid. Is the password correct?", + RESPONSE_ERROR_WRITING_FLASH: ( + "Writing OTA data to flash memory failed. See USB logs for more information." + ), + RESPONSE_ERROR_UPDATE_END: ( + "Finishing update failed. See the MQTT/USB logs for more information." + ), + RESPONSE_ERROR_INVALID_BOOTSTRAPPING: ( + "Please press the reset button on the ESP. A manual reset is " + "required on the first OTA-Update after flashing via USB." + ), + RESPONSE_ERROR_WRONG_CURRENT_FLASH_CONFIG: ( + "ESP has been flashed with wrong flash size. Please choose the " + "correct 'board' option (esp01_1m always works) and then flash over USB." + ), + RESPONSE_ERROR_WRONG_NEW_FLASH_CONFIG: ( + "ESP does not have the requested flash size (wrong board). Please " + "choose the correct 'board' option (esp01_1m always works) and try " + "uploading again." + ), + RESPONSE_ERROR_ESP8266_NOT_ENOUGH_SPACE: ( + "ESP does not have enough space to store OTA file. Please try " + "flashing a minimal firmware (remove everything except ota)" + ), + RESPONSE_ERROR_ESP32_NOT_ENOUGH_SPACE: ( + "The OTA partition on the ESP is too small. ESPHome needs to resize " + "this partition, please flash over USB." + ), + RESPONSE_ERROR_NO_UPDATE_PARTITION: ( + "The OTA partition on the ESP couldn't be found. ESPHome needs to " + "create this partition, please flash over USB." + ), + RESPONSE_ERROR_MD5_MISMATCH: ( + "Application MD5 code mismatch. Please try again " + "or flash over USB with a good quality cable." + ), + RESPONSE_ERROR_SIGNATURE_INVALID: ( + "Firmware signature verification failed. The firmware was not signed " + "with the correct key. Ensure the signing key matches the one used to build " + "the firmware currently running on the device." + ), + RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE: ( + "The requested OTA type is not supported by the device." + ), + RESPONSE_ERROR_UNKNOWN: "Unknown error from ESP", +} + class OTAError(EsphomeError): pass @@ -130,8 +197,10 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None :param expect: Expected response code(s), None to skip validation. :raises OTAError: If an error code is detected or response doesn't match expected. """ - if expect is None: - return + # Detect device errors and connection-closed cases regardless of `expect`. If we + # only ran these checks when expect was set, error bytes returned during + # accept-any-response reads (e.g. feature negotiation, auth nonces) would be + # silently passed through and surface later as cryptic decode/timeout failures. if not data: raise OTAError( "Error: Device closed connection without responding. " @@ -139,69 +208,11 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None "a network issue, or the connection was interrupted." ) dat = data[0] - if dat == RESPONSE_ERROR_MAGIC: - raise OTAError("Error: Invalid magic byte") - if dat == RESPONSE_ERROR_UPDATE_PREPARE: - raise OTAError( - "Error: Couldn't prepare flash memory for update. Is the binary too big? " - "Please try restarting the ESP." - ) - if dat == RESPONSE_ERROR_AUTH_INVALID: - raise OTAError("Error: Authentication invalid. Is the password correct?") - if dat == RESPONSE_ERROR_WRITING_FLASH: - raise OTAError( - "Error: Writing OTA data to flash memory failed. See USB logs for more " - "information." - ) - if dat == RESPONSE_ERROR_UPDATE_END: - raise OTAError( - "Error: Finishing update failed. See the MQTT/USB logs for more " - "information." - ) - if dat == RESPONSE_ERROR_INVALID_BOOTSTRAPPING: - raise OTAError( - "Error: Please press the reset button on the ESP. A manual reset is " - "required on the first OTA-Update after flashing via USB." - ) - if dat == RESPONSE_ERROR_WRONG_CURRENT_FLASH_CONFIG: - raise OTAError( - "Error: ESP has been flashed with wrong flash size. Please choose the " - "correct 'board' option (esp01_1m always works) and then flash over USB." - ) - if dat == RESPONSE_ERROR_WRONG_NEW_FLASH_CONFIG: - raise OTAError( - "Error: ESP does not have the requested flash size (wrong board). Please " - "choose the correct 'board' option (esp01_1m always works) and try " - "uploading again." - ) - if dat == RESPONSE_ERROR_ESP8266_NOT_ENOUGH_SPACE: - raise OTAError( - "Error: ESP does not have enough space to store OTA file. Please try " - "flashing a minimal firmware (remove everything except ota)" - ) - if dat == RESPONSE_ERROR_ESP32_NOT_ENOUGH_SPACE: - raise OTAError( - "Error: The OTA partition on the ESP is too small. ESPHome needs to resize " - "this partition, please flash over USB." - ) - if dat == RESPONSE_ERROR_NO_UPDATE_PARTITION: - raise OTAError( - "Error: The OTA partition on the ESP couldn't be found. ESPHome needs to create " - "this partition, please flash over USB." - ) - if dat == RESPONSE_ERROR_MD5_MISMATCH: - raise OTAError( - "Error: Application MD5 code mismatch. Please try again " - "or flash over USB with a good quality cable." - ) - if dat == RESPONSE_ERROR_SIGNATURE_INVALID: - raise OTAError( - "Error: Firmware signature verification failed. The firmware was not signed " - "with the correct key. Ensure the signing key matches the one used to build " - "the firmware currently running on the device." - ) - if dat == RESPONSE_ERROR_UNKNOWN: - raise OTAError("Unknown error from ESP") + error_msg = _ERROR_MESSAGES.get(dat) + if error_msg is not None: + raise OTAError(f"Error: {error_msg}") + if expect is None: + return if not isinstance(expect, (list, tuple)): expect = [expect] if dat not in expect: @@ -232,8 +243,25 @@ def send_check( def perform_ota( - sock: socket.socket, password: str | None, file_handle: io.IOBase, filename: Path + sock: socket.socket, + password: str | None, + file_handle: io.IOBase, + filename: Path, + ota_type: int = OTA_TYPE_UPDATE_APP, ) -> None: + # Validate ota_type up front. It travels as a single byte on the wire, and + # passing an out-of-range value would only surface as a ValueError from + # bytes([ota_type]) deep inside send_check, bypassing OTAError handling. + if not isinstance(ota_type, int) or not 0 <= ota_type <= 0xFF: + raise OTAError( + f"Invalid ota_type {ota_type!r}; expected an integer in range 0-255" + ) + if ota_type not in _SUPPORTED_OTA_TYPES: + supported = ", ".join(f"0x{t:02X}" for t in sorted(_SUPPORTED_OTA_TYPES)) + raise OTAError( + f"Unsupported OTA type 0x{ota_type:02X}; this ESPHome supports: {supported}" + ) + file_contents = file_handle.read() file_size = len(file_contents) _LOGGER.info("Uploading %s (%s bytes)", filename, file_size) @@ -251,7 +279,11 @@ def perform_ota( ) # Features - send both compression and SHA256 auth support - features_to_send = FEATURE_SUPPORTS_COMPRESSION | FEATURE_SUPPORTS_SHA256_AUTH + features_to_send = ( + CLIENT_FEATURE_SUPPORTS_COMPRESSION + | CLIENT_FEATURE_SUPPORTS_SHA256_AUTH + | CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL + ) send_check(sock, features_to_send, "features") features = receive_exactly( sock, @@ -260,7 +292,36 @@ def perform_ota( None, # Accept any response )[0] - if features == RESPONSE_SUPPORTS_COMPRESSION: + extended_proto = False + if features == RESPONSE_FEATURE_FLAGS: + extended_proto = True + features = receive_exactly( + sock, + 1, + "feature flags", + None, # Accept any response + )[0] + elif features == RESPONSE_SUPPORTS_COMPRESSION: + features = SERVER_FEATURE_SUPPORTS_COMPRESSION + else: + features = 0 + + if ota_type != OTA_TYPE_UPDATE_APP: + # Any non-app OTA type requires the extended protocol and the + # partition-access server feature. Reject up front so the user gets + # a clear capability error instead of a post-auth 0x8E from the device. + if not extended_proto: + raise OTAError( + f"Device does not support extended OTA protocol; " + f"OTA type 0x{ota_type:02X} requires it" + ) + if not (features & SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS): + raise OTAError( + f"Device does not support partition access; " + f"OTA type 0x{ota_type:02X} cannot be used" + ) + + if features & SERVER_FEATURE_SUPPORTS_COMPRESSION: upload_contents = gzip.compress(file_contents, compresslevel=9) _LOGGER.info("Compressed to %s bytes", len(upload_contents)) else: @@ -315,6 +376,9 @@ def perform_ota( # Timeout must match device-side OTA_SOCKET_TIMEOUT_DATA to prevent premature failures sock.settimeout(90.0) + if extended_proto: + send_check(sock, ota_type, "ota type") + upload_size = len(upload_contents) upload_size_encoded = [ (upload_size >> 24) & 0xFF, @@ -375,7 +439,11 @@ def perform_ota( def run_ota_impl_( - remote_host: str | list[str], remote_port: int, password: str | None, filename: Path + remote_host: str | list[str], + remote_port: int, + password: str | None, + filename: Path, + ota_type: int = OTA_TYPE_UPDATE_APP, ) -> tuple[int, str | None]: from esphome.core import CORE @@ -413,7 +481,7 @@ def run_ota_impl_( _LOGGER.info("Connected to %s", sa[0]) with open(filename, "rb") as file_handle: try: - perform_ota(sock, password, file_handle, filename) + perform_ota(sock, password, file_handle, filename, ota_type) except OTAError as err: _LOGGER.error(str(err)) return 1, None @@ -428,10 +496,14 @@ def run_ota_impl_( def run_ota( - remote_host: str | list[str], remote_port: int, password: str | None, filename: Path + remote_host: str | list[str], + remote_port: int, + password: str | None, + filename: Path, + ota_type: int = OTA_TYPE_UPDATE_APP, ) -> tuple[int, str | None]: try: - return run_ota_impl_(remote_host, remote_port, password, filename) + return run_ota_impl_(remote_host, remote_port, password, filename, ota_type) except OTAError as err: _LOGGER.error(err) return 1, None diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index 20ba4b1f76..b114f17e6c 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -185,6 +185,14 @@ def test_receive_exactly_socket_error(mock_socket: Mock) -> None: "Error: The OTA partition on the ESP couldn't be found", ), (espota2.RESPONSE_ERROR_MD5_MISMATCH, "Error: Application MD5 code mismatch"), + ( + espota2.RESPONSE_ERROR_SIGNATURE_INVALID, + "Error: Firmware signature verification failed", + ), + ( + espota2.RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE, + "Error: The requested OTA type is not supported by the device", + ), (espota2.RESPONSE_ERROR_UNKNOWN, "Unknown error from ESP"), ], ) @@ -270,12 +278,13 @@ def test_perform_ota_successful_md5_auth( # Verify magic bytes were sent assert mock_socket.sendall.call_args_list[0] == call(bytes(espota2.MAGIC_BYTES)) - # Verify features were sent (compression + SHA256 support) + # Verify features were sent (compression + SHA256 support + extended protocol) assert mock_socket.sendall.call_args_list[1] == call( bytes( [ - espota2.FEATURE_SUPPORTS_COMPRESSION - | espota2.FEATURE_SUPPORTS_SHA256_AUTH + espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION + | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH + | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL ] ) ) @@ -640,12 +649,13 @@ def test_perform_ota_successful_sha256_auth( # Verify magic bytes were sent assert mock_socket.sendall.call_args_list[0] == call(bytes(espota2.MAGIC_BYTES)) - # Verify features were sent (compression + SHA256 support) + # Verify features were sent (compression + SHA256 support + extended protocol) assert mock_socket.sendall.call_args_list[1] == call( bytes( [ - espota2.FEATURE_SUPPORTS_COMPRESSION - | espota2.FEATURE_SUPPORTS_SHA256_AUTH + espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION + | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH + | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL ] ) ) @@ -699,8 +709,9 @@ def test_perform_ota_sha256_fallback_to_md5( assert mock_socket.sendall.call_args_list[1] == call( bytes( [ - espota2.FEATURE_SUPPORTS_COMPRESSION - | espota2.FEATURE_SUPPORTS_SHA256_AUTH + espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION + | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH + | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL ] ) ) @@ -765,3 +776,220 @@ def test_perform_ota_version_differences( # For v2.0, verify more recv calls due to chunk acknowledgments assert mock_socket.recv.call_count == 9 # v2.0 has 9 recv calls (includes chunk OK) + + +@pytest.mark.usefixtures("mock_time") +def test_perform_ota_extended_protocol_app( + mock_socket: Mock, mock_file: io.BytesIO +) -> None: + """Test OTA extended protocol app update.""" + recv_responses = [ + bytes([espota2.RESPONSE_OK]), # First byte of version response + bytes([espota2.OTA_VERSION_2_0]), # Version number + bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Device supports extended protocol + bytes( + [ + espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION + | espota2.SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS + ] + ), # Device feature flags + bytes([espota2.RESPONSE_AUTH_OK]), # No auth required + bytes([espota2.RESPONSE_UPDATE_PREPARE_OK]), # Binary size OK + bytes([espota2.RESPONSE_BIN_MD5_OK]), # MD5 checksum OK + bytes([espota2.RESPONSE_CHUNK_OK]), # Chunk OK + bytes([espota2.RESPONSE_RECEIVE_OK]), # Receive OK + bytes([espota2.RESPONSE_UPDATE_END_OK]), # Update end OK + ] + + mock_socket.recv.side_effect = recv_responses + + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + espota2.OTA_TYPE_UPDATE_APP, + ) + + # Verify magic bytes were sent + assert mock_socket.sendall.call_args_list[0] == call(bytes(espota2.MAGIC_BYTES)) + + # Verify features were sent (compression + SHA256 support + extended protocol) + assert mock_socket.sendall.call_args_list[1] == call( + bytes( + [ + espota2.CLIENT_FEATURE_SUPPORTS_COMPRESSION + | espota2.CLIENT_FEATURE_SUPPORTS_SHA256_AUTH + | espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL + ] + ) + ) + + # Verify ota type was sent + assert mock_socket.sendall.call_args_list[2] == call( + bytes([espota2.OTA_TYPE_UPDATE_APP]) + ) + + +@pytest.mark.usefixtures("mock_time") +def test_perform_ota_device_rejects_with_unsupported_ota_type( + mock_socket: Mock, mock_file: io.BytesIO +) -> None: + """End-to-end: device returns 0x8E after the size byte; perform_ota must + surface the human-readable 'unsupported OTA type' error from the lookup + table in check_error().""" + recv_responses = [ + bytes([espota2.RESPONSE_OK]), # First byte of version response + bytes([espota2.OTA_VERSION_2_0]), # Version number + bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Extended protocol marker + bytes( + [ + espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION + | espota2.SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS + ] + ), # Feature flags + bytes([espota2.RESPONSE_AUTH_OK]), # No auth required + bytes([espota2.RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE]), # Reject at size step + ] + + mock_socket.recv.side_effect = recv_responses + + with pytest.raises( + espota2.OTAError, + match="The requested OTA type is not supported by the device", + ): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + espota2.OTA_TYPE_UPDATE_APP, + ) + + # Verify the client did send the OTA type byte before the size step + assert mock_socket.sendall.call_args_list[2] == call( + bytes([espota2.OTA_TYPE_UPDATE_APP]) + ) + + +@pytest.mark.usefixtures("mock_time") +def test_perform_ota_unsupported_type_rejected_early( + mock_socket: Mock, mock_file: io.BytesIO +) -> None: + """ota_type values not in _SUPPORTED_OTA_TYPES are rejected before any I/O.""" + with pytest.raises(espota2.OTAError, match="Unsupported OTA type 0xFF"): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + 0xFF, + ) + # No bytes should have been transmitted to the device. + mock_socket.sendall.assert_not_called() + + +@pytest.mark.parametrize("bad_type", [-1, 256, 0x10000, "app", None, 1.5]) +def test_perform_ota_rejects_out_of_range_type( + mock_socket: Mock, mock_file: io.BytesIO, bad_type: object +) -> None: + """Out-of-range or non-int ota_type must raise OTAError, not ValueError.""" + with pytest.raises(espota2.OTAError, match="Invalid ota_type"): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + bad_type, # type: ignore[arg-type] + ) + mock_socket.sendall.assert_not_called() + + +@pytest.mark.usefixtures("mock_time") +def test_perform_ota_non_app_type_requires_extended_protocol( + mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch +) -> None: + """Non-app OTA type must fail when device only supports the legacy protocol.""" + monkeypatch.setattr( + espota2, + "_SUPPORTED_OTA_TYPES", + frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}), + ) + recv_responses = [ + bytes([espota2.RESPONSE_OK]), # First byte of version response + bytes([espota2.OTA_VERSION_2_0]), # Version number + bytes([espota2.RESPONSE_HEADER_OK]), # Legacy single-byte feature ack + ] + + mock_socket.recv.side_effect = recv_responses + + with pytest.raises( + espota2.OTAError, match="Device does not support extended OTA protocol" + ): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + 0xFF, + ) + + +@pytest.mark.usefixtures("mock_time") +def test_perform_ota_non_app_type_requires_partition_access( + mock_socket: Mock, mock_file: io.BytesIO, monkeypatch: pytest.MonkeyPatch +) -> None: + """Non-app OTA type must fail when device advertises extended protocol but + not the partition-access feature.""" + monkeypatch.setattr( + espota2, + "_SUPPORTED_OTA_TYPES", + frozenset({espota2.OTA_TYPE_UPDATE_APP, 0xFF}), + ) + recv_responses = [ + bytes([espota2.RESPONSE_OK]), # First byte of version response + bytes([espota2.OTA_VERSION_2_0]), # Version number + bytes([espota2.RESPONSE_FEATURE_FLAGS]), # Extended protocol marker + bytes( + [espota2.SERVER_FEATURE_SUPPORTS_COMPRESSION] + ), # Compression only, no partition access + ] + + mock_socket.recv.side_effect = recv_responses + + with pytest.raises( + espota2.OTAError, match="Device does not support partition access" + ): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + 0xFF, + ) + + +def test_check_error_detects_errors_when_expect_is_none() -> None: + """check_error must surface device error bytes even when expect is None. + + Regression test: previously, receive_exactly(..., expect=None) calls (used + during feature negotiation and nonce reads) silently passed error bytes + through, turning clean device errors into confusing later failures. + """ + with pytest.raises(espota2.OTAError, match="Error: Authentication invalid"): + espota2.check_error([espota2.RESPONSE_ERROR_AUTH_INVALID], None) + + +def test_check_error_detects_empty_when_expect_is_none() -> None: + """Empty data with expect=None must still raise (connection closed).""" + with pytest.raises( + espota2.OTAError, match="Device closed connection without responding" + ): + espota2.check_error([], None) + + +def test_check_error_passes_non_error_when_expect_is_none() -> None: + """Non-error bytes with expect=None must pass through silently.""" + espota2.check_error([espota2.RESPONSE_OK], None) + espota2.check_error([espota2.RESPONSE_HEADER_OK], None) + espota2.check_error([espota2.RESPONSE_FEATURE_FLAGS], None) diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index fb8f206a1d..186d8a9573 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -83,6 +83,7 @@ from esphome.const import ( PLATFORM_RP2040, ) from esphome.core import CORE, EsphomeError +from esphome.espota2 import OTA_TYPE_UPDATE_APP from esphome.util import BootselResult from esphome.zeroconf import _await_discovery, discover_mdns_devices @@ -1593,7 +1594,7 @@ def test_upload_program_ota_success( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.1.100"], 3232, "secret", expected_firmware + ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP ) @@ -1624,7 +1625,7 @@ def test_upload_program_ota_with_file_arg( assert exit_code == 0 assert host == "192.168.1.100" mock_run_ota.assert_called_once_with( - ["192.168.1.100"], 3232, None, Path("custom.bin") + ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP ) @@ -1682,7 +1683,7 @@ def test_upload_program_ota_with_mqtt_resolution( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.1.100"], 3232, None, expected_firmware + ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP ) @@ -1730,7 +1731,7 @@ def test_upload_program_ota_with_mqtt_empty_broker( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.1.50"], 3232, None, expected_firmware + ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP ) # Verify warning was logged assert "MQTT IP discovery failed" in caplog.text @@ -3207,7 +3208,11 @@ def test_upload_program_ota_static_ip_with_mqttip( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.1.100", "192.168.2.50"], 3232, None, expected_firmware + ["192.168.1.100", "192.168.2.50"], + 3232, + None, + expected_firmware, + OTA_TYPE_UPDATE_APP, ) @@ -3250,7 +3255,11 @@ def test_upload_program_ota_multiple_mqttip_resolves_once( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.2.50", "192.168.2.51", "192.168.1.100"], 3232, None, expected_firmware + ["192.168.2.50", "192.168.2.51", "192.168.1.100"], + 3232, + None, + expected_firmware, + OTA_TYPE_UPDATE_APP, ) @@ -3415,7 +3424,7 @@ def test_upload_program_ota_mqtt_timeout_fallback( tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin" ) mock_run_ota.assert_called_once_with( - ["192.168.1.100"], 3232, None, expected_firmware + ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP ) From fcc6f048054a7759ac2ef5c679a5b1ea963d701c Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 1 May 2026 11:56:19 -0500 Subject: [PATCH 4/5] [api] Don't tear down log connection on stack-trace decode failure When 'esphome logs' processes a crash backtrace from the device, process_stacktrace -> _decode_pc -> _run_idedata can raise EsphomeError if the local build dir hasn't been populated (e.g. the device was flashed from a different machine). on_log runs inside an asyncio protocol callback, so the unhandled exception triggers 'Fatal error: protocol.data_received() call failed.', the loop tears the connection down, and ReconnectLogic immediately reconnects. The device replays the same crash trace and we loop forever. Wrap the per-line decode in a helper that swallows EsphomeError so the connection stays up. Also covered with unit tests for the new helper. --- esphome/components/api/client.py | 39 ++++++--- tests/unit_tests/components/api/__init__.py | 0 .../unit_tests/components/api/test_client.py | 80 +++++++++++++++++++ 3 files changed, 110 insertions(+), 9 deletions(-) create mode 100644 tests/unit_tests/components/api/__init__.py create mode 100644 tests/unit_tests/components/api/test_client.py diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 312d937f01..31ae8163da 100644 --- a/esphome/components/api/client.py +++ b/esphome/components/api/client.py @@ -18,7 +18,7 @@ with warnings.catch_warnings(): import contextlib from esphome.const import CONF_KEY, CONF_PORT, __version__ -from esphome.core import CORE +from esphome.core import CORE, EsphomeError from esphome.platformio_api import process_stacktrace from . import CONF_ENCRYPTION @@ -32,6 +32,32 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) +def _process_stacktrace_line( + config: dict[str, Any], + raw_line: str, + backtrace_state: bool, + platform_process_stacktrace: Any | None, +) -> bool: + """Run the stack-trace decoder for a single log line. + + on_log runs inside an asyncio protocol callback; if an exception + escapes, the loop tears the transport down with "Fatal error: + protocol.data_received() call failed." and ReconnectLogic + immediately reconnects, the device replays the same crash trace, + and we loop forever. Stack-trace decoding requires a populated + build dir for the device, which may not exist (e.g. flashed from + another machine); log and continue instead of killing the + connection. + """ + try: + if platform_process_stacktrace: + return platform_process_stacktrace(config, raw_line, backtrace_state) + return process_stacktrace(config, raw_line, backtrace_state=backtrace_state) + except EsphomeError as exc: + _LOGGER.debug("Stack-trace decoding failed: %s", exc) + return False + + async def async_run_logs( config: dict[str, Any], addresses: list[str], @@ -84,14 +110,9 @@ async def async_run_logs( for parsed_msg in parse_log_message(text, timestamp): print(parsed_msg.replace("\033", "\\033") if dashboard else parsed_msg) for raw_line in text.splitlines(): - if platform_process_stacktrace: - backtrace_state = platform_process_stacktrace( - config, raw_line, backtrace_state - ) - else: - backtrace_state = process_stacktrace( - config, raw_line, backtrace_state=backtrace_state - ) + backtrace_state = _process_stacktrace_line( + config, raw_line, backtrace_state, platform_process_stacktrace + ) # Safe to fall back to plaintext here only for this diagnostics use # case: the stream is one-way from device to client, and this code diff --git a/tests/unit_tests/components/api/__init__.py b/tests/unit_tests/components/api/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit_tests/components/api/test_client.py b/tests/unit_tests/components/api/test_client.py new file mode 100644 index 0000000000..9f038b093d --- /dev/null +++ b/tests/unit_tests/components/api/test_client.py @@ -0,0 +1,80 @@ +"""Tests for esphome.components.api.client.""" + +from __future__ import annotations + +from unittest.mock import patch + +from esphome.components.api import client as api_client +from esphome.core import EsphomeError + + +def test_process_stacktrace_line_swallows_esphome_error() -> None: + """A failing stack-trace decode must not propagate. + + on_log runs inside an asyncio protocol callback; if EsphomeError + escapes, the loop reports "Fatal error: protocol.data_received() + call failed.", tears the connection down, and ReconnectLogic loops + forever as the device replays the same crash trace on every + reconnect. + """ + config = {"esphome": {"name": "test"}} + + with patch.object( + api_client, "process_stacktrace", side_effect=EsphomeError("no idedata") + ) as mock_process: + result = api_client._process_stacktrace_line( + config, "PC: 0x4010496e", True, None + ) + + assert mock_process.called + assert result is False + + +def test_process_stacktrace_line_swallows_platform_handler_error() -> None: + """The same protection must apply to the platform-specific handler.""" + config = {"esphome": {"name": "test"}} + + def platform_handler(_config, _line, _state): + raise EsphomeError("no idedata") + + result = api_client._process_stacktrace_line( + config, "PC: 0x4010496e", True, platform_handler + ) + + assert result is False + + +def test_process_stacktrace_line_returns_handler_result() -> None: + """When decoding succeeds, the handler's result is returned unchanged.""" + config = {"esphome": {"name": "test"}} + + with patch.object( + api_client, "process_stacktrace", return_value=True + ) as mock_process: + result = api_client._process_stacktrace_line( + config, "PC: 0x4010496e", False, None + ) + + mock_process.assert_called_once_with( + config, "PC: 0x4010496e", backtrace_state=False + ) + assert result is True + + +def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None: + """The platform handler is preferred over the generic one.""" + config = {"esphome": {"name": "test"}} + calls: list[tuple[object, str, bool]] = [] + + def platform_handler(cfg, line, state): + calls.append((cfg, line, state)) + return True + + with patch.object(api_client, "process_stacktrace") as mock_generic: + result = api_client._process_stacktrace_line( + config, "BT0: 0x4010496e", False, platform_handler + ) + + assert calls == [(config, "BT0: 0x4010496e", False)] + assert mock_generic.called is False + assert result is True From f4d3fb1a1810b707558db7990fbc4d25b231690b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 1 May 2026 12:01:19 -0500 Subject: [PATCH 5/5] [api] Short-circuit log-line decoder after first failure _decode_pc shells out to PlatformIO via _run_idedata; without a populated build dir for the device that subprocess fails for every PC/BT line in a crash dump. Disable decoding after the first EsphomeError (per logs session) and emit a single user-facing warning instead of retrying on every line. Also rename the helper to _LogLineProcessor since it now owns the per-session decode-enabled state, not just the decode call. --- esphome/components/api/client.py | 69 +++++++++++-------- .../unit_tests/components/api/test_client.py | 68 +++++++++++------- 2 files changed, 85 insertions(+), 52 deletions(-) diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 31ae8163da..9b8b42ab31 100644 --- a/esphome/components/api/client.py +++ b/esphome/components/api/client.py @@ -32,30 +32,47 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) -def _process_stacktrace_line( - config: dict[str, Any], - raw_line: str, - backtrace_state: bool, - platform_process_stacktrace: Any | None, -) -> bool: - """Run the stack-trace decoder for a single log line. +class _LogLineProcessor: + """Feeds incoming log lines to the stack-trace decoder. - on_log runs inside an asyncio protocol callback; if an exception - escapes, the loop tears the transport down with "Fatal error: - protocol.data_received() call failed." and ReconnectLogic - immediately reconnects, the device replays the same crash trace, - and we loop forever. Stack-trace decoding requires a populated - build dir for the device, which may not exist (e.g. flashed from - another machine); log and continue instead of killing the - connection. + Two responsibilities beyond just calling the decoder: + 1. Catch EsphomeError. on_log runs inside an asyncio protocol + callback; if an exception escapes, the loop tears the transport + down with "Fatal error: protocol.data_received() call failed." + and ReconnectLogic immediately reconnects, the device replays + the same crash trace, and we loop forever. + 2. Disable decoding after the first failure. _decode_pc shells out + to PlatformIO via _run_idedata, which is expensive; a single + crash dump can contain many PC/BT lines and we don't want to + retry the failing subprocess for each one. """ - try: - if platform_process_stacktrace: - return platform_process_stacktrace(config, raw_line, backtrace_state) - return process_stacktrace(config, raw_line, backtrace_state=backtrace_state) - except EsphomeError as exc: - _LOGGER.debug("Stack-trace decoding failed: %s", exc) - return False + + def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None: + self._config = config + self._platform_handler = platform_handler + self._decode_enabled = True + self.backtrace_state = False + + def process_line(self, raw_line: str) -> None: + if not self._decode_enabled: + return + try: + if self._platform_handler is not None: + self.backtrace_state = self._platform_handler( + self._config, raw_line, self.backtrace_state + ) + else: + self.backtrace_state = process_stacktrace( + self._config, raw_line, backtrace_state=self.backtrace_state + ) + except EsphomeError as exc: + self._decode_enabled = False + self.backtrace_state = False + _LOGGER.warning( + "Crash trace decoding unavailable (%s). Run " + "'esphome compile' for this device to enable PC decoding.", + exc, + ) async def async_run_logs( @@ -87,7 +104,6 @@ async def async_run_logs( addresses=addresses, # Pass all addresses for automatic retry ) dashboard = CORE.dashboard - backtrace_state = False # Try platform-specific stacktrace handler first, fall back to generic platform_process_stacktrace = None @@ -97,9 +113,10 @@ async def async_run_logs( except (AttributeError, ImportError): pass + processor = _LogLineProcessor(config, platform_process_stacktrace) + def on_log(msg: SubscribeLogsResponse) -> None: """Handle a new log message.""" - nonlocal backtrace_state time_ = datetime.now() message: bytes = msg.message text = message.decode("utf8", "backslashreplace") @@ -110,9 +127,7 @@ async def async_run_logs( for parsed_msg in parse_log_message(text, timestamp): print(parsed_msg.replace("\033", "\\033") if dashboard else parsed_msg) for raw_line in text.splitlines(): - backtrace_state = _process_stacktrace_line( - config, raw_line, backtrace_state, platform_process_stacktrace - ) + processor.process_line(raw_line) # Safe to fall back to plaintext here only for this diagnostics use # case: the stream is one-way from device to client, and this code diff --git a/tests/unit_tests/components/api/test_client.py b/tests/unit_tests/components/api/test_client.py index 9f038b093d..1e16c5ebbc 100644 --- a/tests/unit_tests/components/api/test_client.py +++ b/tests/unit_tests/components/api/test_client.py @@ -8,7 +8,7 @@ from esphome.components.api import client as api_client from esphome.core import EsphomeError -def test_process_stacktrace_line_swallows_esphome_error() -> None: +def test_decoder_swallows_esphome_error() -> None: """A failing stack-trace decode must not propagate. on_log runs inside an asyncio protocol callback; if EsphomeError @@ -18,50 +18,68 @@ def test_process_stacktrace_line_swallows_esphome_error() -> None: reconnect. """ config = {"esphome": {"name": "test"}} + processor = api_client._LogLineProcessor(config, None) with patch.object( api_client, "process_stacktrace", side_effect=EsphomeError("no idedata") ) as mock_process: - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", True, None - ) + processor.process_line("PC: 0x4010496e") assert mock_process.called - assert result is False + assert processor.backtrace_state is False -def test_process_stacktrace_line_swallows_platform_handler_error() -> None: +def test_decoder_swallows_platform_handler_error() -> None: """The same protection must apply to the platform-specific handler.""" config = {"esphome": {"name": "test"}} def platform_handler(_config, _line, _state): raise EsphomeError("no idedata") - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", True, platform_handler - ) + processor = api_client._LogLineProcessor(config, platform_handler) + processor.process_line("PC: 0x4010496e") - assert result is False + assert processor.backtrace_state is False -def test_process_stacktrace_line_returns_handler_result() -> None: - """When decoding succeeds, the handler's result is returned unchanged.""" +def test_decoder_short_circuits_after_failure() -> None: + """After one failure, subsequent lines must not retry the decoder. + + _decode_pc shells out to PlatformIO; a crash dump can contain many + PC/BT lines and retrying the failing subprocess for each one would + stall log streaming. + """ config = {"esphome": {"name": "test"}} + processor = api_client._LogLineProcessor(config, None) with patch.object( - api_client, "process_stacktrace", return_value=True + api_client, "process_stacktrace", side_effect=EsphomeError("no idedata") ) as mock_process: - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", False, None - ) + processor.process_line("PC: 0x4010496e") + processor.process_line("BT0: 0x4010496e") + processor.process_line("BT1: 0x401049aa") - mock_process.assert_called_once_with( - config, "PC: 0x4010496e", backtrace_state=False - ) - assert result is True + assert mock_process.call_count == 1 -def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None: +def test_decoder_threads_backtrace_state() -> None: + """When decoding succeeds, backtrace_state is threaded across calls.""" + config = {"esphome": {"name": "test"}} + processor = api_client._LogLineProcessor(config, None) + + with patch.object( + api_client, "process_stacktrace", side_effect=[True, False] + ) as mock_process: + processor.process_line(">>>stack>>>") + assert processor.backtrace_state is True + processor.process_line("<< None: """The platform handler is preferred over the generic one.""" config = {"esphome": {"name": "test"}} calls: list[tuple[object, str, bool]] = [] @@ -70,11 +88,11 @@ def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None: calls.append((cfg, line, state)) return True + processor = api_client._LogLineProcessor(config, platform_handler) + with patch.object(api_client, "process_stacktrace") as mock_generic: - result = api_client._process_stacktrace_line( - config, "BT0: 0x4010496e", False, platform_handler - ) + processor.process_line("BT0: 0x4010496e") assert calls == [(config, "BT0: 0x4010496e", False)] assert mock_generic.called is False - assert result is True + assert processor.backtrace_state is True