github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-14 02:02:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

[ota] Add old_key so an upload can retry with the previous key (#19398)

Browse Source
This commit is contained in:
J. Nick Koston
2026-10-12 13:35:27 -10:00
committed by GitHub
parent 72ca565a0e
commit 99a4ae1f51
9 changed files with 384 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/__main__.py
+17
View File
@@ -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
esphome/components/esphome/ota/__init__.py
+23 -12
View File
@@ -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,
}
)
esphome/espota2.py
+69 -27
View File
@@ -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)
tests/component_tests/ota/test_esphome_ota.py
+73 -2
View File
@@ -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)
tests/components/ota/old_key.yaml
+12
View File
@@ -0,0 +1,12 @@
wifi:
ssid: MySSID
password: password1
api:
encryption:
key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
ota:
- platform: esphome
encryption:
old_key: "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
tests/components/ota/validate-old_key.esp8266-ard.yaml
+2
View File
@@ -0,0 +1,2 @@
packages:
ota: !include old_key.yaml
tests/unit_tests/test_espota2.py
+17
View File
@@ -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
tests/unit_tests/test_espota2_noise.py
+88 -7
View File
@@ -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:
tests/unit_tests/test_main.py
+83 -46
View File
@@ -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,
)