diff --git a/esphome/espota2.py b/esphome/espota2.py index ff94417f00f..f4c0c73589a 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -59,6 +59,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 @@ -192,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. " @@ -204,6 +211,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: @@ -240,6 +249,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) diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index 23802cff14e..b114f17e6c4 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -873,10 +873,48 @@ def test_perform_ota_device_rejects_with_unsupported_ota_type( @pytest.mark.usefixtures("mock_time") -def test_perform_ota_non_app_type_requires_extended_protocol( +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 @@ -893,16 +931,21 @@ def test_perform_ota_non_app_type_requires_extended_protocol( "testpass", mock_file, "test.bin", - 255, + 0xFF, ) @pytest.mark.usefixtures("mock_time") def test_perform_ota_non_app_type_requires_partition_access( - mock_socket: Mock, mock_file: io.BytesIO + 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 @@ -922,5 +965,31 @@ def test_perform_ota_non_app_type_requires_partition_access( "testpass", mock_file, "test.bin", - 255, + 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)