Move encryption key validation into the noise component

esphome/components/api/__init__.py

@@ -1,4 +1,3 @@
-import base64
 import logging
 from typing import Any
 
@@ -6,6 +5,15 @@ 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 (
@@ -38,6 +46,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"
@@ -136,20 +148,6 @@ def _register_provisioning_source(config: ConfigType) -> ConfigType:
     return config
 
 
-def validate_encryption_key(value: Any) -> 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
-
-
 CONF_SUPPORTS_RESPONSE = "supports_response"
 
 # Enum values in api::enums namespace
@@ -256,18 +254,6 @@ ACTIONS_SCHEMA = automation.validate_automation(
     ),
 )
 
-ENCRYPTION_SCHEMA = cv.Schema(
-    {
-        cv.Optional(CONF_KEY): cv.sensitive(validate_encryption_key),
-    }
-)
-
-
-def _encryption_schema(config: ConfigType | None) -> ConfigType:
-    if config is None:
-        config = {}
-    return ENCRYPTION_SCHEMA(config)
-
 
 def _consume_api_sockets(config: ConfigType) -> ConfigType:
     """Register socket needs for API component."""
@@ -303,7 +289,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)),
@@ -490,7 +476,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:

esphome/components/noise/__init__.py

@@ -1,5 +1,10 @@
+import base64
+import binascii
+from typing import Any
+
 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 +15,53 @@ noise_ns = cg.esphome_ns.namespace("noise")
 CONFIG_SCHEMA = cv.Schema({})
 
 
+def validate_encryption_key(value: Any) -> 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")

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==")