diff --git a/supervisor/backups/backup.py b/supervisor/backups/backup.py index fa330bb7926..32ea5270b59 100644 --- a/supervisor/backups/backup.py +++ b/supervisor/backups/backup.py @@ -296,12 +296,13 @@ def new( if not compressed: self._data[ATTR_COMPRESSED] = False - def set_password(self, password: str) -> bool: + def set_password(self, password: str | None) -> None: """Set the password for an existing backup.""" - if not password: - return False - self._init_password(password) - return True + if password: + self._init_password(password) + else: + self._key = None + self._aes = None def _init_password(self, password: str) -> None: """Set password + init aes cipher.""" @@ -334,6 +335,48 @@ def _decrypt_data(self, data: str) -> str: data = padder.update(decrypt.update(b64decode(data))) + padder.finalize() return data.decode() + async def validate_password(self) -> bool: + """Validate backup password. + + Returns false only when the password is known to be wrong. + """ + + def _validate_file() -> bool: + ending = f".tar{'.gz' if self.compressed else ''}" + + with tarfile.open(self.tarfile, "r:") as backup: + test_tar_name = next( + ( + entry.name + for entry in backup.getmembers() + if entry.name.endswith(ending) + ), + None, + ) + if not test_tar_name: + _LOGGER.warning("No tar file found to validate password with") + return True + + test_tar_file = backup.extractfile(test_tar_name) + try: + with SecureTarFile( + ending, # Not used + gzip=self.compressed, + key=self._key, + mode="r", + fileobj=test_tar_file, + ): + # If we can read the tar file, the password is correct + return True + except tarfile.ReadError: + _LOGGER.debug("Invalid password") + return False + except Exception: # pylint: disable=broad-exception-caught + _LOGGER.exception("Unexpected error validating password") + return True + + return await self.sys_run_in_executor(_validate_file) + async def load(self): """Read backup.json from tar file.""" if not self.tarfile.is_file(): diff --git a/supervisor/backups/manager.py b/supervisor/backups/manager.py index e4e899780eb..7a655af53f2 100644 --- a/supervisor/backups/manager.py +++ b/supervisor/backups/manager.py @@ -687,10 +687,12 @@ async def do_restore_full( f"{backup.slug} is only a partial backup!", _LOGGER.error ) - if backup.protected and not backup.set_password(password): - raise BackupInvalidError( - f"Invalid password for backup {backup.slug}", _LOGGER.error - ) + if backup.protected: + backup.set_password(password) + if not await backup.validate_password(): + raise BackupInvalidError( + f"Invalid password for backup {backup.slug}", _LOGGER.error + ) if backup.supervisor_version > self.sys_supervisor.version: raise BackupInvalidError( @@ -755,10 +757,12 @@ async def do_restore_partial( folder_list.remove(FOLDER_HOMEASSISTANT) homeassistant = True - if backup.protected and not backup.set_password(password): - raise BackupInvalidError( - f"Invalid password for backup {backup.slug}", _LOGGER.error - ) + if backup.protected: + backup.set_password(password) + if not await backup.validate_password(): + raise BackupInvalidError( + f"Invalid password for backup {backup.slug}", _LOGGER.error + ) if backup.homeassistant is None and homeassistant: raise BackupInvalidError( diff --git a/tests/api/test_backups.py b/tests/api/test_backups.py index cb607f59487..7b8710a902d 100644 --- a/tests/api/test_backups.py +++ b/tests/api/test_backups.py @@ -478,7 +478,7 @@ async def test_restore_immediate_errors( with ( patch.object(Backup, "protected", new=PropertyMock(return_value=True)), - patch.object(Backup, "set_password", return_value=False), + patch.object(Backup, "validate_password", return_value=False), ): resp = await api_client.post( f"/backups/{mock_partial_backup.slug}/restore/partial", diff --git a/tests/backups/test_manager.py b/tests/backups/test_manager.py index 1f267228b9a..f20e123b6f2 100644 --- a/tests/backups/test_manager.py +++ b/tests/backups/test_manager.py @@ -330,7 +330,7 @@ async def test_fail_invalid_full_backup( backup_instance = full_backup_mock.return_value backup_instance.protected = True - backup_instance.set_password.return_value = False + backup_instance.validate_password.return_value = False with pytest.raises(BackupInvalidError): await manager.do_restore_full(backup_instance) @@ -359,7 +359,7 @@ async def test_fail_invalid_partial_backup( backup_instance = partial_backup_mock.return_value backup_instance.protected = True - backup_instance.set_password.return_value = False + backup_instance.validate_password.return_value = False with pytest.raises(BackupInvalidError): await manager.do_restore_partial(backup_instance)