Check password early on backup restore

Introduce a validate password method which only peaks into the archive
to validate the password before starting the actual restore process.
This makes sure that a wrong password returns an error even when
restoring the backup in background.

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():

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(

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",

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)