When YAML config editing is enabled, AI-written code creates unauthenticated backups under /local/, exposing configuration.yaml secrets to anyone with web access.

@@ -486,7 +500,213 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

+        yaml_path = call.data["yaml_path"]+        content = call.data.get("content")+        do_backup = call.data.get("backup", True)++        # Validate file path — only configuration.yaml and packages/*.yaml+        normalized = os.path.normpath(rel_path)  # noqa: ASYNC240+        if normalized.startswith("..") or normalized.startswith("/"):+            return {+                "success": False,+                "error": "Path traversal is not allowed.",+            }++        is_config_yaml = normalized in ALLOWED_YAML_CONFIG_FILES+        is_package = fnmatch.fnmatch(normalized, "packages/*.yaml") or fnmatch.fnmatch(+            normalized, "packages/**/*.yaml"+        )+        if not is_config_yaml and not is_package:+            return {+                "success": False,+                "error": (+                    f"File '{rel_path}' is not allowed. "+                    f"Only {', '.join(ALLOWED_YAML_CONFIG_FILES)} and packages/*.yaml are supported."+                ),+            }++        # Validate yaml_path against whitelist+        if yaml_path in BLOCKED_YAML_KEYS:+            return {+                "success": False,+                "error": (+                    f"Key '{yaml_path}' is blocked for safety. "+                    f"Blocked keys: {', '.join(sorted(BLOCKED_YAML_KEYS))}"+                ),+            }+        if yaml_path not in ALLOWED_YAML_KEYS:+            return {+                "success": False,+                "error": (+                    f"Key '{yaml_path}' is not in the allowed list. "+                    f"Allowed keys: {', '.join(sorted(ALLOWED_YAML_KEYS))}"+                ),+            }++        # Validate content is valid YAML for add/replace+        parsed_content: Any = None+        if action in ("add", "replace"):+            if not content:+                return {+                    "success": False,+                    "error": f"'content' is required for action '{action}'.",+                }+            try:+                parsed_content = yaml.safe_load(content)+            except yaml.YAMLError as err:+                return {+                    "success": False,+                    "error": f"Invalid YAML content: {err}",+                }++        target_file = config_dir / normalized+        backup_path_str: str | None = None++        try:+            # Read existing file content (or start with empty dict)+            if target_file.exists():+                raw_content = await hass.async_add_executor_job(target_file.read_text)+                try:+                    data = yaml.safe_load(raw_content) or {}+                except yaml.YAMLError as err:+                    return {+                        "success": False,+                        "error": (+                            f"Cannot parse existing file '{rel_path}': {err}. "+                            "The file may contain !include or !secret directives "+                            "that require Home Assistant's YAML loader. "+                            "This tool only supports standard YAML."+                        ),+                    }+                if not isinstance(data, dict):+                    return {+                        "success": False,+                        "error": f"File '{rel_path}' root is not a YAML mapping.",+                    }+            else:+                if action == "remove":+                    return {+                        "success": False,+                        "error": f"File does not exist: {rel_path}",+                    }+                data = {}+                raw_content = ""

+                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")+                safe_name = normalized.replace(os.sep, "_")+                backup_file = backup_dir / f"{safe_name}.{timestamp}.bak"+                await hass.async_add_executor_job(+                    shutil.copy2, str(target_file), str(backup_file)+                )+                backup_path_str = str(backup_file.relative_to(config_dir))+                _LOGGER.info("Backup created: %s", backup_path_str)++            # Perform the action+            if action == "add":+                if yaml_path in data:+                    existing = data[yaml_path]+                    # Merge: list extends list, dict merges dict, else replace+                    if isinstance(existing, list) and isinstance(parsed_content, list):+                        data[yaml_path] = existing + parsed_content+                    elif isinstance(existing, dict) and isinstance(+                        parsed_content, dict+                    ):+                        existing.update(parsed_content)+                    else:+                        data[yaml_path] = parsed_content+                else:+                    data[yaml_path] = parsed_content+            elif action == "replace":+                data[yaml_path] = parsed_content+            elif action == "remove":+                if yaml_path not in data:+                    return {+                        "success": False,+                        "error": f"Key '{yaml_path}' not found in '{rel_path}'.",+                    }+                del data[yaml_path]++            # Serialize back to YAML+            new_content = yaml.dump(+                data,+                default_flow_style=False,+                allow_unicode=True,+                sort_keys=False,+            )++            # Validate the result parses cleanly+            try:+                yaml.safe_load(new_content)+            except yaml.YAMLError as err:+                return {+                    "success": False,+                    "error": f"Generated YAML failed validation: {err}",+                }++            # Create parent directories if needed (for new package files)+            if not target_file.parent.exists():+                await hass.async_add_executor_job(+                    target_file.parent.mkdir, parents=True, exist_ok=True+                )++            # Write the file+            await hass.async_add_executor_job(target_file.write_text, new_content)++            stat = target_file.stat()+            modified_dt = datetime.fromtimestamp(stat.st_mtime)++            _LOGGER.info(+                "YAML config edited: %s (action=%s, key=%s)",+                rel_path,+                action,+                yaml_path,+            )++            result: dict[str, Any] = {+                "success": True,+                "file": rel_path,+                "action": action,+                "yaml_path": yaml_path,+                "size": stat.st_size,+                "modified": modified_dt.isoformat(),+            }+            if backup_path_str:+                result["backup_path"] = backup_path_str++            return result++        except PermissionError:+            _LOGGER.error("Permission denied editing: %s", rel_path)+            return {+                "success": False,+                "error": f"Permission denied: {rel_path}",+            }+        except OSError as err:+            _LOGGER.error("Error editing YAML config %s: %s", rel_path, err)+            return {+                "success": False,+                "error": str(err),+            }+     # Register all services with response support

+     hass.services.async_register(         DOMAIN,         SERVICE_LIST_FILES,

@@ -8,0 +9 @@ from __future__ import annotations+import errno@@ -12,0 +14 @@ import re+import shutil@@ -18,0 +21 @@ import voluptuous as vol

+    legacy_dir = config_dir / "www" / "yaml_backups"

+    failed = 0+    for src in legacy_dir.iterdir():+        if not src.is_file() or src.is_symlink() or src.suffix != ".bak":+            continue++        dest = new_dir / src.name+        if dest.exists():+            dest = new_dir / f"{src.stem}.legacy{src.suffix}"+            counter = 1+            while dest.exists():+                dest = new_dir / f"{src.stem}.legacy{counter}{src.suffix}"+                counter += 1++        try:+            shutil.move(str(src), str(dest))+            moved += 1+        except OSError as err:+            failed += 1+            _LOGGER.error(+                "Failed to migrate %s out of www/: %s. File remains "+                "exposed via /local/yaml_backups/.",+                src.name,+                err,+            )++    try:+        legacy_dir.rmdir()+    except OSError as err:+        if err.errno not in (errno.ENOTEMPTY, errno.EEXIST):+            _LOGGER.warning(+                "Could not remove legacy backup dir %s: [%s] %s",+                legacy_dir,+                type(err).__name__,+                err,+            )++    return moved, failed++