Backup protected status can vary per location

supervisor/api/backups.py

@@ -34,6 +34,7 @@
     ATTR_LOCATION,
     ATTR_NAME,
     ATTR_PASSWORD,
+    ATTR_PATH,
     ATTR_PROTECTED,
     ATTR_REPOSITORIES,
     ATTR_SIZE,
@@ -55,6 +56,7 @@
     ATTR_ADDITIONAL_LOCATIONS,
     ATTR_BACKGROUND,
     ATTR_LOCATIONS,
+    ATTR_PROTECTED_LOCATIONS,
     ATTR_SIZE_BYTES,
     CONTENT_TYPE_TAR,
 )
@@ -165,6 +167,11 @@ def _list_backups(self):
                 ATTR_LOCATION: backup.location,
                 ATTR_LOCATIONS: backup.locations,
                 ATTR_PROTECTED: backup.protected,
+                ATTR_PROTECTED_LOCATIONS: [
+                    loc
+                    for loc in backup.locations
+                    if backup.all_locations[loc][ATTR_PROTECTED]
+                ],
                 ATTR_COMPRESSED: backup.compressed,
                 ATTR_CONTENT: {
                     ATTR_HOMEASSISTANT: backup.homeassistant_version is not None,
@@ -236,6 +243,11 @@ async def backup_info(self, request):
             ATTR_SIZE_BYTES: backup.size_bytes,
             ATTR_COMPRESSED: backup.compressed,
             ATTR_PROTECTED: backup.protected,
+            ATTR_PROTECTED_LOCATIONS: [
+                loc
+                for loc in backup.locations
+                if backup.all_locations[loc][ATTR_PROTECTED]
+            ],
             ATTR_SUPERVISOR_VERSION: backup.supervisor_version,
             ATTR_HOMEASSISTANT: backup.homeassistant_version,
             ATTR_LOCATION: backup.location,
@@ -460,7 +472,7 @@ async def download(self, request: web.Request):
             raise APIError(f"Backup {backup.slug} is not in location {location}")
 
         _LOGGER.info("Downloading backup %s", backup.slug)
-        filename = backup.all_locations[location]
+        filename = backup.all_locations[location][ATTR_PATH]
         response = web.FileResponse(filename)
         response.content_type = CONTENT_TYPE_TAR
 

supervisor/api/const.py

@@ -53,6 +53,7 @@
 ATTR_MOUNTS = "mounts"
 ATTR_MOUNT_POINTS = "mount_points"
 ATTR_PANEL_PATH = "panel_path"
+ATTR_PROTECTED_LOCATIONS = "protected_locations"
 ATTR_REMOVABLE = "removable"
 ATTR_REMOVE_CONFIG = "remove_config"
 ATTR_REVISION = "revision"

supervisor/backups/backup.py

@@ -39,6 +39,7 @@
     ATTR_HOMEASSISTANT,
     ATTR_NAME,
     ATTR_PASSWORD,
+    ATTR_PATH,
     ATTR_PROTECTED,
     ATTR_REGISTRIES,
     ATTR_REPOSITORIES,
@@ -91,7 +92,12 @@ def __init__(
         self._outer_secure_tarfile: SecureTarFile | None = None
         self._key: bytes | None = None
         self._aes: Cipher | None = None
-        self._locations: dict[str | None, Path] = {location: tar_file}
+        self._locations: dict[str | None, dict[str, Path | bool]] = {
+            location: {
+                ATTR_PATH: tar_file,
+                ATTR_PROTECTED: data.get(ATTR_PROTECTED, False) if data else False,
+            }
+        }
 
     @property
     def version(self) -> int:
@@ -121,7 +127,7 @@ def date(self) -> str:
     @property
     def protected(self) -> bool:
         """Return backup date."""
-        return self._data[ATTR_PROTECTED]
+        return self._locations[self.location][ATTR_PROTECTED]
 
     @property
     def compressed(self) -> bool:
@@ -198,7 +204,7 @@ def location(self) -> str | None:
         return self.locations[0]
 
     @property
-    def all_locations(self) -> dict[str | None, Path]:
+    def all_locations(self) -> dict[str | None, dict[str, Path | bool]]:
         """Return all locations this backup was found in."""
         return self._locations
 
@@ -236,7 +242,7 @@ def is_new(self) -> bool:
     @property
     def tarfile(self) -> Path:
         """Return path to backup tarfile."""
-        return self._locations[self.location]
+        return self._locations[self.location][ATTR_PATH]
 
     @property
     def is_current(self) -> bool:
@@ -252,7 +258,27 @@ def data(self) -> dict[str, Any]:
 
     def __eq__(self, other: Any) -> bool:
         """Return true if backups have same metadata."""
-        return isinstance(other, Backup) and self._data == other._data
+        if not isinstance(other, Backup):
+            return False
+
+        # Compare all fields except ones about protection. Current encryption status does not affect equality
+        keys = self._data.keys() | other._data.keys()
+        for k in keys - {ATTR_PROTECTED, ATTR_CRYPTO, ATTR_DOCKER}:
+            if (
+                k not in self._data
+                or k not in other._data
+                or self._data[k] != other._data[k]
+            ):
+                _LOGGER.debug(
+                    "Backup %s and %s not equal because %s field has different value: %s and %s",
+                    self.slug,
+                    other.slug,
+                    k,
+                    self._data.get(k),
+                    other._data.get(k),
+                )
+                return False
+        return True
 
     def consolidate(self, backup: Self) -> None:
         """Consolidate two backups with same slug in different locations."""
@@ -264,6 +290,20 @@ def consolidate(self, backup: Self) -> None:
             raise BackupInvalidError(
                 f"Backup in {backup.location} and {self.location} both have slug {self.slug} but are not the same!"
             )
+
+        # In case of conflict we always ignore the ones from the first one. But log them to let the user know
+
+        if conflict := {
+            loc: val[ATTR_PATH]
+            for loc, val in self.all_locations.items()
+            if loc in backup.all_locations and backup.all_locations[loc] != val
+        }:
+            _LOGGER.warning(
+                "Backup %s exists in two files in locations %s. Ignoring %s",
+                self.slug,
+                ", ".join(str(loc) for loc in conflict),
+                ", ".join([path.as_posix() for path in conflict.values()]),
+            )
         self._locations.update(backup.all_locations)
 
     def new(
@@ -292,6 +332,7 @@ def new(
             self._init_password(password)
             self._data[ATTR_PROTECTED] = True
             self._data[ATTR_CRYPTO] = CRYPTO_AES128
+            self._locations[self.location][ATTR_PROTECTED] = True
 
         if not compressed:
             self._data[ATTR_COMPRESSED] = False
@@ -418,6 +459,9 @@ def _load_file():
             )
             return False
 
+        if self._data[ATTR_PROTECTED]:
+            self._locations[self.location][ATTR_PROTECTED] = True
+
         return True
 
     @asynccontextmanager
@@ -452,7 +496,9 @@ async def open(self, location: str | None | type[DEFAULT]) -> AsyncGenerator[Non
             )
 
         backup_tarfile = (
-            self.tarfile if location == DEFAULT else self.all_locations[location]
+            self.tarfile
+            if location == DEFAULT
+            else self.all_locations[location][ATTR_PATH]
         )
         if not backup_tarfile.is_file():
             raise BackupError(

supervisor/backups/manager.py

@@ -12,6 +12,8 @@
 from ..addons.addon import Addon
 from ..const import (
     ATTR_DAYS_UNTIL_STALE,
+    ATTR_PATH,
+    ATTR_PROTECTED,
     FILE_HASSIO_BACKUPS,
     FOLDER_HOMEASSISTANT,
     CoreState,
@@ -291,7 +293,7 @@ def remove(
         )
         for location in targets:
             try:
-                backup.all_locations[location].unlink()
+                backup.all_locations[location][ATTR_PATH].unlink()
                 del backup.all_locations[location]
             except OSError as err:
                 if err.errno == errno.EBADMSG and location in {
@@ -345,13 +347,20 @@ def copy_to_additional_locations() -> dict[str | None, Path]:
             return all_locations
 
         try:
-            backup.all_locations.update(
-                await self.sys_run_in_executor(copy_to_additional_locations)
+            all_new_locations = await self.sys_run_in_executor(
+                copy_to_additional_locations
             )
         except BackupDataDiskBadMessageError:
             self.sys_resolution.unhealthy = UnhealthyReason.OSERROR_BAD_MESSAGE
             raise
 
+        backup.all_locations.update(
+            {
+                loc: {ATTR_PATH: path, ATTR_PROTECTED: backup.protected}
+                for loc, path in all_new_locations.items()
+            }
+        )
+
     @Job(name="backup_manager_import_backup")
     async def import_backup(
         self,
@@ -676,6 +685,30 @@ async def _do_restore(
                         _job_override__cleanup=False
                     )
 
+    async def _validate_location_password(
+        self,
+        backup: Backup,
+        password: str | None = None,
+        location: str | None | type[DEFAULT] = DEFAULT,
+    ) -> None:
+        """Validate location and password for backup, raise if invalid."""
+        if location != DEFAULT and location not in backup.all_locations:
+            raise BackupInvalidError(
+                f"Backup {backup.slug} does not exist in {location}", _LOGGER.error
+            )
+
+        if (
+            location == DEFAULT
+            and backup.protected
+            or location != DEFAULT
+            and backup.all_locations[location][ATTR_PROTECTED]
+        ):
+            backup.set_password(password)
+            if not await backup.validate_password():
+                raise BackupInvalidError(
+                    f"Invalid password for backup {backup.slug}", _LOGGER.error
+                )
+
     @Job(
         name=JOB_FULL_RESTORE,
         conditions=[
@@ -704,12 +737,7 @@ async def do_restore_full(
                 f"{backup.slug} is only a partial backup!", _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
-                )
+        await self._validate_location_password(backup, password, location)
 
         if backup.supervisor_version > self.sys_supervisor.version:
             raise BackupInvalidError(
@@ -774,12 +802,7 @@ async def do_restore_partial(
             folder_list.remove(FOLDER_HOMEASSISTANT)
             homeassistant = True
 
-        if backup.protected:
-            backup.set_password(password)
-            if not await backup.validate_password():
-                raise BackupInvalidError(
-                    f"Invalid password for backup {backup.slug}", _LOGGER.error
-                )
+        await self._validate_location_password(backup, password, location)
 
         if backup.homeassistant is None and homeassistant:
             raise BackupInvalidError(