diff --git a/esphome/__main__.py b/esphome/__main__.py index 82ef585e1a..1e2c4b0b3a 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -2232,6 +2232,12 @@ def command_analyze_memory(args: ArgsProtocol, config: ConfigType) -> int: return 0 +def command_rotate_key(args: ArgsProtocol, config: ConfigType) -> int | None: + from esphome.cli.rotate_key import command_rotate_key as run + + return run(args, config) + + def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None: from esphome.cli.rename import command_rename as run @@ -2258,6 +2264,7 @@ POST_CONFIG_ACTIONS = { "clean-mqtt": command_clean_mqtt, "idedata": command_idedata, "rename": command_rename, + "rotate-key": command_rotate_key, "discover": command_discover, "analyze-memory": command_analyze_memory, "bundle": command_bundle, @@ -2655,6 +2662,37 @@ def parse_args(argv): ) parser_rename.add_argument("name", help="The new name for the device.", type=str) + parser_rotate_key = subparsers.add_parser( + "rotate-key", + help=( + "Change the OTA encryption key over the air: write a new key to the " + "yaml, build with it, upload with the current key, confirm." + ), + parents=[mqtt_options], + ) + parser_rotate_key.add_argument( + "configuration", help="Your YAML configuration file.", nargs=1 + ) + parser_rotate_key.add_argument( + "--device", + action="append", + help=ARGUMENT_HELP_DEVICE, + ) + parser_rotate_key.add_argument( + "--prompt-new-key", + action="store_true", + help=( + "Ask for the new key instead of generating one; read without echo, " + "or as one line from stdin when that is not a terminal." + ), + ) + parser_rotate_key.add_argument( + "-y", + "--yes", + action="store_true", + help="Do not ask for confirmation when the api encryption key changes too.", + ) + parser_analyze_memory = subparsers.add_parser( "analyze-memory", help="Analyze memory usage by component.", diff --git a/esphome/cli/rotate_key.py b/esphome/cli/rotate_key.py new file mode 100644 index 0000000000..f0d407e335 --- /dev/null +++ b/esphome/cli/rotate_key.py @@ -0,0 +1,266 @@ +"""``esphome rotate-key``.""" + +from __future__ import annotations + +import argparse +from pathlib import Path +import sys + +from esphome import espota2 +from esphome.components.noise import ( + generate_encryption_key, + static_encryption_key, + validate_encryption_key, +) +from esphome.config_validation import Invalid +from esphome.const import CONF_API, CONF_PASSWORD, CONF_PORT +from esphome.core import CORE, EsphomeError +from esphome.log import AnsiFore, color +from esphome.ota_key_edit import locate_key_edits, old_key_edit, with_sharers +from esphome.types import ConfigType +from esphome.upload_targets import PortType, get_port_type +from esphome.util import ( + ESPHOME_COMMAND, + read_secret_line, + run_external_process, + safe_input, + safe_print, +) +from esphome.yaml_edit import RestoreError, Snapshot, apply_line_edits, restore_files + +PRECHECK_TIMEOUT = 15.0 + + +def command_rotate_key(args: argparse.Namespace, config: ConfigType) -> int | None: + """Write a new key, build with it, upload with the current one, confirm.""" + from esphome.__main__ import ( + Purpose, + _esphome_ota_conf, + _resolve_network_devices, + choose_upload_log_host, + ) + + def fail(message: str) -> int: + safe_print(color(AnsiFore.BOLD_RED, message)) + return 1 + + ota_conf = _esphome_ota_conf(config) + if ota_conf is None: + return fail("rotate-key needs the esphome OTA platform in the configuration") + old_key = static_encryption_key(ota_conf) + if old_key is None: + return fail( + "rotate-key needs 'encryption:' under the esphome OTA platform with a " + "key in the yaml; a key provisioned at runtime is rotated from Home " + "Assistant" + ) + if CORE.is_host: + return fail("rotate-key is for devices updated over the air") + old_key = str(old_key) + remote_port = int(ota_conf[CONF_PORT]) + + devices = choose_upload_log_host( + default=args.device, check_default=None, purpose=Purpose.UPLOADING + ) + if get_port_type(devices[0]) in (PortType.SERIAL, PortType.BOOTSEL): + return fail( + "rotate-key works over the air; on a serial connection change the " + "key in the yaml and run 'esphome run' instead" + ) + network_devices = _resolve_network_devices(devices, config, args) + + if args.prompt_new_key: + new_key = read_secret_line("New OTA encryption key: ") + else: + new_key = generate_encryption_key() + try: + validate_encryption_key(new_key) + except Invalid as err: + return fail(f"Invalid new key: {err}") + if new_key == old_key: + return fail("The new key is the same as the current one") + + try: + edits = locate_key_edits(old_key, new_key) + # The previous key stays in the config as old_key so an install still + # reaches the device if it ends up running either key + secret = next((edit.secret for edit in edits if edit.secret), None) + edits += old_key_edit(old_key, secret) + unchecked = with_sharers(edits) + except EsphomeError as err: + return fail(str(err)) + + warnings = [] + # With the recommended shape the ota inherits the api key, so rotating + # it changes what Home Assistant connects with + if str(static_encryption_key(config.get(CONF_API) or {})) == old_key: + warnings.append( + "The api encryption key changes too; Home Assistant will ask for " + "the new key after the install." + ) + shared: dict[str, set[str]] = {} + for edit in edits: + what = f"secret '{edit.secret}'" if edit.secret else edit.path.name + for path in edit.shared_with: + shared.setdefault( + str(path.relative_to(CORE.config_dir.resolve())), set() + ).add(what) + if unchecked: + warnings.append( + "Could not check every configuration for shared use: " + + "; ".join(unchecked) + ) + if shared: + warnings.append( + "Other configurations share what is rewritten: " + + "; ".join( + f"{p} ({', '.join(sorted(w))})" for p, w in sorted(shared.items()) + ) + + ". Their devices keep the previous key until their next install; " + "make sure those configurations carry it as old_key." + ) + if warnings: + for warning in warnings: + safe_print(color(AnsiFore.BOLD_YELLOW, warning)) + if not args.yes: + if not sys.stdin.isatty(): + return fail("Confirm with --yes when there is no terminal") + if safe_input("Continue? [y/N] ").strip().lower() not in ("y", "yes"): + return 1 + + safe_print("Checking that the device accepts the current key...") + if not espota2.probe_ota_key( + network_devices, + remote_port, + old_key, + timeout=PRECHECK_TIMEOUT, + retry_rejected=False, + ): + return fail( + "The device did not complete an encrypted handshake with the current " + "key (see the warning above for the reason). If it is reachable, it " + "has to run the key in the yaml on ESPHome 2026.9.0 or newer; install " + "the current configuration first, then rotate." + ) + + # Both keys are shown only when a file could not be put back + keys = f"Previous key: {old_key}\nNew key: {new_key}" + try: + originals = apply_line_edits(edits) + except RestoreError as err: + safe_print(keys) + return fail(str(err)) + except EsphomeError as err: + return fail(str(err)) + except KeyboardInterrupt as err: + if notes := getattr(err, "__notes__", None): # a rollback that failed too + safe_print(keys) + return fail("Interrupted; " + "; ".join(notes)) + return fail("Interrupted before the upload") + # From here every exit restores or reports, so nothing sits outside the try + uploaded = False + try: + safe_print( + "Wrote the new key to " + + ", ".join(color(AnsiFore.CYAN, str(p)) for p in originals) + ) + # Global options go before the subcommand; the compile parser is strict + cli_args = ["--dashboard"] if CORE.dashboard else [] + if args.toolchain: + cli_args += ["--toolchain", str(args.toolchain)] + for key, value in args.substitution or []: + cli_args += ["-s", key, value] + cli_args += ["compile", str(CORE.config_path)] + if run_external_process(*ESPHOME_COMMAND, *cli_args) != 0: + return fail("Compiling with the new key failed") + + def device_reached() -> None: + # The upload may commit from here on; the edited config reaches + # the device on either key + nonlocal uploaded + uploaded = True + + rc, _ = espota2.run_ota( + network_devices, + remote_port, + ota_conf.get(CONF_PASSWORD), + CORE.firmware_bin, + noise_psk=old_key, + plaintext_fallback=False, + on_connect=device_reached, + ) + if rc != 0: + return fail( + "Uploading with the current key failed. The configuration keeps " + "the new key with the previous one as old_key, so the next " + "install reaches the device whichever key it runs." + if uploaded + else "The upload did not reach the device." + ) + safe_print("Waiting for the device to come back with the new key...") + if not espota2.probe_ota_key(network_devices, remote_port, new_key): + return fail( + "The device did not answer with the new key. The configuration " + "keeps it, with the previous key as old_key, so the next install " + "reaches the device either way." + ) + except KeyboardInterrupt: + return fail( + "Interrupted; the device's key was not confirmed" + if uploaded + else "Interrupted before the upload" + ) + finally: + # Before the upload nothing changed on the device; after it the device + # most likely runs the new key, and old_key covers the other case + if uploaded: + safe_print(f"New OTA encryption key: {color(AnsiFore.CYAN, new_key)}") + else: + _restore(originals, keys) + + safe_print(color(AnsiFore.BOLD_GREEN, "SUCCESS")) + return 0 + + +def _restore(originals: dict[Path, Snapshot], keys: str) -> None: + """Put the previous key back, and drop the images a build made with the + new key; both keys are shown when a step fails.""" + files = ", ".join(str(p) for p in originals) + try: + restore_files(originals) + except (EsphomeError, KeyboardInterrupt) as err: + safe_print( + color(AnsiFore.BOLD_RED, str(err) or f"Interrupted restoring {files}") + ) + safe_print(keys) + else: + safe_print(color(AnsiFore.BOLD_YELLOW, f"Restored the previous key in {files}")) + build_dir = CORE.firmware_bin.parent + failed: list[str] = [] + images: list[Path] = [] + try: + images = [p for s in (".bin", ".uf2") for p in build_dir.glob(f"*{s}")] + for image in images: # every one is tried, whatever the others did + try: + image.unlink() + except OSError as err: + failed.append(f"{image}: {err}") + except (OSError, KeyboardInterrupt) as err: + failed.append(f"{build_dir}: {str(err) or 'interrupted'}") + if failed: + safe_print( + color( + AnsiFore.BOLD_RED, + "Could not remove images that may hold the new key: " + + "; ".join(failed), + ) + ) + safe_print(keys) + elif images: + safe_print( + color( + AnsiFore.BOLD_YELLOW, + f"Removed the images in {build_dir}, which may hold the new key; " + "compile again before the next upload.", + ) + ) diff --git a/esphome/ota_key_edit.py b/esphome/ota_key_edit.py new file mode 100644 index 0000000000..9bf4404958 --- /dev/null +++ b/esphome/ota_key_edit.py @@ -0,0 +1,352 @@ +"""Rewrite the OTA encryption key in place: the `key:` line the loader read +it from, or the secrets.yaml line a `!secret` on it points to.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import logging +import os +from pathlib import Path +import re + +from esphome.components.noise import static_encryption_key +from esphome.const import CONF_API, CONF_ENCRYPTION, CONF_KEY +from esphome.core import CORE, EsphomeError +from esphome.espota2 import CONF_OLD_KEY, esphome_ota_items +from esphome.types import ConfigType +from esphome.yaml_edit import ( + LineEdit, + editable_file, + field_line_re, + own_documents, + read_text, + rewrite, + source_line, + source_of, +) +from esphome.yaml_util import secrets_path_for + +_LOGGER = logging.getLogger(__name__) +_DOMAIN = "ota_key_edit" + + +@dataclass +class KeyEdit(LineEdit): + """A line edit with the secret it holds and the other configs using it.""" + + secret: str | None = None + shared_with: list[Path] = field(default_factory=list) + + +def _key_blocks(raw: ConfigType, key: str) -> list[ConfigType]: + """The api and esphome ota ``encryption:`` mappings whose key is ``key``.""" + blocks = [raw.get(CONF_API) or {}, *esphome_ota_items(raw)] + # A bare ota block gets the api key at validation; it has no key line + return [ + block[CONF_ENCRYPTION] + for block in blocks + if (node := static_encryption_key(block)) is not None + and str(node) == key + and source_of(block[CONF_ENCRYPTION], CONF_KEY) is not None + ] + + +_INDENT_RE = re.compile(r"\s*") +_KEY_FIELD_RE = re.compile(rf"^\s*{CONF_KEY}\s*:") +_BARE_BLOCK_RE = re.compile(rf"^\s*{CONF_ENCRYPTION}\s*:\s*(#.*)?$") +_YAML_SUFFIXES = (".yaml", ".yml") +_COMMENT_RE = re.compile(r"(?:^|\s)#.*$") +_INCLUDE_DIR_RE = re.compile(r"!include_dir_\w+\s+(?P[^\s#]+)") +# An include or package file whose path holds a substitution +_SUBSTITUTED_INCLUDE_RE = re.compile( + r"(?:!include\w*\s+|\bfile:\s*)[\"']?(?P[^\s#\"']*\$[^\s#\"']*)" +) + + +def _secret_ref(name: str = r"[^\s#\"']+") -> str: + """`!secret name` as the whole scalar; `!secret ota key` is not `ota`.""" + return rf"!secret\s+(?P[\"']?)(?P{name})(?P=q)(?=\s*(?:#|$))" + + +def _secret_field_re(name: str) -> re.Pattern[str]: + """``name: !secret ``.""" + return re.compile(rf"^\s*{name}\s*:\s*{_secret_ref()}") + + +def _secret_use_re(name: str) -> re.Pattern[str]: + """A use of the secret ``name`` anywhere on a line.""" + return re.compile(_secret_ref(re.escape(name))) + + +_KEY_SECRET_RE = _secret_field_re(CONF_KEY) +_OLD_KEY_SECRET_RE = _secret_field_re(CONF_OLD_KEY) + + +def _indent(text: str) -> str: + return _INDENT_RE.match(text).group() + + +def _without_comment(text: str) -> str: + """The line up to its comment, which starts at a `#` after whitespace.""" + return _COMMENT_RE.sub("", text) + + +def _file_use_re(file_name: str) -> re.Pattern[str]: + """The file name at a path boundary; `wifi_api.yaml` is not `api.yaml`.""" + return re.compile(rf"(? str: + """The indent of the first content line; a root mapping may be indented.""" + content = ( + t for t in lines if t.strip() and not t.strip().startswith(("#", "---", "%")) + ) + return next((_indent(t) for t in content), "") + + +def _config_texts() -> tuple[list[tuple[Path, str]], list[str]]: + """Every yaml file under the config dir with its text, and what could + not be read (a symlinked directory is not walked); walked once per run.""" + if (found := CORE.data.get(_DOMAIN)) is None: + found = CORE.data[_DOMAIN] = _walk_config_texts() + return found + + +def _walk_config_texts() -> tuple[list[tuple[Path, str]], list[str]]: + data_dir = CORE.data_dir.resolve() + found = [] + skipped: list[str] = [] + + def skip(what: object) -> None: + skipped.append(str(what)) + _LOGGER.debug("Skipping %s", what) + + for dirpath, dirnames, filenames in os.walk( + CORE.config_dir.resolve(), onerror=skip + ): + for name in list(dirnames): + path = Path(dirpath, name) + if path == data_dir: + dirnames.remove(name) + elif path.is_symlink(): + dirnames.remove(name) + skip(f"{path} is a link") + for filename in filenames: + path = Path(dirpath, filename) + if path.suffix not in _YAML_SUFFIXES: + continue + try: + text = read_text(path) + except EsphomeError as err: + skip(err) + continue + # Searched, never edited: comments do not count + found.append((path, "\n".join(map(_without_comment, text.splitlines())))) + return sorted(found), skipped + + +def with_sharers(edits: list[KeyEdit]) -> list[str]: + """Fill in ``shared_with``: the other configs that use each rewritten + line's secret or include, directly or through further includes. Returns + what the scan could not follow (unreadable files, directory and + substituted includes).""" + main = CORE.config_path.resolve() + own = own_documents() + # This config's own includes stay in: another device may reach a line + # through one of them + texts, skipped = _config_texts() + texts = [(path, text) for path, text in texts if path.resolve() != main] + unchecked = [*skipped] + for path, text in texts: + if match := _INCLUDE_DIR_RE.search(text): + unchecked.append(f"{path} includes the directory {match['dir']}") + if match := _SUBSTITUTED_INCLUDE_RE.search(text): + unchecked.append( + f"{path} includes {match['target']} through a substitution" + ) + for edit in edits: + if edit.insert_after: + continue + ref = ( + _secret_use_re(edit.secret) if edit.secret else _file_use_re(edit.path.name) + ) + users = {path for path, text in texts if ref.search(text)} + frontier = users + while frontier: + frontier = { + path + for name in {p.name for p in frontier} + for path, text in texts + if path not in users and _file_use_re(name).search(text) + } + users |= frontier + edit.shared_with = sorted(p for p in users if p.resolve() not in own) + return unchecked + + +def _secret_file(doc: Path) -> Path: + return editable_file(secrets_path_for(doc)) + + +def _secret_line( + secrets_path: Path, name: str, value: str | None = None +) -> tuple[int, re.Match[str]] | None: + """The root level ``name:`` line, None when absent.""" + lines = read_text(secrets_path).splitlines() + line_re = field_line_re(name, value, re.escape(_root_indent(lines))) + return next( + ((i, m) for i, text in enumerate(lines) if (m := line_re.match(text))), None + ) + + +def _secret_rewrite( + secrets_path: Path, name: str, value: str, expect: str | None = None +) -> KeyEdit | None: + """Set the root level ``name:`` line to ``value``; None when it is not a + plain scalar holding ``expect``.""" + if (hit := _secret_line(secrets_path, name, expect)) is None: + return None + i, m = hit + return KeyEdit(secrets_path, i, m.string, rewrite(m, value), secret=name) + + +def _secret_insert(secrets_path: Path, after: str, name: str, value: str) -> KeyEdit: + """Add ``name: value`` under the root level ``after`` line.""" + if (hit := _secret_line(secrets_path, after)) is None: + raise EsphomeError( + f"No plain '{after}:' line in {secrets_path}; edit it by hand" + ) + i, m = hit + return KeyEdit( + secrets_path, + i, + m.string, + f'{_indent(m.string)}{name}: "{value}"', + insert_after=True, + ) + + +def locate_key_edits(old_key: str, new_key: str) -> list[KeyEdit]: + """The edits that replace the current key; refuses a substitution, flow + mapping or remote package.""" + literal_re = field_line_re(CONF_KEY, old_key) + blocks = _key_blocks(CORE.raw_config or {}, old_key) + if not blocks: + raise EsphomeError("The current key was not found on a line of the yaml") + edits: list[KeyEdit] = [] + key_lines: set[tuple[Path, int]] = set() + for block in blocks: + doc, line_no, text = source_line(block, CONF_KEY) + key_lines.add((doc.resolve(), line_no)) + if match := _KEY_SECRET_RE.match(text): + name, secrets_path = match["name"], _secret_file(doc) + edit = _secret_rewrite(secrets_path, name, new_key, expect=old_key) + if edit is None: + raise EsphomeError( + f"No '{name}:' line with the current key in {secrets_path}" + ) + elif match := literal_re.match(text): + edit = KeyEdit(doc, line_no, text, rewrite(match, new_key)) + else: + raise EsphomeError( + f"{doc}:{line_no + 1} does not hold the key as a plain value or " + "a !secret; edit the key by hand" + ) + # The api and ota blocks may share one !secret line + if edit not in edits: + edits.append(edit) + for edit in edits: + if edit.secret: + _refuse_other_own_uses(edit.secret, key_lines) + return edits + + +def _refuse_other_own_uses(name: str, allowed: set[tuple[Path, int]]) -> None: + """The secret may only feed the ``allowed`` lines; a wifi password on it + would change too.""" + ref = _secret_use_re(name) + for doc in own_documents(): + for i, text in enumerate(read_text(doc).splitlines()): + if ref.search(_without_comment(text)) and (doc, i) not in allowed: + raise EsphomeError( + f"'{name}' is also used at {doc}:{i + 1}; a rotation would " + "change that value too, edit the key by hand" + ) + + +def old_key_edit(old_key: str, secret: str | None = None) -> list[KeyEdit]: + """Set ``old_key:`` on the esphome ota block, rewriting or adding it. A + key from secrets.yaml (``secret``) stays there as ``_old``. Of + several entries the one with its own old_key or key line gets it.""" + items = [ + (item, item[CONF_ENCRYPTION] or {}) + for item in esphome_ota_items(CORE.raw_config or {}) + if CONF_ENCRYPTION in item + ] + if not items: + raise EsphomeError("The esphome OTA platform has no 'encryption:' block") + item, encryption = next( + ( + pair + for name in (CONF_OLD_KEY, CONF_KEY) + for pair in items + if source_of(pair[1], name) is not None + ), + items[0], + ) + if (existing := encryption.get(CONF_OLD_KEY)) is not None: + doc, line_no, text = source_line(encryption, CONF_OLD_KEY) + if match := _OLD_KEY_SECRET_RE.match(text): + name, secrets_path = match["name"], _secret_file(doc) + if (edit := _secret_rewrite(secrets_path, name, old_key)) is None: + raise EsphomeError( + f"No plain '{name}:' line in {secrets_path}; edit it by hand" + ) + _refuse_other_own_uses(name, {(doc.resolve(), line_no)}) + return [edit] + match = field_line_re(CONF_OLD_KEY, str(existing)).match(text) + if match is None: + raise EsphomeError( + f"{doc}:{line_no + 1} does not hold '{CONF_OLD_KEY}' as a plain " + "value; edit it by hand" + ) + return [KeyEdit(doc, line_no, text, rewrite(match, old_key))] + # Indent from the block's own key line, else one level under `encryption:` + if source_of(encryption, CONF_KEY) is not None: + mapping, anchor, anchor_re, step = encryption, CONF_KEY, _KEY_FIELD_RE, "" + else: + mapping, anchor, anchor_re, step = item, CONF_ENCRYPTION, _BARE_BLOCK_RE, " " + doc, line_no, text = source_line(mapping, anchor) + if not anchor_re.match(text): + raise EsphomeError( + f"{doc}:{line_no + 1} does not hold '{anchor}:' as a block line; " + "edit it by hand" + ) + if secret is None: + rendered = f'{CONF_OLD_KEY}: "{old_key}"' + extra = [] + else: + rendered = f"{CONF_OLD_KEY}: !secret {secret}_old" + secrets_path = _secret_file(doc) # the file that gets old_key + # An existing `_old` line is reused only when nothing uses it + if ( + kept := _secret_rewrite(secrets_path, f"{secret}_old", old_key) + ) is not None: + ref = _secret_use_re(f"{secret}_old") + texts, skipped = _config_texts() + if skipped: # an unread file might use the line + raise EsphomeError( + "Could not read every configuration to check the secret: " + + skipped[0] + ) + if users := [p for p, t in texts if ref.search(t)]: + raise EsphomeError( + f"'{secret}_old:' in {secrets_path} is used by " + + ", ".join(str(p) for p in users) + + f"; rename it or add '{CONF_OLD_KEY}' by hand" + ) + extra = [kept or _secret_insert(secrets_path, secret, f"{secret}_old", old_key)] + edit = KeyEdit( + doc, line_no, text, f"{_indent(text)}{step}{rendered}", insert_after=True + ) + return [edit, *extra] diff --git a/esphome/yaml_util.py b/esphome/yaml_util.py index 0d6196e93f..0331eb1bd8 100644 --- a/esphome/yaml_util.py +++ b/esphome/yaml_util.py @@ -926,9 +926,7 @@ class ESPHomeLoaderMixin: if self.name == CORE.config_path: raise e try: - main_config_dir = CORE.config_path.parent - main_secret_yml = main_config_dir / SECRET_YAML - secrets = self.yaml_loader(main_secret_yml) + secrets = self.yaml_loader(CORE.config_path.parent / SECRET_YAML) except EsphomeError as er: raise EsphomeError(f"{e}\n{er}") from er @@ -1301,6 +1299,20 @@ def find_files(directory: Path, pattern: str) -> Iterator[Path]: yield filename +def secrets_path_for(document: Path) -> Path: + """The secrets.yaml a ``!secret`` in ``document`` resolves against: the + one beside the document when it loads, else the main config's, the + way construct_secret falls back.""" + beside = document.parent / SECRET_YAML + if document == CORE.config_path: + return beside + try: + load_yaml(beside, clear_secrets=False, track_document_range=False) + except EsphomeError: + return CORE.config_path.parent / SECRET_YAML + return beside + + def is_secret(value): try: return _SECRET_VALUES[str(value)] diff --git a/tests/unit_tests/cli/test_rotate_key.py b/tests/unit_tests/cli/test_rotate_key.py new file mode 100644 index 0000000000..a2e4a1e8ab --- /dev/null +++ b/tests/unit_tests/cli/test_rotate_key.py @@ -0,0 +1,471 @@ +"""Tests for ``esphome rotate-key``.""" + +from __future__ import annotations + +from collections.abc import Generator +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from unittest.mock import ANY, Mock, patch + +import pytest + +from esphome import yaml_util +from esphome.cli.rotate_key import command_rotate_key +from esphome.const import ( + CONF_API, + CONF_ENCRYPTION, + CONF_ESPHOME, + CONF_KEY, + CONF_OTA, + CONF_PLATFORM, + CONF_PORT, + CONF_WEB_SERVER, + KEY_CORE, + KEY_TARGET_PLATFORM, + PLATFORM_ESP32, +) +from esphome.core import CORE, EsphomeError, Toolchain +from esphome.ota_key_edit import KeyEdit +from esphome.upload_targets import PortType +from esphome.util import ESPHOME_COMMAND +from esphome.yaml_edit import RestoreError + + +@dataclass +class MockArgs: + device: list[str] | None = None + prompt_new_key: bool = False + yes: bool = False + substitution: list[list[str]] | None = None + toolchain: Toolchain | None = None + + +def _interrupt_after(writes: int) -> Any: + """A write_keeping_mode that lets ``writes`` calls through, then Ctrl-C.""" + from esphome.yaml_edit import write_keeping_mode + + calls = 0 + + def write(*args: Any, **kwargs: Any) -> None: + nonlocal calls + calls += 1 + if calls > writes: + raise KeyboardInterrupt + write_keeping_mode(*args, **kwargs) + + return write + + +def setup_core(tmp_path: Path) -> None: + CORE.reset() + CORE.config_path = tmp_path / "test.yaml" + CORE.name = "test" + CORE.build_path = tmp_path / ".esphome" / "build" / "test" + CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32} + + +OLD_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" +NEW_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA=" +API_YAML = f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" + +ota: + - platform: esphome + encryption: +""" +OTA_CONF = { + CONF_PLATFORM: CONF_ESPHOME, + CONF_PORT: 3232, + CONF_ENCRYPTION: {CONF_KEY: OLD_KEY}, +} + + +@pytest.fixture +def env(tmp_path: Path) -> Generator[dict[str, Mock]]: + """A config with an inline api key, loaded the way read_config does so + the key node carries its source range, and every network step mocked.""" + setup_core(tmp_path) + CORE.config_path.write_text(API_YAML, encoding="utf-8") + CORE.raw_config = yaml_util.load_yaml(CORE.config_path) + CORE.config = { + CONF_API: {CONF_ENCRYPTION: {CONF_KEY: OLD_KEY}}, + CONF_OTA: [OTA_CONF], + } + with ( + patch("esphome.__main__.choose_upload_log_host", return_value=["dev.local"]), + patch("esphome.__main__._resolve_network_devices", return_value=["dev.local"]), + patch("esphome.cli.rotate_key.get_port_type", return_value=PortType.NETWORK), + patch("esphome.espota2.probe_ota_key", return_value=True) as probe, + patch( + "esphome.cli.rotate_key.run_external_process", return_value=0 + ) as compile_, + patch("esphome.espota2.run_ota") as upload, + patch("esphome.cli.rotate_key.safe_input", return_value="y") as confirm, + patch("esphome.cli.rotate_key.sys.stdin") as stdin, + patch( + "esphome.components.noise.secrets.token_bytes", + return_value=bytes(range(1, 33)), + ), + ): + stdin.isatty.return_value = True + # The real upload reports the connection before its result + upload.return_value = (0, "dev.local") + + def upload_after_connecting( + *_args: Any, **kwargs: Any + ) -> tuple[int, str | None]: + kwargs["on_connect"]() + return upload.return_value + + upload.side_effect = upload_after_connecting + yield { + "probe": probe, + "compile": compile_, + "upload": upload, + "confirm": confirm, + "stdin": stdin, + } + + +def test_success(env: dict[str, Mock], capfd: pytest.CaptureFixture[str]) -> None: + """Generate, write, compile in a child, upload with the old key, confirm + with the new one, print it.""" + args = MockArgs() + assert command_rotate_key(args, CORE.config) == 0 + + assert ( + CORE.config_path.read_text() + == API_YAML.replace(OLD_KEY, NEW_KEY) + f' old_key: "{OLD_KEY}"\n' + ) + env["confirm"].assert_called_once() + precheck, confirm = env["probe"].call_args_list + assert precheck.args[2] == OLD_KEY + assert precheck.kwargs["retry_rejected"] is False + assert confirm.args[2] == NEW_KEY + compile_args = env["compile"].call_args.args + assert compile_args[-2:] == ("compile", str(CORE.config_path)) + env["upload"].assert_called_once_with( + ["dev.local"], + 3232, + None, + CORE.firmware_bin, + noise_psk=OLD_KEY, + plaintext_fallback=False, + on_connect=ANY, + ) + out = capfd.readouterr().out + assert NEW_KEY in out + assert "SUCCESS" in out + + +def test_yes_skips_confirmation( + env: dict[str, Mock], +) -> None: + assert command_rotate_key(MockArgs(yes=True), CORE.config) == 0 + env["confirm"].assert_not_called() + + +def test_declined(env: dict[str, Mock]) -> None: + env["confirm"].return_value = "n" + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + env["probe"].assert_not_called() + + +def test_warns_about_a_shared_secret( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + """Other configurations using the rewritten secret are named before the + confirmation.""" + + def scan(edits: list[KeyEdit]) -> list[str]: + edits[0].secret = "device_key" + edits[0].shared_with = [CORE.config_dir / "b.yaml"] + return ["c.yaml is a link"] + + with patch("esphome.cli.rotate_key.with_sharers", side_effect=scan): + env["confirm"].return_value = "n" + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert "b.yaml (secret 'device_key')" in out + assert "Could not check every configuration for shared use: c.yaml is a link" in out + + +def test_no_terminal_needs_yes( + env: dict[str, Mock], +) -> None: + env["stdin"].isatty.return_value = False + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + + +def test_own_ota_key_asks_nothing( + env: dict[str, Mock], +) -> None: + """Without an api key nothing changes for Home Assistant.""" + config = {CONF_OTA: [OTA_CONF]} + assert command_rotate_key(MockArgs(), config) == 0 + env["confirm"].assert_not_called() + assert NEW_KEY in CORE.config_path.read_text() + + +def test_prompted_new_key(env: dict[str, Mock]) -> None: + key = "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE=" + with patch("esphome.cli.rotate_key.read_secret_line", return_value=key): + assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 0 + assert key in CORE.config_path.read_text() + assert env["probe"].call_args_list[1].args[2] == key + + +@pytest.mark.parametrize("key", ["not-base64", OLD_KEY], ids=["invalid", "same"]) +def test_rejects_bad_new_key(env: dict[str, Mock], key: str) -> None: + with patch("esphome.cli.rotate_key.read_secret_line", return_value=key): + assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + env["probe"].assert_not_called() + + +def test_precheck_fails_writes_nothing( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + env["probe"].return_value = False + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + env["compile"].assert_not_called() + assert "did not complete an encrypted handshake" in capfd.readouterr().out + + +@pytest.mark.parametrize("step", ["compile", "interrupt"]) +def test_restores_before_the_upload( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str], step: str +) -> None: + if step == "compile": + env["compile"].return_value = 1 + else: + env["compile"].side_effect = KeyboardInterrupt + image = CORE.firmware_bin + image.parent.mkdir(parents=True) + factory = image.with_name("firmware.factory.bin") + for built in (image, factory): + built.write_bytes(b"built with the new key") + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + assert env["probe"].call_count == 1 + out = capfd.readouterr().out + assert ("Interrupted before the upload" in out) is (step == "interrupt") + # The images may hold the new key, so no upload can install them + assert not image.exists() and not factory.exists() + assert "compile again before the next upload" in out + assert NEW_KEY not in out + + +def test_a_build_that_cannot_be_removed_shows_the_new_key( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + """Every image is tried; the ones that stay are all named.""" + env["compile"].return_value = 1 + image = CORE.firmware_bin + image.mkdir(parents=True) # unlink fails on a directory + stuck = image.with_name("firmware.factory.bin") + stuck.mkdir() + removable = image.with_name("firmware.ota.bin") + removable.write_bytes(b"") + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert str(image) in out and str(stuck) in out + assert not removable.exists() + assert OLD_KEY in out and NEW_KEY in out + + +def test_an_interrupt_during_the_image_cleanup_shows_the_keys( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + env["compile"].return_value = 1 + CORE.firmware_bin.parent.mkdir(parents=True) + CORE.firmware_bin.write_bytes(b"") + with patch("pathlib.Path.glob", side_effect=KeyboardInterrupt): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert f"{CORE.firmware_bin.parent}: interrupted" in out + assert OLD_KEY in out and NEW_KEY in out + + +def test_a_build_from_before_the_rotation_is_kept( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + env["probe"].return_value = False + image = CORE.firmware_bin + image.parent.mkdir(parents=True) + image.write_bytes(b"built with the current key") + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert image.exists() + assert "compile again" not in capfd.readouterr().out + + +@pytest.mark.parametrize("rollback", ["succeeded", "failed"]) +def test_an_interrupt_while_writing_reports_a_failed_rollback( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str], rollback: str +) -> None: + """apply_line_edits keeps an interrupt an interrupt and notes a rollback + that failed too; then a file still holds the new key, so both are shown.""" + interrupt = KeyboardInterrupt() + if rollback == "failed": + interrupt.add_note("Could not restore test.yaml: disk") + with patch("esphome.cli.rotate_key.apply_line_edits", side_effect=interrupt): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert ("Could not restore test.yaml: disk" in out) is (rollback == "failed") + assert (NEW_KEY in out) is (rollback == "failed") + assert ("Interrupted before the upload" in out) is (rollback == "succeeded") + + +def test_a_second_interrupt_during_the_restore_shows_the_keys( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + env["compile"].return_value = 1 + with patch("esphome.yaml_edit.write_keeping_mode", side_effect=_interrupt_after(1)): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert f"Could not restore {CORE.config_path}: interrupted" in out + assert OLD_KEY in out and NEW_KEY in out + + +def test_restores_when_the_device_was_never_reached( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + """An upload that fails before a connection opens cannot have committed.""" + env["upload"].side_effect = None + env["upload"].return_value = (1, None) + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert CORE.config_path.read_text() == API_YAML + out = capfd.readouterr().out + assert "did not reach the device" in out + assert "put back" not in out + assert "Restored the previous key" in out + + +def test_keeps_the_edit_after_a_failed_upload( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + """An attempted upload may have committed, so the config keeps both keys + and the new one is printed.""" + env["upload"].return_value = (1, None) + assert command_rotate_key(MockArgs(), CORE.config) == 1 + text = CORE.config_path.read_text() + assert NEW_KEY in text and f'old_key: "{OLD_KEY}"' in text + assert NEW_KEY in capfd.readouterr().out + assert env["probe"].call_count == 1 + + +def test_shows_both_keys_only_when_a_file_stayed_rewritten( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + with patch( + "esphome.cli.rotate_key.apply_line_edits", side_effect=RestoreError("stuck") + ): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert "stuck" in out and OLD_KEY in out and NEW_KEY in out + + +def test_reports_a_failed_restore( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str] +) -> None: + env["compile"].return_value = 1 + with patch( + "esphome.cli.rotate_key.restore_files", + side_effect=RestoreError("Could not restore x"), + ): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert "Could not restore x" in out + assert OLD_KEY in out and NEW_KEY in out + + +@pytest.mark.parametrize("outcome", ["unconfirmed", "interrupted"]) +def test_keeps_the_new_key_after_the_upload( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str], outcome: str +) -> None: + """Once the upload went through the device most likely runs the new key + and old_key covers the other case, so the files stay and the key is + printed.""" + env["probe"].side_effect = [ + True, + KeyboardInterrupt if outcome == "interrupted" else False, + ] + assert command_rotate_key(MockArgs(), CORE.config) == 1 + assert NEW_KEY in CORE.config_path.read_text() + assert f'old_key: "{OLD_KEY}"' in CORE.config_path.read_text() + out = capfd.readouterr().out + assert NEW_KEY in out + assert ("Interrupted" in out) is (outcome == "interrupted") + + +def test_child_compile_gets_global_options_first( + env: dict[str, Mock], +) -> None: + """The compile parser is strict, so -s and --toolchain precede it.""" + args = MockArgs(substitution=[["name", "kitchen"]]) + CORE.dashboard = True + args.toolchain = Toolchain.PLATFORMIO + assert command_rotate_key(args, CORE.config) == 0 + assert env["compile"].call_args.args[len(ESPHOME_COMMAND) :] == ( + "--dashboard", + "--toolchain", + "platformio", + "-s", + "name", + "kitchen", + "compile", + str(CORE.config_path), + ) + + +@pytest.mark.parametrize( + ("config", "port_type", "match"), + [ + ( + {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]}, + "NETWORK", + "needs 'encryption:'", + ), + ({CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]}, "NETWORK", "esphome OTA"), + ({CONF_OTA: [OTA_CONF]}, PortType.SERIAL, "over the air"), + ], + ids=["no_encryption", "web_server_only", "serial"], +) +def test_refuses( + env: dict[str, Mock], + capfd: pytest.CaptureFixture[str], + config: dict[str, Any], + port_type: str, + match: str, +) -> None: + with patch("esphome.cli.rotate_key.get_port_type", return_value=port_type): + assert command_rotate_key(MockArgs(), config) == 1 + assert match in capfd.readouterr().out + env["probe"].assert_not_called() + + +def test_refuses_host(env: dict[str, Mock]) -> None: + CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = "host" + assert command_rotate_key(MockArgs(), CORE.config) == 1 + env["probe"].assert_not_called() + + +@pytest.mark.parametrize("helper", ["locate_key_edits", "apply_line_edits"]) +def test_reports_edit_errors( + env: dict[str, Mock], capfd: pytest.CaptureFixture[str], helper: str +) -> None: + with patch(f"esphome.cli.rotate_key.{helper}", side_effect=EsphomeError("nope")): + assert command_rotate_key(MockArgs(), CORE.config) == 1 + out = capfd.readouterr().out + assert "nope" in out + # A refusal, or an edit rolled back cleanly, shows no key + assert NEW_KEY not in out + assert CORE.config_path.read_text() == API_YAML diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 801eda4496..5ccfcfd469 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -8087,3 +8087,32 @@ def test_command_rename_is_dispatched_to_the_cli_module() -> None: with patch("esphome.cli.rename.command_rename", return_value=7) as run: assert main.command_rename(args, {}) == 7 run.assert_called_once_with(args, {}) + + +def test_command_rotate_key_is_dispatched_to_the_cli_module() -> None: + args = MockArgs() + with patch("esphome.cli.rotate_key.command_rotate_key", return_value=7) as run: + assert main.command_rotate_key(args, {}) == 7 + run.assert_called_once_with(args, {}) + + +def test_parse_args_rotate_key() -> None: + args = parse_args( + [ + "esphome", + "rotate-key", + "--device", + "dev.local", + "--prompt-new-key", + "-y", + "--username", + "mqtt-user", + "device.yaml", + ] + ) + assert args.command == "rotate-key" + assert args.device == ["dev.local"] + assert args.prompt_new_key is True + assert args.yes is True + # MQTT device resolution reads these like upload does + assert args.username == "mqtt-user" diff --git a/tests/unit_tests/test_ota_key_edit.py b/tests/unit_tests/test_ota_key_edit.py new file mode 100644 index 0000000000..88610d0fb6 --- /dev/null +++ b/tests/unit_tests/test_ota_key_edit.py @@ -0,0 +1,1114 @@ +"""Tests for rewriting the OTA encryption key in place.""" + +from __future__ import annotations + +from pathlib import Path +import sys +from unittest.mock import patch + +import pytest + +from esphome import yaml_util +from esphome.config import do_substitution_pass +from esphome.core import CORE, EsphomeError +from esphome.ota_key_edit import KeyEdit, locate_key_edits, old_key_edit, with_sharers +from esphome.yaml_edit import Snapshot, apply_line_edits, restore_files + +OLD_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=" +OLDER_KEY = "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE=" +NEW_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA=" + +API_YAML = f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" # shared with ota + +ota: + - platform: esphome + encryption: +""" + +SECRET_YAML = """esphome: + name: test + +api: + encryption: + key: !secret device_key + +ota: + - platform: esphome + encryption: + key: !secret device_key +""" + + +def _setup(tmp_path: Path, yaml_text: str, secrets: str | None = None) -> Path: + """Write the yaml (and secrets.yaml), point CORE at it and load the raw + config the way read_config does, so key nodes carry their source range.""" + CORE.reset() + CORE.config_path = tmp_path / "test.yaml" + # Bytes, so Windows does not turn the newlines into CRLF on the way in + CORE.config_path.write_bytes(yaml_text.encode()) + if secrets is not None: + (tmp_path / "secrets.yaml").write_bytes(secrets.encode()) + CORE.raw_config = do_substitution_pass(yaml_util.load_yaml(CORE.config_path), None) + return CORE.config_path + + +def _edits() -> list[KeyEdit]: + """Everything a rotation writes, the way the command collects it.""" + edits = locate_key_edits(OLD_KEY, NEW_KEY) + secret = next((e.secret for e in edits if e.secret), None) + edits += old_key_edit(OLD_KEY, secret) + with_sharers(edits) + return edits + + +def _rotate() -> list[KeyEdit]: + edits = _edits() + apply_line_edits(edits) + return edits + + +def test_inline_api_key(tmp_path: Path) -> None: + """A bare ota block inherits the api key, so the api line is the one + rewritten, quotes and comment kept.""" + path = _setup(tmp_path, API_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(path, 5)] + assert edits[0].new_line == f' key: "{NEW_KEY}" # shared with ota' + originals = apply_line_edits(edits) + assert originals == {path: Snapshot(API_YAML, API_YAML.replace(OLD_KEY, NEW_KEY))} + assert path.read_text() == API_YAML.replace(OLD_KEY, NEW_KEY) + restore_files(originals) + assert path.read_text() == API_YAML + + +def test_shared_secret(tmp_path: Path) -> None: + """Both blocks pointing at one secret give one edit in secrets.yaml, and + only that line changes: another secret with the same value stays.""" + secrets = f"""wifi_password: hunter2 +device_key: '{OLD_KEY}' # keep +other_key: "{OLD_KEY}" +""" + _setup(tmp_path, SECRET_YAML, secrets) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(tmp_path / "secrets.yaml", 1)] + apply_line_edits(edits) + assert (tmp_path / "secrets.yaml").read_text() == secrets.replace( + f"device_key: '{OLD_KEY}'", f"device_key: '{NEW_KEY}'" + ) + assert CORE.config_path.read_text() == SECRET_YAML + + +def test_explicit_ota_key_next_to_api(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +api: + encryption: + key: {OLD_KEY} + +ota: + - platform: esphome + encryption: + key: {OLD_KEY} +""" + path = _setup(tmp_path, yaml_text) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert sorted(e.line for e in edits) == [5, 10] + apply_line_edits(edits) + assert path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY) + + +def test_secret_api_key_next_to_a_literal_ota_key(tmp_path: Path) -> None: + """An explicit ota key must match the api key at validation; both the + secrets line and the literal line change so they still do.""" + yaml_text = SECRET_YAML.replace( + " key: !secret device_key\n", f' key: "{OLD_KEY}"\n' + ) + _setup(tmp_path, yaml_text, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert sorted(e.path.name for e in edits) == ["secrets.yaml", "test.yaml"] + apply_line_edits(edits) + assert (tmp_path / "secrets.yaml").read_text() == f"device_key: {NEW_KEY}\n" + assert CORE.config_path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY) + + +def test_key_in_an_included_file(tmp_path: Path) -> None: + """A key that lives in an included file is rewritten there.""" + (tmp_path / "api.yaml").write_text( + f"""encryption: + key: "{OLD_KEY}" +""", + encoding="utf-8", + ) + yaml_text = """esphome: + name: test + +api: !include api.yaml + +ota: + - platform: esphome + encryption: +""" + _setup(tmp_path, yaml_text) + # Includes are deferred until the substitution pass, as in read_config + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(tmp_path / "api.yaml", 1)] + apply_line_edits(edits) + assert NEW_KEY in (tmp_path / "api.yaml").read_text() + assert CORE.config_path.read_text() == yaml_text + + +def test_own_ota_key_without_api(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +mqtt: + broker: broker.local + +ota: + - platform: esphome + encryption: + key: "{OLD_KEY}" +""" + _setup(tmp_path, yaml_text) + assert [e.line for e in locate_key_edits(OLD_KEY, NEW_KEY)] == [9] + + +@pytest.mark.parametrize( + "yaml_text", + [ + f"""esphome: + name: test +substitutions: + k: "{OLD_KEY}" +api: + encryption: + key: ${{k}} +ota: + - platform: esphome + encryption: +""", + f"""esphome: + name: test +api: + encryption: {{key: "{OLD_KEY}"}} +ota: + - platform: esphome + encryption: +""", + ], + ids=["substitution", "flow_mapping"], +) +def test_refuses_what_it_cannot_rewrite(tmp_path: Path, yaml_text: str) -> None: + _setup(tmp_path, yaml_text) + with pytest.raises(EsphomeError, match="edit the key by hand"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_refuses_a_secret_defined_elsewhere(tmp_path: Path) -> None: + """A secret whose value the line regex cannot find, for example a folded + scalar, is refused rather than guessed at.""" + secrets = f"device_key: >-\n {OLD_KEY}\n" + _setup(tmp_path, SECRET_YAML, secrets) + with pytest.raises(EsphomeError, match="No 'device_key:' line"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_refuses_a_line_that_changed_since_it_was_located(tmp_path: Path) -> None: + path = _setup(tmp_path, API_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + path.write_bytes(API_YAML.replace(" # shared with ota", "").encode()) + with pytest.raises(EsphomeError, match="changed since it was read"): + apply_line_edits(edits) + + +def test_old_key_added_to_a_bare_block(tmp_path: Path) -> None: + """A bare block gets old_key one level in; the api line is rewritten.""" + path = _setup(tmp_path, API_YAML) + edits = _edits() + apply_line_edits(edits) + assert path.read_text() == ( + API_YAML.replace(OLD_KEY, NEW_KEY) + f' old_key: "{OLD_KEY}"\n' + ) + + +def test_old_key_added_after_an_explicit_key(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +mqtt: + broker: broker.local + +ota: + - platform: esphome + encryption: + key: "{OLD_KEY}" + port: 3232 +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace( + f' key: "{OLD_KEY}"\n', + f' key: "{NEW_KEY}"\n old_key: "{OLD_KEY}"\n', + ) + + +def test_old_key_rewritten_when_present(tmp_path: Path) -> None: + """A second rotation replaces the previous old_key in place.""" + yaml_text = f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" + +ota: + - platform: esphome + encryption: + old_key: '{OLDER_KEY}' # from the last rotation +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY).replace( + OLDER_KEY, OLD_KEY + ) + + +def test_ota_written_as_a_mapping(tmp_path: Path) -> None: + """A single-platform ota: written without the list dash is still found.""" + yaml_text = f"""esphome: + name: test + +ota: + platform: esphome + encryption: + key: "{OLD_KEY}" +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY) + ( + f' old_key: "{OLD_KEY}"\n' + ) + + +def test_keeps_windows_line_endings_and_a_bare_last_line(tmp_path: Path) -> None: + text = API_YAML.replace("\n", "\r\n").rstrip("\r\n") + path = _setup(tmp_path, text) + _rotate() + assert ( + path.read_bytes() + == (text.replace(OLD_KEY, NEW_KEY) + f'\r\n old_key: "{OLD_KEY}"').encode() + ) + + +def test_key_missing_from_the_yaml(tmp_path: Path) -> None: + _setup(tmp_path, API_YAML) + with pytest.raises(EsphomeError, match="was not found"): + locate_key_edits(NEW_KEY, OLD_KEY) + + +def test_key_without_a_source_location(tmp_path: Path) -> None: + """A raw config built in code has no ranges to edit from.""" + _setup(tmp_path, API_YAML) + CORE.raw_config = {"api": {"encryption": {"key": OLD_KEY}}} + with pytest.raises(EsphomeError, match="not found on a line"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_inherited_key_in_a_bare_block_is_not_a_line(tmp_path: Path) -> None: + """Final validate writes the api key into a bare ota block; that entry + has no line to rewrite and the api line is the only edit.""" + path = _setup(tmp_path, API_YAML) + CORE.raw_config["ota"][0]["encryption"] = ( + CORE.raw_config["ota"][0]["encryption"] or {} + ) + CORE.raw_config["ota"][0]["encryption"]["key"] = OLD_KEY + edits = _edits() + assert [(e.line, e.insert_after) for e in edits] == [(5, False), (9, True)] + apply_line_edits(edits) + assert path.read_text() == ( + API_YAML.replace(OLD_KEY, NEW_KEY) + f' old_key: "{OLD_KEY}"\n' + ) + + +def test_values_replaced_by_validation_still_locate(tmp_path: Path) -> None: + """Validation may swap a value for a plain string; the mapping key keeps + the range the edit needs.""" + yaml_text = API_YAML.replace( + " encryption:\n", f' encryption:\n old_key: "{NEW_KEY}"\n', 1 + ) + path = _setup(tmp_path, yaml_text) + api = CORE.raw_config["api"]["encryption"] + api["key"] = str(api["key"]) + ota = CORE.raw_config["ota"][0]["encryption"] + ota["old_key"] = str(ota["old_key"]) + _rotate() + assert path.read_text() == yaml_text.replace(OLD_KEY, "TMP").replace( + NEW_KEY, OLD_KEY + ).replace("TMP", NEW_KEY) + + +def test_key_from_the_data_dir_is_refused(tmp_path: Path) -> None: + """Remote packages are checked out under .esphome and are not the user's.""" + (tmp_path / ".esphome").mkdir() + (tmp_path / ".esphome" / "api.yaml").write_text( + f'encryption:\n key: "{OLD_KEY}"\n', encoding="utf-8" + ) + _setup( + tmp_path, + """esphome: + name: test + +api: !include .esphome/api.yaml + +ota: + - platform: esphome + encryption: +""", + ) + with pytest.raises(EsphomeError, match="not an editable file"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_secret_in_an_include_falls_back_to_the_main_secrets(tmp_path: Path) -> None: + (tmp_path / "sub").mkdir() + (tmp_path / "sub" / "api.yaml").write_text( + "encryption:\n key: !secret device_key\n", encoding="utf-8" + ) + _setup( + tmp_path, + """esphome: + name: test + +api: !include sub/api.yaml + +ota: + - platform: esphome + encryption: +""", + f"device_key: {OLD_KEY}\n", + ) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(tmp_path / "secrets.yaml", 0)] + # A file beside the include that does not parse is skipped by the + # loader too, so the main one is still the target + (tmp_path / "sub" / "secrets.yaml").write_bytes(b"a: [\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(tmp_path / "secrets.yaml", 0)] + # One beside the include that loads is the loader's first choice + (tmp_path / "sub" / "secrets.yaml").write_bytes(f"device_key: {OLD_KEY}\n".encode()) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(tmp_path / "sub" / "secrets.yaml", 0)] + + +def test_split_with_key_first_and_old_key_later(tmp_path: Path) -> None: + """old_key lands on the entry that already carries it, whichever entry + comes first; a second old_key would make the merged config inconsistent.""" + yaml_text = f"""esphome: + name: test + +ota: + - platform: esphome + encryption: + key: "{OLD_KEY}" + - platform: esphome + encryption: + old_key: "{OLDER_KEY}" +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY).replace( + OLDER_KEY, OLD_KEY + ) + + +def test_old_key_needs_an_encryption_block(tmp_path: Path) -> None: + _setup( + tmp_path, + f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" + +ota: + - platform: esphome +""", + ) + with pytest.raises(EsphomeError, match="no 'encryption:' block"): + old_key_edit(OLD_KEY) + + +def test_key_from_an_include_tag_is_refused(tmp_path: Path) -> None: + """`key: !include file` is indirect like a substitution; the value it + loads is not a line of this yaml.""" + (tmp_path / "key.yaml").write_bytes(f'"{OLD_KEY}"\n'.encode()) + _setup(tmp_path, API_YAML.replace(f'key: "{OLD_KEY}"', "key: !include key.yaml")) + with pytest.raises(EsphomeError, match="edit the key by hand"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_old_key_from_a_substitution_is_refused(tmp_path: Path) -> None: + _setup( + tmp_path, + f"""esphome: + name: test +substitutions: + prev: "{NEW_KEY}" +api: + encryption: + key: "{OLD_KEY}" +ota: + - platform: esphome + encryption: + old_key: ${{prev}} +""", + ) + with pytest.raises(EsphomeError, match="edit it by hand"): + old_key_edit(OLD_KEY) + + +def test_old_key_on_a_block_without_source(tmp_path: Path) -> None: + """An ota block built in code has no line to add old_key after.""" + _setup(tmp_path, API_YAML) + CORE.raw_config = { + "api": {"encryption": {"key": OLD_KEY}}, + "ota": [{"platform": "esphome", "encryption": {"old_key": NEW_KEY}}], + } + with pytest.raises(EsphomeError, match="was not read from a file"): + old_key_edit(OLD_KEY) + + +def test_other_platform_listed_first(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" + +ota: + - platform: web_server + - platform: esphome + encryption: +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace(OLD_KEY, NEW_KEY) + ( + f' old_key: "{OLD_KEY}"\n' + ) + + +def test_block_scalar_key_in_the_config_is_refused(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +api: + encryption: + key: >- + {OLD_KEY} + +ota: + - platform: esphome + encryption: +""" + _setup(tmp_path, yaml_text) + with pytest.raises(EsphomeError, match="edit the key by hand"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_comment_after_the_key_line_stays_below_old_key(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +mqtt: + broker: broker.local + +ota: + - platform: esphome + encryption: + key: "{OLD_KEY}" + # rotated on install + port: 3232 +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace( + f' key: "{OLD_KEY}"\n', + f' key: "{NEW_KEY}"\n old_key: "{OLD_KEY}"\n', + ) + + +def test_quoted_and_prefixed_secret_names(tmp_path: Path) -> None: + """A quoted name still matches, a longer name that starts the same and a + nested same-named key do not.""" + secrets = f"""device_key_old: {OLD_KEY} +nested: + device_key: {OLD_KEY} +"device_key": {OLD_KEY} +""" + _setup(tmp_path, SECRET_YAML, secrets) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.line, e.new_line) for e in edits] == [(3, f'"device_key": {NEW_KEY}')] + + +def test_quoted_secret_reference(tmp_path: Path) -> None: + """`!secret "name"` is valid yaml; the quotes are not part of the name.""" + _setup( + tmp_path, + SECRET_YAML.replace("!secret device_key", '!secret "device_key"'), + f"device_key: {OLD_KEY}\n", + ) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path.name, e.new_line) for e in edits] == [ + ("secrets.yaml", f"device_key: {NEW_KEY}") + ] + + +@pytest.mark.skipif(sys.platform == "win32", reason="posix file modes") +def test_keeps_the_secrets_file_mode(tmp_path: Path) -> None: + """A 0600 secrets.yaml stays 0600 through the rewrite and the restore.""" + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + secrets = tmp_path / "secrets.yaml" + secrets.chmod(0o600) + originals = apply_line_edits(locate_key_edits(OLD_KEY, NEW_KEY)) + assert secrets.stat().st_mode & 0o777 == 0o600 + assert NEW_KEY in secrets.read_text() + restore_files(originals) + assert secrets.stat().st_mode & 0o777 == 0o600 + assert OLD_KEY in secrets.read_text() + + +def test_symlinked_secrets_file_is_edited_in_place(tmp_path: Path) -> None: + """The rewrite lands on the link's target, so the link survives and + other configurations sharing the file see the new key.""" + target = tmp_path / "shared" / "secrets.yaml" + target.parent.mkdir() + target.write_bytes(f"device_key: {OLD_KEY}\n".encode()) + (tmp_path / "secrets.yaml").symlink_to(target) + _setup(tmp_path, SECRET_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [e.path for e in edits] == [target.resolve()] + apply_line_edits(edits) + assert (tmp_path / "secrets.yaml").is_symlink() + assert NEW_KEY in target.read_text() + + +def test_refuses_a_secrets_file_linked_outside_the_config_dir( + tmp_path: Path, +) -> None: + config_dir = tmp_path / "config" + config_dir.mkdir() + outside = tmp_path / "secrets.yaml" + outside.write_bytes(f"device_key: {OLD_KEY}\n".encode()) + (config_dir / "secrets.yaml").symlink_to(outside) + _setup(config_dir, SECRET_YAML) + with pytest.raises(EsphomeError, match="not an editable file"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_old_key_goes_into_the_included_block(tmp_path: Path) -> None: + """`encryption: !include enc.yaml` keeps the key in another file; old_key + is inserted after that file's key line, not at the same line number of + the main yaml.""" + (tmp_path / "enc.yaml").write_bytes(f'key: "{OLD_KEY}"\n'.encode()) + _setup( + tmp_path, + "esphome:\n name: test\n\nota:\n - platform: esphome\n" + " encryption: !include enc.yaml\n", + ) + (edit,) = old_key_edit(OLD_KEY) + assert (edit.path, edit.line, edit.insert_after) == ( + (tmp_path / "enc.yaml").resolve(), + 0, + True, + ) + assert edit.new_line == f'old_key: "{OLD_KEY}"' + apply_line_edits([*locate_key_edits(OLD_KEY, NEW_KEY), edit]) + assert (tmp_path / "enc.yaml").read_text() == ( + f'key: "{NEW_KEY}"\nold_key: "{OLD_KEY}"\n' + ) + + +def test_truncated_file_is_a_stale_edit(tmp_path: Path) -> None: + path = _setup(tmp_path, API_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + path.write_bytes(b"esphome:\n name: test\n") + with pytest.raises(EsphomeError, match="changed since it was read"): + apply_line_edits(edits) + + +def test_secret_shared_with_other_configs_is_reported(tmp_path: Path) -> None: + """Other yaml files using the same secret are listed; this config's own + files, the build data and other secret names are not.""" + (tmp_path / "other.yaml").write_bytes( + b"api:\n encryption:\n key: !secret device_key\n" + ) + (tmp_path / "quoted.yaml").write_bytes( + b'ota:\n - platform: esphome\n encryption:\n key: !secret "device_key"\n' + ) + (tmp_path / "unrelated.yaml").write_bytes( + b"wifi:\n password: !secret device_key_old\n" + ) + build = tmp_path / ".esphome" / "build" + build.mkdir(parents=True) + (build / "copy.yaml").write_bytes(b"key: !secret device_key\n") + (tmp_path / "packages").mkdir() + (tmp_path / "packages" / "base.yaml").write_bytes(b"key: !secret device_key\n") + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\ndevice_key_old: x\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + root = tmp_path.resolve() + assert [e.shared_with for e in edits] == [ + [root / "other.yaml", root / "packages" / "base.yaml", root / "quoted.yaml"] + ] + + +def test_secret_key_keeps_the_previous_key_in_secrets(tmp_path: Path) -> None: + """A key from secrets.yaml never lands in the device yaml as text: the + ota block gets `old_key: !secret device_key_old` and the secrets file + the line it points to, right under the rewritten one.""" + _setup(tmp_path, SECRET_YAML, f"wifi: hunter2\ndevice_key: {OLD_KEY}\n") + edits = _edits() + apply_line_edits(edits) + assert CORE.config_path.read_text() == SECRET_YAML.replace( + " key: !secret device_key\n", + " key: !secret device_key\n old_key: !secret device_key_old\n", + ) + assert (tmp_path / "secrets.yaml").read_text() == ( + f'wifi: hunter2\ndevice_key: {NEW_KEY}\ndevice_key_old: "{OLD_KEY}"\n' + ) + + +def test_secret_old_key_rewritten_in_secrets(tmp_path: Path) -> None: + """A second rotation replaces the value the `!secret` old_key points to.""" + yaml_text = SECRET_YAML.replace( + " key: !secret device_key\n", + " key: !secret device_key\n old_key: !secret device_key_old\n", + ) + secrets = f"device_key: {OLD_KEY}\ndevice_key_old: '{OLDER_KEY}' # keep\n" + _setup(tmp_path, yaml_text, secrets) + _rotate() + assert CORE.config_path.read_text() == yaml_text + assert (tmp_path / "secrets.yaml").read_text() == ( + f"device_key: {NEW_KEY}\ndevice_key_old: '{OLD_KEY}' # keep\n" + ) + + +def test_bare_block_with_a_secret_api_key(tmp_path: Path) -> None: + """The inherited key is a secret too, so old_key points at secrets.yaml.""" + yaml_text = SECRET_YAML.replace( + " encryption:\n key: !secret device_key\n", " encryption:\n" + ) + _setup(tmp_path, yaml_text, f"device_key: {OLD_KEY}\n") + _rotate() + assert CORE.config_path.read_text() == yaml_text.replace( + " encryption:\n", " encryption:\n old_key: !secret device_key_old\n" + ) + + +def test_shared_secret_scan_covers_yml_and_skips_unreadable_files( + tmp_path: Path, +) -> None: + (tmp_path / "other.yml").write_bytes( + b"api:\n encryption:\n key: !secret device_key\n" + ) + (tmp_path / "latin1.yaml").write_bytes(b"caf\xe9: !secret device_key\n") + (tmp_path / "notes.txt").write_bytes(b"key: !secret device_key\n") + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + assert [e.shared_with for e in edits] == [[tmp_path.resolve() / "other.yml"]] + + +def test_literal_key_in_a_shared_include_is_reported(tmp_path: Path) -> None: + """Another configuration including the file that holds the key is + listed, like a shared secret.""" + (tmp_path / "common.yaml").write_bytes( + f'encryption:\n key: "{OLD_KEY}"\n'.encode() + ) + (tmp_path / "other.yaml").write_bytes(b"api: !include common.yaml\n") + _setup( + tmp_path, + "esphome:\n name: test\n\napi: !include common.yaml\n\n" + "ota:\n - platform: esphome\n encryption:\n", + ) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + assert [(e.path.name, e.shared_with) for e in edits] == [ + ("common.yaml", [tmp_path.resolve() / "other.yaml"]) + ] + + +def test_mode_failure_is_reported(tmp_path: Path) -> None: + path = _setup(tmp_path, API_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with ( + patch("pathlib.Path.chmod", side_effect=OSError("denied")), + pytest.raises( + EsphomeError, match="was written but could not get its mode back" + ), + ): + apply_line_edits(edits) + assert path.read_text() == API_YAML + + +def test_indented_secrets_root_and_space_before_colon(tmp_path: Path) -> None: + """A secrets file may indent its whole root mapping, and `key :` is + valid yaml; the added line follows the root indent.""" + secrets = f" wifi : hunter2\n device_key : {OLD_KEY}\n" + _setup(tmp_path, SECRET_YAML.replace("key: !secret", "key : !secret"), secrets) + _rotate() + assert (tmp_path / "secrets.yaml").read_text() == ( + f' wifi : hunter2\n device_key : {NEW_KEY}\n device_key_old: "{OLD_KEY}"\n' + ) + assert "old_key: !secret device_key_old" in CORE.config_path.read_text() + + +def test_old_key_refuses_a_flow_style_key(tmp_path: Path) -> None: + """`encryption: {key: ...}` puts the key on the block's own line.""" + _setup( + tmp_path, + "esphome:\n name: test\n\nota:\n - platform: esphome\n" + f' encryption: {{key: "{OLD_KEY}"}}\n', + ) + with pytest.raises(EsphomeError, match="does not hold 'key:' as a block line"): + old_key_edit(OLD_KEY) + + +def test_own_includes_and_similar_names_are_not_shared_users(tmp_path: Path) -> None: + """A file this configuration reaches through another include is its own + but still a path for other devices, and `wifi_common.yaml` does not + mention `common.yaml`.""" + (tmp_path / "common.yaml").write_bytes(f'key: "{OLD_KEY}"\n'.encode()) + (tmp_path / "base.yaml").write_bytes(b"encryption: !include common.yaml\n") + (tmp_path / "other.yaml").write_bytes(b"api: !include base.yaml\n") + (tmp_path / "lookalike.yaml").write_bytes(b"wifi: !include wifi_common.yaml\n") + _setup( + tmp_path, + "esphome:\n name: test\n\napi: !include base.yaml\n\n" + "ota:\n - platform: esphome\n encryption:\n", + ) + # other.yaml reaches common.yaml through this configuration's own + # base.yaml, so it is a sharer; base.yaml itself and the lookalike are not + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + assert [(e.path.name, [p.name for p in e.shared_with]) for e in edits] == [ + ("common.yaml", ["other.yaml"]) + ] + + +def test_existing_old_secret_used_elsewhere_is_not_overwritten( + tmp_path: Path, +) -> None: + """A `_old` line belongs to whoever references it, this + configuration included since it has no old_key; a leftover nobody uses + is reused.""" + secrets = f"device_key: {OLD_KEY}\ndevice_key_old: hunter2\n" + _setup(tmp_path, SECRET_YAML, secrets) + (edit, kept) = old_key_edit(OLD_KEY, "device_key") + assert kept.new_line == f"device_key_old: {OLD_KEY}" + (tmp_path / "other.yaml").write_bytes( + b"wifi:\n password: !secret device_key_old\n" + ) + _setup(tmp_path, SECRET_YAML, secrets) # the scan runs once per run + with pytest.raises( + EsphomeError, match="'device_key_old:' in .* is used by .*other.yaml" + ): + old_key_edit(OLD_KEY, "device_key") + (tmp_path / "other.yaml").unlink() + _setup( + tmp_path, SECRET_YAML + "wifi:\n password: !secret device_key_old\n", secrets + ) + with pytest.raises( + EsphomeError, match="'device_key_old:' in .* is used by .*test.yaml" + ): + old_key_edit(OLD_KEY, "device_key") + + +def test_key_secret_with_a_folded_value_gets_no_old_line(tmp_path: Path) -> None: + """The `_old` line goes under the key's own line, which must be plain.""" + _setup(tmp_path, SECRET_YAML, f"device_key: >-\n {OLD_KEY}\n") + with pytest.raises(EsphomeError, match="No plain 'device_key:' line"): + old_key_edit(OLD_KEY, "device_key") + + +def test_old_secret_collision_check_refuses_an_unreadable_file( + tmp_path: Path, +) -> None: + """A file the scan cannot read might use the line, so it is not reused.""" + (tmp_path / "latin1.yaml").write_bytes(b"caf\xe9: 1\n") + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\ndevice_key_old: hunter2\n") + with pytest.raises(EsphomeError, match="Could not read every configuration"): + old_key_edit(OLD_KEY, "device_key") + + +def test_old_key_secret_with_a_folded_value_is_refused(tmp_path: Path) -> None: + yaml_text = SECRET_YAML.replace( + " key: !secret device_key\n", + " key: !secret device_key\n old_key: !secret device_key_old\n", + ) + _setup( + tmp_path, yaml_text, f"device_key: {OLD_KEY}\ndevice_key_old: >-\n {NEW_KEY}\n" + ) + with pytest.raises(EsphomeError, match="No plain 'device_key_old:' line"): + old_key_edit(OLD_KEY) + + +def test_file_shortened_after_the_load_is_reported(tmp_path: Path) -> None: + path = _setup(tmp_path, API_YAML) + path.write_bytes(b"esphome:\n name: test\n") + with pytest.raises(EsphomeError, match="changed since it was read"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_two_same_port_entries_from_a_package_split(tmp_path: Path) -> None: + """A bare block from a package and a keyed block from the device: the + keyed one is rewritten and gets old_key, the bare one is left alone.""" + yaml_text = f"""esphome: + name: test + +ota: + - platform: esphome + encryption: + - platform: esphome + encryption: + key: "{OLD_KEY}" +""" + path = _setup(tmp_path, yaml_text) + _rotate() + assert path.read_text() == yaml_text.replace( + f' key: "{OLD_KEY}"\n', + f' key: "{NEW_KEY}"\n old_key: "{OLD_KEY}"\n', + ) + + +def test_flow_style_bare_block_is_refused(tmp_path: Path) -> None: + yaml_text = f"""esphome: + name: test + +api: + encryption: + key: "{OLD_KEY}" + +ota: + - platform: esphome + encryption: {{}} +""" + _setup(tmp_path, yaml_text) + with pytest.raises( + EsphomeError, match="does not hold 'encryption:' as a block line" + ): + old_key_edit(OLD_KEY) + + +def test_secret_beside_a_symlinked_main_config(tmp_path: Path) -> None: + """The loader looks beside the config path as given, not its target.""" + target = tmp_path / "shared" / "test.yaml" + target.parent.mkdir() + target.write_bytes(SECRET_YAML.encode()) + (target.parent / "secrets.yaml").write_bytes(b"device_key: other\n") + (tmp_path / "secrets.yaml").write_bytes(f"device_key: {OLD_KEY}\n".encode()) + CORE.reset() + CORE.config_path = tmp_path / "test.yaml" + CORE.config_path.symlink_to(target) + CORE.raw_config = yaml_util.load_yaml(CORE.config_path) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [e.path for e in edits] == [(tmp_path / "secrets.yaml").resolve()] + + +def test_secret_in_a_symlinked_include_resolves_beside_the_link(tmp_path: Path) -> None: + """The loader looks beside the include as written, not beside its target.""" + shared = tmp_path / "shared" + shared.mkdir() + (shared / "api.yaml").write_bytes(b"encryption:\n key: !secret device_key\n") + (shared / "secrets.yaml").write_bytes(b"device_key: other\n") + sub = tmp_path / "sub" + sub.mkdir() + (sub / "api.yaml").symlink_to(shared / "api.yaml") + (sub / "secrets.yaml").write_bytes(f"device_key: {OLD_KEY}\n".encode()) + _setup( + tmp_path, + "esphome:\n name: test\n\napi: !include sub/api.yaml\n\n" + "ota:\n - platform: esphome\n encryption:\n", + ) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [e.path for e in edits] == [(sub / "secrets.yaml").resolve()] + + +def test_scan_reports_what_it_could_not_read(tmp_path: Path) -> None: + """An unreadable file or a linked directory makes the shared-use check + incomplete; the warning says so, and the collision check refuses.""" + (tmp_path / "latin1.yaml").write_bytes(b"caf\xe9: 1\n") + elsewhere = tmp_path.parent / "elsewhere_configs" + elsewhere.mkdir(exist_ok=True) + (tmp_path / "linked").symlink_to(elsewhere) + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\ndevice_key_old: x\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + unchecked = with_sharers(edits) + assert len(unchecked) == 2 + assert any("latin1.yaml" in u for u in unchecked) + assert any("is a link" in u for u in unchecked) + with pytest.raises(EsphomeError, match="Could not read every configuration"): + old_key_edit(OLD_KEY, "device_key") + + +def test_secret_used_elsewhere_in_this_configuration_is_refused( + tmp_path: Path, +) -> None: + """A wifi password on the key's secret would change with it.""" + _setup( + tmp_path, + SECRET_YAML + "wifi:\n password: !secret device_key\n", + f"device_key: {OLD_KEY}\n", + ) + with pytest.raises(EsphomeError, match="also used at .*test.yaml:13"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_sharers_include_the_files_that_include_a_sharer(tmp_path: Path) -> None: + """`other.yaml` reaches the secret through `base.yaml`.""" + (tmp_path / "base.yaml").write_bytes( + b"api:\n encryption:\n key: !secret device_key\n" + ) + (tmp_path / "other.yaml").write_bytes(b"packages:\n base: !include base.yaml\n") + (tmp_path / "third.yaml").write_bytes(b"packages:\n base: !include other.yaml\n") + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + assert [p.name for p in edits[0].shared_with] == [ + "base.yaml", + "other.yaml", + "third.yaml", + ] + + +def test_old_secret_goes_to_the_secrets_file_of_the_block_that_gets_old_key( + tmp_path: Path, +) -> None: + """`!secret device_key_old` in the main file resolves from the main + secrets; when the key only lives beside an include, that is refused + rather than left to fail at compile time.""" + sub = tmp_path / "sub" + sub.mkdir() + (sub / "api.yaml").write_bytes(b"encryption:\n key: !secret device_key\n") + (sub / "secrets.yaml").write_bytes(f"device_key: {OLD_KEY}\n".encode()) + _setup( + tmp_path, + "esphome:\n name: test\n\napi: !include sub/api.yaml\n\n" + "ota:\n - platform: esphome\n encryption:\n", + "wifi: x\n", + ) + with pytest.raises( + EsphomeError, match="No plain 'device_key:' line in .*secrets.yaml" + ): + old_key_edit(OLD_KEY, "device_key") + + +def test_a_commented_out_secret_use_does_not_count(tmp_path: Path) -> None: + _setup( + tmp_path, + SECRET_YAML + + "wifi:\n # password: !secret device_key\n password: x # !secret device_key\n", + f"device_key: {OLD_KEY}\n", + ) + assert len(locate_key_edits(OLD_KEY, NEW_KEY)) == 1 + + +def test_a_directory_include_is_reported_as_unchecked(tmp_path: Path) -> None: + """`!include_dir_*` pulls in files by directory, which the name scan + cannot follow.""" + (tmp_path / "other.yaml").write_bytes(b"packages: !include_dir_named packages\n") + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + unchecked = with_sharers(edits) + assert unchecked == [ + f"{tmp_path.resolve() / 'other.yaml'} includes the directory packages" + ] + + +def test_secret_name_with_spaces_is_not_a_plain_reference(tmp_path: Path) -> None: + """`!secret ota key` is one name to the loader, not the secret `ota`.""" + _setup( + tmp_path, + SECRET_YAML.replace("!secret device_key", "!secret device key"), + f"device key: {OLD_KEY}\n", + ) + with pytest.raises(EsphomeError, match="edit the key by hand"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_comments_in_other_configurations_are_not_uses(tmp_path: Path) -> None: + (tmp_path / "other.yaml").write_bytes( + b"# key: !secret device_key\n# packages: !include_dir_named packages\nwifi: x\n" + ) + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + unchecked = with_sharers(edits) + assert (edits[0].shared_with, unchecked) == ([], []) + + +def test_existing_old_key_secret_used_elsewhere_is_refused(tmp_path: Path) -> None: + yaml_text = ( + SECRET_YAML.replace( + " key: !secret device_key\n", + " key: !secret device_key\n old_key: !secret device_key_old\n", + ) + + "wifi:\n password: !secret device_key_old\n" + ) + _setup(tmp_path, yaml_text, f"device_key: {OLD_KEY}\ndevice_key_old: {OLDER_KEY}\n") + with pytest.raises(EsphomeError, match="'device_key_old' is also used at"): + old_key_edit(OLD_KEY, "device_key") + + +def test_a_key_merged_from_an_anchor_is_refused(tmp_path: Path) -> None: + """The anchor's line may be merged into other mappings as well.""" + _setup( + tmp_path, + f'named: &named\n key: "{OLD_KEY}"\n\napi:\n encryption:\n <<: *named\n\n' + "ota:\n - platform: esphome\n encryption:\n", + ) + with pytest.raises(EsphomeError, match="was not found on a line of the yaml"): + locate_key_edits(OLD_KEY, NEW_KEY) + + +def test_key_overridden_on_top_of_a_package_block(tmp_path: Path) -> None: + """The device yaml's line wins the merge and is the one rewritten.""" + from esphome.config_helpers import merge_config + + (tmp_path / "package.yaml").write_bytes( + f'api:\n encryption:\n key: "{OLDER_KEY}"\n'.encode() + ) + path = _setup(tmp_path, API_YAML) + package = yaml_util.load_yaml(tmp_path / "package.yaml") + CORE.raw_config = merge_config(package, CORE.raw_config) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.path, e.line) for e in edits] == [(path, 5)] + + +def test_a_substituted_include_path_is_reported_as_unchecked(tmp_path: Path) -> None: + (tmp_path / "other.yaml").write_bytes(b"packages:\n base: !include ${common}\n") + (tmp_path / "third.yaml").write_bytes( + b"packages:\n base:\n url: x\n file: ${pkg}\n" + ) + _setup(tmp_path, SECRET_YAML, f"device_key: {OLD_KEY}\n") + edits = locate_key_edits(OLD_KEY, NEW_KEY) + unchecked = with_sharers(edits) + assert sorted(unchecked) == [ + f"{tmp_path.resolve() / 'other.yaml'} includes ${{common}} through a substitution", + f"{tmp_path.resolve() / 'third.yaml'} includes ${{pkg}} through a substitution", + ] + + +def test_another_device_including_the_main_file_is_a_sharer(tmp_path: Path) -> None: + (tmp_path / "other.yaml").write_bytes(b"packages:\n base: !include test.yaml\n") + _setup(tmp_path, API_YAML) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + with_sharers(edits) + assert [p.name for p in edits[0].shared_with] == ["other.yaml"]