Read the encryption key from the esphome OTA platform and rename the command

docs/API.md

@@ -120,6 +120,7 @@ Connections that arrive on the trusted ingress site (HA add-on supervisor proxy)
 |---------|------|----------|-------------|
 | `devices/list` | — | `DevicesResponse` | List configured + importable devices |
 | `devices/get_states` | — | `dict` | Get device online/offline states |
+| `devices/get_encryption_key` | `{configuration}` | `{key}` | The device's one encryption key, resolved through ESPHome's YAML loader (`!secret` / `!include` / packages) with a `esphome config --show-secrets` fallback: the `api: encryption: key`, else the esphome OTA item's `encryption: key` (esphome 2026.9+ shares one key between api and OTA). `""` when neither resolves; the frontend treats that as "open the editor and check" |
 | `devices/create` | `{name, friendly_name?, board_id?, ssid?, psk?, file_content?, overwrite?}` | `WizardResponse` | Create device. With `friendly_name`, `name` is the hostname — validated (lowercase letters, digits, hyphens, underscores; no leading/trailing hyphen; at most 31 chars; `INVALID_ARGS` otherwise) and used verbatim, never rewritten — and the cleaned `friendly_name` becomes `esphome.friendly_name:`. Without it (or when it cleans to empty), `name` is the user's raw display label — capitalisation, inter-word spaces, and unicode are preserved; surrounding whitespace is trimmed. The backend slugifies the cleaned value for `esphome.name:` and the YAML filename and writes the cleaned original into `esphome.friendly_name:`. Callers that already pass a slugified value get the same hostname/friendly_name pair as before (slug-of-slug is a no-op). Three flows: `file_content` writes the supplied YAML as-is; `board_id` generates from a board template; with neither, emits a minimal esp32 stub for the "empty configuration" path. **Wi-Fi handling:** a supplied `ssid` / `psk` is written to `secrets.yaml` (validated, shared with `config/set_wifi_credentials`) and the generated YAML references `!secret wifi_ssid` / `!secret wifi_password` — bare credentials are never written into the device YAML, and the next device reuses the shared secret. With no `ssid`, the generator emits `!secret` when Wi-Fi secrets already exist, else (for a board with no other network) a no-network stub. A board offering onboard-ethernet suggested hardware is **wired by default** — its `ethernet:` block is auto-pulled and the `wifi:` block dropped (see the network-providers note under [Boards](#boards)). A filename collision returns `ALREADY_EXISTS`; pass `overwrite: true` to replace the YAML in place, preserving the existing device's metadata (labels / comment / board_id) and StorageJSON. For a package board, `WizardResponse.warning` is set when the config was kept despite a validation failure confined to remote package resolution (same contract as `devices/import`). A pre-write validation failure rooted in the config dir's `secrets.yaml` (a duplicate key, a parse error) is `INVALID_ARGS` phrased `Can't <action>: secrets.yaml has a duplicate key "<key>" (lines A and B). …` or `Can't <action>: secrets.yaml doesn't parse: …` (a rewrite that can't reach the live definition says `secrets.yaml defines "<key>" where the dashboard can't rewrite it`), never the generator-bug `INTERNAL_ERROR`; the same attribution applies to every mutation that validates (`clone`, `rename`, `import`, …), and the frontend keys its "Open secrets" action on that phrasing. |
 | `devices/import_bundle_token` | — | `{token}` | Mint a single-use token for the HTTP upload of one bundle. Bundles are uploaded over HTTP, not the WebSocket (see the note below); the response's `ImportBundleResponse` shape and the `overwrite` semantics are documented there. |
 | `devices/update` | `{configuration, friendly_name?, comment?, board_id?}` | `UpdateDeviceResponse` | Update device metadata (sidecar JSON) |
