From 4165dbb53469cce45e112f58b5ee9ba7b079d8da Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 13 Oct 2026 20:16:38 -1000 Subject: [PATCH] A comment needs whitespace before its hash in a secret name, old_key takes its secret from the ota block it lands on --- esphome/cli/rotate_key.py | 3 +- esphome/ota_key_edit.py | 29 +++++++++--- tests/unit_tests/test_ota_key_edit.py | 63 ++++++++++++++++++++++----- 3 files changed, 75 insertions(+), 20 deletions(-) diff --git a/esphome/cli/rotate_key.py b/esphome/cli/rotate_key.py index 877aa621927..bd37a64c756 100644 --- a/esphome/cli/rotate_key.py +++ b/esphome/cli/rotate_key.py @@ -103,8 +103,7 @@ def command_rotate_key(args: argparse.Namespace, config: ConfigType) -> int | No 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) + edits += old_key_edit(old_key) unchecked = with_sharers(edits) except EsphomeError as err: return fail(str(err)) diff --git a/esphome/ota_key_edit.py b/esphome/ota_key_edit.py index d866e24bc91..8465ee05b62 100644 --- a/esphome/ota_key_edit.py +++ b/esphome/ota_key_edit.py @@ -38,6 +38,10 @@ class KeyEdit(LineEdit): shared_with: list[Path] = field(default_factory=list) +def _api_encryption() -> ConfigType: + return ((CORE.raw_config or {}).get(CONF_API) or {}).get(CONF_ENCRYPTION) or {} + + 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)] @@ -71,9 +75,10 @@ _SUBSTITUTED_INCLUDE_RE = re.compile( ) -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_ref(name: str = r"[^\s\"']+") -> str: + """`!secret name` as the whole scalar: `!secret ota key` is not `ota`, and + a comment needs whitespace before its `#`, so `a#b` is one name.""" + return rf"!secret\s+(?P[\"']?)(?P{name})(?P=q)(?=\s+#|\s*$)" def _secret_field_re(name: str) -> re.Pattern[str]: @@ -283,10 +288,18 @@ def _refuse_other_own_uses(name: str, allowed: set[tuple[Path, int]]) -> None: ) -def old_key_edit(old_key: str, secret: str | None = None) -> list[KeyEdit]: +def _secret_name(block: ConfigType) -> str | None: + """The secrets file name a ``key: !secret `` 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 (``secret``) stays there as ``_old``. Of - several entries the one with its own old_key or key line gets it.""" + key from secrets.yaml 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 {}) @@ -331,7 +344,9 @@ def old_key_edit(old_key: str, secret: str | None = None) -> list[KeyEdit]: f"{doc}:{line_no + 1} does not hold '{anchor}:' as a block line; " "edit it by hand" ) - if secret is None: + # The block's own key line decides; a bare block takes after the api key + keyed = encryption if mapping is encryption else _api_encryption() + if (secret := _secret_name(keyed)) is None: rendered = f'{CONF_OLD_KEY}: "{old_key}"' extra = [] else: diff --git a/tests/unit_tests/test_ota_key_edit.py b/tests/unit_tests/test_ota_key_edit.py index f1cf544b6de..74022a9de8f 100644 --- a/tests/unit_tests/test_ota_key_edit.py +++ b/tests/unit_tests/test_ota_key_edit.py @@ -59,9 +59,7 @@ def _setup(tmp_path: Path, yaml_text: str, secrets: str | None = None) -> 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) + edits = [*locate_key_edits(OLD_KEY, NEW_KEY), *old_key_edit(OLD_KEY)] with_sharers(edits) return edits @@ -816,7 +814,7 @@ def test_existing_old_secret_used_elsewhere_is_not_overwritten( 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") + (edit, kept) = old_key_edit(OLD_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" @@ -825,7 +823,7 @@ def test_existing_old_secret_used_elsewhere_is_not_overwritten( with pytest.raises( EsphomeError, match="'device_key_old:' in .* is used by .*other.yaml" ): - old_key_edit(OLD_KEY, "device_key") + old_key_edit(OLD_KEY) (tmp_path / "other.yaml").unlink() _setup( tmp_path, SECRET_YAML + "wifi:\n password: !secret device_key_old\n", secrets @@ -833,14 +831,14 @@ def test_existing_old_secret_used_elsewhere_is_not_overwritten( with pytest.raises( EsphomeError, match="'device_key_old:' in .* is used by .*test.yaml" ): - old_key_edit(OLD_KEY, "device_key") + old_key_edit(OLD_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") + old_key_edit(OLD_KEY) def test_old_secret_collision_check_refuses_an_unreadable_file( @@ -850,7 +848,7 @@ def test_old_secret_collision_check_refuses_an_unreadable_file( (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") + old_key_edit(OLD_KEY) def test_old_key_secret_with_a_folded_value_is_refused(tmp_path: Path) -> None: @@ -960,7 +958,7 @@ def test_scan_reports_what_it_could_not_read(tmp_path: Path) -> None: 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") + old_key_edit(OLD_KEY) def test_secret_used_elsewhere_in_this_configuration_is_refused( @@ -1012,7 +1010,7 @@ def test_old_secret_goes_to_the_secrets_file_of_the_block_that_gets_old_key( with pytest.raises( EsphomeError, match="No plain 'device_key:' line in .*secrets.yaml" ): - old_key_edit(OLD_KEY, "device_key") + old_key_edit(OLD_KEY) def test_a_commented_out_secret_use_does_not_count(tmp_path: Path) -> None: @@ -1037,6 +1035,49 @@ def test_a_directory_include_is_reported_as_unchecked(tmp_path: Path) -> None: ] +def test_a_hash_inside_a_secret_name_is_part_of_it(tmp_path: Path) -> None: + """A comment needs whitespace before its `#`; `device_key#prod` is the name.""" + _setup( + tmp_path, + SECRET_YAML.replace("!secret device_key", "!secret device_key#prod"), + f"device_key: other\ndevice_key#prod: {OLD_KEY}\n", + ) + edits = locate_key_edits(OLD_KEY, NEW_KEY) + assert [(e.line, e.secret) for e in edits] == [(1, "device_key#prod")] + + +def test_old_key_follows_the_ota_block_not_the_api_secret(tmp_path: Path) -> None: + """An api key from a secret next to an ota block with its own literal + key: the literal goes under the ota key line, no `_old` secret.""" + (tmp_path / "api.yaml").write_bytes(b"encryption:\n key: !secret device_key\n") + yaml_text = f"""esphome: + name: test + +api: !include api.yaml + +ota: + - platform: esphome + encryption: + key: "{OLD_KEY}" +""" + path = _setup(tmp_path, yaml_text, f"device_key: {OLD_KEY}\n") + _rotate() + assert path.read_text() == yaml_text.replace( + f' key: "{OLD_KEY}"\n', + f' key: "{NEW_KEY}"\n old_key: "{OLD_KEY}"\n', + ) + assert (tmp_path / "secrets.yaml").read_text() == f"device_key: {NEW_KEY}\n" + + +def test_bare_block_after_an_api_key_without_a_line(tmp_path: Path) -> None: + """An api key validation rebuilt has no line to read a secret from, so + the inherited key is kept as a literal.""" + path = _setup(tmp_path, API_YAML) + CORE.raw_config["api"]["encryption"] = {"key": OLD_KEY} + (edit,) = old_key_edit(OLD_KEY) + assert (edit.path, edit.new_line) == (path, f' old_key: "{OLD_KEY}"') + + 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( @@ -1068,7 +1109,7 @@ def test_existing_old_key_secret_used_elsewhere_is_refused(tmp_path: Path) -> No ) _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") + old_key_edit(OLD_KEY) def test_a_key_merged_from_an_anchor_is_refused(tmp_path: Path) -> None: