A comment needs whitespace before its hash in a secret name, old_key takes its secret from the ota block it lands on

This commit is contained in:
J. Nick Koston
2026-10-13 20:16:38 -10:00
parent 3bc9f74f46
commit 4165dbb534
3 changed files with 75 additions and 20 deletions
+1 -2
View File
@@ -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))
+22 -7
View File
@@ -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<q>[\"']?)(?P<name>{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<q>[\"']?)(?P<name>{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 <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 (``secret``) stays there as ``<secret>_old``. Of
several entries the one with its own old_key or key line gets it."""
key from secrets.yaml stays there as ``<name>_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:
+52 -11
View File
@@ -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 `<name>_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: