mirror of
https://github.com/esphome/esphome.git
synced 2026-06-24 12:53:26 +00:00
[core] Sensitive redaction via yaml_util representer
cv.sensitive(...) now returns a SensitiveStr (thin str subclass) so the tag travels with the validated value. yaml_util.dump constructs a per-call ESPHomeDumper subclass with a class-attribute redaction flag; the PyYAML representer for SensitiveStr renders values wrapped in literal \\033[8m...\\033[28m text when show_secrets is False and raw when True. The post-dump regex in command_config is deleted. Also tags wifi.ssid sites with cv.sensitive so SSID coverage isn't lost when the regex (which matched 'ssid:' via substring) goes away. No module-level mutable state; the per-call subclass keeps each dump invocation self-contained and thread-safe by construction.
This commit is contained in:
@@ -1412,11 +1412,6 @@ def command_config(args: ArgsProtocol, config: ConfigType) -> int | None:
|
||||
if not CORE.verbose:
|
||||
config = strip_default_ids(config)
|
||||
output = yaml_util.dump(config, args.show_secrets)
|
||||
# add the console decoration so the front-end can hide the secrets
|
||||
if not args.show_secrets:
|
||||
output = re.sub(
|
||||
r"(password|key|psk|ssid)\: (.+)", r"\1: \\033[8m\2\\033[28m", output
|
||||
)
|
||||
if not CORE.quiet:
|
||||
safe_print(output)
|
||||
_LOGGER.info("Configuration is valid!")
|
||||
|
||||
@@ -271,7 +271,7 @@ EAP_AUTH_SCHEMA = cv.All(
|
||||
WIFI_NETWORK_BASE = cv.Schema(
|
||||
{
|
||||
cv.GenerateID(): cv.declare_id(WiFiAP),
|
||||
cv.Optional(CONF_SSID): cv.ssid,
|
||||
cv.Optional(CONF_SSID): cv.sensitive(cv.ssid),
|
||||
cv.Optional(CONF_PASSWORD): cv.sensitive(validate_password),
|
||||
cv.Optional(CONF_CHANNEL): validate_channel,
|
||||
cv.Optional(CONF_MANUAL_IP): STA_MANUAL_IP_SCHEMA,
|
||||
@@ -434,7 +434,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
cv.Optional(CONF_NETWORKS): cv.All(
|
||||
cv.ensure_list(WIFI_NETWORK_STA), cv.Length(max=MAX_WIFI_NETWORKS)
|
||||
),
|
||||
cv.Optional(CONF_SSID): cv.ssid,
|
||||
cv.Optional(CONF_SSID): cv.sensitive(cv.ssid),
|
||||
cv.Optional(CONF_PASSWORD): cv.sensitive(validate_password),
|
||||
cv.Optional(CONF_MANUAL_IP): STA_MANUAL_IP_SCHEMA,
|
||||
cv.Optional(CONF_EAP): EAP_AUTH_SCHEMA,
|
||||
@@ -850,7 +850,7 @@ async def final_step():
|
||||
WiFiConfigureAction,
|
||||
cv.Schema(
|
||||
{
|
||||
cv.Required(CONF_SSID): cv.templatable(cv.ssid),
|
||||
cv.Required(CONF_SSID): cv.sensitive(cv.templatable(cv.ssid)),
|
||||
cv.Required(CONF_PASSWORD): cv.sensitive(cv.templatable(validate_password)),
|
||||
cv.Optional(CONF_SAVE, default=True): cv.templatable(cv.boolean),
|
||||
cv.Optional(CONF_TIMEOUT, default="30000ms"): cv.templatable(
|
||||
|
||||
@@ -101,7 +101,7 @@ from esphome.schema_extractors import (
|
||||
)
|
||||
from esphome.util import parse_esphome_version
|
||||
from esphome.voluptuous_schema import _Schema
|
||||
from esphome.yaml_util import make_data_base
|
||||
from esphome.yaml_util import SensitiveStr, make_data_base
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@@ -514,7 +514,13 @@ class SensitiveValidator:
|
||||
self.inner = inner
|
||||
|
||||
def __call__(self, value: typing.Any) -> typing.Any:
|
||||
return self.inner(value)
|
||||
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):
|
||||
return SensitiveStr(validated)
|
||||
return validated
|
||||
|
||||
def __repr__(self) -> str:
|
||||
# Mirror the inner validator's repr so ``build_language_schema``'s
|
||||
|
||||
@@ -52,6 +52,16 @@ _load_listeners: list[Callable[[Path], None]] = []
|
||||
DocumentPath = list[str | int]
|
||||
|
||||
|
||||
class SensitiveStr(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
|
||||
needing a post-process regex.
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
|
||||
@contextmanager
|
||||
def track_yaml_loads() -> Generator[list[Path]]:
|
||||
"""Context manager that records every file loaded by the YAML loader.
|
||||
@@ -808,11 +818,18 @@ def dump(dict_, show_secrets=False, sort_keys=False):
|
||||
if show_secrets:
|
||||
_SECRET_VALUES.clear()
|
||||
_SECRET_CACHE.clear()
|
||||
|
||||
# Per-call subclass so the redaction flag lives on the dumper class itself,
|
||||
# not in module-level mutable state. Cheap (one type object per call),
|
||||
# encapsulated, and thread-safe by construction.
|
||||
class _Dumper(ESPHomeDumper):
|
||||
_redact_sensitive = not show_secrets
|
||||
|
||||
return yaml.dump(
|
||||
dict_,
|
||||
default_flow_style=False,
|
||||
allow_unicode=True,
|
||||
Dumper=ESPHomeDumper,
|
||||
Dumper=_Dumper,
|
||||
sort_keys=sort_keys,
|
||||
)
|
||||
|
||||
@@ -958,6 +975,10 @@ def format_path(path: DocumentPath, current_obj: Any) -> str:
|
||||
|
||||
|
||||
class ESPHomeDumper(yaml.SafeDumper):
|
||||
# Default for the base class; per-call subclass in ``dump()`` overrides.
|
||||
# When True, ``represent_sensitive`` wraps values in ANSI conceal codes.
|
||||
_redact_sensitive: bool = False
|
||||
|
||||
def represent_mapping(self, tag, mapping, flow_style=None):
|
||||
value = []
|
||||
node = yaml.MappingNode(tag, value, flow_style=flow_style)
|
||||
@@ -992,6 +1013,23 @@ class ESPHomeDumper(yaml.SafeDumper):
|
||||
return self.represent_secret(value)
|
||||
return self.represent_scalar(tag="tag:yaml.org,2002:str", value=str(value))
|
||||
|
||||
def represent_sensitive(self, value):
|
||||
# ``!secret`` references win: keep the original representation so the
|
||||
# dumped YAML round-trips back to ``!secret name`` instead of leaking
|
||||
# the resolved value.
|
||||
if is_secret(value):
|
||||
return self.represent_secret(value)
|
||||
if self._redact_sensitive:
|
||||
# Emit the conceal sequence as literal ``\033`` text (not actual
|
||||
# ESC bytes) so the dump matches the previous regex-based output
|
||||
# and downstream consumers like device-builder, which match
|
||||
# ``\033[8m...\033[28m`` against the rendered text, keep working.
|
||||
return self.represent_scalar(
|
||||
tag="tag:yaml.org,2002:str",
|
||||
value=f"\\033[8m{value}\\033[28m",
|
||||
)
|
||||
return self.represent_scalar(tag="tag:yaml.org,2002:str", value=str(value))
|
||||
|
||||
# pylint: disable=arguments-renamed
|
||||
def represent_bool(self, value):
|
||||
return self.represent_scalar(
|
||||
@@ -1063,6 +1101,9 @@ ESPHomeDumper.add_multi_representer(
|
||||
)
|
||||
ESPHomeDumper.add_multi_representer(bool, ESPHomeDumper.represent_bool)
|
||||
ESPHomeDumper.add_multi_representer(str, ESPHomeDumper.represent_stringify)
|
||||
# Registered after ``str`` so ``add_multi_representer``'s MRO lookup prefers
|
||||
# the more specific ``SensitiveStr`` representer over the bare-``str`` one.
|
||||
ESPHomeDumper.add_multi_representer(SensitiveStr, 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)
|
||||
|
||||
@@ -145,6 +145,46 @@ def test_sensitive__custom_inner_delegates_validation() -> None:
|
||||
validator(123)
|
||||
|
||||
|
||||
def test_sensitive__wraps_string_result_in_sensitive_str() -> None:
|
||||
from esphome.yaml_util import SensitiveStr
|
||||
|
||||
validator = config_validation.sensitive()
|
||||
result = validator("hunter2")
|
||||
|
||||
assert isinstance(result, SensitiveStr)
|
||||
assert isinstance(result, str)
|
||||
assert result == "hunter2"
|
||||
|
||||
|
||||
def test_sensitive__does_not_double_tag_already_sensitive() -> None:
|
||||
from esphome.yaml_util import SensitiveStr
|
||||
|
||||
# If the inner validator already returns a SensitiveStr (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")
|
||||
|
||||
def inner(_value):
|
||||
return pre_tagged
|
||||
|
||||
validator = config_validation.sensitive(inner)
|
||||
result = validator("anything")
|
||||
|
||||
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()
|
||||
|
||||
def inner(_value):
|
||||
return sentinel
|
||||
|
||||
validator = config_validation.sensitive(inner)
|
||||
assert validator("anything") is sentinel
|
||||
|
||||
|
||||
def test_sensitive__is_detectable_via_isinstance() -> None:
|
||||
validator = config_validation.sensitive()
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ from esphome.yaml_util import (
|
||||
DiscoveredYamlFiles,
|
||||
ESPHomeDataBase,
|
||||
ESPLiteralValue,
|
||||
SensitiveStr,
|
||||
discover_user_yaml_files,
|
||||
force_load_include_files,
|
||||
format_path,
|
||||
@@ -1340,3 +1341,57 @@ def test_frontmatter_included_file_stored(tmp_path: Path) -> None:
|
||||
assert main.resolve() not in core.CORE.frontmatter
|
||||
# Included file's frontmatter is captured
|
||||
assert core.CORE.frontmatter[inc.resolve()]["child_meta"] == "hello"
|
||||
|
||||
|
||||
def test_sensitive_str__is_a_str_subclass() -> None:
|
||||
value = SensitiveStr("hunter2")
|
||||
assert isinstance(value, str)
|
||||
assert value == "hunter2"
|
||||
|
||||
|
||||
def test_dump__redacts_sensitive_str_by_default() -> None:
|
||||
out = yaml_util.dump({"password": SensitiveStr("hunter2")})
|
||||
assert "\\033[8mhunter2\\033[28m" in out
|
||||
assert "hunter2" not in out.replace(
|
||||
"\\033[8mhunter2\\033[28m", ""
|
||||
) # 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
|
||||
assert "\\033[8m" not in out
|
||||
assert "\\033[28m" not in out
|
||||
|
||||
|
||||
def test_dump__plain_str_is_not_redacted() -> None:
|
||||
out = yaml_util.dump({"hostname": "myserver"})
|
||||
assert "myserver" in out
|
||||
assert "\\033[8m" not in out
|
||||
|
||||
|
||||
def test_dump__secret_reference_wins_over_redaction() -> 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()
|
||||
|
||||
|
||||
def test_dump__redaction_flag_does_not_leak_between_calls() -> None:
|
||||
# Per-call _Dumper subclass means show_secrets in one call doesn't
|
||||
# affect another. Run them in both orders to catch any leakage.
|
||||
redacted = yaml_util.dump({"password": SensitiveStr("hunter2")})
|
||||
raw = yaml_util.dump({"password": SensitiveStr("hunter2")}, show_secrets=True)
|
||||
redacted_again = yaml_util.dump({"password": SensitiveStr("hunter2")})
|
||||
|
||||
assert "\\033[8m" in redacted
|
||||
assert "\\033[8m" not in raw
|
||||
assert "\\033[8m" in redacted_again
|
||||
|
||||
Reference in New Issue
Block a user