diff --git a/tests/integration/fixtures/api_zero_psk_provisioning_persists_across_restart.yaml b/tests/integration/fixtures/api_zero_psk_provisioning_persists_across_restart.yaml new file mode 100644 index 0000000000..1bb2a43e71 --- /dev/null +++ b/tests/integration/fixtures/api_zero_psk_provisioning_persists_across_restart.yaml @@ -0,0 +1,6 @@ +esphome: + name: zero-psk-provision-test +host: +api: + encryption: +logger: diff --git a/tests/integration/test_api_zero_psk_provisioning.py b/tests/integration/test_api_zero_psk_provisioning.py index bcea2a2471..2468729b39 100644 --- a/tests/integration/test_api_zero_psk_provisioning.py +++ b/tests/integration/test_api_zero_psk_provisioning.py @@ -10,11 +10,19 @@ from __future__ import annotations import asyncio import base64 +import socket from aioesphomeapi import InvalidEncryptionKeyAPIError, RequiresEncryptionAPIError import pytest -from .types import APIClientConnectedFactory, RunCompiledFunction +from .conftest import run_binary_and_wait_for_port +from .const import LOCALHOST +from .types import ( + APIClientConnectedFactory, + CompileFunction, + ConfigWriter, + RunCompiledFunction, +) # The well-known provisioning PSK: base64 of 32 zero bytes ZERO_PSK = base64.b64encode(bytes(32)).decode() @@ -125,3 +133,40 @@ async def test_api_zero_psk_provisioning_plaintext( with pytest.raises(InvalidEncryptionKeyAPIError): async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client: await client.device_info() + + +@pytest.mark.asyncio +async def test_api_zero_psk_provisioning_persists_across_restart( + yaml_config: str, + write_yaml_config: ConfigWriter, + compile_esphome: CompileFunction, + reserved_tcp_port: tuple[int, socket.socket], + api_client_connected: APIClientConnectedFactory, +) -> None: + """A key provisioned over the api is loaded from preferences on the next + boot, so the device comes back requiring that key.""" + port, port_socket = reserved_tcp_port + config_path = await write_yaml_config(yaml_config) + binary_path = await compile_esphome(config_path) + port_socket.close() + + async with ( + run_binary_and_wait_for_port(binary_path, LOCALHOST, port), + api_client_connected(noise_psk=ZERO_PSK) as client, + ): + # The key is saved and synced before the response is sent + assert await client.noise_encryption_set_key(NEW_KEY) is True + + lines: list[str] = [] + async with run_binary_and_wait_for_port( + binary_path, LOCALHOST, port, line_callback=lines.append + ): + async with api_client_connected(noise_psk=NEW_KEY.decode()) as client: + device_info = await client.device_info() + assert device_info.api_encryption_provisionable is False + + with pytest.raises(InvalidEncryptionKeyAPIError): + async with api_client_connected(noise_psk=ZERO_PSK, timeout=5) as client: + await client.device_info() + + assert any("Loaded saved Noise PSK" in line for line in lines)