diff --git a/docs/API.md b/docs/API.md index 311a94ae8..88cf261e0 100644 --- a/docs/API.md +++ b/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 : secrets.yaml has a duplicate key "" (lines A and B). …` or `Can't : secrets.yaml doesn't parse: …` (a rewrite that can't reach the live definition says `secrets.yaml defines "" 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. diff --git a/esphome_device_builder/controllers/devices/__init__.py b/esphome_device_builder/controllers/devices/__init__.py index 50c5ffe33..8af69c76d 100644 --- a/esphome_device_builder/controllers/devices/__init__.py +++ b/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 diff --git a/esphome_device_builder/controllers/devices/controller.py b/esphome_device_builder/controllers/devices/controller.py index c8dd33d66..b72b30c17 100644 --- a/esphome_device_builder/controllers/devices/controller.py +++ b/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( diff --git a/esphome_device_builder/controllers/devices/api_key.py b/esphome_device_builder/controllers/devices/encryption_key_lookup.py similarity index 65% rename from esphome_device_builder/controllers/devices/api_key.py rename to esphome_device_builder/controllers/devices/encryption_key_lookup.py index 0b3d81cdb..a51968db6 100644 --- a/esphome_device_builder/controllers/devices/api_key.py +++ b/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,22 +18,13 @@ 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} @@ -40,9 +32,10 @@ async def get_api_connection(controller: DevicesController, configuration: str) """ 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) diff --git a/esphome_device_builder/controllers/devices/importable.py b/esphome_device_builder/controllers/devices/importable.py index 3505f09bc..92f87d7cc 100644 --- a/esphome_device_builder/controllers/devices/importable.py +++ b/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." diff --git a/esphome_device_builder/helpers/device_yaml/__init__.py b/esphome_device_builder/helpers/device_yaml/__init__.py index 75a531653..995a2de47 100644 --- a/esphome_device_builder/helpers/device_yaml/__init__.py +++ b/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,7 +111,9 @@ "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", @@ -118,7 +121,6 @@ "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", diff --git a/esphome_device_builder/helpers/device_yaml/_loading.py b/esphome_device_builder/helpers/device_yaml/_loading.py index 1b53f9ae1..b5e716d0c 100644 --- a/esphome_device_builder/helpers/device_yaml/_loading.py +++ b/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, diff --git a/esphome_device_builder/helpers/device_yaml/_parsing.py b/esphome_device_builder/helpers/device_yaml/_parsing.py index 80d7520f6..57a3a3350 100644 --- a/esphome_device_builder/helpers/device_yaml/_parsing.py +++ b/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 "" diff --git a/esphome_device_builder/models/devices.py b/esphome_device_builder/models/devices.py index 5a2bf15d7..766b8a905 100644 --- a/esphome_device_builder/models/devices.py +++ b/esphome_device_builder/models/devices.py @@ -262,9 +262,12 @@ class Device(DashboardModel): # as a flatten-to-False signal. # # The actual key is fetched on demand via - # ``devices/get_api_key``. + # ``devices/get_encryption_key``. api_enabled: bool = False api_encrypted: bool = False + # esphome shares one key between api and OTA, so a device whose key + # lives only under ``ota:`` still has one for ``devices/get_encryption_key``. + ota_encryption_required: bool = False # Encryption status as observed from the device's # ``_esphomelib._tcp.local.`` mDNS broadcast. # None → mDNS not seen yet. The frontend trusts ``api_encrypted`` diff --git a/tests/controllers/devices/test_branches_coverage.py b/tests/controllers/devices/test_branches_coverage.py index 264cb8576..f36d9c77b 100644 --- a/tests/controllers/devices/test_branches_coverage.py +++ b/tests/controllers/devices/test_branches_coverage.py @@ -8,7 +8,7 @@ Grouped by surface: -- **API command wiring** (delete / delete_bulk / get_api_key / +- **API command wiring** (delete / delete_bulk / get_encryption_key / add_component error branches) — these are the public commands that go through the WS layer; pin both the happy-path return shape and the typed-error branches the dashboard relies on. @@ -217,14 +217,14 @@ async def test_archive_bulk_returns_per_device_success_with_mixed_outcomes( # --------------------------------------------------------------------------- -# get_api_key public-API wiring +# get_encryption_key public-API wiring # --------------------------------------------------------------------------- -async def test_get_api_key_resolves_through_yaml_loader( +async def test_get_encryption_key_resolves_through_yaml_loader( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: - """``devices/get_api_key`` returns the resolved encryption key. + """``devices/get_encryption_key`` returns the resolved encryption key. The handler runs through ESPHome's YAML loader so ``!secret`` references resolve the same way they do at compile time — @@ -239,15 +239,15 @@ async def test_get_api_key_resolves_through_yaml_loader( encoding="utf-8", ) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": "a/c+inline-key=="} -async def test_get_api_key_resolves_substitution_from_secret( +async def test_get_encryption_key_resolves_substitution_from_secret( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: - """``devices/get_api_key`` expands ``${api_key}`` over a ``!secret`` substitution (#1691).""" + """``devices/get_encryption_key`` expands ``${api_key}`` over a ``!secret`` substitution.""" controller = make_controller(tmp_path) (tmp_path / "secrets.yaml").write_text("api_key: a/c+secret-key==\n", encoding="utf-8") (tmp_path / "kitchen.yaml").write_text( @@ -257,7 +257,7 @@ async def test_get_api_key_resolves_substitution_from_secret( encoding="utf-8", ) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": "a/c+secret-key=="} @@ -307,7 +307,7 @@ async def test_resolve_device_api_connection_raises_on_unloadable_config( await controller._resolve_device_api_connection("kitchen.yaml") -async def test_get_api_key_returns_empty_when_no_encryption( +async def test_get_encryption_key_returns_empty_when_no_encryption( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """A device without ``api.encryption`` returns ``{"key": ""}``. @@ -323,12 +323,33 @@ async def test_get_api_key_returns_empty_when_no_encryption( encoding="utf-8", ) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": ""} -async def test_get_api_key_falls_back_to_esphome_config_subprocess( +@pytest.mark.parametrize( + "ota_block", + [ + pytest.param( + "ota:\n - platform: esphome\n encryption:\n key: ota-key==\n", id="list" + ), + pytest.param("ota:\n platform: esphome\n encryption:\n key: ota-key==\n", id="legacy"), + ], +) +async def test_get_encryption_key_reads_the_esphome_ota_key_without_api( + tmp_path: Path, make_controller: MakeControllerFactory, ota_block: str +) -> None: + """A key only under the esphome OTA item is the device's key.""" + controller = make_controller(tmp_path) + (tmp_path / "gate.yaml").write_text("esphome:\n name: gate\n" + ota_block, encoding="utf-8") + + result = await controller.get_encryption_key(configuration="gate.yaml") + + assert result == {"key": "ota-key=="} + + +async def test_get_encryption_key_falls_back_to_esphome_config_subprocess( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: r"""When the in-process loader misses, ``esphome config`` subprocess wins. @@ -370,12 +391,12 @@ async def _fake_create_subprocess(*_args: Any, **_kwargs: Any) -> Any: with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", _fake_create_subprocess) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": "ZGFzaGJvYXJkLWtleS1mcm9tLWVzcGhvbWUtY29uZmln"} -async def test_get_api_key_subprocess_returns_empty_on_nonzero_exit( +async def test_get_encryption_key_subprocess_returns_empty_on_nonzero_exit( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """A subprocess that exits non-zero still returns ``{"key": ""}``. @@ -403,12 +424,12 @@ async def _fake_create_subprocess(*_args: Any, **_kwargs: Any) -> Any: with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", _fake_create_subprocess) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": ""} -async def test_get_api_key_subprocess_returns_empty_on_unparsable_yaml( +async def test_get_encryption_key_subprocess_returns_empty_on_unparsable_yaml( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """Subprocess output that doesn't parse as YAML degrades to ``""``. @@ -434,12 +455,12 @@ async def _fake_create_subprocess(*_args: Any, **_kwargs: Any) -> Any: with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", _fake_create_subprocess) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": ""} -async def test_get_api_key_subprocess_returns_empty_on_oserror( +async def test_get_encryption_key_subprocess_returns_empty_on_oserror( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """A subprocess startup failure (``OSError``) still returns ``""``. @@ -461,12 +482,12 @@ async def _boom(*_args: Any, **_kwargs: Any) -> Any: with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", _boom) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": ""} -async def test_get_api_key_skips_subprocess_when_fast_path_finds_key( +async def test_get_encryption_key_skips_subprocess_when_fast_path_finds_key( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """The fast path's hit short-circuits — no subprocess overhead. @@ -486,13 +507,13 @@ async def test_get_api_key_skips_subprocess_when_fast_path_finds_key( with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", spawn_spy) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": "a/c+inline-key=="} spawn_spy.assert_not_called() -async def test_get_api_key_fallback_skipped_when_esphome_cmd_unset( +async def test_get_encryption_key_fallback_skipped_when_esphome_cmd_unset( tmp_path: Path, make_controller: MakeControllerFactory ) -> None: """The subprocess fallback is a no-op without ``_esphome_cmd``. @@ -517,7 +538,7 @@ async def test_get_api_key_fallback_skipped_when_esphome_cmd_unset( with pytest.MonkeyPatch.context() as mp: mp.setattr(resolve_mod, "create_subprocess_exec", spawn_spy) - result = await controller.get_api_key(configuration="kitchen.yaml") + result = await controller.get_encryption_key(configuration="kitchen.yaml") assert result == {"key": ""} spawn_spy.assert_not_called() diff --git a/tests/test_api_key.py b/tests/test_encryption_key_extraction.py similarity index 72% rename from tests/test_api_key.py rename to tests/test_encryption_key_extraction.py index 7422e2783..f02aef09b 100644 --- a/tests/test_api_key.py +++ b/tests/test_encryption_key_extraction.py @@ -1,4 +1,4 @@ -"""Tests for the Native API encryption-key extraction + scanner flag. +"""Tests for the encryption-key extraction (api and esphome OTA) + scanner flags. Covers the helper layer (resolves through ESPHome's YAML loader so ``!secret`` / ``!include`` / packages all work) and the scan-time @@ -20,10 +20,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_yaml, ) +from esphome_device_builder.helpers.device_yaml._parsing import ( + resolved_ota_has_encryption, + yaml_has_ota_encryption, +) from esphome_device_builder.models import Device # --------------------------------------------------------------------------- @@ -53,6 +59,94 @@ def test_get_api_encryption_block_handles_non_dict_inputs() -> None: assert get_api_encryption_block({"api": {"encryption": "not-a-dict"}}) is None +@pytest.mark.parametrize( + ("config", "expected"), + [ + pytest.param( + { + "ota": [ + {"platform": "web_server"}, + {"platform": "esphome", "encryption": {"key": "k"}}, + ] + }, + "k", + id="list", + ), + pytest.param( + {"ota": {"platform": "esphome", "encryption": {"key": "k"}}}, "k", id="legacy" + ), + pytest.param({"ota": {"encryption": {"key": "k"}}}, "k", id="mapping-default-platform"), + pytest.param( + {"ota": [{"platform": "esphome", "encryption": None}, {"encryption": {"key": "k"}}]}, + "k", + id="second-entry", + ), + pytest.param({"ota": [{"platform": "esphome", "encryption": None}]}, "", id="bare"), + pytest.param({"ota": [{"platform": "esphome", "encryption": "x"}]}, "", id="non-dict"), + pytest.param( + {"ota": [{"platform": "web_server", "encryption": {"key": "k"}}]}, "", id="other" + ), + pytest.param(None, "", id="no-config"), + ], +) +def test_get_ota_encryption_key(config: dict | None, expected: str) -> None: + assert get_ota_encryption_key(config) == expected + + +def test_resolved_ota_has_encryption_counts_a_bare_block() -> None: + assert resolved_ota_has_encryption({"ota": [{"platform": "esphome", "encryption": None}]}) + assert resolved_ota_has_encryption({"ota": {"platform": "esphome", "encryption": {"key": "k"}}}) + assert not resolved_ota_has_encryption({"ota": [{"platform": "esphome"}]}) + other = {"ota": [{"platform": "web_server", "encryption": None}]} + assert not resolved_ota_has_encryption(other) + assert not resolved_ota_has_encryption(None) + + +def test_get_resolved_encryption_key_prefers_api() -> None: + config = {"api": {"encryption": {"key": "api=="}}, "ota": {"encryption": {"key": "ota=="}}} + assert get_resolved_encryption_key(config) == "api==" + + +def test_get_resolved_encryption_key_falls_back_to_ota() -> None: + config = {"ota": [{"platform": "esphome", "encryption": {"key": "ota=="}}]} + assert get_resolved_encryption_key(config) == "ota==" + unresolved_api = {"api": {"encryption": {"key": "${missing}"}}, **config} + assert get_resolved_encryption_key(unresolved_api) == "ota==" + + +def test_get_resolved_encryption_key_expands_ota_substitution() -> None: + config = { + "substitutions": {"psk": "ZGFzaA=="}, + "ota": [{"platform": "esphome", "encryption": {"key": "${psk}"}}], + } + assert get_resolved_encryption_key(config) == "ZGFzaA==" + + +def test_get_resolved_encryption_key_empty_when_unresolved_or_missing() -> None: + assert get_resolved_encryption_key({"ota": {"encryption": {"key": "${missing}"}}}) == "" + assert get_resolved_encryption_key({"api": {}}) == "" + assert get_resolved_encryption_key(None) == "" + + +_OTA_LIST = "ota:\n - platform: esphome\n encryption:\n" +_API_KEYED = "api:\n encryption:\n key: k\n" + + +@pytest.mark.parametrize( + ("yaml_text", "expected"), + [ + pytest.param(_OTA_LIST + " key: k\n", True, id="list"), + pytest.param("ota:\n platform: esphome\n encryption:\n key: k\n", True, id="mapping"), + pytest.param(_OTA_LIST, True, id="bare"), + pytest.param("ota:\n - platform: esphome\n password: x\n", False, id="no-encryption"), + pytest.param("ota:\n - platform: esphome\n" + _API_KEYED, False, id="api-block"), + pytest.param(_API_KEYED, False, id="no-ota"), + ], +) +def test_yaml_has_ota_encryption(yaml_text: str, expected: bool) -> None: + assert yaml_has_ota_encryption(yaml_text) is expected + + def test_get_api_encryption_key_returns_resolved_string() -> None: config = {"api": {"encryption": {"key": "ZGFzaA=="}}} assert get_api_encryption_key(config) == "ZGFzaA==" @@ -176,7 +270,7 @@ def test_load_device_yaml_resolves_secrets(tmp_path: Path) -> None: """``!secret`` references resolve through the sibling ``secrets.yaml``. The regex-on-raw-YAML approach the frontend used to do gave up - here — backend resolution is the whole reason ``devices/get_api_key`` + here — backend resolution is the whole reason ``devices/get_encryption_key`` exists. """ (tmp_path / "secrets.yaml").write_text("api_key: 'AAAA=='\n") @@ -341,6 +435,62 @@ def test_load_device_from_storage_sets_api_encrypted_from_resolved_yaml( assert device.api_encrypted is True +@pytest.mark.parametrize( + "ota_block", + [ + pytest.param( + 'ota:\n - platform: esphome\n encryption:\n key: "ZGFzaA=="\n', id="list" + ), + pytest.param( + 'ota:\n platform: esphome\n encryption:\n key: "ZGFzaA=="\n', id="legacy" + ), + ], +) +def test_load_device_from_storage_sets_ota_encryption_required( + isolated_storage: Path, ota_block: str +) -> None: + """A key only under the esphome OTA item sets the OTA flag, not the api ones.""" + device = _scan(isolated_storage / "gate.yaml", "esphome:\n name: gate\n" + ota_block) + assert device.api_enabled is False + assert device.api_encrypted is False + assert device.ota_encryption_required is True + + +def test_load_device_from_storage_ota_encryption_required_for_bare_block( + isolated_storage: Path, +) -> None: + device = _scan( + isolated_storage / "inherit.yaml", + "esphome:\n name: inherit\n" + 'api:\n encryption:\n key: "ZGFzaA=="\n' + "ota:\n - platform: esphome\n encryption:\n", + ) + assert device.api_encrypted is True + assert device.ota_encryption_required is True + + +def test_load_device_from_storage_ota_encryption_required_false_without_block( + isolated_storage: Path, +) -> None: + device = _scan( + isolated_storage / "plain.yaml", + "esphome:\n name: plain\nota:\n - platform: esphome\n password: x\n", + ) + assert device.ota_encryption_required is False + + +def test_load_device_from_storage_ota_encryption_required_falls_back_for_invalid_draft( + isolated_storage: Path, +) -> None: + device = _scan( + isolated_storage / "broken-ota.yaml", + "esphome:\n name: broken-ota\n" + 'ota:\n - platform: esphome\n encryption:\n key: "ZGFzaA=="\n' + "sensor:\n - platform: !\n bad: [unterminated\n", + ) + assert device.ota_encryption_required is True + + def test_load_device_from_storage_api_disabled_for_mqtt_only( isolated_storage: Path, ) -> None: diff --git a/tests/test_run_esphome_config.py b/tests/test_run_esphome_config.py index f40cdd7be..e1ace9136 100644 --- a/tests/test_run_esphome_config.py +++ b/tests/test_run_esphome_config.py @@ -1,7 +1,7 @@ """Tests for ``helpers.device_yaml.run_esphome_config``. The subprocess primitive behind ``/json-config`` and the -``devices/get_api_key`` package fallback — it runs ``esphome config +``devices/get_encryption_key`` package fallback — it runs ``esphome config --show-secrets`` and parses the fully-resolved output. """