Merge remote-tracking branch 'origin/dev' into HEAD

# Conflicts:
#	tests/unit_tests/test_main.py

esphome/__main__.py

@@ -2393,6 +2393,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
 
@@ -2420,6 +2426,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,
@@ -2862,6 +2869,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.",
@@ -3100,6 +3138,12 @@ def run_esphome(argv):
     # and rewrite conf_path to the extracted YAML config. The suffix check
     # stays inline so the ordinary run never imports esphome.bundle.
     if conf_path.name.lower().endswith(BUNDLE_EXTENSION):
+        if args.command == "rotate-key":
+            # The edits would land in the extracted copy, which the next run replaces
+            _LOGGER.error(
+                "rotate-key edits the yaml; run it on the source configuration"
+            )
+            return 1
         from esphome.bundle import prepare_bundle_for_compile
 
         _LOGGER.info("Extracting config bundle %s...", conf_path)
@@ -3148,8 +3192,10 @@ def run_esphome(argv):
         config = read_config(
             command_line_substitutions,
             skip_external_update=skip_external,
-            # Snapshot only needed by `esphome config --no-defaults`.
-            snapshot_user_config=getattr(args, "no_defaults", False),
+            # `esphome config --no-defaults` prints it; rotate-key edits the
+            # source lines it keeps
+            snapshot_user_config=getattr(args, "no_defaults", False)
+            or args.command == "rotate-key",
         )
         if config is None:
             return 2

esphome/cli/rotate_key.py

@@ -0,0 +1,291 @@
+"""``esphome rotate-key``."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+from esphome import espota2
+from esphome.components.noise import (
+    decode_encryption_key,
+    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.espota2 import ProbeResult
+from esphome.log import AnsiFore, color
+from esphome.ota_key_edit import rotation_edits
+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
+# What to do when the device does not accept the key in the yaml before anything is written
+PRECHECK_ADVICE = {
+    ProbeResult.UNREACHABLE: (
+        "The device could not be reached; check that it is online, then try again."
+    ),
+    ProbeResult.REJECTED: (
+        "The device runs a different key than the yaml. Install the current "
+        "configuration first with 'esphome run', which also tries old_key; when "
+        "neither key works, present the device's key with --prompt-ota-key."
+    ),
+    ProbeResult.NOT_OFFERED: (
+        "The device does not offer encryption: it runs a firmware older than "
+        "ESPHome 2026.9.0 or without a key. Install the current configuration "
+        "first, then rotate."
+    ),
+    ProbeResult.DEVICE_ERROR: (
+        "The device answered with an error (see the warning above); fix that first."
+    ),
+}
+# Why the device did not confirm the new key after the upload
+CONFIRM_FAILURE = {
+    ProbeResult.REJECTED: "The device still answers with the previous key.",
+    ProbeResult.NOT_OFFERED: "The device came back without encryption.",
+    ProbeResult.DEVICE_ERROR: "The device answered with an error (see the warning above).",
+    ProbeResult.UNREACHABLE: "The device did not come back to confirm the new key.",
+}
+
+
+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,
+        _app_ota_image,
+        _esphome_ota_conf,
+        _resolve_network_devices,
+        choose_upload_log_host,
+    )
+
+    def fail(message: str) -> int:
+        safe_print(color(AnsiFore.BOLD_RED, message))
+        return 1
+
+    # Validation replaces the raw config's list entries in place, losing their
+    # source lines; the snapshot taken before it keeps them
+    if (user_config := getattr(config, "user_config", None)) is not None:
+        CORE.raw_config = user_config
+    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])
+
+    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}")
+    # Two base64 spellings can decode to the same bytes
+    if decode_encryption_key(new_key) == decode_encryption_key(old_key):
+        return fail("The new key is the same as the current one")
+
+    try:
+        # The previous key stays in the config as old_key so an install still
+        # reaches the device if it ends up running either key
+        edits = rotation_edits(old_key, new_key)
+    except EsphomeError as err:
+        return fail(str(err))
+    # 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:
+        safe_print(
+            color(
+                AnsiFore.BOLD_YELLOW,
+                "The api encryption key changes too; Home Assistant will ask for "
+                "the new key after the install.",
+            )
+        )
+        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
+
+    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)
+
+    safe_print("Checking that the device accepts the current key...")
+    found = espota2.probe_ota_key(
+        network_devices,
+        remote_port,
+        old_key,
+        timeout=PRECHECK_TIMEOUT,
+        retry_rejected=False,
+    )
+    if found is not ProbeResult.ACCEPTED:
+        return fail(PRECHECK_ADVICE[found])
+
+    # 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")
+    except BaseException as err:
+        if notes := getattr(err, "__notes__", None):
+            safe_print(color(AnsiFore.BOLD_RED, "; ".join(notes)))
+            safe_print(keys)
+        raise
+    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 args.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 image_sent() -> None:
+            # The device accepted the session and may commit from here on;
+            # the edited config reaches it on either key
+            nonlocal uploaded
+            uploaded = True
+
+        rc, _ = espota2.run_ota(
+            network_devices,
+            remote_port,
+            ota_conf.get(CONF_PASSWORD),
+            _app_ota_image(args),
+            noise_psk=old_key,
+            plaintext_fallback=False,
+            on_upload_start=image_sent,
+        )
+        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 failed before the image was sent (see the error "
+                "above); nothing changed on the device."
+            )
+        # A successful upload sent the image even if the callback did not run
+        uploaded = True
+        safe_print("Waiting for the device to come back with the new key...")
+        found = espota2.probe_ota_key(network_devices, remote_port, new_key)
+        if found is not ProbeResult.ACCEPTED:
+            return fail(
+                CONFIRM_FAILURE[found]
+                + " The configuration keeps the new key, with the previous one 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:
+        # An interrupted restore carries its report as a note
+        why = str(err) or "; ".join(getattr(err, "__notes__", ()))
+        safe_print(color(AnsiFore.BOLD_RED, why or f"Interrupted restoring {files}"))
+        safe_print(keys)
+    except BaseException as err:
+        if notes := getattr(err, "__notes__", None):
+            safe_print(color(AnsiFore.BOLD_RED, "; ".join(notes)))
+        safe_print(keys)
+        raise
+    else:
+        safe_print(color(AnsiFore.BOLD_YELLOW, f"Restored the previous key in {files}"))
+    finally:
+        _drop_images(keys)
+
+
+def _drop_images(keys: str) -> None:
+    """Delete the images a build made with the new key."""
+    build_dir = CORE.firmware_bin.parent
+    failed: list[str] = []
+    images: list[Path] = []
+    try:
+        # The serial and BOOTSEL uploads also take the elf and the signed bin
+        suffixes = (".bin", ".uf2", ".elf", ".signed")
+        images = [p for s in suffixes 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.",
+            )
+        )

esphome/ota_key_edit.py

@@ -0,0 +1,521 @@
+"""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
+
+import yaml
+
+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.util import filter_yaml_files
+from esphome.yaml_edit import (
+    LineEdit,
+    editable_file,
+    field_line_re,
+    field_prefix,
+    own_documents,
+    read_text,
+    rewrite,
+    secret_field_re,
+    source_line,
+    source_of,
+)
+from esphome.yaml_util import named_path, 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 yaml lines rewritten to
+    point at it, the only uses of it allowed."""
+
+    secret: str | None = None
+    secret_uses: frozenset[tuple[Path, int]] = frozenset()
+
+
+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)]
+    found = []
+    for block in blocks:
+        if (node := static_encryption_key(block)) is None or str(node) != key:
+            continue
+        encryption = block[CONF_ENCRYPTION]
+        if source_of(encryption, CONF_KEY) is not None:
+            found.append(encryption)
+        elif hasattr(next(k for k in encryption if k == CONF_KEY), "esp_range"):
+            # Read but not on a line of its own, as with a merge key; a key
+            # validation added has no range and is passed over
+            raise EsphomeError(
+                f"The '{CONF_KEY}' of an '{CONF_ENCRYPTION}:' block is not on a "
+                "line of its own; edit the key by hand"
+            )
+    return found
+
+
+_INDENT_RE = re.compile(r"\s*")
+_KEY_FIELD_RE = re.compile(rf"^{field_prefix(CONF_KEY)}")
+_BARE_BLOCK_RE = re.compile(rf"^{field_prefix(CONF_ENCRYPTION)}\s*(#.*)?$")
+_COLLECTION_STARTS = (
+    yaml.BlockMappingStartToken,
+    yaml.BlockSequenceStartToken,
+    yaml.FlowMappingStartToken,
+    yaml.FlowSequenceStartToken,
+)
+_COLLECTION_ENDS = (
+    yaml.BlockEndToken,
+    yaml.FlowMappingEndToken,
+    yaml.FlowSequenceEndToken,
+)
+_KEY_SECRET_RE = secret_field_re(CONF_KEY)
+
+
+def _indent(text: str) -> str:
+    return _INDENT_RE.match(text).group()
+
+
+@dataclass
+class _Scan:
+    """What a yaml file names, from its tokens alone; comments never count."""
+
+    secrets: list[tuple[str, int]] = field(default_factory=list)  # name, line
+    targets: set[Path] = field(default_factory=set)  # local files it includes, as named
+    include_dir: str | None = None
+    unresolved: str | None = None  # an include path only known when loaded
+
+    def uses(self, name: str) -> bool:
+        return any(secret == name for secret, _ in self.secrets)
+
+
+@dataclass
+class _Configs:
+    files: list[tuple[Path, _Scan]]
+    skipped: list[str]
+    named_by: dict[Path, set[Path]]  # a resolved file to the files naming it
+
+
+@dataclass
+class _State:
+    scans: dict[Path, _Scan] = field(default_factory=dict)
+    configs: _Configs | None = None
+    secrets_paths: dict[Path, Path] = field(default_factory=dict)  # by directory
+
+
+def _state() -> _State:
+    return CORE.data.setdefault(_DOMAIN, _State())
+
+
+def _scan(path: Path) -> _Scan:
+    """``path`` scanned once per run; nothing is constructed, included or loaded."""
+    scans = _state().scans
+    if (found := scans.get(path)) is not None:
+        return found
+    if "!" not in (text := read_text(path)):  # names come only from tags
+        found = scans[path] = _Scan()
+        return found
+    try:
+        # An anchor may sit between a tag and its value
+        tokens = [
+            t
+            for t in yaml.scan(text, Loader=yaml.SafeLoader)
+            if not isinstance(t, yaml.AnchorToken)
+        ]
+    except yaml.YAMLError as err:
+        raise EsphomeError(
+            f"Could not scan {path}: {getattr(err, 'problem', err)}"
+        ) from err
+    found = scans[path] = _Scan()
+    depth = 0
+    include_maps: set[int] = set()  # depths of `!include {file: ...}` mappings
+    keys: dict[int, str] = {}  # the last key seen at each depth
+    for prev2, prev, token in zip(
+        [None, None, *tokens], [None, *tokens], tokens, strict=False
+    ):
+        tag = prev.value[1] if isinstance(prev, yaml.TagToken) else ""
+        if isinstance(token, _COLLECTION_STARTS):
+            depth += 1
+            if tag == "include":
+                include_maps.add(depth)
+            continue
+        if isinstance(token, _COLLECTION_ENDS):
+            include_maps.discard(depth)
+            keys.pop(depth, None)
+            depth -= 1
+            continue
+        # An include map's `file:` value, possibly behind a tag
+        file_value = (
+            depth in include_maps
+            and keys.get(depth) == "file"
+            and (
+                isinstance(prev, yaml.ValueToken)
+                or (isinstance(prev2, yaml.ValueToken) and tag)
+            )
+        )
+        if isinstance(token, yaml.AliasToken) and (tag == "include" or file_value):
+            found.unresolved = found.unresolved or f"*{token.value}"
+            continue
+        if not isinstance(token, yaml.ScalarToken):
+            continue
+        value = token.value
+        if isinstance(prev, yaml.KeyToken):
+            keys[depth] = value
+            continue
+        # Only an include loads a local file; a remote package's file is not one
+        included = tag == "include" or (file_value and not tag)
+        if file_value and tag:  # !env_var, !secret, ...: only known when loaded
+            found.unresolved = found.unresolved or f"!{tag} {value}"
+        if tag == "secret":
+            found.secrets.append((value, prev.start_mark.line))
+        elif tag.startswith("include_dir_") and found.include_dir is None:
+            found.include_dir = value
+        elif included and ("$" in value or "<%" in value):
+            found.unresolved = found.unresolved or value
+        elif included:
+            found.targets.add(named_path(path.parent / value))
+    return found
+
+
+def _child_indent(doc: Path, line_no: int, indent: str) -> str:
+    """The indent of the first option under the block line, or one level more."""
+    for text in read_text(doc).splitlines()[line_no + 1 :]:
+        if not (stripped := text.strip()) or stripped.startswith("#"):
+            continue
+        if len(child := _indent(text)) > len(indent):
+            return child
+        break
+    return f"{indent}  "
+
+
+def _root_indent(lines: list[str]) -> 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 _configs() -> _Configs:
+    """Every yaml file under the config dir and the files they include scanned,
+    and what could not be; walked once per run."""
+    state = _state()
+    if state.configs is not None:
+        return state.configs
+    data_dir = CORE.data_dir.resolve()
+    found: list[tuple[Path, _Scan]] = []
+    skipped: list[str] = []
+
+    def skip(what: object) -> None:
+        skipped.append(str(what))
+        _LOGGER.debug("Skipping %s", what)
+
+    # Remote packages are checked out under the data dir, and their secrets fall back here
+    packages = data_dir / "packages"
+    roots = [CORE.config_dir.resolve(), *([packages] if packages.is_dir() else [])]
+    for root in roots:
+        for dirpath, dirnames, filenames in os.walk(root, onerror=skip):
+            for name in list(dirnames):
+                path = Path(dirpath, name)
+                # Hidden trees (.git, .venv) hold no configurations, as in find_files
+                if name.startswith(".") or path.resolve() == data_dir:
+                    dirnames.remove(name)
+                elif path.is_symlink():
+                    dirnames.remove(name)
+                    skip(f"{path} is a link")
+            for path in filter_yaml_files(Path(dirpath, f) for f in filenames):
+                try:
+                    found.append((path, _scan(path)))
+                except EsphomeError as err:
+                    skip(err)
+    # An include from outside these trees is scanned too, as are its own includes
+    seen = {named_path(path) for path, _ in found}
+    pending = [target for _, scan in found for target in scan.targets]
+    while pending:
+        if (target := pending.pop()) in seen:
+            continue
+        seen.add(target)
+        try:
+            found.append((target, scan := _scan(target)))
+        except EsphomeError as err:
+            skip(err)
+            continue
+        pending.extend(scan.targets)
+    found.sort(key=lambda ps: ps[0])
+    named_by: dict[Path, set[Path]] = {}
+    for path, scan in found:
+        for target in scan.targets:
+            named_by.setdefault(target.resolve(), set()).add(path)
+    state.configs = _Configs(found, skipped, named_by)
+    return state.configs
+
+
+def _secrets_path(doc: Path) -> Path:
+    """``secrets_path_for`` once per directory; it loads the secrets file."""
+    paths = _state().secrets_paths
+    if (found := paths.get(doc.parent)) is None:
+        found = paths[doc.parent] = secrets_path_for(doc)
+    return found
+
+
+def _secret_users(name: str) -> set[Path]:
+    """The scanned files that use the secret ``name``, from any secrets file."""
+    return {path for path, scan in _configs().files if scan.uses(name)}
+
+
+_HAND_EDIT = "only a key unique to this device can be rotated, change it by hand"
+
+
+def _refuse_shared(edits: list[KeyEdit]) -> None:
+    """Refuses unless every edited line and secret is used only by this
+    configuration, on the lines being rewritten."""
+    main = CORE.config_path.resolve()
+    root = CORE.config_dir.resolve()
+    own = own_documents()
+    configs = _configs()
+    allowed = {(e.secret, *site) for e in edits for site in e.secret_uses}
+
+    def including(files: set[Path]) -> set[Path]:
+        return {p for f in files for p in configs.named_by.get(f.resolve(), ())}
+
+    def with_includers(users: set[Path]) -> set[Path]:
+        frontier = users
+        while frontier := including(frontier) - users:
+            users = users | frontier
+        return users
+
+    def shown(path: Path) -> str:
+        # The same text on every OS
+        return (
+            path.relative_to(root) if path.is_relative_to(root) else path
+        ).as_posix()
+
+    what: dict[str, None] = {}
+    users: set[str] = set()
+    for edit in edits:
+        # This config's own files stay in the search: another device may
+        # reach a line through one of them, the main file included
+        files = including({edit.path})
+        found = []
+        if edit.secret:
+            files |= _secret_users(edit.secret)
+            found = [
+                f"{shown(doc)}:{i + 1}"
+                for doc in own
+                for name, i in _scan(doc).secrets
+                if name == edit.secret and (name, doc, i) not in allowed
+            ]
+        found += [shown(p) for p in with_includers(files) if p.resolve() not in own]
+        if found:
+            what[f"secret '{edit.secret}'" if edit.secret else edit.path.name] = None
+            users.update(found)
+    # A device that includes the main file loads everything this one does
+    if includers := [
+        shown(p) for p in with_includers(including({main})) if p.resolve() not in own
+    ]:
+        what[main.name] = None
+        users.update(includers)
+    if what:
+        raise EsphomeError(
+            f"{', '.join(what)} {'is' if len(what) == 1 else 'are'} also used by "
+            f"{', '.join(sorted(users))}; {_HAND_EDIT}"
+        )
+    unchecked = [*configs.skipped]
+    for path, scan in configs.files:
+        if path.resolve() == main:
+            continue
+        if scan.include_dir is not None:
+            unchecked.append(f"{path} includes the directory {scan.include_dir}")
+        if scan.unresolved is not None:
+            unchecked.append(
+                f"{path} includes {scan.unresolved}, a path only known when loaded"
+            )
+    if unchecked:
+        more = f" (and {len(unchecked) - 1} more)" if len(unchecked) > 1 else ""
+        raise EsphomeError(
+            "Could not check whether other configurations use this key: "
+            f"{unchecked[0]}{more}; {_HAND_EDIT}"
+        )
+
+
+def rotation_edits(old_key: str, new_key: str) -> list[KeyEdit]:
+    """Every edit of a rotation: the new key and the previous one as old_key.
+    Refuses unless every edited line and secret is this device's alone."""
+    edits = [*locate_key_edits(old_key, new_key), *old_key_edit(old_key)]
+    _refuse_shared(edits)
+    return edits
+
+
+def _secret_line(
+    secrets_path: Path, name: str, value: str | None = None
+) -> tuple[int, re.Match[str]]:
+    """The root level plain ``name:`` line, holding ``value`` if given."""
+    lines = read_text(secrets_path).splitlines()
+    line_re = field_line_re(name, value, re.escape(_root_indent(lines)))
+    for i, text in enumerate(lines):
+        if m := line_re.match(text):
+            return i, m
+    holding = "" if value is None else " with the current key"
+    raise EsphomeError(
+        f"No plain '{name}:' line{holding} in {secrets_path}; edit it by hand"
+    )
+
+
+def _has_root_key(secrets_path: Path, name: str) -> bool:
+    """Whether a root level ``name:`` key exists, whatever its value."""
+    lines = read_text(secrets_path).splitlines()
+    key_re = re.compile(field_prefix(name, re.escape(_root_indent(lines))))
+    return any(key_re.match(text) for text in lines)
+
+
+def _secret_rewrite(
+    secrets_path: Path, name: str, value: str, expect: str | None = None
+) -> KeyEdit:
+    """Set the root level plain ``name:`` line, holding ``expect`` if given,
+    to ``value``."""
+    i, m = _secret_line(secrets_path, name, expect)
+    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."""
+    i, m = _secret_line(secrets_path, after)
+    return KeyEdit(
+        secrets_path,
+        i,
+        m.string,
+        f'{_indent(m.string)}{name}: "{value}"',
+        insert_after=True,
+        secret=name,
+    )
+
+
+_SECRET_AS_LITERAL = (
+    "The current key is kept in secrets.yaml, so it cannot go into the yaml as a "
+    f"plain '{CONF_OLD_KEY}'; use a !secret for it or edit it by hand"
+)
+
+
+def _field_edits(
+    blocks: list[ConfigType],
+    option: str,
+    new_value: str,
+    expect: bool,
+    literal_ok: bool,
+) -> list[KeyEdit]:
+    """Rewrite the ``option:`` line of each block, or the secret it names, to
+    ``new_value``; with ``expect`` the secret must hold the block's value."""
+    secret_re = secret_field_re(option)
+    what = "the key" if option == CONF_KEY else f"'{option}'"
+    edits: dict[tuple[Path, int], KeyEdit] = {}
+    for block in blocks:
+        doc, line_no, text = source_line(block, option)
+        value = str(block[option])
+        if match := secret_re.match(text):
+            name = match["name"]
+            edit = _secret_rewrite(
+                editable_file(_secrets_path(doc)),
+                name,
+                new_value,
+                expect=value if expect else None,
+            )
+            edit.secret_uses = frozenset({(doc.resolve(), line_no)})
+        elif match := field_line_re(option, value).match(text):
+            if not literal_ok:
+                raise EsphomeError(_SECRET_AS_LITERAL)
+            edit = KeyEdit(doc, line_no, text, rewrite(match, new_value))
+        else:
+            raise EsphomeError(
+                f"{doc}:{line_no + 1} does not hold {what} as a plain value or a "
+                f"!secret; edit {what} by hand"
+            )
+        # The api and ota blocks may share one !secret line
+        where = (edit.path.resolve(), edit.line)
+        if (kept := edits.setdefault(where, edit)) is not edit:
+            kept.secret_uses |= edit.secret_uses
+    return list(edits.values())
+
+
+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."""
+    if not (blocks := _key_blocks(CORE.raw_config or {}, old_key)):
+        raise EsphomeError("The current key was not found on a line of the yaml")
+    return _field_edits(blocks, CONF_KEY, new_key, expect=True, literal_ok=True)
+
+
+def _secret_name(block: ConfigType) -> str | None:
+    """The secrets file name a ``key: !secret <name>`` line of ``block`` holds."""
+    if source_of(block, CONF_KEY) is None:
+        return None
+    match = _KEY_SECRET_RE.match(source_line(block, CONF_KEY)[2])
+    return match["name"] if match else None
+
+
+def old_key_edit(old_key: str) -> list[KeyEdit]:
+    """Set ``old_key:`` on the esphome ota block, rewriting or adding it. A
+    key from secrets.yaml stays there as ``<name>_old``. Of several entries
+    the one with its own key line gets it."""
+    raw = CORE.raw_config or {}
+    items = [item for item in esphome_ota_items(raw) if CONF_ENCRYPTION in item]
+    if not items:
+        raise EsphomeError("The esphome OTA platform has no 'encryption:' block")
+    encs = [item[CONF_ENCRYPTION] or {} for item in items]
+    api_enc = (raw.get(CONF_API) or {}).get(CONF_ENCRYPTION) or {}
+    # Only a key that never appears in the yaml itself must stay out of it
+    key_blocks = _key_blocks(raw, old_key)
+    secret_backed = bool(key_blocks) and all(map(_secret_name, key_blocks))
+    # Same-port entries must agree, so every existing old_key is rewritten
+    if with_old := [enc for enc in encs if enc.get(CONF_OLD_KEY) is not None]:
+        return _field_edits(
+            with_old, CONF_OLD_KEY, old_key, expect=False, literal_ok=not secret_backed
+        )
+    if keyed := next((enc for enc in encs if source_of(enc, CONF_KEY)), None):
+        doc, line_no, text = source_line(keyed, CONF_KEY)
+        anchor, anchor_re, indent = CONF_KEY, _KEY_FIELD_RE, _indent(text)
+        if (api_key := source_of(api_enc, CONF_KEY)) and (
+            api_key[0].resolve() == doc.resolve() and api_key[1] == line_no
+        ):
+            # The api encryption schema takes no old_key
+            raise EsphomeError(
+                f"{doc}:{line_no + 1} is the api key too; add '{CONF_OLD_KEY}' to "
+                "an ota-only encryption block by hand"
+            )
+    else:
+        # A bare block takes after the api key
+        keyed = api_enc
+        # Package entries merge in first, so the last one is the device's own
+        doc, line_no, text = source_line(items[-1], CONF_ENCRYPTION)
+        anchor, anchor_re = CONF_ENCRYPTION, _BARE_BLOCK_RE
+        indent = _child_indent(doc, line_no, _indent(text))
+    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 := _secret_name(keyed)) is None:
+        if secret_backed:
+            raise EsphomeError(_SECRET_AS_LITERAL)
+        rendered = f'{CONF_OLD_KEY}: "{old_key}"'
+        extra = []
+    else:
+        rendered = f"{CONF_OLD_KEY}: !secret {secret}_old"
+        secrets_path = editable_file(_secrets_path(doc))
+        name = f"{secret}_old"
+        # A leftover line is reused; a second one would not load
+        extra = [
+            _secret_rewrite(secrets_path, name, old_key)
+            if _has_root_key(secrets_path, name)
+            else _secret_insert(secrets_path, secret, name, old_key)
+        ]
+    edit = KeyEdit(doc, line_no, text, f"{indent}{rendered}", insert_after=True)
+    return [edit, *extra]

esphome/yaml_edit.py

@@ -63,10 +63,24 @@ def field_line_re(
     ``indent``; the name may be quoted. Keeps the prefix, the quotes and the
     trailer for rewrite."""
     scalar = PLAIN_SCALAR if value is None else re.escape(value)
-    prefix = rf"{indent}[\"']?{re.escape(name)}[\"']?\s*:\s*"
+    prefix = rf"{field_prefix(name, indent)}\s*"
     return re.compile(rf"^(?P<prefix>{prefix}){TRAILER.format(value=scalar)}")
 
 
+def field_prefix(name: str, indent: str = r"\s*") -> str:
+    """The pattern of ``name:`` at ``indent``; the name may be quoted."""
+    return rf"{indent}[\"']?{re.escape(name)}[\"']?\s*:"
+
+
+def secret_field_re(name: str) -> re.Pattern[str]:
+    """Match ``name: !secret <secret>``; the secret is the ``name`` group. It is
+    the whole scalar (`!secret ota key` is not `ota`) and `a#b` is one name."""
+    return re.compile(
+        rf"^{field_prefix(name)}\s*!secret\s+(?P<q>[\"']?)(?P<name>[^\s\"']+)(?P=q)"
+        r"(?=\s+#|\s*$)"
+    )
+
+
 def rewrite(match: re.Match[str], value: str, quote: str | None = None) -> str:
     """The matched line with ``value`` in place of the scalar; the source
     quotes stay unless ``quote`` is given."""
@@ -134,9 +148,10 @@ def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
         if rng is None:
             return None
         doc, line = Path(rng.start_mark.document), rng.start_mark.line
