mirror of
https://github.com/esphome/esphome.git
synced 2026-10-14 02:02:51 +00:00
[core] Mask integer-valued credentials without changing code generation (#20430)
This commit is contained in:
@@ -81,8 +81,8 @@ CONFIG_SCHEMA = cv.All(
|
||||
cv.Optional(
|
||||
CONF_IDLE_PERIOD_TO_SLEEP
|
||||
): cv.positive_time_period_milliseconds,
|
||||
cv.Optional(CONF_PASSWORD): cv.uint32_t,
|
||||
cv.Optional(CONF_NEW_PASSWORD): cv.uint32_t,
|
||||
cv.Optional(CONF_PASSWORD): cv.sensitive(cv.uint32_t),
|
||||
cv.Optional(CONF_NEW_PASSWORD): cv.sensitive(cv.uint32_t),
|
||||
cv.Optional(CONF_ON_FINGER_SCAN_START): automation.validate_automation({}),
|
||||
cv.Optional(CONF_ON_FINGER_SCAN_MATCHED): automation.validate_automation(
|
||||
{}
|
||||
|
||||
@@ -166,10 +166,10 @@ _CONNECTION_SCHEMA = cv.Schema(
|
||||
{
|
||||
cv.Optional(CONF_PAN_ID): cv.hex_int,
|
||||
cv.Optional(CONF_CHANNEL): cv.int_range(min=11, max=26),
|
||||
cv.Optional(CONF_NETWORK_KEY): cv.hex_int,
|
||||
cv.Optional(CONF_NETWORK_KEY): cv.sensitive(cv.hex_int),
|
||||
cv.Optional(CONF_EXT_PAN_ID): cv.hex_int,
|
||||
cv.Optional(CONF_NETWORK_NAME): cv.string_strict,
|
||||
cv.Optional(CONF_PSKC): cv.hex_int,
|
||||
cv.Optional(CONF_PSKC): cv.sensitive(cv.hex_int),
|
||||
cv.Optional(CONF_MESH_LOCAL_PREFIX): cv.ipv6network,
|
||||
}
|
||||
)
|
||||
@@ -243,7 +243,9 @@ CONFIG_SCHEMA = cv.All(
|
||||
*CONF_DEVICE_TYPES, upper=True
|
||||
),
|
||||
cv.Optional(CONF_FORCE_DATASET): cv.boolean,
|
||||
cv.Optional(CONF_TLV): cv.All(cv.string_strict, _validate_tlv_hex),
|
||||
cv.Optional(CONF_TLV): cv.sensitive(
|
||||
cv.All(cv.string_strict, _validate_tlv_hex)
|
||||
),
|
||||
cv.Optional(CONF_USE_ADDRESS): cv.string_strict,
|
||||
cv.Optional(CONF_OUTPUT_POWER): cv.All(
|
||||
cv.decibel,
|
||||
|
||||
@@ -114,7 +114,13 @@ from esphome.schema_extractors import (
|
||||
# pylint: disable-next=unused-import
|
||||
from esphome.util import parse_esphome_version # noqa: F401
|
||||
from esphome.voluptuous_schema import KeyInvalid, _Schema
|
||||
from esphome.yaml_util import SensitiveStr, make_data_base
|
||||
from esphome.yaml_util import (
|
||||
Sensitive,
|
||||
SensitiveHexInt,
|
||||
SensitiveInt,
|
||||
SensitiveStr,
|
||||
make_data_base,
|
||||
)
|
||||
|
||||
if typing.TYPE_CHECKING:
|
||||
from esphome.types import ConfigType
|
||||
@@ -562,11 +568,17 @@ class SensitiveValidator:
|
||||
|
||||
def __call__(self, value: typing.Any) -> typing.Any:
|
||||
validated = self.inner(value)
|
||||
# Tag string results so yaml_util.dump can mask them. Non-string
|
||||
# results pass through unchanged; already-tagged values are not
|
||||
# re-wrapped to keep nested cv.sensitive applications idempotent.
|
||||
if isinstance(validated, str) and not isinstance(validated, SensitiveStr):
|
||||
# Already-tagged values are returned as-is so nested cv.sensitive is
|
||||
# idempotent; bool is an int subclass and must stay bool. Other types
|
||||
# (e.g. lambdas) pass through unmasked.
|
||||
if isinstance(validated, (Sensitive, bool)):
|
||||
return validated
|
||||
if isinstance(validated, str):
|
||||
return SensitiveStr(validated)
|
||||
if isinstance(validated, int):
|
||||
return (SensitiveHexInt if isinstance(validated, HexInt) else SensitiveInt)(
|
||||
validated
|
||||
)
|
||||
return validated
|
||||
|
||||
def __repr__(self) -> str:
|
||||
|
||||
+27
-7
@@ -81,7 +81,13 @@ def _raise_invalid(message: str) -> None:
|
||||
raise Invalid(message)
|
||||
|
||||
|
||||
class SensitiveStr(str):
|
||||
class Sensitive:
|
||||
"""Marker base for validated values masked in user-visible YAML output."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
|
||||
class SensitiveStr(Sensitive, str):
|
||||
"""Marker subclass for validated strings that should be masked in
|
||||
user-visible YAML output. ``cv.sensitive`` wraps validated values in this
|
||||
type so ``dump()`` can render them with ANSI conceal codes without
|
||||
@@ -91,6 +97,16 @@ class SensitiveStr(str):
|
||||
__slots__ = ()
|
||||
|
||||
|
||||
class SensitiveInt(Sensitive, int):
|
||||
"""Marker subclass for integer credentials masked in user-visible YAML."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
|
||||
class SensitiveHexInt(core.HexInt, SensitiveInt):
|
||||
"""Keep hexadecimal formatting and code generation for sensitive integers."""
|
||||
|
||||
|
||||
@contextmanager
|
||||
def track_yaml_loads() -> Generator[list[Path]]:
|
||||
"""Context manager that records every file loaded by the YAML loader.
|
||||
@@ -1361,11 +1377,11 @@ class ESPHomeDumper(yaml.SafeDumper):
|
||||
return self.represent_stringify(path.as_posix())
|
||||
return self.represent_stringify(value)
|
||||
|
||||
def represent_sensitive(self, value: SensitiveStr) -> yaml.ScalarNode:
|
||||
def represent_sensitive(self, value: Sensitive) -> yaml.ScalarNode:
|
||||
# Only the redact-and-not-a-secret branch is unique to sensitive
|
||||
# values; otherwise let ``represent_stringify`` handle ``!secret``
|
||||
# precedence and the plain-str fallthrough. Conceal sequence is
|
||||
# emitted as literal ``\033`` text (not actual ESC bytes) so the
|
||||
# values; otherwise use the plain string/integer representer to
|
||||
# preserve ``!secret`` references and scalar types. Conceal sequence
|
||||
# is emitted as literal ``\033`` text (not actual ESC bytes) so the
|
||||
# output matches the prior regex format and device-builder's
|
||||
# ``\033[8m...\033[28m`` parser keeps working.
|
||||
if self._redact_sensitive and not is_secret(value):
|
||||
@@ -1373,6 +1389,10 @@ class ESPHomeDumper(yaml.SafeDumper):
|
||||
tag="tag:yaml.org,2002:str",
|
||||
value=f"\\033[8m{value}\\033[28m",
|
||||
)
|
||||
# MRO dispatch picks this representer over represent_int, so
|
||||
# integers have to be forwarded explicitly.
|
||||
if isinstance(value, int):
|
||||
return self.represent_int(value)
|
||||
return self.represent_stringify(value)
|
||||
|
||||
# pylint: disable=arguments-renamed
|
||||
@@ -1452,8 +1472,8 @@ ESPHomeDumper.add_multi_representer(
|
||||
)
|
||||
ESPHomeDumper.add_multi_representer(bool, ESPHomeDumper.represent_bool)
|
||||
ESPHomeDumper.add_multi_representer(str, ESPHomeDumper.represent_stringify)
|
||||
# MRO-walked dispatch; SensitiveStr's own entry wins over the str one.
|
||||
ESPHomeDumper.add_multi_representer(SensitiveStr, ESPHomeDumper.represent_sensitive)
|
||||
# MRO-walked dispatch; Sensitive precedes str and int in every marker's MRO.
|
||||
ESPHomeDumper.add_multi_representer(Sensitive, ESPHomeDumper.represent_sensitive)
|
||||
ESPHomeDumper.add_multi_representer(int, ESPHomeDumper.represent_int)
|
||||
ESPHomeDumper.add_multi_representer(float, ESPHomeDumper.represent_float)
|
||||
ESPHomeDumper.add_multi_representer(_BaseAddress, ESPHomeDumper.represent_stringify)
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the fingerprint_grow component."""
|
||||
@@ -0,0 +1,19 @@
|
||||
"""Tests for the fingerprint_grow component schema."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components.fingerprint_grow import CONFIG_SCHEMA
|
||||
from esphome.const import PlatformFramework
|
||||
from esphome.yaml_util import Sensitive
|
||||
from tests.component_tests.types import SetCoreConfigCallable
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["password", "new_password"])
|
||||
def test_password_is_sensitive(
|
||||
field: str, set_core_config: SetCoreConfigCallable
|
||||
) -> None:
|
||||
"""Numeric sensor passwords are tagged for redaction and keep their value."""
|
||||
set_core_config(PlatformFramework.ESP32_IDF)
|
||||
config = CONFIG_SCHEMA({field: 305419896})
|
||||
assert isinstance(config[field], Sensitive)
|
||||
assert config[field] == 305419896
|
||||
@@ -0,0 +1 @@
|
||||
"""Tests for the openthread component."""
|
||||
@@ -0,0 +1,39 @@
|
||||
"""Tests for the openthread component schema."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components.esp32 import KEY_VARIANT, VARIANT_ESP32C6
|
||||
from esphome.components.openthread import CONFIG_SCHEMA
|
||||
from esphome.const import PlatformFramework
|
||||
from esphome.core import HexInt
|
||||
from esphome.yaml_util import Sensitive
|
||||
from tests.component_tests.types import SetCoreConfigCallable
|
||||
|
||||
NETWORK_KEY = 0xDFD34F0F05CAD978EC4E32B0413038FF
|
||||
PSKC = 0xC23A76E98F1A6483639B1AC1271E2E27
|
||||
TLV = "0e080000000000010000000300001435060004001fffe0"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("config", "field", "expected"),
|
||||
[
|
||||
({"network_key": hex(NETWORK_KEY)}, "network_key", NETWORK_KEY),
|
||||
({"network_key": hex(NETWORK_KEY), "pskc": hex(PSKC)}, "pskc", PSKC),
|
||||
({"tlv": TLV}, "tlv", TLV),
|
||||
],
|
||||
)
|
||||
def test_credentials_are_sensitive(
|
||||
config: dict[str, str],
|
||||
field: str,
|
||||
expected: int | str,
|
||||
set_core_config: SetCoreConfigCallable,
|
||||
) -> None:
|
||||
"""Thread credentials are tagged for redaction without changing their value
|
||||
or the hexadecimal integer type used to build sdkconfig options."""
|
||||
set_core_config(
|
||||
PlatformFramework.ESP32_IDF, platform_data={KEY_VARIANT: VARIANT_ESP32C6}
|
||||
)
|
||||
validated = CONFIG_SCHEMA(config)
|
||||
assert isinstance(validated[field], Sensitive)
|
||||
assert validated[field] == expected
|
||||
assert isinstance(validated[field], HexInt) == isinstance(expected, int)
|
||||
@@ -1,3 +1,4 @@
|
||||
from collections.abc import Callable
|
||||
import importlib
|
||||
import io
|
||||
import json
|
||||
@@ -11,7 +12,7 @@ from hypothesis.strategies import builds, integers, ip_addresses, one_of, text
|
||||
import pytest
|
||||
import voluptuous as vol
|
||||
|
||||
from esphome import config_validation as cv
|
||||
from esphome import config_validation as cv, cpp_generator
|
||||
from esphome.components.esp32 import (
|
||||
VARIANT_ESP32,
|
||||
VARIANT_ESP32C2,
|
||||
@@ -69,6 +70,8 @@ from esphome.schema_extractors import SCHEMA_EXTRACT
|
||||
from esphome.util import Registry
|
||||
from esphome.yaml_util import (
|
||||
ESPHomeDataBase,
|
||||
SensitiveHexInt,
|
||||
SensitiveInt,
|
||||
SensitiveStr,
|
||||
load_yaml,
|
||||
make_data_base,
|
||||
@@ -201,12 +204,13 @@ def test_sensitive__wraps_string_result_in_sensitive_str() -> None:
|
||||
assert result == "hunter2"
|
||||
|
||||
|
||||
def test_sensitive__does_not_double_tag_already_sensitive() -> None:
|
||||
# If the inner validator already returns a SensitiveStr (e.g., nested
|
||||
@pytest.mark.parametrize(
|
||||
"pre_tagged", [SensitiveStr("hunter2"), SensitiveInt(5), SensitiveHexInt(0x12)]
|
||||
)
|
||||
def test_sensitive__does_not_double_tag_already_sensitive(pre_tagged: object) -> None:
|
||||
# If the inner validator already returns a tagged value (e.g., nested
|
||||
# cv.sensitive wrappers), re-tagging is a no-op rather than a new
|
||||
# SensitiveStr around the same value.
|
||||
pre_tagged = SensitiveStr("hunter2")
|
||||
|
||||
# wrapper around the same value.
|
||||
def inner(_value):
|
||||
return pre_tagged
|
||||
|
||||
@@ -216,16 +220,33 @@ def test_sensitive__does_not_double_tag_already_sensitive() -> None:
|
||||
assert result is pre_tagged
|
||||
|
||||
|
||||
def test_sensitive__non_string_result_passes_through() -> None:
|
||||
# If an inner validator returns something other than a string (e.g., a
|
||||
# Lambda template), the sensitive wrapper must not coerce it.
|
||||
sentinel = object()
|
||||
|
||||
@pytest.mark.parametrize("passthrough", [object(), True, False])
|
||||
def test_sensitive__non_string_result_passes_through(passthrough: object) -> None:
|
||||
# If an inner validator returns something other than a string or integer
|
||||
# (e.g., a Lambda template or a bool), the sensitive wrapper must not
|
||||
# coerce it.
|
||||
def inner(_value):
|
||||
return sentinel
|
||||
return passthrough
|
||||
|
||||
validator = cv.sensitive(inner)
|
||||
assert validator("anything") is sentinel
|
||||
assert validator("anything") is passthrough
|
||||
|
||||
|
||||
@pytest.mark.parametrize("inner", [cv.uint32_t, cv.hex_int])
|
||||
def test_sensitive__integer_preserves_codegen(inner: Callable[[str], int]) -> None:
|
||||
"""Tag credentials without changing their firmware expressions."""
|
||||
original = inner("0x12345678")
|
||||
tagged = cv.sensitive(inner)("0x12345678")
|
||||
assert isinstance(tagged, SensitiveInt)
|
||||
assert tagged == original
|
||||
assert str(cpp_generator.safe_exp(tagged)) == str(cpp_generator.safe_exp(original))
|
||||
assert isinstance(tagged, HexInt) == isinstance(original, HexInt)
|
||||
|
||||
|
||||
def test_sensitive__integer_validation_still_rejects_out_of_range() -> None:
|
||||
"""Keep the inner validator's range checks."""
|
||||
with pytest.raises(Invalid):
|
||||
cv.sensitive(cv.uint32_t)(1 << 32)
|
||||
|
||||
|
||||
def test_sensitive__is_detectable_via_isinstance() -> None:
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from collections.abc import Callable
|
||||
import io
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
@@ -1901,9 +1902,39 @@ def test_dump__redacts_sensitive_str_by_default() -> None:
|
||||
) # the raw value is only present inside the wrap
|
||||
|
||||
|
||||
def test_dump__show_secrets_emits_sensitive_str_raw() -> None:
|
||||
out = yaml_util.dump({"password": SensitiveStr("hunter2")}, show_secrets=True)
|
||||
assert "hunter2" in out
|
||||
@pytest.mark.parametrize(
|
||||
("validator", "value", "rendered"),
|
||||
[
|
||||
(cv.uint32_t, 305419896, "305419896"),
|
||||
(
|
||||
cv.hex_int,
|
||||
0x00112233445566778899AABBCCDDEEFF,
|
||||
"0x112233445566778899AABBCCDDEEFF",
|
||||
),
|
||||
(cv.int_, -42, "-42"),
|
||||
],
|
||||
)
|
||||
def test_dump__redacts_sensitive_integer(
|
||||
validator: Callable[[int], int], value: int, rendered: str
|
||||
) -> None:
|
||||
"""Mask integer credentials without changing their numeric value or format."""
|
||||
sensitive = cv.sensitive(validator)(value)
|
||||
assert sensitive == value
|
||||
assert str(sensitive) == rendered
|
||||
out = yaml_util.dump({"credential": sensitive})
|
||||
assert f"\\033[8m{rendered}\\033[28m" in out
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
(SensitiveStr("hunter2"), "password: hunter2"),
|
||||
(cv.sensitive(cv.hex_int)(0x12345678), "password: 0x12345678"),
|
||||
],
|
||||
)
|
||||
def test_dump__show_secrets_emits_sensitive_raw(value: object, expected: str) -> None:
|
||||
out = yaml_util.dump({"password": value}, show_secrets=True)
|
||||
assert out.strip() == expected
|
||||
assert "\\033[8m" not in out
|
||||
assert "\\033[28m" not in out
|
||||
|
||||
@@ -1914,19 +1945,20 @@ def test_dump__plain_str_is_not_redacted() -> None:
|
||||
assert "\\033[8m" not in out
|
||||
|
||||
|
||||
def test_dump__secret_reference_wins_over_redaction() -> None:
|
||||
@pytest.mark.parametrize(
|
||||
"value", [SensitiveStr("hunter2"), cv.sensitive(cv.uint32_t)(305419896)]
|
||||
)
|
||||
def test_dump__secret_reference_wins_over_redaction(
|
||||
value: object, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
# If the value also has an entry in _SECRET_VALUES (i.e., it was loaded
|
||||
# via !secret), the dump should render it as !secret <name>, not as a
|
||||
# redacted scalar. SensitiveStr layered on top must not change that.
|
||||
value = SensitiveStr("hunter2")
|
||||
yaml_util._SECRET_VALUES[str(value)] = "my_secret_name"
|
||||
try:
|
||||
out = yaml_util.dump({"password": value})
|
||||
assert "!secret" in out
|
||||
assert "my_secret_name" in out
|
||||
assert "\\033[8m" not in out
|
||||
finally:
|
||||
yaml_util._SECRET_VALUES.clear()
|
||||
# redacted scalar. The sensitive marker layered on top must not change that.
|
||||
monkeypatch.setitem(yaml_util._SECRET_VALUES, str(value), "my_secret_name")
|
||||
out = yaml_util.dump({"password": value})
|
||||
assert "!secret" in out
|
||||
assert "my_secret_name" in out
|
||||
assert "\\033[8m" not in out
|
||||
|
||||
|
||||
def test_dump__redaction_flag_does_not_leak_between_calls() -> None:
|
||||
|
||||
Reference in New Issue
Block a user