|
| 1 | +"""Home Assistant module to handle restoring backups.""" |
| 2 | + |
| 3 | +from dataclasses import dataclass |
| 4 | +import json |
| 5 | +import logging |
| 6 | +from pathlib import Path |
| 7 | +import shutil |
| 8 | +import sys |
| 9 | +from tempfile import TemporaryDirectory |
| 10 | + |
| 11 | +from awesomeversion import AwesomeVersion |
| 12 | +import securetar |
| 13 | + |
| 14 | +from .const import __version__ as HA_VERSION |
| 15 | + |
| 16 | +RESTORE_BACKUP_FILE = ".HA_RESTORE" |
| 17 | +KEEP_PATHS = ("backups",) |
| 18 | + |
| 19 | +_LOGGER = logging.getLogger(__name__) |
| 20 | + |
| 21 | + |
| 22 | +@dataclass |
| 23 | +class RestoreBackupFileContent: |
| 24 | + """Definition for restore backup file content.""" |
| 25 | + |
| 26 | + backup_file_path: Path |
| 27 | + |
| 28 | + |
| 29 | +def restore_backup_file_content(config_dir: Path) -> RestoreBackupFileContent | None: |
| 30 | + """Return the contents of the restore backup file.""" |
| 31 | + instruction_path = config_dir.joinpath(RESTORE_BACKUP_FILE) |
| 32 | + try: |
| 33 | + instruction_content = instruction_path.read_text(encoding="utf-8") |
| 34 | + return RestoreBackupFileContent( |
| 35 | + backup_file_path=Path(instruction_content.split(";")[0]) |
| 36 | + ) |
| 37 | + except FileNotFoundError: |
| 38 | + return None |
| 39 | + |
| 40 | + |
| 41 | +def _clear_configuration_directory(config_dir: Path) -> None: |
| 42 | + """Delete all files and directories in the config directory except for the backups directory.""" |
| 43 | + keep_paths = [config_dir.joinpath(path) for path in KEEP_PATHS] |
| 44 | + config_contents = sorted( |
| 45 | + [entry for entry in config_dir.iterdir() if entry not in keep_paths] |
| 46 | + ) |
| 47 | + |
| 48 | + for entry in config_contents: |
| 49 | + entrypath = config_dir.joinpath(entry) |
| 50 | + |
| 51 | + if entrypath.is_file(): |
| 52 | + entrypath.unlink() |
| 53 | + elif entrypath.is_dir(): |
| 54 | + shutil.rmtree(entrypath) |
| 55 | + |
| 56 | + |
| 57 | +def _extract_backup(config_dir: Path, backup_file_path: Path) -> None: |
| 58 | + """Extract the backup file to the config directory.""" |
| 59 | + with ( |
| 60 | + TemporaryDirectory() as tempdir, |
| 61 | + securetar.SecureTarFile( |
| 62 | + backup_file_path, |
| 63 | + gzip=False, |
| 64 | + mode="r", |
| 65 | + ) as ostf, |
| 66 | + ): |
| 67 | + ostf.extractall( |
| 68 | + path=Path(tempdir, "extracted"), |
| 69 | + members=securetar.secure_path(ostf), |
| 70 | + filter="fully_trusted", |
| 71 | + ) |
| 72 | + backup_meta_file = Path(tempdir, "extracted", "backup.json") |
| 73 | + backup_meta = json.loads(backup_meta_file.read_text(encoding="utf8")) |
| 74 | + |
| 75 | + if ( |
| 76 | + backup_meta_version := AwesomeVersion( |
| 77 | + backup_meta["homeassistant"]["version"] |
| 78 | + ) |
| 79 | + ) > HA_VERSION: |
| 80 | + raise ValueError( |
| 81 | + f"You need at least Home Assistant version {backup_meta_version} to restore this backup" |
| 82 | + ) |
| 83 | + |
| 84 | + with securetar.SecureTarFile( |
| 85 | + Path( |
| 86 | + tempdir, |
| 87 | + "extracted", |
| 88 | + f"homeassistant.tar{'.gz' if backup_meta["compressed"] else ''}", |
| 89 | + ), |
| 90 | + gzip=backup_meta["compressed"], |
| 91 | + mode="r", |
| 92 | + ) as istf: |
| 93 | + for member in istf.getmembers(): |
| 94 | + if member.name == "data": |
| 95 | + continue |
| 96 | + member.name = member.name.replace("data/", "") |
| 97 | + _clear_configuration_directory(config_dir) |
| 98 | + istf.extractall( |
| 99 | + path=config_dir, |
| 100 | + members=[ |
| 101 | + member |
| 102 | + for member in securetar.secure_path(istf) |
| 103 | + if member.name != "data" |
| 104 | + ], |
| 105 | + filter="fully_trusted", |
| 106 | + ) |
| 107 | + |
| 108 | + |
| 109 | +def restore_backup(config_dir_path: str) -> bool: |
| 110 | + """Restore the backup file if any. |
| 111 | +
|
| 112 | + Returns True if a restore backup file was found and restored, False otherwise. |
| 113 | + """ |
| 114 | + config_dir = Path(config_dir_path) |
| 115 | + if not (restore_content := restore_backup_file_content(config_dir)): |
| 116 | + return False |
| 117 | + |
| 118 | + logging.basicConfig(stream=sys.stdout, level=logging.INFO) |
| 119 | + backup_file_path = restore_content.backup_file_path |
| 120 | + _LOGGER.info("Restoring %s", backup_file_path) |
| 121 | + try: |
| 122 | + _extract_backup(config_dir, backup_file_path) |
| 123 | + except FileNotFoundError as err: |
| 124 | + raise ValueError(f"Backup file {backup_file_path} does not exist") from err |
| 125 | + _LOGGER.info("Restore complete, restarting") |
| 126 | + return True |
0 commit comments