-        if doc.resolve() not in own_documents() or not field_line_re(name).match(
-            line_at(doc, line)
-        ):
+        if doc.resolve() not in own_documents():
+            return None
+        text = line_at(doc, line)
+        if not (field_line_re(name).match(text) or secret_field_re(name).match(text)):
             return None
         return doc, line
     # An included mapping carries its parent's `!include` line, so only a

esphome/yaml_util.py

@@ -966,9 +966,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
 
@@ -1364,6 +1362,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)]

tests/unit_tests/cli/test_rotate_key.py

@@ -0,0 +1,618 @@
+"""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 PRECHECK_ADVICE, command_rotate_key
+from esphome.config import do_substitution_pass
+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.espota2 import ProbeResult
+from esphome.upload_targets import PortType
+from esphome.util import ESPHOME_COMMAND
+from esphome.yaml_edit import RestoreError
+from tests.unit_tests.test_ota_key_edit import API_YAML, NEW_KEY, OLD_KEY, SECRET_YAML
+from tests.unit_tests.test_yaml_edit import _setup
+
+
+@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
+    dashboard: bool = False
+
+
+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
+
+
+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(tmp_path, API_YAML)
+    CORE.name = "test"
+    CORE.build_path = tmp_path / ".esphome" / "build" / "test"
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
+    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"]
+        ) as choose,
+        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=ProbeResult.ACCEPTED
+        ) 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_upload_start"]()
+            return upload.return_value
+
+        upload.side_effect = upload_after_connecting
+        yield {
+            "probe": probe,
+            "compile": compile_,
+            "upload": upload,
+            "confirm": confirm,
+            "stdin": stdin,
+            "choose": choose,
+        }
+
+
+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_upload_start=ANY,
+    )
+    out = capfd.readouterr().out
+    assert NEW_KEY in out
+    assert "SUCCESS" in out
+
+
+def test_uploads_the_platform_ota_image(env: dict[str, Mock]) -> None:
+    """The upload sends what the platform's ota_image hook picks, as installs do."""
+    image = Path("firmware.ota.uf2")
+    with patch("esphome.__main__._app_ota_image", return_value=image) as pick:
+        assert command_rotate_key(MockArgs(), CORE.config) == 0
+    pick.assert_called_once()
+    assert env["upload"].call_args.args[3] == image
+
+
+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()
+
+
+@pytest.mark.parametrize(
+    ("yaml_text", "files", "message"),
+    [
+        (
+            API_YAML,
+            {"other.yaml": "packages:\n  base: !include test.yaml\n"},
+            "test.yaml is also used by other.yaml",
+        ),
+        (
+            # Without a secrets.yaml beside it the device may fall back to this one
+            SECRET_YAML,
+            {
+                "secrets.yaml": f"device_key: {OLD_KEY}\n",
+                "sub/device.yaml": "api:\n  encryption:\n    key: !secret device_key\n",
+            },
+            "secret 'device_key' is also used by sub/device.yaml",
+        ),
+        (
+            API_YAML,
+            {"broken.yaml": "key: !secret device_key\nbad: @x\n"},
+            "Could not check whether other configurations use this key: Could not scan",
+        ),
+        (
+            f'api:\n  encryption:\n    key: "{OLD_KEY}"\n\nota:\n  - !include ota.yaml\n',
+            {
+                "ota.yaml": "platform: esphome\nencryption:\n",
+                "other.yaml": "ota:\n  - !include ota.yaml\n",
+            },
+            "ota.yaml is also used by other.yaml",
+        ),
+        (
+            SECRET_YAML,
+            {
+                "secrets.yaml": f"device_key: {OLD_KEY}\ndevice_key_old: x\n",
+                "other.yaml": "wifi:\n  password: !secret device_key_old\n",
+            },
+            "secret 'device_key_old' is also used by other.yaml",
+        ),
+    ],
+    ids=["includer", "secrets_fallback", "unchecked", "old_key_file", "old_secret"],
+)
+def test_refuses_a_key_that_is_not_unique(
+    env: dict[str, Mock],
+    capfd: pytest.CaptureFixture[str],
+    yaml_text: str,
+    files: dict[str, str],
+    message: str,
+) -> None:
+    """Anything another configuration uses or may use stops the rotation
+    before any question, device choice, probe or write."""
+    CORE.config_path.write_text(yaml_text, encoding="utf-8")
+    for name, text in files.items():
+        (CORE.config_dir / name).parent.mkdir(exist_ok=True)
+        (CORE.config_dir / name).write_text(text, encoding="utf-8")
+    CORE.raw_config = do_substitution_pass(yaml_util.load_yaml(CORE.config_path), None)
+    before = {p: p.read_bytes() for p in CORE.config_dir.rglob("*.yaml")}
+    assert command_rotate_key(MockArgs(yes=True), CORE.config) == 1
+    out = capfd.readouterr().out
+    assert message in out
+    assert "only a key unique to this device can be rotated, change it by hand" in out
+    assert {p: p.read_bytes() for p in CORE.config_dir.rglob("*.yaml")} == before
+    env["confirm"].assert_not_called()
+    env["choose"].assert_not_called()
+    env["probe"].assert_not_called()
+
+
+def test_uses_the_snapshot_taken_before_validation(env: dict[str, Mock]) -> None:
+    """Validation rewrites the raw config; the snapshot keeps the source lines."""
+
+    class Validated(dict):
+        user_config = CORE.raw_config
+
+    CORE.raw_config = {}
+    assert command_rotate_key(MockArgs(yes=True), Validated(CORE.config)) == 0
+    assert NEW_KEY in CORE.config_path.read_text()
+
+
+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",
+    # The last one spells OLD_KEY's bytes with other padding bits
+    ["not-base64", OLD_KEY, OLD_KEY[:-2] + "9="],
+    ids=["invalid", "same", "same_bytes"],
+)
+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 = ProbeResult.REJECTED
+    assert command_rotate_key(MockArgs(), CORE.config) == 1
+    assert CORE.config_path.read_text() == API_YAML
+    env["compile"].assert_not_called()
+    assert "runs a different key than the yaml" in capfd.readouterr().out
+    for found in (
+        ProbeResult.UNREACHABLE,
+        ProbeResult.NOT_OFFERED,
+        ProbeResult.DEVICE_ERROR,
+    ):
+        env["probe"].return_value = found
+        assert command_rotate_key(MockArgs(), CORE.config) == 1
+        assert PRECHECK_ADVICE[found] 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)
+    built = [
+        image,
+        image.with_name("firmware.factory.bin"),
+        image.with_name("firmware.elf"),
+        image.with_name("firmware.bin.signed"),
+    ]
+    for path in built:
+        path.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 any(path.exists() for path in built)
+    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_images_are_dropped_when_the_restore_raises_anything(
+    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("esphome.cli.rotate_key.restore_files", side_effect=SystemExit(3)),
+        pytest.raises(SystemExit),
+    ):
+        command_rotate_key(MockArgs(), CORE.config)
+    assert not CORE.firmware_bin.exists()
+    out = capfd.readouterr().out
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+def test_a_restore_exit_shows_its_report(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    env["compile"].return_value = 1
+    err = SystemExit(3)
+    err.add_note("Could not restore x")
+    with (
+        patch("esphome.cli.rotate_key.restore_files", side_effect=err),
+        pytest.raises(SystemExit),
+    ):
+        command_rotate_key(MockArgs(), CORE.config)
+    out = capfd.readouterr().out
+    assert "Could not restore x" in out
+    assert OLD_KEY in out and NEW_KEY in out
+
+
+@pytest.mark.parametrize("failed_rollback", [True, False])
+def test_an_exit_while_writing_shows_the_keys_only_when_the_rollback_failed(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], failed_rollback: bool
+) -> None:
+    err = SystemExit(3)
+    if failed_rollback:
+        err.add_note("Could not restore x")
+    with (
+        patch("esphome.cli.rotate_key.apply_line_edits", side_effect=err),
+        pytest.raises(SystemExit),
+    ):
+        command_rotate_key(MockArgs(), CORE.config)
+    out = capfd.readouterr().out
+    assert ("Could not restore x" in out) is failed_rollback
+    assert (NEW_KEY in out) is failed_rollback
+
+
+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 = ProbeResult.UNREACHABLE
+    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 the image is sent 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 "failed before the image was sent" in out
+    assert "put back" not in out
+    assert "Restored the previous key" in out
+
+
+def test_a_successful_upload_without_the_callback_keeps_the_new_key(
+    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
+) -> None:
+    """A finished upload is never rolled back, even if the callback did not run."""
+    env["upload"].side_effect = None
+    assert command_rotate_key(MockArgs(), CORE.config) == 0
+    assert NEW_KEY in CORE.config_path.read_text()
+    assert "SUCCESS" in capfd.readouterr().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", ["unreachable", "rejected", "not_offered", "device_error", "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 = [
+        ProbeResult.ACCEPTED,
+        {
+            "interrupted": KeyboardInterrupt,
+            "unreachable": ProbeResult.UNREACHABLE,
+            "rejected": ProbeResult.REJECTED,
+            "not_offered": ProbeResult.NOT_OFFERED,
+            "device_error": ProbeResult.DEVICE_ERROR,
+        }[outcome],
+    ]
+    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")
+    assert ("still answers with the previous key" in out) is (outcome == "rejected")
+    assert ("did not come back" in out) is (outcome == "unreachable")
+    assert ("came back without encryption" in out) is (outcome == "not_offered")
+    assert ("answered with an error" in out) is (outcome == "device_error")
+
+
+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"]],
+        toolchain=Toolchain.PLATFORMIO,
+        dashboard=True,
+    )
+    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", ["rotation_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

tests/unit_tests/test_main.py

@@ -6881,6 +6881,19 @@ def test_run_esphome_bundle_detection(tmp_path: Path) -> None:
     assert result == 2
 
 
+def test_run_esphome_rotate_key_refuses_a_bundle(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The edits would land in an extracted copy that the next run replaces."""
+    bundle_path = tmp_path / f"device{BUNDLE_EXTENSION}"
+    bundle_path.write_bytes(b"fake-bundle")
+    with patch("esphome.bundle.prepare_bundle_for_compile") as mock_prepare:
+        result = run_esphome(["esphome", "rotate-key", str(bundle_path)])
+    mock_prepare.assert_not_called()
+    assert result == 1
+    assert "run it on the source configuration" in caplog.text
+
+
 def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
     """Test run_esphome does not extract for regular .yaml files."""
     yaml_file = tmp_path / "device.yaml"
@@ -8297,6 +8310,35 @@ def test_command_rename_is_dispatched_to_the_cli_module() -> None:
     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"
+
+
 VALID_OTA_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
 # Headers of real ESP32-S3 and ESP8266 app images
 ESP32_IMAGE = bytes.fromhex("e905022f645f3740ee00000009000000")