Compare commits

..

4 Commits

Author SHA1 Message Date
Stefan Agner 1e457600f1 Harden backup tar extraction with Python tar_filter (#172252) 2026-05-27 18:10:04 +02:00
Franck Nijhof 51d1d4aa9e Update MDI icons from frontend for 2026.6.0 beta (#172366) 2026-05-27 18:04:08 +02:00
Alex Romanov 8184b93151 Add Tuya smart kettle select entities (#171897)
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
2026-05-27 17:32:01 +02:00
Bram Kragten 403cb85bc8 Bump frontend to 20260527.0 (#172355) 2026-05-27 17:16:46 +02:00
15 changed files with 223 additions and 125 deletions
+2 -4
View File
@@ -92,8 +92,7 @@ def _extract_backup(
):
ostf.tar.extractall(
path=Path(tempdir, "extracted"),
members=securetar.secure_path(ostf.tar),
filter="fully_trusted",
filter="tar",
)
backup_meta_file = Path(tempdir, "extracted", "backup.json")
backup_meta = json.loads(backup_meta_file.read_text(encoding="utf8"))
@@ -119,8 +118,7 @@ def _extract_backup(
) as istf:
istf.extractall(
path=Path(tempdir, "homeassistant"),
members=securetar.secure_path(istf),
filter="fully_trusted",
filter="tar",
)
if restore_content.restore_homeassistant:
keep = list(KEEP_BACKUPS)
+3 -16
View File
@@ -11,7 +11,7 @@ from homeassistant.helpers.hassio import is_hassio
from .agent import BackupAgent, LocalBackupAgent, OnProgressCallback
from .const import DOMAIN, LOGGER
from .models import AgentBackup, BackupNotFound, InvalidBackupFilename
from .models import AgentBackup, BackupNotFound
from .util import read_backup, suggested_filename
@@ -54,13 +54,7 @@ class CoreLocalBackupAgent(LocalBackupAgent):
try:
backup = read_backup(backup_path)
backups[backup.backup_id] = (backup, backup_path)
except (
OSError,
TarError,
json.JSONDecodeError,
KeyError,
InvalidBackupFilename,
) as err:
except (OSError, TarError, json.JSONDecodeError, KeyError) as err:
LOGGER.warning("Unable to read backup %s: %s", backup_path, err)
return backups
@@ -128,14 +122,7 @@ class CoreLocalBackupAgent(LocalBackupAgent):
def get_new_backup_path(self, backup: AgentBackup) -> Path:
"""Return the local path to a new backup."""
candidate = self._backup_dir / suggested_filename(backup)
# suggested_filename does not strip separators; refuse paths that would
# land outside the backup directory.
if candidate.parent != self._backup_dir:
raise InvalidBackupFilename(
f"Refusing to write outside {self._backup_dir}: {candidate}"
)
return candidate
return self._backup_dir / suggested_filename(backup)
async def async_delete_backup(self, backup_id: str, **kwargs: Any) -> None:
"""Delete a backup file."""
+1 -7
View File
@@ -1978,13 +1978,7 @@ class CoreBackupReaderWriter(BackupReaderWriter):
try:
backup = await async_add_executor_job(read_backup, temp_file)
except (
OSError,
tarfile.TarError,
json.JSONDecodeError,
KeyError,
InvalidBackupFilename,
) as err:
except (OSError, tarfile.TarError, json.JSONDecodeError, KeyError) as err:
LOGGER.warning("Unable to parse backup %s: %s", temp_file, err)
raise
+3 -10
View File
@@ -6,7 +6,7 @@ import copy
from dataclasses import dataclass, replace
from io import BytesIO
import json
from pathlib import Path, PurePath, PureWindowsPath
from pathlib import Path, PurePath
from queue import SimpleQueue
import tarfile
import threading
@@ -34,7 +34,7 @@ from homeassistant.util.async_iterator import (
from homeassistant.util.json import JsonObjectType, json_loads_object
from .const import BUF_SIZE, LOGGER, SECURETAR_CREATE_VERSION
from .models import AddonInfo, AgentBackup, Folder, InvalidBackupFilename
from .models import AddonInfo, AgentBackup, Folder
class DecryptError(HomeAssistantError):
@@ -109,13 +109,6 @@ def read_backup(backup_path: Path) -> AgentBackup:
extra_metadata = cast(dict[str, bool | str], data.get("extra", {}))
date = extra_metadata.get("supervisor.backup_request_date", data["date"])
name = cast(str, data["name"])
# The name is used to derive the on-disk filename via suggested_filename;
# reject anything that could escape the backup directory.
safe_name = PureWindowsPath(name).name
if safe_name != name or name in ("", ".", ".."):
raise InvalidBackupFilename(f"Invalid backup name: {name!r}")
return AgentBackup(
addons=addons,
backup_id=cast(str, data["slug"]),
@@ -125,7 +118,7 @@ def read_backup(backup_path: Path) -> AgentBackup:
folders=folders,
homeassistant_included=homeassistant_included,
homeassistant_version=homeassistant_version,
name=name,
name=cast(str, data["name"]),
protected=cast(bool, data.get("protected", False)),
size=backup_path.stat().st_size,
)
@@ -21,5 +21,5 @@
"integration_type": "system",
"preview_features": { "winter_mode": {} },
"quality_scale": "internal",
"requirements": ["home-assistant-frontend==20260429.4"]
"requirements": ["home-assistant-frontend==20260527.0"]
}
+2
View File
@@ -939,6 +939,7 @@ class DPCode(StrEnum):
TEMP_INDOOR = "temp_indoor" # Indoor temperature in °C
TEMP_SET = "temp_set" # Set the temperature in °C
TEMP_SET_F = "temp_set_f" # Set the temperature in °F
TEMP_SETTING_QUICK_C = "temp_setting_quick_c"
TEMP_UNIT_CONVERT = "temp_unit_convert" # Temperature unit switching
TEMP_VALUE = "temp_value" # Color temperature
TEMP_VALUE_V2 = "temp_value_v2"
@@ -992,6 +993,7 @@ class DPCode(StrEnum):
WORK_POWER = "work_power"
WORK_STATE = "work_state"
WORK_STATE_E = "work_state_e"
WORK_TYPE = "work_type"
@dataclass
+12
View File
@@ -19,6 +19,18 @@ from .entity import TuyaEntity
# All descriptions can be found here. Mostly the Enum data types in the
# default instructions set of each category end up being a select.
SELECTS: dict[DeviceCategory, tuple[SelectEntityDescription, ...]] = {
DeviceCategory.BH: (
SelectEntityDescription(
key=DPCode.TEMP_SETTING_QUICK_C,
entity_category=EntityCategory.CONFIG,
translation_key="quick_heat_temperature",
),
SelectEntityDescription(
key=DPCode.WORK_TYPE,
entity_category=EntityCategory.CONFIG,
translation_key="kettle_work_mode",
),
),
DeviceCategory.CL: (
SelectEntityDescription(
key=DPCode.CONTROL_BACK_MODE,
@@ -478,6 +478,15 @@
"1": "Continuous working mode"
}
},
"kettle_work_mode": {
"name": "Work mode",
"state": {
"boiling_quick": "Quick boil",
"setting_quick": "Quick heat",
"temp_boiling": "Boil and keep warm",
"temp_setting": "Heat and keep warm"
}
},
"led_type": {
"name": "Light source type",
"state": {
@@ -515,6 +524,16 @@
"smart": "Smart"
}
},
"quick_heat_temperature": {
"name": "Quick heat temperature",
"state": {
"80": "80 °C",
"85": "85 °C",
"90": "90 °C",
"95": "95 °C",
"100": "100 °C"
}
},
"record_mode": {
"name": "Record mode",
"state": {
+1 -1
View File
@@ -39,7 +39,7 @@ habluetooth==6.7.9
hass-nabucasa==2.2.0
hassil==3.5.0
home-assistant-bluetooth==2.0.0
home-assistant-frontend==20260429.4
home-assistant-frontend==20260527.0
home-assistant-intents==2026.5.5
httpx==0.28.1
ifaddr==0.2.0
+1 -1
View File
@@ -5,7 +5,7 @@ To update, run python3 -m script.hassfest
from typing import Final
FRONTEND_VERSION: Final[str] = "20260429.4"
FRONTEND_VERSION: Final[str] = "20260527.0"
MDI_ICONS: Final[set[str]] = {
"ab-testing",
+1 -1
View File
@@ -1266,7 +1266,7 @@ hole==0.9.0
holidays==0.97
# homeassistant.components.frontend
home-assistant-frontend==20260429.4
home-assistant-frontend==20260527.0
# homeassistant.components.conversation
home-assistant-intents==2026.5.5
-30
View File
@@ -2088,36 +2088,6 @@ async def test_receive_backup_path_traversal(
assert resp.status == 400
@pytest.mark.parametrize(
"name",
[
"/absolute/path",
"../parent",
"with/slash",
],
)
async def test_receive_backup_rejects_unsafe_inner_name(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
name: str,
) -> None:
"""Test receive backup rejects an inner name that would escape the backup dir."""
await setup_backup_integration(hass)
client = await hass_client()
backup = replace(TEST_BACKUP_ABC123, name=name)
with patch(
"homeassistant.components.backup.manager.read_backup",
return_value=backup,
):
resp = await client.post(
"/api/backup/upload?agent_id=backup.local",
data={"file": StringIO("test")},
)
assert resp.status == 400
async def test_receive_backup_busy_manager(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
-32
View File
@@ -14,7 +14,6 @@ import pytest
import securetar
from homeassistant.components.backup import DOMAIN, AddonInfo, AgentBackup, Folder
from homeassistant.components.backup.models import InvalidBackupFilename
from homeassistant.components.backup.util import (
DecryptedBackupStreamer,
EncryptedBackupStreamer,
@@ -159,37 +158,6 @@ def test_read_backup(backup_json_content: bytes, expected_backup: AgentBackup) -
assert backup == expected_backup
@pytest.mark.parametrize(
"name",
[
"/absolute/path",
"../parent",
"with/slash",
"with\\backslash",
"C:\\drive\\path",
"",
".",
"..",
],
)
def test_read_backup_rejects_unsafe_name(name: str) -> None:
"""Test that read_backup rejects names that could escape the backup directory."""
backup_json_content = (
b'{"compressed":true,"date":"2024-12-02T07:23:58.261875-05:00","homeassistant":'
b'{"exclude_database":true,"version":"2024.12.0.dev0"},"name":"'
+ name.encode().replace(b"\\", b"\\\\")
+ b'","protected":true,"slug":"455645fe","type":"partial","version":2}'
)
mock_path = Mock()
mock_path.stat.return_value.st_size = 1234
with patch("homeassistant.components.backup.util.tarfile.open") as mock_open_tar:
tar_ctx = mock_open_tar.return_value.__enter__.return_value
tar_ctx.extractfile.return_value.read.return_value = backup_json_content
with pytest.raises(InvalidBackupFilename):
read_backup(mock_path)
@pytest.mark.parametrize(
("backup", "password", "validation_result", "expected_messages"),
[
@@ -5637,6 +5637,128 @@
'state': 'low',
})
# ---
# name: test_platform_setup_and_discovery[select.smart_kettle_quick_heat_temperature-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'85',
'90',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.smart_kettle_quick_heat_temperature',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Quick heat temperature',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Quick heat temperature',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'quick_heat_temperature',
'unique_id': 'tuya.s5ah3novtabe4tfdhbtemp_setting_quick_c',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[select.smart_kettle_quick_heat_temperature-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Smart Kettle Quick heat temperature',
'options': list([
'85',
'90',
]),
}),
'context': <ANY>,
'entity_id': 'select.smart_kettle_quick_heat_temperature',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '85',
})
# ---
# name: test_platform_setup_and_discovery[select.smart_kettle_work_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
'options': list([
'setting_quick',
'boiling_quick',
'temp_setting',
'temp_boiling',
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.smart_kettle_work_mode',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Work mode',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Work mode',
'platform': 'tuya',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'kettle_work_mode',
'unique_id': 'tuya.s5ah3novtabe4tfdhbwork_type',
'unit_of_measurement': None,
})
# ---
# name: test_platform_setup_and_discovery[select.smart_kettle_work_mode-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Smart Kettle Work mode',
'options': list([
'setting_quick',
'boiling_quick',
'temp_setting',
'temp_boiling',
]),
}),
'context': <ANY>,
'entity_id': 'select.smart_kettle_work_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'setting_quick',
})
# ---
# name: test_platform_setup_and_discovery[select.smart_odor_eliminator_pro_odor_elimination_mode-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
+55 -22
View File
@@ -1,5 +1,6 @@
"""Test methods in backup_restore."""
from io import BytesIO
import json
from pathlib import Path
import tarfile
@@ -386,8 +387,8 @@ def test_restore_backup(
}
def test_restore_backup_filter_files(tmp_path: Path) -> None:
"""Test filtering dangerous files when restoring a backup."""
def test_restore_backup_rejects_unsafe_files(tmp_path: Path) -> None:
"""Test that a backup with unsafe paths is rejected."""
backup_file_path = tmp_path / "backups" / "test.tar"
backup_file_path.parent.mkdir()
get_fixture_path(
@@ -410,7 +411,55 @@ def test_restore_backup_filter_files(tmp_path: Path) -> None:
"data/home-assistant_v2.db-wal",
}
real_extractone = tarfile.TarFile._extract_one
with (
mock.patch(
"homeassistant.backup_restore.restore_backup_file_content",
return_value=backup_restore.RestoreBackupFileContent(
backup_file_path=backup_file_path,
password=None,
remove_after_restore=False,
restore_database=True,
restore_homeassistant=True,
),
),
pytest.raises(tarfile.FilterError),
):
backup_restore.restore_backup(tmp_path.as_posix())
result = restore_result_file_content(tmp_path)
assert result is not None
assert result["success"] is False
assert result["error_type"] in {"AbsolutePathError", "OutsideDestinationError"}
def test_restore_backup_rejects_absolute_symlink(tmp_path: Path) -> None:
"""Test rejection of a symlink whose linkname escapes the destination.
A SYMTYPE entry followed by a regular file whose name traverses the
symlink would otherwise land attacker-controlled bytes outside the
extraction directory. The tar filter resolves the path after the
symlink and rejects the entry.
"""
backup_file_path = tmp_path / "backups" / "test.tar"
backup_file_path.parent.mkdir()
with tarfile.open(backup_file_path, "w") as tar:
backup_json = json.dumps(
{"homeassistant": {"version": "0.0.0"}, "compressed": False}
).encode()
info = tarfile.TarInfo(name="./backup.json")
info.size = len(backup_json)
tar.addfile(info, BytesIO(backup_json))
symlink = tarfile.TarInfo(name="pwn")
symlink.type = tarfile.SYMTYPE
symlink.linkname = "/tmp" # noqa: S108
tar.addfile(symlink)
payload = b"pwned"
evil = tarfile.TarInfo(name="pwn/ha_escape_target")
evil.size = len(payload)
tar.addfile(evil, BytesIO(payload))
with (
mock.patch(
@@ -423,27 +472,11 @@ def test_restore_backup_filter_files(tmp_path: Path) -> None:
restore_homeassistant=True,
),
),
mock.patch(
"tarfile.TarFile._extract_one", autospec=True, wraps=real_extractone
) as extractone_mock,
pytest.raises(tarfile.FilterError),
):
assert backup_restore.restore_backup(tmp_path.as_posix()) is True
backup_restore.restore_backup(tmp_path.as_posix())
# Check the unsafe files are not extracted, and that the safe files are extracted
extracted_files = {call.args[1].name for call in extractone_mock.mock_calls}
assert extracted_files == {
"./backup.json", # From the outer tar
"homeassistant.tar.gz", # From the outer tar
".",
"data",
"data/home-assistant_v2.db",
"data/home-assistant_v2.db-wal",
}
assert restore_result_file_content(tmp_path) == {
"error": None,
"error_type": None,
"success": True,
}
assert not Path("/tmp/ha_escape_target").exists() # noqa: S108
@pytest.mark.parametrize(("remove_after_restore"), [True, False])