@@ -160,6 +161,7 @@ Web Serial log clients also read `Device.logger_baud_rate` (0 ⇒ serial logging
 `Device.has_pending_changes`: `true` = config changed since last compile, `false` = up to date, `null` = never compiled.
 `Device.pending_changes_via_hash`: `true` when `has_pending_changes` came from the mDNS-sourced config-hash compare (vs the local mtime fallback). The frontend gates only this case on a live mDNS, so a local YAML edit still cues "install" when mDNS is dark.
 `Device.update_available`: `true` = device was compiled with a different ESPHome version than the server.
+`Device.ota_encryption_required`: `true` when the esphome OTA item declares `encryption:` (its own key or a bare block inheriting the api key), from the resolved YAML or the raw-text draft scan. With `api_encrypted` it gates the dashboard's "Show encryption key" action, so a device whose key lives only under `ota:` still offers it.
 `Device.migration_available`: `true` = the raw main-file YAML carries a legacy spelling the migration fold would respell (`editor/migrate_config` would return a non-null diff). Computed at scanner load from the top-level file only — packages/`!include` contents are not scanned, same scope as the editor nudge. Always serialised (`false` when clean). Drives the dashboard's migration dot; the one-click apply still goes through `editor/migrate_config`.
 `Device.loaded_platforms`: dotted `domain.platform` pairs (`ota.esphome`, `time.homeassistant`) from the last compile's StorageJSON, sorted; empty until first compile. The pair-shaped companion to `loaded_integrations` (bare component names) — package-resolved platforms included, which is what lets the frontend satisfy dotted dependency checks on packages-based configs.
 `Device.runtime_state.active_source`: `ReachabilitySource` — channel currently driving online state (`mdns` > `mqtt` > `ping`); `unknown` until a source claims it (also the transient default after a restart). The frontend gates an api device's mDNS-sourced out-of-sync / update indicators on `active_source == "mdns"` (or `deployed_identity_live`, for identity the backend confirmed over a direct Native API connection where mDNS is dark). mDNS ownership rides the browser lifecycle: a `Removed` (goodbye or PTR expiry) withdraws it and ping arbitrates, and ownership returns via the device's next announce.

esphome_device_builder/controllers/devices/__init__.py

@@ -12,7 +12,7 @@
 - ``add_component`` — ``devices/add_component`` WS command
   body (featured-id resolution + manifest-driven preset
   merge + atomic YAML rewrite).
-- ``api_key`` — Native API encryption-key resolver
+- ``encryption_key_lookup`` — encryption-key and Native API connection resolver
   (in-process YAML loader fast path + ``esphome config``
   subprocess fallback).
 - ``archive`` — archive / unarchive / delete helpers + the

esphome_device_builder/controllers/devices/controller.py

@@ -58,10 +58,10 @@
 from ..version_history import GIT_COMMIT_ERRORS
 from . import (
     add_component,
-    api_key,
     archive,
     backtrace,
     encryption_key,
+    encryption_key_lookup,
     firmware_sync,
     importable,
     logs,
@@ -905,17 +905,14 @@ def _stamp_regen_failure(self, configuration: str, mtime: float) -> int:
     async def _finalize_regen_success(self, configuration: str) -> None:
         await storage_regen.finalize_success(self, configuration)
 
-    @api_command("devices/get_api_key")
-    async def get_api_key(self, *, configuration: str, **kwargs: Any) -> dict[str, str]:
-        """Return the resolved Native API encryption key for *configuration*."""
-        return await api_key.get_api_key(self, configuration)
-
-    async def _resolve_api_key_via_esphome_config(self, configuration: str) -> str:
-        return await api_key.resolve_via_esphome_config(self, configuration)
+    @api_command("devices/get_encryption_key")
+    async def get_encryption_key(self, *, configuration: str, **kwargs: Any) -> dict[str, str]:
+        """Return the resolved encryption key (api, else esphome OTA) for *configuration*."""
+        return await encryption_key_lookup.get_encryption_key(self, configuration)
 
     async def _resolve_device_api_connection(self, configuration: str) -> tuple[str, int]:
         """Native API (encryption key, port) for the state monitor's API info fallback."""
-        return await api_key.get_api_connection(self, configuration)
+        return await encryption_key_lookup.get_api_connection(self, configuration)
 
     @api_command("devices/add_component")
     async def add_component(

esphome_device_builder/controllers/devices/api_key.py → esphome_device_builder/controllers/devices/encryption_key_lookup.py

@@ -1,4 +1,4 @@
-"""Native API encryption-key resolution for the devices controller."""
+"""Encryption-key and Native API connection resolution for the devices controller."""
 
 from __future__ import annotations
 
@@ -9,6 +9,7 @@
     EsphomeConfigUnavailableError,
     get_api_port,
     get_resolved_api_encryption_key,
+    get_resolved_encryption_key,
     load_device_yaml,
     run_esphome_config,
 )
@@ -17,32 +18,24 @@
     from .controller import DevicesController
 
 
-async def get_api_key(controller: DevicesController, configuration: str) -> dict[str, str]:
-    """
-    Return the resolved Native API encryption key for *configuration*.
-
-    Tries the in-process YAML loader first, then falls back to
-    ``esphome config --show-secrets`` for configs whose key is
-    constructed by Jinja-templated ``packages`` (issue #437).
-    Returns ``{"key": ""}`` when both paths fail; the caller
-    treats that as the "open the editor and check" signal.
-    """
+async def get_encryption_key(controller: DevicesController, configuration: str) -> dict[str, str]:
+    """Return ``{"key": ...}`` for *configuration*, api key else esphome OTA key, ``""`` if none."""
     path = controller._db.settings.rel_path(configuration)
     config = await run_in_executor(load_device_yaml, path)
-    key = get_resolved_api_encryption_key(config)
-    if key:
-        return {"key": key}
-    key = await resolve_via_esphome_config(controller, configuration)
+    key = get_resolved_encryption_key(config) or await _resolve_via_esphome_config(
+        controller, configuration
+    )
     return {"key": key}
 
 
 async def get_api_connection(controller: DevicesController, configuration: str) -> tuple[str, int]:
     """
     Resolve the Native API ``(encryption_key, port)`` from the on-disk YAML.
 
-    In-process only — unlike :func:`get_api_key` this never shells out
+    In-process only — unlike :func:`get_encryption_key` this never shells out
     to ``esphome config``, so the background API-info sweep pays no
-    per-device subprocess. A device whose key resolves only through
+    per-device subprocess. The key is the api one only: an OTA-side key
+    never encrypts the Native API. A device whose key resolves only through
     Jinja-templated ``packages`` returns an empty key here and is left
     for mDNS. Raises :class:`ValueError` when the YAML is missing or
     unparsable so the caller records a miss instead of dialing a doomed
@@ -55,9 +48,9 @@ async def get_api_connection(controller: DevicesController, configuration: str)
     return get_resolved_api_encryption_key(config), get_api_port(config)
 
 
-async def resolve_via_esphome_config(controller: DevicesController, configuration: str) -> str:
+async def _resolve_via_esphome_config(controller: DevicesController, configuration: str) -> str:
     """
-    Subprocess fallback for :func:`get_api_key`.
+    Subprocess fallback for :func:`get_encryption_key`.
 
     Delegates to :func:`helpers.device_yaml.run_esphome_config`, which fully
     resolves substitutions / packages / secrets. Returns ``""`` on every
@@ -74,4 +67,4 @@ async def resolve_via_esphome_config(controller: DevicesController, configuratio
         return ""
     if config is None:
         return ""
-    return get_resolved_api_encryption_key(config)
+    return get_resolved_encryption_key(config)

esphome_device_builder/controllers/devices/importable.py

@@ -15,7 +15,7 @@
 from ...helpers.device_yaml import (
     EsphomeConfigUnavailableError,
     generate_adoption_yaml,
-    resolved_ota_has_own_key,
+    get_ota_encryption_key,
     run_esphome_config,
 )
 from ...helpers.json import JSONDecodeError, dumps_indent, loads
@@ -336,7 +336,7 @@ async def _mint_key_unless_package_encrypts(
     if isinstance(api_block, dict) and "encryption" in api_block:
         return None
     # A package's own OTA key would have to match a baked api key; leave both out.
-    if resolved_ota_has_own_key(config):
+    if get_ota_encryption_key(config):
         return (
             "The package gives the OTA platform its own encryption key, so no API "
             "encryption key was generated; edit the device to use one key for both."

esphome_device_builder/helpers/device_yaml/__init__.py

@@ -60,12 +60,13 @@
     get_api_encryption_block,
     get_api_encryption_key,
     get_api_port,
+    get_ota_encryption_key,
     get_resolved_api_encryption_key,
+    get_resolved_encryption_key,
     has_top_level_block,
     parse_esphome_meta,
     parse_platform_from_yaml,
     resolved_device_name,
-    resolved_ota_has_own_key,
     retarget_fallback_ap_ssid,
     safe_stat_key,
     yaml_has_api_encryption,
@@ -110,15 +111,16 @@
     "get_api_encryption_block",
     "get_api_encryption_key",
     "get_api_port",
+    "get_ota_encryption_key",
     "get_resolved_api_encryption_key",
+    "get_resolved_encryption_key",
     "has_top_level_block",
     "load_device_from_storage",
     "load_device_yaml",
     "parse_esphome_meta",
     "parse_platform_from_yaml",
     "pending_changes_via_hash",
     "resolved_device_name",
-    "resolved_ota_has_own_key",
     "retarget_fallback_ap_ssid",
     "run_esphome_config",
     "safe_stat_key",

esphome_device_builder/helpers/device_yaml/_loading.py

@@ -42,6 +42,7 @@
     has_top_level_block,
     mdns_disabled_enabled,
     name_add_mac_suffix_enabled,
+    ota_encryption_declared,
     parse_esphome_meta,
     safe_stat_key,
     yaml_has_api_encryption,
@@ -369,6 +370,7 @@ def load_device_from_storage(
         mdns_disabled=mdns_disabled_enabled(resolved_config, yaml_content),
         api_enabled=api_enabled,
         api_encrypted=api_encrypted,
+        ota_encryption_required=ota_encryption_declared(resolved_config, yaml_content),
         mac_address=mac_address,
         ethernet_mac=ethernet_mac,
         bluetooth_mac=bluetooth_mac,

esphome_device_builder/helpers/device_yaml/_parsing.py

@@ -5,6 +5,7 @@
 import hashlib
 import logging
 import re
+from collections.abc import Iterator
 from pathlib import Path
 from typing import NamedTuple
 
@@ -225,8 +226,8 @@ def extract_component_source_fingerprint(yaml_content: str) -> str:
     return _digest_lines(lines)
 
 
-_RAW_API_ENCRYPTION_RE = re.compile(
-    # Matches an ``encryption:`` line that's indented under ``api:``
+def _nested_key_re(block: str, key: str) -> re.Pattern[str]:
+    # Matches a *key* line that's indented under top-level *block*
     # (any depth ≥ 1 space). Used as a draft-time heuristic — once
     # ``load_device_yaml`` succeeds, the resolved-config check wins.
     #
@@ -235,9 +236,19 @@ def extract_component_source_fingerprint(yaml_content: str) -> str:
     # line. No overlap, so the engine can't backtrack between them on a
     # long run of newlines (the previous ``\s*\n`` alternative could
     # also consume a bare ``\n``, which CodeQL flagged as exponential).
-    r"^api:[^\n]*\n(?:[ \t][^\n]*\n|\n)*[ \t]+encryption:(?:\s|$)",
-    re.MULTILINE,
-)
+    return re.compile(
+        rf"^{block}:[^\n]*\n(?:[ \t][^\n]*\n|\n)*[ \t]+{key}:(?:\s|$)",
+        re.MULTILINE,
+    )
+
+
+_RAW_API_ENCRYPTION_RE = _nested_key_re("api", "encryption")
+_RAW_OTA_ENCRYPTION_RE = _nested_key_re("ota", "encryption")
+
+
+def yaml_has_ota_encryption(yaml_content: str) -> bool:
+    """Heuristic: True when raw YAML appears to declare ``encryption:`` under ``ota:``."""
+    return "encryption:" in yaml_content and bool(_RAW_OTA_ENCRYPTION_RE.search(yaml_content))
 
 
 def yaml_has_api_encryption(yaml_content: str) -> bool:
@@ -248,7 +259,7 @@ def yaml_has_api_encryption(yaml_content: str) -> bool:
     a syntax error. The resolved-config check is preferred whenever
     available (catches ``!include`` / packages this regex can't see).
     """
-    return bool(_RAW_API_ENCRYPTION_RE.search(yaml_content))
+    return "encryption:" in yaml_content and bool(_RAW_API_ENCRYPTION_RE.search(yaml_content))
 
 
 def _truthy_child_re(block: str, key: str) -> re.Pattern[str]:
@@ -314,6 +325,17 @@ def mdns_disabled_enabled(resolved_config: dict | None, yaml_content: str) -> bo
     return bool(_RAW_MDNS_DISABLED_RE.search(yaml_content))
 
 
+def ota_encryption_declared(resolved_config: dict | None, yaml_content: str) -> bool:
+    """
+    Detect an esphome OTA ``encryption:`` block: resolved config wins, raw text fills in.
+
+    The raw-text fallback applies only when resolution failed.
+    """
+    if resolved_config is not None:
+        return resolved_ota_has_encryption(resolved_config)
+    return yaml_has_ota_encryption(yaml_content)
+
+
 def config_has_top_level_block(config: dict | None, key: str) -> bool:
     """Return True when *config* (a resolved device YAML) defines top-level *key*.
 
@@ -621,35 +643,35 @@ def resolve_esp32_variant(
     return None
 
 
-def resolved_ota_has_own_key(config: dict | None) -> bool:
-    """Whether an esphome OTA entry in a resolved config carries its own ``encryption: key``."""
-    ota = config.get(const.CONF_OTA) if isinstance(config, dict) else None
-    entries = ota if isinstance(ota, list) else [ota]
-    for entry in entries:
-        if not isinstance(entry, dict) or entry.get(const.CONF_PLATFORM, "esphome") != "esphome":
-            continue
+def get_ota_encryption_key(config: dict | None) -> str:
+    """Return the first esphome OTA entry's own ``encryption: key`` (``${var}`` kept) or ``""``."""
+    for entry in _ota_esphome_entries(config):
         encryption = entry.get("encryption")
-        if isinstance(encryption, dict) and encryption.get("key"):
-            return True
-    return False
+        key = encryption.get("key") if isinstance(encryption, dict) else None
+        if isinstance(key, str) and key:
+            return key
+    return ""
+
+
+def resolved_ota_has_encryption(config: dict | None) -> bool:
+    """Whether an esphome OTA entry declares ``encryption:``; a bare block parses to ``None``."""
+    return any("encryption" in entry for entry in _ota_esphome_entries(config))
 
 
 def extract_ota_partition_access(config: dict | None) -> bool:
-    """
-    Report whether an ``ota: platform: esphome`` entry sets ``allow_partition_access``.
+    """Report whether an esphome OTA entry sets ``allow_partition_access``."""
+    return any(
+        entry.get(_CONF_ALLOW_PARTITION_ACCESS) is True for entry in _ota_esphome_entries(config)
+    )
 
-    Accepts both the list-of-platforms form and the legacy single-mapping
-    form (which implies the esphome platform).
-    """
+
+def _ota_esphome_entries(config: dict | None) -> Iterator[dict]:
+    """Yield the esphome-platform ``ota:`` entries; list form and the legacy single mapping."""
     ota = config.get(const.CONF_OTA) if isinstance(config, dict) else None
     entries = ota if isinstance(ota, list) else [ota]
     for entry in entries:
-        if not isinstance(entry, dict):
-            continue
-        platform = entry.get(const.CONF_PLATFORM, "esphome")
-        if platform == "esphome" and entry.get(_CONF_ALLOW_PARTITION_ACCESS) is True:
-            return True
-    return False
+        if isinstance(entry, dict) and entry.get(const.CONF_PLATFORM, "esphome") == "esphome":
+            yield entry
 
 
 def _str_or_none(value: object) -> str | None:
@@ -860,9 +882,20 @@ def get_api_encryption_key(config: dict | None) -> str:
 
 def get_resolved_api_encryption_key(config: dict | None) -> str:
     """Native API encryption key with ``${var}`` resolved; ``""`` if absent or unresolved."""
-    key = get_api_encryption_key(config)
-    if not key:
-        return ""
+    return _resolve_key(config, get_api_encryption_key(config))
+
+
+def get_resolved_encryption_key(config: dict | None) -> str:
+    """Return the device's one key, api or esphome OTA, ``${var}`` resolved; ``""`` if none."""
+    return get_resolved_api_encryption_key(config) or _resolve_key(
+        config, get_ota_encryption_key(config)
+    )
+
+
+def _resolve_key(config: dict | None, key: str) -> str:
+    """Expand ``${var}`` in *key* against *config*'s substitutions; ``""`` if unresolved."""
+    if "$" not in key:
+        return key
     key = _resolve_substitutions(key, _extract_resolved_substitutions(config)) or ""
     if _UNRESOLVED_SUBSTITUTION_RE.search(key):
         return ""