From 8d4e2bed49eb7cab5a9ef6d9fdb4f5fc88960997 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 18 Aug 2026 11:08:24 -0500 Subject: [PATCH] Move encryption key validation into the noise component --- esphome/components/api/__init__.py | 44 ++++++---------- esphome/components/noise/__init__.py | 51 +++++++++++++++++++ tests/component_tests/noise/__init__.py | 0 .../noise/test_encryption_key.py | 37 ++++++++++++++ 4 files changed, 103 insertions(+), 29 deletions(-) create mode 100644 tests/component_tests/noise/__init__.py create mode 100644 tests/component_tests/noise/test_encryption_key.py diff --git a/esphome/components/api/__init__.py b/esphome/components/api/__init__.py index 81f8d51abe..1ecc292cdd 100644 --- a/esphome/components/api/__init__.py +++ b/esphome/components/api/__init__.py @@ -1,10 +1,18 @@ -import base64 import logging from esphome import automation from esphome.automation import Condition import esphome.codegen as cg from esphome.components.logger import request_log_listener + +# ENCRYPTION_SCHEMA and validate_encryption_key are re-exported for external +# components and downstream consumers that import them from api +from esphome.components.noise import ( # noqa: F401 + ENCRYPTION_SCHEMA, + decode_encryption_key, + encryption_schema, + validate_encryption_key, +) from esphome.config_helpers import get_logger_level import esphome.config_validation as cv from esphome.const import ( @@ -37,6 +45,10 @@ from esphome.core import CORE, ID, CoroPriority, EsphomeError, coroutine_with_pr from esphome.cpp_generator import MockObj, TemplateArgsType from esphome.types import ConfigFragmentType, ConfigType +# Compat alias: downstream consumers (e.g. device-builder) referenced the +# schema by its old private name before it moved to the noise component +_encryption_schema = encryption_schema + _LOGGER = logging.getLogger(__name__) DOMAIN = "api" @@ -134,20 +146,6 @@ def _register_provisioning_source(config: ConfigType) -> ConfigType: return config -def validate_encryption_key(value): - value = cv.string_strict(value) - try: - decoded = base64.b64decode(value, validate=True) - except ValueError as err: - raise cv.Invalid("Invalid key format, please check it's using base64") from err - - if len(decoded) != 32: - raise cv.Invalid("Encryption key must be base64 and 32 bytes long") - - # Return original data for roundtrip conversion - return value - - CONF_SUPPORTS_RESPONSE = "supports_response" # Enum values in api::enums namespace @@ -254,18 +252,6 @@ ACTIONS_SCHEMA = automation.validate_automation( ), ) -ENCRYPTION_SCHEMA = cv.Schema( - { - cv.Optional(CONF_KEY): cv.sensitive(validate_encryption_key), - } -) - - -def _encryption_schema(config): - if config is None: - config = {} - return ENCRYPTION_SCHEMA(config) - def _consume_api_sockets(config: ConfigType) -> ConfigType: """Register socket needs for API component.""" @@ -301,7 +287,7 @@ CONFIG_SCHEMA = cv.All( CONF_SERVICES, group_of_exclusion=CONF_ACTIONS ): ACTIONS_SCHEMA, cv.Exclusive(CONF_ACTIONS, group_of_exclusion=CONF_ACTIONS): ACTIONS_SCHEMA, - cv.Optional(CONF_ENCRYPTION): _encryption_schema, + cv.Optional(CONF_ENCRYPTION): encryption_schema, cv.Optional(CONF_BATCH_DELAY, default="100ms"): cv.All( cv.positive_time_period_milliseconds, cv.Range(max=cv.TimePeriod(milliseconds=65535)), @@ -488,7 +474,7 @@ async def to_code(config: ConfigType) -> None: if (encryption_config := config.get(CONF_ENCRYPTION, None)) is not None: if key := encryption_config.get(CONF_KEY): - decoded = base64.b64decode(key) + decoded = decode_encryption_key(key) cg.add(var.set_noise_psk(list(decoded))) cg.add_define("USE_API_NOISE_PSK_FROM_YAML") else: diff --git a/esphome/components/noise/__init__.py b/esphome/components/noise/__init__.py index 9f4707188a..697931591a 100644 --- a/esphome/components/noise/__init__.py +++ b/esphome/components/noise/__init__.py @@ -1,5 +1,9 @@ +import base64 +import binascii + import esphome.codegen as cg import esphome.config_validation as cv +from esphome.const import CONF_KEY from esphome.core import CORE from esphome.types import ConfigType @@ -10,6 +14,53 @@ noise_ns = cg.esphome_ns.namespace("noise") CONFIG_SCHEMA = cv.Schema({}) +def validate_encryption_key(value: str) -> str: + value = cv.string_strict(value) + try: + decoded = base64.b64decode(value, validate=True) + except ValueError as err: + raise cv.Invalid("Invalid key format, please check it's using base64") from err + + if len(decoded) != 32: + raise cv.Invalid("Encryption key must be base64 and 32 bytes long") + + # Return original data for roundtrip conversion + return value + + +def decode_encryption_key(value: str) -> bytes: + """Decode a base64 encryption key to its 32 raw bytes. + + a2b_base64 matches the decode the clients use (aioesphomeapi + decode_noise_psk), so both ends derive the same bytes. The length is + re-checked so a caller cannot turn an unvalidated short decode into a + zero-padded PSK. + """ + try: + decoded = binascii.a2b_base64(value) + except ValueError as err: + raise cv.Invalid("Invalid key format, please check it's using base64") from err + if len(decoded) != 32: + raise cv.Invalid("Encryption key must be base64 and 32 bytes long") + return decoded + + +ENCRYPTION_SCHEMA = cv.Schema( + { + cv.Optional(CONF_KEY): cv.sensitive(validate_encryption_key), + } +) + + +def encryption_schema(config: ConfigType | None) -> ConfigType: + # A bare `encryption:` block is valid; a missing key means the consumer + # falls back to its keyless behavior (api provisioning, ota inheriting + # the api key). + if config is None: + config = {} + return ENCRYPTION_SCHEMA(config) + + async def to_code(config: ConfigType) -> None: cg.add_define("USE_NOISE") cg.add_library("esphome/noise-c", "0.1.21") diff --git a/tests/component_tests/noise/__init__.py b/tests/component_tests/noise/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/component_tests/noise/test_encryption_key.py b/tests/component_tests/noise/test_encryption_key.py new file mode 100644 index 0000000000..62abae6487 --- /dev/null +++ b/tests/component_tests/noise/test_encryption_key.py @@ -0,0 +1,37 @@ +"""Tests for the shared noise encryption key helpers.""" + +from __future__ import annotations + +import pytest + +from esphome import config_validation as cv +from esphome.components.noise import decode_encryption_key, validate_encryption_key + +KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" + + +def test_validate_encryption_key_roundtrips() -> None: + assert validate_encryption_key(KEY) == KEY + + +@pytest.mark.parametrize("value", ["not-base64!!!", "AAECAw=="]) +def test_validate_encryption_key_rejects_bad_input(value: str) -> None: + with pytest.raises(cv.Invalid): + validate_encryption_key(value) + + +def test_decode_encryption_key_returns_32_bytes() -> None: + assert decode_encryption_key(KEY) == bytes(range(32)) + + +def test_decode_encryption_key_rejects_invalid_base64() -> None: + """The shared helper raises cv.Invalid, not binascii.Error.""" + with pytest.raises(cv.Invalid, match="base64"): + decode_encryption_key("A") + + +def test_decode_encryption_key_rejects_short_decode() -> None: + """a2b_base64 stops at embedded padding; a short decode must not become + a zero padded PSK on the device.""" + with pytest.raises(cv.Invalid, match="32 bytes"): + decode_encryption_key("AAECAw==")