backup_restore.py

Go to the documentation of this file.

"""Home Assistant module to handle restoring backups."""
 
from dataclasses import dataclass
import json
import logging
from pathlib import Path
import shutil
import sys
from tempfile import TemporaryDirectory
 
from awesomeversion import AwesomeVersion
import securetar
 
from .const import __version__ as HA_VERSION
 
RESTORE_BACKUP_FILE = ".HA_RESTORE"
KEEP_PATHS = ("backups",)
 
_LOGGER = logging.getLogger(__name__)
 
 
@dataclass
class RestoreBackupFileContent:
    """Definition for restore backup file content."""
 
    backup_file_path: Path
 
 
def restore_backup_file_content(config_dir: Path) -> RestoreBackupFileContent | None:
    """Return the contents of the restore backup file."""
    instruction_path = config_dir.joinpath(RESTORE_BACKUP_FILE)
    try:
        instruction_content = json.loads(instruction_path.read_text(encoding="utf-8"))
        return RestoreBackupFileContent(
            backup_file_path=Path(instruction_content["path"])
        )
    except (FileNotFoundError, json.JSONDecodeError):
        return None
 
 
def _clear_configuration_directory(config_dir: Path) -> None:
    """Delete all files and directories in the config directory except for the backups directory."""
    keep_paths = [config_dir.joinpath(path) for path in KEEP_PATHS]
    config_contents = sorted(
        [entry for entry in config_dir.iterdir() if entry not in keep_paths]
    )
 
    for entry in config_contents:
        entrypath = config_dir.joinpath(entry)
 
        if entrypath.is_file():
            entrypath.unlink()
        elif entrypath.is_dir():
            shutil.rmtree(entrypath)
 
 
def _extract_backup(config_dir: Path, backup_file_path: Path) -> None:
    """Extract the backup file to the config directory."""
    with (
        TemporaryDirectory() as tempdir,
        securetar.SecureTarFile(
            backup_file_path,
            gzip=False,
            mode="r",
        ) as ostf,
    ):
        ostf.extractall(
            path=Path(tempdir, "extracted"),
            members=securetar.secure_path(ostf),
            filter="fully_trusted",
        )
        backup_meta_file = Path(tempdir, "extracted", "backup.json")
        backup_meta = json.loads(backup_meta_file.read_text(encoding="utf8"))
 
        if (
            backup_meta_version := AwesomeVersion(
                backup_meta["homeassistant"]["version"]
            )
        ) > HA_VERSION:
            raise ValueError(
                f"You need at least Home Assistant version {backup_meta_version} to restore this backup"
            )
 
        with securetar.SecureTarFile(
            Path(
                tempdir,
                "extracted",
                f"homeassistant.tar{'.gz' if backup_meta["compressed"] else ''}",
            ),
            gzip=backup_meta["compressed"],
            mode="r",
        ) as istf:
            for member in istf.getmembers():
                if member.name == "data":
                    continue
                member.name = member.name.replace("data/", "")
            _clear_configuration_directory(config_dir)
            istf.extractall(
                path=config_dir,
                members=[
                    member
                    for member in securetar.secure_path(istf)
                    if member.name != "data"
                ],
                filter="fully_trusted",
            )
 
 
def restore_backup(config_dir_path: str) -> bool:
    """Restore the backup file if any.
 
    Returns True if a restore backup file was found and restored, False otherwise.
    """
    config_dir = Path(config_dir_path)
    if not (restore_content := restore_backup_file_content(config_dir)):
        return False
 
    logging.basicConfig(stream=sys.stdout, level=logging.INFO)
    backup_file_path = restore_content.backup_file_path
    _LOGGER.info("Restoring %s", backup_file_path)
    try:
        _extract_backup(config_dir, backup_file_path)
    except FileNotFoundError as err:
        raise ValueError(f"Backup file {backup_file_path} does not exist") from err
    _LOGGER.info("Restore complete, restarting")
    return True