Fix restoring unencrypted backup in corner case (#5600)

* Fix restoring unencrypted backup in corner case

If a backup has a encrypted and unencrypted location, and the encrypted
location is beeing restored first, the encryption key is still cached.
When the user restores the unencrypted backup next, it will fail because
the Supervisor tries to use encryption key still.

* Add integration test for restoring backups with and without encryption

* Rename _validate_location_password to _set_location_password

* Reload backup metadata from restore location

* Revert "Reload backup metadata from restore location"

This reverts commit 9b47a1cfe9a2682a0908e08cd143373744084fb7.

* Make pytest work/punt the ball on docker config restore issue

* Address pylint error
This commit is contained in:
Stefan Agner 2025-02-04 17:53:22 +01:00 committed by GitHub
parent 58df65541c
commit 9164d35615
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 68 additions and 4 deletions

View File

@ -708,7 +708,7 @@ class BackupManager(FileConfiguration, JobGroup):
_job_override__cleanup=False
)
async def _validate_location_password(
async def _set_location_password(
self,
backup: Backup,
password: str | None = None,
@ -727,6 +727,8 @@ class BackupManager(FileConfiguration, JobGroup):
raise BackupInvalidError(
f"Invalid password for backup {backup.slug}", _LOGGER.error
)
else:
backup.set_password(None)
@Job(
name=JOB_FULL_RESTORE,
@ -756,7 +758,7 @@ class BackupManager(FileConfiguration, JobGroup):
f"{backup.slug} is only a partial backup!", _LOGGER.error
)
await self._validate_location_password(backup, password, location)
await self._set_location_password(backup, password, location)
if backup.supervisor_version > self.sys_supervisor.version:
raise BackupInvalidError(
@ -821,7 +823,7 @@ class BackupManager(FileConfiguration, JobGroup):
folder_list.remove(FOLDER_HOMEASSISTANT)
homeassistant = True
await self._validate_location_password(backup, password, location)
await self._set_location_password(backup, password, location)
if backup.homeassistant is None and homeassistant:
raise BackupInvalidError(

View File

@ -4,7 +4,7 @@ import asyncio
from pathlib import Path, PurePath
from shutil import copy
from typing import Any
from unittest.mock import ANY, AsyncMock, PropertyMock, patch
from unittest.mock import ANY, AsyncMock, MagicMock, PropertyMock, patch
from aiohttp import MultipartWriter
from aiohttp.test_utils import TestClient
@ -955,6 +955,68 @@ async def test_restore_backup_from_location(
assert test_file.is_file()
@pytest.mark.usefixtures("tmp_supervisor_data")
async def test_restore_backup_unencrypted_after_encrypted(
api_client: TestClient,
coresys: CoreSys,
):
"""Test restoring an unencrypted backup after an encrypted backup and vis-versa."""
enc_tar = copy(get_fixture_path("test_consolidate.tar"), coresys.config.path_backup)
unc_tar = copy(
get_fixture_path("test_consolidate_unc.tar"), coresys.config.path_core_backup
)
await coresys.backups.reload()
backup = coresys.backups.get("d9c48f8b")
assert backup.all_locations == {
None: {"path": Path(enc_tar), "protected": True},
".cloud_backup": {"path": Path(unc_tar), "protected": False},
}
# pylint: disable=fixme
# TODO: There is a bug in the restore code that causes the restore to fail
# if the backup contains a Docker registry configuration and one location
# is encrypted and the other is not (just like our test fixture).
# We punt the ball on this one for this PR since this is a rare edge case.
backup.restore_dockerconfig = MagicMock()
coresys.core.state = CoreState.RUNNING
coresys.hardware.disk.get_disk_free_space = lambda x: 5000
# Restore encrypted backup
(test_file := coresys.config.path_ssl / "test.txt").touch()
resp = await api_client.post(
f"/backups/{backup.slug}/restore/partial",
json={"location": None, "password": "test", "folders": ["ssl"]},
)
assert resp.status == 200
body = await resp.json()
assert body["result"] == "ok"
assert not test_file.is_file()
# Restore unencrypted backup
test_file.touch()
resp = await api_client.post(
f"/backups/{backup.slug}/restore/partial",
json={"location": ".cloud_backup", "folders": ["ssl"]},
)
assert resp.status == 200
body = await resp.json()
assert body["result"] == "ok"
assert not test_file.is_file()
# Restore encrypted backup
test_file.touch()
resp = await api_client.post(
f"/backups/{backup.slug}/restore/partial",
json={"location": None, "password": "test", "folders": ["ssl"]},
)
assert resp.status == 200
body = await resp.json()
assert body["result"] == "ok"
assert not test_file.is_file()
@pytest.mark.parametrize(
("backup_type", "postbody"), [("partial", {"homeassistant": True}), ("full", {})]
)