From 99a4ae1f518454b7d95da31e14b6a8d2470b2615 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Mon, 12 Oct 2026 13:35:27 -1000 Subject: [PATCH] [ota] Add old_key so an upload can retry with the previous key (#19398) --- esphome/__main__.py | 17 +++ esphome/components/esphome/ota/__init__.py | 35 +++-- esphome/espota2.py | 96 +++++++++---- tests/component_tests/ota/test_esphome_ota.py | 75 +++++++++- tests/components/ota/old_key.yaml | 12 ++ .../ota/validate-old_key.esp8266-ard.yaml | 2 + tests/unit_tests/test_espota2.py | 17 +++ tests/unit_tests/test_espota2_noise.py | 95 ++++++++++++- tests/unit_tests/test_main.py | 129 +++++++++++------- 9 files changed, 384 insertions(+), 94 deletions(-) create mode 100644 tests/components/ota/old_key.yaml create mode 100644 tests/components/ota/validate-old_key.esp8266-ard.yaml diff --git a/esphome/__main__.py b/esphome/__main__.py index 0c164893df2..2b094e67a14 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1376,6 +1376,7 @@ def _upload_via_native_api( # Fail closed: an encryption block whose key did not resolve must never # fall back to a plaintext upload noise_psk = None + old_noise_psk = None plaintext_fallback = False allow_plaintext_upload = False if ota_key := getattr(args, "ota_key", None): @@ -1396,6 +1397,8 @@ def _upload_via_native_api( # Ensure the key is a string, as required by the underlying OTA implementation. # It arrives here as a SensitiveStr which aioesphomeapi rejects. noise_psk = str(noise_psk) + if old_key := encryption_conf.get(espota2.CONF_OLD_KEY): + old_noise_psk = str(old_key) elif api_key := static_encryption_key(config.get(CONF_API) or {}): # Remove before 2027.3.0: the api key is tried, falling back to plaintext noise_psk = str(api_key) @@ -1445,9 +1448,22 @@ def _upload_via_native_api( noise_psk, plaintext_fallback=plaintext_fallback, allow_plaintext_upload=allow_plaintext_upload, + old_noise_psk=old_noise_psk, ) +def _old_key_notice(config: ConfigType) -> None: + """After `run`, whose image was just built from this config; a plain + `upload` may send a stale image that still carries the previous key.""" + from esphome import espota2 + + for item in config.get(CONF_OTA, []): + if item.get(CONF_PLATFORM) == CONF_ESPHOME and ( + item.get(CONF_ENCRYPTION) or {} + ).get(espota2.CONF_OLD_KEY): + _LOGGER.warning(espota2.OLD_KEY_REMOVE_NOTICE) + + def _upload_via_web_server( config: ConfigType, network_devices: list[str], binary: Path ) -> tuple[int, str | None]: @@ -1892,6 +1908,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None: exit_code, successful_device = upload_program(config, args, devices) if exit_code == 0: _LOGGER.info("Successfully uploaded program.") + _old_key_notice(config) else: _LOGGER.warning("Failed to upload to %s", devices) return exit_code diff --git a/esphome/components/esphome/ota/__init__.py b/esphome/components/esphome/ota/__init__.py index dc4fc5b47ed..7a0d03df529 100644 --- a/esphome/components/esphome/ota/__init__.py +++ b/esphome/components/esphome/ota/__init__.py @@ -5,6 +5,7 @@ from esphome.components.noise import ( ENCRYPTION_SCHEMA, new_psk_progmem, static_encryption_key, + validate_encryption_key, ) from esphome.components.ota import ( BASE_OTA_SCHEMA, @@ -32,7 +33,7 @@ from esphome.const import ( ) from esphome.core import CORE, coroutine_with_priority from esphome.coroutine import CoroPriority -from esphome.espota2 import CONF_ALLOW_PLAINTEXT_UPLOAD +from esphome.espota2 import CONF_ALLOW_PLAINTEXT_UPLOAD, CONF_OLD_KEY import esphome.final_validate as fv from esphome.types import ConfigType @@ -99,16 +100,18 @@ def ota_esphome_final_validate(config: ConfigType) -> None: # Encryption blocks conflict only when both pin a key; a bare # `encryption:` (a package/device split) is compatible with a # keyed one, and merge_config yields the keyed result - merged_key = ( - merged_ota_esphome_configs_by_port[conf_port] - .get(CONF_ENCRYPTION, {}) - .get(CONF_KEY) + merged_enc = merged_ota_esphome_configs_by_port[conf_port].get( + CONF_ENCRYPTION, {} ) - other_key = ota_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY) - if merged_key and other_key and merged_key != other_key: - raise cv.Invalid( - f"Found multiple configurations but {CONF_ENCRYPTION} is inconsistent" - ) + other_enc = ota_conf.get(CONF_ENCRYPTION, {}) + for option in (CONF_KEY, CONF_OLD_KEY): + merged_key = merged_enc.get(option) + other_key = other_enc.get(option) + if merged_key and other_key and merged_key != other_key: + raise cv.Invalid( + f"Found multiple configurations but {CONF_ENCRYPTION} " + f"{option} is inconsistent" + ) ports_with_merged_configs.append(conf_port) merged_ota_esphome_configs_by_port[conf_port] = merge_config( @@ -206,7 +209,8 @@ def _resolve_encryption_key(encryption_conf: ConfigType, api_conf: ConfigType) - """Resolve the one encryption key per device into the ota block. An explicit ota key must match the api key, a bare block inherits it, - a runtime provisioned api key cannot be inherited. + a runtime provisioned api key cannot be inherited, and an old_key equal + to the resolved key is a leftover from a finished rotation. """ api_key = api_conf.get(CONF_ENCRYPTION, {}).get(CONF_KEY) if ota_key := encryption_conf.get(CONF_KEY): @@ -229,12 +233,19 @@ def _resolve_encryption_key(encryption_conf: ConfigType, api_conf: ConfigType) - ) else: encryption_conf[CONF_KEY] = api_key + if encryption_conf.get(CONF_OLD_KEY) == encryption_conf[CONF_KEY]: + raise cv.Invalid( + f"'{CONF_OLD_KEY}' is the same as '{CONF_KEY}'; remove '{CONF_OLD_KEY}' " + "once the device runs the new key" + ) # Uploader side options live only on the ota block; the api block keeps the -# shared schema +# shared schema. The firmware is built with `key` alone, `old_key` never +# reaches the device _ENCRYPTION_SCHEMA = ENCRYPTION_SCHEMA.extend( { + cv.Optional(CONF_OLD_KEY): cv.sensitive(validate_encryption_key), cv.Optional(CONF_ALLOW_PLAINTEXT_UPLOAD): cv.boolean, } ) diff --git a/esphome/espota2.py b/esphome/espota2.py index dbdf3233660..f775afe7f66 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -215,14 +215,14 @@ class OTANetworkError(OTAError): """Network-level OTA failure (timeout, reset, closed connection); retrying may succeed.""" -# Remove before 2027.3.0 -class OTAEncryptionFallback(OTAError): - """The encrypted attempt failed and the caller may retry in plaintext.""" +class OTAKeyRejected(OTAError): + """The Noise handshake failed on the key; another key may succeed.""" -# Uploader side option under `ota: encryption:`; the ota component imports the -# name so the upload path never loads the component module +# Uploader side options under `ota: encryption:`; the ota component imports +# the names so the upload path never loads the component module CONF_ALLOW_PLAINTEXT_UPLOAD = "allow_plaintext_upload" +CONF_OLD_KEY = "old_key" ALLOW_PLAINTEXT_UPLOAD_NOTICE = ( f"'{CONF_ALLOW_PLAINTEXT_UPLOAD}' is set; expected once, on the install that " "migrates a device which never encrypted. If this device encrypted before, " @@ -249,20 +249,43 @@ PLAINTEXT_FALLBACK_NOTICE = ( ) -# Remove before 2027.3.0 -class _EncryptionAttempt: - """The key an upload tries and whether it may fall back to plaintext; - a rejected handshake falls back at once, a transport fault only on repeat.""" +OLD_KEY_REMOVE_NOTICE = ( + "Once the device has booted this firmware it runs the current key; remove " + f"'{CONF_OLD_KEY}' from 'ota: encryption:'" +) - def __init__(self, noise_psk: str | None, plaintext_fallback: bool) -> None: + +class _EncryptionAttempt: + """The key an upload tries, the previous key to retry with once, and + (until 2027.3.0) whether it may fall back to plaintext.""" + + def __init__( + self, noise_psk: str | None, plaintext_fallback: bool, old_noise_psk: str | None + ) -> None: self.noise_psk = noise_psk + self.old_noise_psk = old_noise_psk + self.tried_old_key = False self.plaintext_fallback = plaintext_fallback self.handshake_faults = 0 + def retry_with_old_key(self) -> bool: + """Swap in the previous key once; the device may not run the new one yet.""" + if not self.old_noise_psk: + return False + _LOGGER.warning( + "Device rejected the OTA encryption key; retrying with '%s'", CONF_OLD_KEY + ) + self.noise_psk = self.old_noise_psk + self.old_noise_psk = None + self.tried_old_key = True + return True + + # Remove before 2027.3.0 def handshake_fault_falls_back(self) -> bool: self.handshake_faults += 1 return self.plaintext_fallback and self.handshake_faults >= 2 + # Remove before 2027.3.0 def downgrade(self, reason: str) -> None: _LOGGER.warning( "%s. Retrying in plaintext; a device that requires encryption " @@ -459,18 +482,20 @@ class NoiseSocketWrapper: if payload[0] != NOISE_HANDSHAKE_OK: reason = payload[1:].decode("utf-8", "replace") if reason == NOISE_MAC_FAILURE_REASON: - raise OTAError( + raise OTAKeyRejected( "Device rejected the handshake; is the OTA encryption key correct?" ) raise OTAError(f"Device rejected the noise handshake: {reason}") try: self._handshake.read_message(payload[1:]) - except (ValueError, self._invalid_tag) as err: - # InvalidTag is a wrong key; ValueError covers a device sending an - # invalid curve point, which cryptography rejects during the DH - raise OTAError( + except self._invalid_tag as err: + raise OTAKeyRejected( "Noise handshake failed; is the OTA encryption key correct?" ) from err + except ValueError as err: + # An invalid curve point, rejected by cryptography during the DH; + # the device is broken, not the key + raise OTAError(f"Noise handshake failed: {err}") from err self._encrypt, self._decrypt = self._handshake.get_ciphers() def sendall(self, data: bytes) -> None: @@ -652,11 +677,6 @@ def perform_ota( except OTANetworkError as err: # A transport fault: retry encrypted before considering plaintext raise OTAHandshakeNetworkError(str(err)) from err - except OTAError as err: - # Remove before 2027.3.0 - if plaintext_fallback: - raise OTAEncryptionFallback(str(err)) from err - raise _LOGGER.info("Encrypted connection established") if allow_plaintext_upload: _LOGGER.warning(ALLOW_PLAINTEXT_UPLOAD_REMOVE_WARNING) @@ -874,6 +894,7 @@ def run_ota_impl_( noise_psk: str | None = None, plaintext_fallback: bool = False, allow_plaintext_upload: bool = False, + old_noise_psk: str | None = None, ) -> tuple[int, str | None]: from esphome.core import CORE @@ -912,11 +933,12 @@ def run_ota_impl_( total_attempts = len(res) + EXTRA_UPLOAD_ATTEMPTS last_error = "" reached_device = False + reconnect = False # the device closed on purpose; nothing to wait for attempt = 0 - encryption = _EncryptionAttempt(noise_psk, plaintext_fallback) + encryption = _EncryptionAttempt(noise_psk, plaintext_fallback, old_noise_psk) while attempt < total_attempts: af, socktype, _, _, sa = res[attempt % len(res)] - if reached_device or attempt >= len(res): + if not reconnect and (reached_device or attempt >= len(res)): _LOGGER.info( "Retrying in %.0f seconds (attempt %d of %d)...", UPLOAD_RETRY_DELAY, @@ -925,6 +947,7 @@ def run_ota_impl_( ) time.sleep(UPLOAD_RETRY_DELAY) reached_device = False + reconnect = False _LOGGER.info("Connecting to %s port %s...", sa[0], sa[1]) sock = socket.socket(af, socktype) sock.settimeout(20.0) @@ -951,11 +974,22 @@ def run_ota_impl_( encryption.plaintext_fallback, allow_plaintext_upload=allow_plaintext_upload, ) - except OTAEncryptionFallback as err: - # Same address and attempt budget: not a network retry + except OTAKeyRejected as err: + # The device sent a reject and closed on its own, so reconnect + # at once, on the same address and attempt budget last_error = str(err) - encryption.downgrade(last_error) - continue + reconnect = True + if encryption.retry_with_old_key(): + continue + # Remove before 2027.3.0 + if encryption.plaintext_fallback: + encryption.downgrade(last_error) + continue + if encryption.tried_old_key: + raise OTAKeyRejected( + f"The device rejected both 'key' and '{CONF_OLD_KEY}'" + ) from err + raise except OTAHandshakeNetworkError as err: last_error = str(err) if encryption.handshake_fault_falls_back(): @@ -976,7 +1010,13 @@ def run_ota_impl_( _LOGGER.error(str(err)) return 1, None - # Successfully uploaded to sa[0] + if encryption.tried_old_key: + # Kept next to the result: a device that should already run the + # current key but answered to the previous one is worth a look + _LOGGER.warning( + "The device accepted '%s'; it was still running the previous key", + CONF_OLD_KEY, + ) return 0, sa[0] _LOGGER.error("Upload failed after %d attempts: %s", total_attempts, last_error) @@ -992,6 +1032,7 @@ def run_ota( noise_psk: str | None = None, plaintext_fallback: bool = False, allow_plaintext_upload: bool = False, + old_noise_psk: str | None = None, ) -> tuple[int, str | None]: try: return run_ota_impl_( @@ -1003,6 +1044,7 @@ def run_ota( noise_psk, plaintext_fallback, allow_plaintext_upload, + old_noise_psk, ) except OTAError as err: _LOGGER.error(err) diff --git a/tests/component_tests/ota/test_esphome_ota.py b/tests/component_tests/ota/test_esphome_ota.py index 77db1499231..39920ff4363 100644 --- a/tests/component_tests/ota/test_esphome_ota.py +++ b/tests/component_tests/ota/test_esphome_ota.py @@ -30,7 +30,7 @@ from esphome.const import ( CONF_VERSION, ) from esphome.core import CORE, ID -from esphome.espota2 import CONF_ALLOW_PLAINTEXT_UPLOAD +from esphome.espota2 import CONF_ALLOW_PLAINTEXT_UPLOAD, CONF_OLD_KEY import esphome.final_validate as fv @@ -215,6 +215,77 @@ def test_encryption_schema_allow_plaintext_upload() -> None: encryption_schema({CONF_ALLOW_PLAINTEXT_UPLOAD: True}) +def test_encryption_old_key_schema() -> None: + """old_key is validated like key and is an ota only option.""" + conf = _encryption_schema({CONF_KEY: API_KEY, CONF_OLD_KEY: OTHER_KEY}) + assert conf[CONF_OLD_KEY] == OTHER_KEY + with pytest.raises(cv.Invalid, match="base64"): + _encryption_schema({CONF_OLD_KEY: "short"}) + with pytest.raises(cv.Invalid): + encryption_schema({CONF_OLD_KEY: OTHER_KEY}) + + +def test_encryption_old_key_keeps_inherited_key() -> None: + """A bare block with old_key inherits the api key as the current one.""" + full_conf = { + CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}}, + CONF_OTA: [ + _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_OLD_KEY: OTHER_KEY}}) + ], + } + token = fv.full_config.set(full_conf) + try: + ota_esphome_final_validate({}) + updated = fv.full_config.get() + assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY + assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_OLD_KEY] == OTHER_KEY + finally: + fv.full_config.reset(token) + + +def test_encryption_old_key_mismatch_between_merged_configs_rejected() -> None: + """Same-port blocks that pin different old keys raise like differing keys.""" + full_conf = { + CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}}, + CONF_OTA: [ + _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_OLD_KEY: OTHER_KEY}}), + _make_ota_config( + port=3232, + **{ + CONF_ENCRYPTION: { + CONF_OLD_KEY: "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE=" + } + }, + ), + ], + } + token = fv.full_config.set(full_conf) + try: + with pytest.raises(cv.Invalid, match="old_key is inconsistent"): + ota_esphome_final_validate({}) + finally: + fv.full_config.reset(token) + + +def test_encryption_old_key_equal_to_key_rejected() -> None: + """An old_key that matches key is a leftover from a finished rotation.""" + full_conf = { + CONF_API: {}, + CONF_OTA: [ + _make_ota_config( + port=3232, + **{CONF_ENCRYPTION: {CONF_KEY: API_KEY, CONF_OLD_KEY: API_KEY}}, + ) + ], + } + token = fv.full_config.set(full_conf) + try: + with pytest.raises(cv.Invalid, match="'old_key' is the same as 'key'"): + ota_esphome_final_validate({}) + finally: + fv.full_config.reset(token) + + def test_encryption_key_mismatch_between_merged_configs_rejected() -> None: """Same-port configs with different encryption keys raise.""" full_conf = { @@ -225,7 +296,7 @@ def test_encryption_key_mismatch_between_merged_configs_rejected() -> None: } token = fv.full_config.set(full_conf) try: - with pytest.raises(cv.Invalid, match="encryption is inconsistent"): + with pytest.raises(cv.Invalid, match="encryption key is inconsistent"): ota_esphome_final_validate({}) finally: fv.full_config.reset(token) diff --git a/tests/components/ota/old_key.yaml b/tests/components/ota/old_key.yaml new file mode 100644 index 00000000000..8074ca06f6f --- /dev/null +++ b/tests/components/ota/old_key.yaml @@ -0,0 +1,12 @@ +wifi: + ssid: MySSID + password: password1 + +api: + encryption: + key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + +ota: + - platform: esphome + encryption: + old_key: "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA=" diff --git a/tests/components/ota/validate-old_key.esp8266-ard.yaml b/tests/components/ota/validate-old_key.esp8266-ard.yaml new file mode 100644 index 00000000000..f58187f4556 --- /dev/null +++ b/tests/components/ota/validate-old_key.esp8266-ard.yaml @@ -0,0 +1,2 @@ +packages: + ota: !include old_key.yaml diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py index f4c5935059c..6fa6c616b67 100644 --- a/tests/unit_tests/test_espota2.py +++ b/tests/unit_tests/test_espota2.py @@ -799,6 +799,23 @@ def test_run_ota_impl_connection_failed( mock_sleep.assert_called_with(espota2.UPLOAD_RETRY_DELAY) +@pytest.mark.usefixtures("mock_socket_constructor", "mock_resolve_ip") +def test_run_ota_impl_old_key_reconnect_never_waits( + mock_socket: Mock, firmware_file: Path, mock_perform_ota: Mock, mock_sleep: Mock +) -> None: + """The reconnect with old_key skips the retry delay even once the + address list has been cycled by an earlier connect failure.""" + mock_socket.connect.side_effect = [OSError("timed out"), None, None] + mock_perform_ota.side_effect = [espota2.OTAKeyRejected("rejected"), None] + + result_code, _ = espota2.run_ota_impl_( + "test.local", 3232, None, str(firmware_file), noise_psk="a", old_noise_psk="b" + ) + + assert result_code == 0 + assert mock_sleep.call_count == 1 # only the connect failure waited + + @pytest.mark.usefixtures("mock_socket_constructor", "mock_resolve_ip") def test_run_ota_impl_connect_retry_succeeds( mock_socket: Mock, firmware_file: Path, mock_perform_ota: Mock, mock_sleep: Mock diff --git a/tests/unit_tests/test_espota2_noise.py b/tests/unit_tests/test_espota2_noise.py index 60f6b28e076..ca50641309b 100644 --- a/tests/unit_tests/test_espota2_noise.py +++ b/tests/unit_tests/test_espota2_noise.py @@ -70,10 +70,12 @@ class FakeEncryptedDevice(threading.Thread): prologue_features_override: int | None = None, connections: int = 1, drop_handshakes: int = 0, + reject_reason: str | None = None, ) -> None: super().__init__(daemon=True) self.connections = connections self.drop_handshakes = drop_handshakes # hang up mid-handshake this many times + self.reject_reason = reject_reason # refuse every handshake with this reason self.psk = psk self.version = version self.offer_noise = offer_noise @@ -160,6 +162,9 @@ class FakeEncryptedDevice(threading.Thread): if self.drop_handshakes > 0: self.drop_handshakes -= 1 return # a transport fault: the socket closes with no reply + if self.reject_reason is not None: + _send_frame(sock, b"\x01" + self.reject_reason.encode()) + return try: proto.read_message(msg1[1:]) except InvalidTag: @@ -244,8 +249,9 @@ def _run_ota( tmp_path: Path, noise_psk: str, plaintext_fallback: bool = True, + old_noise_psk: str | None = None, ) -> int: - """Drive the retry loop, which is where the plaintext fallback reconnects.""" + """Drive the retry loop, which is where the fallback and the old key reconnect.""" path = tmp_path / "firmware.bin" path.write_bytes(firmware) device.start() @@ -256,10 +262,62 @@ def _run_ota( path, noise_psk=noise_psk, plaintext_fallback=plaintext_fallback, + old_noise_psk=old_noise_psk, ) return rc +THIRD_PSK = base64.b64encode(bytes(range(2, 34))).decode() + + +@pytest.mark.parametrize( + ("device_psk", "old_noise_psk", "expected_rc", "retried"), + [ + # The device still runs the previous key: one reconnect with old_key + (OTHER_PSK, OTHER_PSK, 0, True), + # The device already runs the new key: old_key is never presented + (PSK, OTHER_PSK, 0, False), + # Neither key matches: the retry is spent, then it fails closed + (THIRD_PSK, OTHER_PSK, 1, True), + # No old_key configured: a rejected key is a device error + (OTHER_PSK, None, 1, False), + ], + ids=["old_key_accepted", "key_accepted", "both_rejected", "no_old_key"], +) +def test_old_key_retry( + caplog: pytest.LogCaptureFixture, + tmp_path: Path, + device_psk: str, + old_noise_psk: str | None, + expected_rc: int, + retried: bool, +) -> None: + pytest.importorskip("aioesphomeapi.noise") + firmware = b"firmware" + device = FakeEncryptedDevice(psk=device_psk, connections=2 if retried else 1) + with patch("time.sleep"), caplog.at_level(logging.WARNING): + rc = _run_ota( + device, + firmware, + tmp_path, + PSK, + plaintext_fallback=False, + old_noise_psk=old_noise_psk, + ) + device.join_and_check() + assert rc == expected_rc + assert (device.received == firmware) is (expected_rc == 0) + assert ( + any("retrying with 'old_key'" in r.message for r in caplog.records) is retried + ) + assert any("accepted 'old_key'" in r.message for r in caplog.records) is ( + retried and expected_rc == 0 + ) + assert not any("plaintext" in r.message for r in caplog.records) + if expected_rc == 1 and old_noise_psk is not None: + assert any("rejected both" in r.message for r in caplog.records) + + def test_encrypted_upload_success() -> None: """A full encrypted v2 upload spanning several 8192-byte blocks.""" pytest.importorskip("aioesphomeapi.noise") @@ -495,21 +553,21 @@ def test_handshake_reject_with_other_reason() -> None: def test_handshake_garbage_second_message() -> None: - """A valid-looking point with a garbage MAC fails cleanly.""" + """A valid-looking point with a garbage MAC is a key failure.""" wrapper = _wrapper(_frame(b"\x00" + bytes(range(48)))) with pytest.raises( - espota2.OTAError, match="handshake failed; is the OTA encryption key" + espota2.OTAKeyRejected, match="handshake failed; is the OTA encryption key" ): wrapper.do_handshake() def test_handshake_invalid_curve_point() -> None: - """An all-zero x25519 point is rejected as a clean error, not a crash.""" + """An all-zero x25519 point is a clean error, not a crash, and not a + key failure: it must not spend the old_key retry.""" wrapper = _wrapper(_frame(b"\x00" + bytes(48))) - with pytest.raises( - espota2.OTAError, match="handshake failed; is the OTA encryption key" - ): + with pytest.raises(espota2.OTAError, match="handshake failed: ") as info: wrapper.do_handshake() + assert not isinstance(info.value, espota2.OTAKeyRejected) def test_recv_closed_at_frame_boundary_returns_empty() -> None: @@ -585,6 +643,29 @@ def test_recv_serves_buffered_plaintext_without_new_frame() -> None: wrapper._decrypt.decrypt.assert_called_once() +@pytest.mark.parametrize("plaintext_fallback", [False, True]) +def test_non_key_reject_reason_is_a_device_error( + caplog: pytest.LogCaptureFixture, tmp_path: Path, plaintext_fallback: bool +) -> None: + """Only a key failure spends the old_key retry or, until 2027.3.0, the + plaintext fallback; another reject reason fails at once.""" + pytest.importorskip("aioesphomeapi.noise") + device = FakeEncryptedDevice(reject_reason="Busy") + with patch("time.sleep"), caplog.at_level(logging.WARNING): + rc = _run_ota( + device, + b"firmware", + tmp_path, + PSK, + plaintext_fallback=plaintext_fallback, + old_noise_psk=OTHER_PSK, + ) + device.join_and_check() + assert rc == 1 + assert not any("retrying with 'old_key'" in r.message for r in caplog.records) + assert not any("Retrying in plaintext" in r.message for r in caplog.records) + + def test_bare_block_refuses_a_device_that_cannot_encrypt( caplog: pytest.LogCaptureFixture, tmp_path: Path ) -> None: diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 3b10c9cb3c2..801eda4496a 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -1558,6 +1558,14 @@ class MockArgs: env_ota_key: str | None = None +# The keyword half of every run_ota call; a test overrides the one it is about +RUN_OTA_DEFAULT_KWARGS = { + "plaintext_fallback": False, + "allow_plaintext_upload": False, + "old_noise_psk": None, +} + + def test_upload_program_serial_esp32( mock_upload_using_esptool: Mock, mock_get_port_type: Mock, @@ -2120,32 +2128,39 @@ def test_upload_program_ota_success( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) +@pytest.mark.parametrize( + "old_key", + [None, "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="], + ids=["key", "old_key"], +) def test_upload_program_ota_encryption_key( mock_run_ota: Mock, mock_get_port_type: Mock, tmp_path: Path, + old_key: str | None, + caplog: pytest.LogCaptureFixture, ) -> None: - """The resolved encryption key is passed through to run_ota.""" + """The resolved encryption key, and the previous one when set, are passed + through to run_ota.""" setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path) mock_get_port_type.return_value = "NETWORK" mock_run_ota.return_value = (0, "192.168.1.100") key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + encryption = {CONF_KEY: key} + if old_key is not None: + encryption["old_key"] = old_key config = { CONF_OTA: [ - { - CONF_PLATFORM: CONF_ESPHOME, - CONF_PORT: 3232, - CONF_ENCRYPTION: {CONF_KEY: key}, - } + {CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232, CONF_ENCRYPTION: encryption} ] } - exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"]) + with caplog.at_level(logging.WARNING): + exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"]) assert exit_code == 0 assert host == "192.168.1.100" @@ -2159,9 +2174,10 @@ def test_upload_program_ota_encryption_key( expected_firmware, OTA_TYPE_UPDATE_APP, key, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS | {"old_noise_psk": old_key}, ) + # The notice belongs to run, whose image was just built from this config + assert not any("remove 'old_key'" in r.message for r in caplog.records) def test_upload_program_bare_encryption_block_never_falls_back( @@ -2191,10 +2207,7 @@ def test_upload_program_bare_encryption_block_never_falls_back( with patch("esphome.espota2.run_ota", return_value=(0, "192.168.1.100")) as run_ota: upload_program(config, MockArgs(), ["192.168.1.100"]) assert run_ota.call_args.args[5] == key - assert run_ota.call_args.kwargs == { - "plaintext_fallback": False, - "allow_plaintext_upload": False, - } + assert run_ota.call_args.kwargs == RUN_OTA_DEFAULT_KWARGS def test_upload_program_ota_allow_plaintext_upload( @@ -2223,9 +2236,8 @@ def test_upload_program_ota_allow_plaintext_upload( assert exit_code == 0 assert mock_run_ota.call_args.args[2] == "pw" assert mock_run_ota.call_args.args[5] == key - assert mock_run_ota.call_args.kwargs == { - "plaintext_fallback": False, - "allow_plaintext_upload": True, + assert mock_run_ota.call_args.kwargs == RUN_OTA_DEFAULT_KWARGS | { + "allow_plaintext_upload": True } @@ -2235,6 +2247,7 @@ def test_upload_program_ota_allow_plaintext_upload( { CONF_ENCRYPTION: { CONF_KEY: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=", + "old_key": "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE=", "allow_plaintext_upload": True, } }, @@ -2265,10 +2278,7 @@ def test_upload_program_ota_prompted_key_is_presented( assert exit_code == 0 assert mock_run_ota.call_args.args[5] == old_key - assert mock_run_ota.call_args.kwargs == { - "plaintext_fallback": False, - "allow_plaintext_upload": False, - } + assert mock_run_ota.call_args.kwargs == RUN_OTA_DEFAULT_KWARGS def test_upload_program_serial_ignores_prompted_key( @@ -2320,6 +2330,46 @@ def test_command_run_prompts_for_ota_key_before_compile() -> None: assert calls == ["prompt", "compile"] +@pytest.mark.parametrize( + "old_key", [None, "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="] +) +def test_command_run_says_when_old_key_can_go( + caplog: pytest.LogCaptureFixture, old_key: str | None +) -> None: + """Only run has just built the image from this config, so only run can + say the device will run `key` alone once it boots.""" + encryption = {CONF_KEY: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="} + if old_key is not None: + encryption["old_key"] = old_key + setup_core( + config={ + CONF_OTA: [ + { + CONF_PLATFORM: CONF_ESPHOME, + CONF_PORT: 3232, + CONF_ENCRYPTION: encryption, + } + ] + }, + platform=PLATFORM_ESP32, + ) + args = MockArgs() + args.device = ["dev.local"] + args.no_logs = True + with ( + patch("esphome.__main__.write_cpp", return_value=0), + patch("esphome.__main__.compile_program", return_value=0), + patch("esphome.__main__.choose_upload_log_host", return_value=["dev.local"]), + patch("esphome.__main__.get_serial_ports", return_value=[]), + patch("esphome.__main__.upload_program", return_value=(0, "dev.local")), + caplog.at_level(logging.WARNING), + ): + assert command_run(args, CORE.config) == 0 + assert any("remove 'old_key'" in r.message for r in caplog.records) is ( + old_key is not None + ) + + def test_upload_program_platform_hook_ignores_prompted_key( mock_get_port_type: Mock, tmp_path: Path, @@ -2636,8 +2686,7 @@ def test_upload_program_ota_api_key_opportunistic( expected_firmware, OTA_TYPE_UPDATE_APP, key, - plaintext_fallback=True, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS | {"plaintext_fallback": True}, ) @@ -2666,10 +2715,7 @@ def test_upload_program_ota_no_usable_api_key_stays_plaintext( assert exit_code == 0 assert mock_run_ota.call_args.args[5] is None - assert mock_run_ota.call_args.kwargs == { - "plaintext_fallback": False, - "allow_plaintext_upload": False, - } + assert mock_run_ota.call_args.kwargs == RUN_OTA_DEFAULT_KWARGS def test_upload_program_ota_encryption_without_key_fails_closed( @@ -2728,8 +2774,7 @@ def test_upload_program_ota_with_file_arg( Path("custom.bin"), OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -2785,8 +2830,7 @@ def test_upload_program_ota_partition_table_with_file_arg( partition_file, OTA_TYPE_UPDATE_PARTITION_TABLE, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -2849,8 +2893,7 @@ def test_upload_program_ota_partition_table_mqttip( partition_file, OTA_TYPE_UPDATE_PARTITION_TABLE, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -3039,8 +3082,7 @@ def test_upload_program_ota_bootloader_with_file_arg( bootloader_file, OTA_TYPE_UPDATE_BOOTLOADER, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -3551,8 +3593,7 @@ def test_upload_program_ota_with_mqtt_resolution( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -3608,8 +3649,7 @@ def test_upload_program_ota_with_mqtt_empty_broker( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) # Verify warning was logged assert "MQTT IP discovery failed" in caplog.text @@ -5161,8 +5201,7 @@ def test_upload_program_ota_static_ip_with_mqttip( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -5213,8 +5252,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, ) @@ -5398,8 +5436,7 @@ def test_upload_program_ota_mqtt_timeout_fallback( expected_firmware, OTA_TYPE_UPDATE_APP, None, - plaintext_fallback=False, - allow_plaintext_upload=False, + **RUN_OTA_DEFAULT_KWARGS, )