diff --git a/esphome/components/esphome/ota/ota_esphome.cpp b/esphome/components/esphome/ota/ota_esphome.cpp index ca5a382875..4261b4c91e 100644 --- a/esphome/components/esphome/ota/ota_esphome.cpp +++ b/esphome/components/esphome/ota/ota_esphome.cpp @@ -242,6 +242,7 @@ void ESPHomeOTAComponent::handle_handshake_() { // 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); #ifdef USE_OTA_PARTITIONS @@ -379,6 +380,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 diff --git a/esphome/components/esphome/ota/ota_esphome.h b/esphome/components/esphome/ota/ota_esphome.h index f612451ab0..0431bd98e0 100644 --- a/esphome/components/esphome/ota/ota_esphome.h +++ b/esphome/components/esphome/ota/ota_esphome.h @@ -98,8 +98,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}; @@ -107,6 +108,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/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() {} 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. diff --git a/esphome/espota2.py b/esphome/espota2.py index 578ea92ce6..85e24abcb7 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -62,6 +62,11 @@ 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 @@ -204,8 +209,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. " @@ -216,6 +223,8 @@ def check_error(data: list[int] | bytes, expect: int | list[int] | None) -> None 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: @@ -252,6 +261,19 @@ def perform_ota( 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) @@ -296,14 +318,20 @@ def perform_ota( else: features = 0 - if ota_type not in (OTA_TYPE_UPDATE_APP, OTA_TYPE_UPDATE_PARTITION_TABLE): - raise OTAError(f"Unsupported OTA type: 0x{ota_type:02X}") - - if ( - ota_type != OTA_TYPE_UPDATE_APP - and not features & SERVER_FEATURE_SUPPORTS_PARTITION_ACCESS - ): - raise OTAError("Device only supports app updates") + 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) diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index 9ccc459a42..d56f9cb6a5 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -786,6 +786,59 @@ def test_perform_ota_version_differences( 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_successful_partition_table( mock_socket: Mock, mock_file: io.BytesIO @@ -861,3 +914,167 @@ def test_perform_ota_extended_protocol_unsupported( "partitions.bin", espota2.OTA_TYPE_UPDATE_PARTITION_TABLE, ) + + +@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) \ No newline at end of file