diff --git a/esphome/__main__.py b/esphome/__main__.py index ff4b560eb10..174f9c3c363 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1126,15 +1126,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 db7dd136149..66ce7a9be46 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,26 +203,29 @@ 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_) { + 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: { - // No password: switch to blocking before writing so the feature ACK - // byte is flushed to the wire before entering the blocking data phase. - // lwIP may not transmit data written in non-blocking mode when the - // socket immediately switches to blocking and blocks in recv(). -#ifdef USE_OTA_PASSWORD - if (this->password_.empty()) -#endif - { - this->client_->setblocking(true); - } - // 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 @@ -306,6 +311,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 @@ -321,6 +327,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")); @@ -340,6 +356,10 @@ void ESPHomeOTAComponent::handle_data_() { this->notify_state_(ota::OTA_STARTED, 0.0f, 0); #endif + if (ota_type != ota::OTA_TYPE_UPDATE_APP) { + error_code = ota::OTA_RESPONSE_ERROR_UNSUPPORTED_OTA_TYPE; + goto error; // NOLINT(cppcoreguidelines-avoid-goto) + } // This will block for a few seconds as it locks flash error_code = this->backend_->begin(ota_size); if (error_code != ota::OTA_RESPONSE_OK) @@ -626,7 +646,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 53288fc0005..f612451ab03 100644 --- a/esphome/components/esphome/ota/ota_esphome.h +++ b/esphome/components/esphome/ota/ota_esphome.h @@ -91,6 +91,7 @@ class ESPHomeOTAComponent final : public ota::OTAComponent { std::string password_; std::unique_ptr auth_buf_; #endif // USE_OTA_PASSWORD + bool extended_proto_{false}; socket::ListenSocket *server_{nullptr}; std::unique_ptr client_; diff --git a/esphome/components/lvgl/lvgl_esphome.cpp b/esphome/components/lvgl/lvgl_esphome.cpp index 0308e6b783f..eb85faa16cd 100644 --- a/esphome/components/lvgl/lvgl_esphome.cpp +++ b/esphome/components/lvgl/lvgl_esphome.cpp @@ -454,10 +454,12 @@ void LVTouchListener::update(const touchscreen::TouchPoints_t &tpoints) { #ifdef USE_LVGL_METER -int16_t lv_get_needle_angle_for_value(lv_obj_t *obj, int value) { +int16_t lv_get_needle_angle_for_value(lv_obj_t *obj, int32_t value) { auto *scale = lv_obj_get_parent(obj); auto min_value = lv_scale_get_range_min_value(scale); - return ((value - min_value) * lv_scale_get_angle_range(scale) / (lv_scale_get_range_max_value(scale) - min_value) + + auto max_value = lv_scale_get_range_max_value(scale); + value = clamp(value, min_value, max_value); + return ((value - min_value) * lv_scale_get_angle_range(scale) / (max_value - min_value) + lv_scale_get_rotation((scale))) % 360; } diff --git a/esphome/components/lvgl/lvgl_esphome.h b/esphome/components/lvgl/lvgl_esphome.h index 83cf9cc0995..be1f150affe 100644 --- a/esphome/components/lvgl/lvgl_esphome.h +++ b/esphome/components/lvgl/lvgl_esphome.h @@ -112,7 +112,7 @@ inline void lv_animimg_set_src(lv_obj_t *img, std::vector images #endif // USE_LVGL_ANIMIMG #ifdef USE_LVGL_METER -int16_t lv_get_needle_angle_for_value(lv_obj_t *obj, int value); +int16_t lv_get_needle_angle_for_value(lv_obj_t *obj, int32_t value); #endif #ifdef USE_LVGL_GRADIENT diff --git a/esphome/components/ota/ota_backend.h b/esphome/components/ota/ota_backend.h index bd9c4819010..7e7b0f6523c 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 39f51e02e93..e1cc52e7ce2 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,11 @@ 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 UPLOAD_BLOCK_SIZE = 8192 UPLOAD_BUFFER_SIZE = UPLOAD_BLOCK_SIZE * 8 @@ -64,6 +70,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 @@ -139,69 +201,9 @@ 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 not isinstance(expect, (list, tuple)): expect = [expect] if dat not in expect: @@ -232,7 +234,11 @@ 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: file_contents = file_handle.read() file_size = len(file_contents) @@ -251,7 +257,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 +270,24 @@ 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: + raise OTAError(f"Unsupported OTA type: 0x{ota_type:02X}") + + 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 +342,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 +405,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 +447,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 +462,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/esphome/platformio_runner.py b/esphome/platformio_runner.py index 599c9408a49..5b14a725577 100644 --- a/esphome/platformio_runner.py +++ b/esphome/platformio_runner.py @@ -101,7 +101,7 @@ def patch_file_downloader() -> None: FileDownloader.__init__ = patched_init -_IGNORE_LIB_WARNINGS = f"(?:{'|'.join(['Hash', 'Update'])})" +_IGNORE_LIB_WARNINGS = "(?:Hash|Update)" # Regex patterns matched against each line of PlatformIO output. Lines that # match are dropped by RedirectText before they reach the parent process. # Patterns are anchored at the start of the line (RedirectText uses diff --git a/pyproject.toml b/pyproject.toml index dc6785001d3..d16bf2b6255 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -113,6 +113,7 @@ exclude = ['generated'] select = [ "E", # pycodestyle "F", # pyflakes/autoflake + "FLY", # flynt: convert string formatting to f-strings "FURB", # refurb "I", # isort "PERF", # performance diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index 20ba4b1f760..1baf3f5db43 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,80 @@ 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_extended_protocol_unsupported_type( + mock_socket: Mock, mock_file: io.BytesIO +) -> None: + """Test OTA fails when OTA type is unsupported by the client.""" + # Setup socket responses for recv calls + recv_responses = [ + bytes([espota2.RESPONSE_OK]), # First byte of version response + bytes([espota2.OTA_VERSION_2_0]), # Version number + bytes([espota2.RESPONSE_HEADER_OK]), # Features response + ] + + mock_socket.recv.side_effect = recv_responses + + with pytest.raises(espota2.OTAError, match="Unsupported OTA type"): + espota2.perform_ota( + mock_socket, + "testpass", + mock_file, + "test.bin", + 255, + ) diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index fb8f206a1d2..186d8a9573c 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 )