Add an admin only device wipe API (#4934)

* Add an admin only device wipe API

* Fix pylint issue

supervisor/api/__init__.py

@@ -182,6 +182,7 @@ class RestAPI(CoreSysAttributes):
                 web.post("/os/config/sync", api_os.config_sync),
                 web.post("/os/datadisk/move", api_os.migrate_data),
                 web.get("/os/datadisk/list", api_os.list_data),
+                web.post("/os/datadisk/wipe", api_os.wipe_data),
             ]
         )
 

supervisor/api/middleware/security.py

@@ -118,7 +118,7 @@ ADDONS_ROLE_ACCESS: dict[str, re.Pattern] = {
         r"|/multicast/.+"
         r"|/network/.+"
         r"|/observer/.+"
-        r"|/os/.+"
+        r"|/os/(?!datadisk/wipe).+"
         r"|/refresh_updates"
         r"|/resolution/.+"
         r"|/security/.+"

supervisor/api/os.py

@@ -96,6 +96,11 @@ class APIOS(CoreSysAttributes):
 
         await asyncio.shield(self.sys_os.datadisk.migrate_disk(body[ATTR_DEVICE]))
 
+    @api_process
+    def wipe_data(self, request: web.Request) -> Awaitable[None]:
+        """Trigger data disk wipe on Host."""
+        return asyncio.shield(self.sys_os.datadisk.wipe_disk())
+
     @api_process
     async def list_data(self, request: web.Request) -> dict[str, Any]:
         """Return possible data targets."""

supervisor/dbus/agent/system.py

@@ -12,6 +12,6 @@ class System(DBusInterface):
     object_path: str = DBUS_OBJECT_HAOS_SYSTEM
 
     @dbus_connected
-    async def schedule_wipe_device(self) -> None:
+    async def schedule_wipe_device(self) -> bool:
         """Schedule a factory reset on next system boot."""
-        await self.dbus.System.call_schedule_wipe_device()
+        return await self.dbus.System.call_schedule_wipe_device()

supervisor/os/data_disk.py

@@ -272,6 +272,35 @@ class DataDisk(CoreSysAttributes):
                 _LOGGER.warning,
             ) from err
 
+    @Job(
+        name="data_disk_wipe",
+        conditions=[JobCondition.HAOS, JobCondition.OS_AGENT, JobCondition.HEALTHY],
+        limit=JobExecutionLimit.ONCE,
+        on_condition=HassOSJobError,
+    )
+    async def wipe_disk(self) -> None:
+        """Wipe the current data disk."""
+        _LOGGER.info("Scheduling wipe of data disk on next reboot")
+        try:
+            if not await self.sys_dbus.agent.system.schedule_wipe_device():
+                raise HassOSDataDiskError(
+                    "Can't schedule wipe of data disk, check host logs for details",
+                    _LOGGER.error,
+                )
+        except DBusError as err:
+            raise HassOSDataDiskError(
+                f"Can't schedule wipe of data disk: {err!s}", _LOGGER.error
+            ) from err
+
+        _LOGGER.info("Rebooting the host to finish the wipe")
+        try:
+            await self.sys_host.control.reboot()
+        except (HostError, DBusError) as err:
+            raise HassOSError(
+                f"Can't restart device to finish data disk wipe: {err!s}",
+                _LOGGER.warning,
+            ) from err
+
     async def _format_device_with_single_partition(
         self, new_disk: Disk
     ) -> UDisks2Block:

tests/api/middleware/test_security.py

@@ -179,6 +179,7 @@ async def test_bad_requests(
         ("post", "/addons/abc123/options", {"admin", "manager"}),
         ("post", "/addons/abc123/restart", {"admin", "manager"}),
         ("post", "/addons/abc123/security", {"admin"}),
+        ("post", "/os/datadisk/wipe", {"admin"}),
     ],
 )
 async def test_token_validation(

tests/api/test_os.py

@@ -17,6 +17,7 @@ from tests.dbus_service_mocks.agent_boards import Boards as BoardsService
 from tests.dbus_service_mocks.agent_boards_green import Green as GreenService
 from tests.dbus_service_mocks.agent_boards_yellow import Yellow as YellowService
 from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
+from tests.dbus_service_mocks.agent_system import System as SystemService
 from tests.dbus_service_mocks.base import DBusServiceMock
 
 
@@ -113,6 +114,23 @@ async def test_api_os_datadisk_migrate(
         reboot.assert_called_once()
 
 
+async def test_api_os_datadisk_wipe(
+    api_client: TestClient,
+    os_agent_services: dict[str, DBusServiceMock],
+    os_available,
+):
+    """Test datadisk wipe."""
+    system_service: SystemService = os_agent_services["agent_system"]
+    system_service.ScheduleWipeDevice.calls.clear()
+
+    with patch.object(SystemControl, "reboot") as reboot:
+        resp = await api_client.post("/os/datadisk/wipe")
+        assert resp.status == 200
+
+        assert system_service.ScheduleWipeDevice.calls == [()]
+        reboot.assert_called_once()
+
+
 async def test_api_board_yellow_info(api_client: TestClient, coresys: CoreSys):
     """Test yellow board info."""
     resp = await api_client.get("/os/boards/yellow")

tests/dbus/agent/test_system.py

@@ -30,5 +30,5 @@ async def test_dbus_osagent_system_wipe(
 
     await os_agent.connect(dbus_session_bus)
 
-    assert await os_agent.system.schedule_wipe_device() is None
+    assert await os_agent.system.schedule_wipe_device() is True
     assert system_service.ScheduleWipeDevice.calls == [()]

tests/dbus_service_mocks/agent_system.py

@@ -1,5 +1,7 @@
 """Mock of OS Agent System dbus service."""
 
+from dbus_fast import DBusError
+
 from .base import DBusServiceMock, dbus_method
 
 BUS_NAME = "io.hass.os"
@@ -18,11 +20,14 @@ class System(DBusServiceMock):
 
     object_path = "/io/hass/os/System"
     interface = "io.hass.os.System"
+    response_schedule_wipe_device: bool | DBusError = True
 
     @dbus_method()
     def ScheduleWipeDevice(self) -> "b":
         """Schedule wipe device."""
-        return True
+        if isinstance(self.response_schedule_wipe_device, DBusError):
+            raise self.response_schedule_wipe_device  # pylint: disable=raising-bad-type
+        return self.response_schedule_wipe_device
 
     @dbus_method()
     def WipeDevice(self) -> "b":

tests/dbus_service_mocks/logind.py

@@ -1,5 +1,7 @@
 """Mock of logind dbus service."""
 
+from dbus_fast import DBusError
+
 from .base import DBusServiceMock, dbus_method
 
 BUS_NAME = "org.freedesktop.login1"
@@ -18,10 +20,13 @@ class Logind(DBusServiceMock):
 
     object_path = "/org/freedesktop/login1"
     interface = "org.freedesktop.login1.Manager"
+    side_effect_reboot: DBusError | None = None
 
     @dbus_method()
     def Reboot(self, interactive: "b") -> None:
         """Reboot."""
+        if self.side_effect_reboot:
+            raise self.side_effect_reboot  # pylint: disable=raising-bad-type
 
     @dbus_method()
     def PowerOff(self, interactive: "b") -> None:

tests/os/test_data_disk.py

@@ -3,16 +3,17 @@ from dataclasses import replace
 from pathlib import PosixPath
 from unittest.mock import patch
 
-from dbus_fast import Variant
+from dbus_fast import DBusError, ErrorType, Variant
 import pytest
 
 from supervisor.core import Core
 from supervisor.coresys import CoreSys
-from supervisor.exceptions import HassOSDataDiskError
+from supervisor.exceptions import HassOSDataDiskError, HassOSError
 from supervisor.os.data_disk import Disk
 
 from tests.common import mock_dbus_services
 from tests.dbus_service_mocks.agent_datadisk import DataDisk as DataDiskService
+from tests.dbus_service_mocks.agent_system import System as SystemService
 from tests.dbus_service_mocks.base import DBusServiceMock
 from tests.dbus_service_mocks.logind import Logind as LogindService
 from tests.dbus_service_mocks.udisks2_block import Block as BlockService
@@ -270,3 +271,45 @@ async def test_datadisk_migrate_between_external_renames(
     assert sdb1_partition_service.SetName.calls == [
         ("hassos-data-external-old", {"auth.no_user_interaction": Variant("b", True)})
     ]
+
+
+async def test_datadisk_wipe_errors(
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
+    os_available,
+):
+    """Test errors during datadisk wipe."""
+    system_service: SystemService = all_dbus_services["agent_system"]
+    system_service.ScheduleWipeDevice.calls.clear()
+    logind_service: LogindService = all_dbus_services["logind"]
+    logind_service.Reboot.calls.clear()
+
+    system_service.response_schedule_wipe_device = False
+    with pytest.raises(
+        HassOSDataDiskError, match="Can't schedule wipe of data disk, check host logs"
+    ):
+        await coresys.os.datadisk.wipe_disk()
+
+    assert system_service.ScheduleWipeDevice.calls == [()]
+    assert not logind_service.Reboot.calls
+
+    system_service.ScheduleWipeDevice.calls.clear()
+    system_service.response_schedule_wipe_device = DBusError(ErrorType.FAILED, "fail")
+    with pytest.raises(
+        HassOSDataDiskError, match="Can't schedule wipe of data disk: fail"
+    ):
+        await coresys.os.datadisk.wipe_disk()
+
+    assert system_service.ScheduleWipeDevice.calls == [()]
+    assert not logind_service.Reboot.calls
+
+    system_service.ScheduleWipeDevice.calls.clear()
+    system_service.response_schedule_wipe_device = True
+    logind_service.side_effect_reboot = DBusError(ErrorType.FAILED, "fail")
+    with patch.object(Core, "shutdown"), pytest.raises(
+        HassOSError, match="Can't restart device"
+    ):
+        await coresys.os.datadisk.wipe_disk()
+
+    assert system_service.ScheduleWipeDevice.calls == [()]
+    assert logind_service.Reboot.calls == [(False,)]