Disable owning integrations for the entire firmware interaction process (home-assistant#157082)

Author: puddly

homeassistant/components/homeassistant_hardware/firmware_config_flow.py

@@ -33,13 +33,14 @@
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 from homeassistant.helpers.hassio import is_hassio
 
-from .const import OTBR_DOMAIN, Z2M_EMBER_DOCS_URL, ZHA_DOMAIN
+from .const import DOMAIN, OTBR_DOMAIN, Z2M_EMBER_DOCS_URL, ZHA_DOMAIN
 from .util import (
     ApplicationType,
     FirmwareInfo,
     OwningAddon,
     OwningIntegration,
     ResetTarget,
+    async_firmware_flashing_context,
     async_flash_silabs_firmware,
     get_otbr_addon_manager,
     guess_firmware_info,
@@ -228,83 +229,95 @@ async def _install_firmware(
         # Keep track of the firmware we're working with, for error messages
         self.installing_firmware_name = firmware_name
 
-        # Installing new firmware is only truly required if the wrong type is
-        # installed: upgrading to the latest release of the current firmware type
-        # isn't strictly necessary for functionality.
-        self._probed_firmware_info = await probe_silabs_firmware_info(
-            self._device,
-            bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
-            application_probe_methods=self.APPLICATION_PROBE_METHODS,
-        )
-
-        firmware_install_required = self._probed_firmware_info is None or (
-            self._probed_firmware_info.firmware_type != expected_installed_firmware_type
-        )
-
-        session = async_get_clientsession(self.hass)
-        client = FirmwareUpdateClient(fw_update_url, session)
-
-        try:
-            manifest = await client.async_update_data()
-            fw_manifest = next(
-                fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
+        # For the duration of firmware flashing, hint to other integrations (i.e. ZHA)
+        # that the hardware is in use and should not be accessed. This is separate from
+        # locking the serial port itself, since a momentary release of the port may
+        # still allow for ZHA to reclaim the device.
+        async with async_firmware_flashing_context(self.hass, self._device, DOMAIN):
+            # Installing new firmware is only truly required if the wrong type is
+            # installed: upgrading to the latest release of the current firmware type
+            # isn't strictly necessary for functionality.
+            self._probed_firmware_info = await probe_silabs_firmware_info(
+                self._device,
+                bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
+                application_probe_methods=self.APPLICATION_PROBE_METHODS,
             )
-        except (StopIteration, TimeoutError, ClientError, ManifestMissing) as err:
-            _LOGGER.warning("Failed to fetch firmware update manifest", exc_info=True)
 
-            # Not having internet access should not prevent setup
-            if not firmware_install_required:
-                _LOGGER.debug("Skipping firmware upgrade due to index download failure")
-                return
-
-            raise AbortFlow(
-                reason="fw_download_failed",
-                description_placeholders=self._get_translation_placeholders(),
-            ) from err
-
-        if not firmware_install_required:
-            assert self._probed_firmware_info is not None
+            firmware_install_required = self._probed_firmware_info is None or (
+                self._probed_firmware_info.firmware_type
+                != expected_installed_firmware_type
+            )
 
-            # Make sure we do not downgrade the firmware
-            fw_metadata = NabuCasaMetadata.from_json(fw_manifest.metadata)
-            fw_version = fw_metadata.get_public_version()
-            probed_fw_version = Version(self._probed_firmware_info.firmware_version)
+            session = async_get_clientsession(self.hass)
+            client = FirmwareUpdateClient(fw_update_url, session)
 
-            if probed_fw_version >= fw_version:
-                _LOGGER.debug(
-                    "Not downgrading firmware, installed %s is newer than available %s",
-                    probed_fw_version,
-                    fw_version,
+            try:
+                manifest = await client.async_update_data()
+                fw_manifest = next(
+                    fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
+                )
+            except (StopIteration, TimeoutError, ClientError, ManifestMissing) as err:
+                _LOGGER.warning(
+                    "Failed to fetch firmware update manifest", exc_info=True
                 )
-                return
 
-        try:
-            fw_data = await client.async_fetch_firmware(fw_manifest)
-        except (TimeoutError, ClientError, ValueError) as err:
-            _LOGGER.warning("Failed to fetch firmware update", exc_info=True)
+                # Not having internet access should not prevent setup
+                if not firmware_install_required:
+                    _LOGGER.debug(
+                        "Skipping firmware upgrade due to index download failure"
+                    )
+                    return
+
+                raise AbortFlow(
+                    reason="fw_download_failed",
+                    description_placeholders=self._get_translation_placeholders(),
+                ) from err
 
-            # If we cannot download new firmware, we shouldn't block setup
             if not firmware_install_required:
-                _LOGGER.debug("Skipping firmware upgrade due to image download failure")
-                return
+                assert self._probed_firmware_info is not None
+
+                # Make sure we do not downgrade the firmware
+                fw_metadata = NabuCasaMetadata.from_json(fw_manifest.metadata)
+                fw_version = fw_metadata.get_public_version()
+                probed_fw_version = Version(self._probed_firmware_info.firmware_version)
+
+                if probed_fw_version >= fw_version:
+                    _LOGGER.debug(
+                        "Not downgrading firmware, installed %s is newer than available %s",
+                        probed_fw_version,
+                        fw_version,
+                    )
+                    return
 
-            # Otherwise, fail
-            raise AbortFlow(
-                reason="fw_download_failed",
-                description_placeholders=self._get_translation_placeholders(),
-            ) from err
+            try:
+                fw_data = await client.async_fetch_firmware(fw_manifest)
+            except (TimeoutError, ClientError, ValueError) as err:
+                _LOGGER.warning("Failed to fetch firmware update", exc_info=True)
 
-        self._probed_firmware_info = await async_flash_silabs_firmware(
-            hass=self.hass,
-            device=self._device,
-            fw_data=fw_data,
-            expected_installed_firmware_type=expected_installed_firmware_type,
-            bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
-            application_probe_methods=self.APPLICATION_PROBE_METHODS,
-            progress_callback=lambda offset, total: self.async_update_progress(
-                offset / total
-            ),
-        )
+                # If we cannot download new firmware, we shouldn't block setup
+                if not firmware_install_required:
+                    _LOGGER.debug(
+                        "Skipping firmware upgrade due to image download failure"
+                    )
+                    return
+
+                # Otherwise, fail
+                raise AbortFlow(
+                    reason="fw_download_failed",
+                    description_placeholders=self._get_translation_placeholders(),
+                ) from err
+
+            self._probed_firmware_info = await async_flash_silabs_firmware(
+                hass=self.hass,
+                device=self._device,
+                fw_data=fw_data,
+                expected_installed_firmware_type=expected_installed_firmware_type,
+                bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
+                application_probe_methods=self.APPLICATION_PROBE_METHODS,
+                progress_callback=lambda offset, total: self.async_update_progress(
+                    offset / total
+                ),
+            )
 
     async def _configure_and_start_otbr_addon(self) -> None:
         """Configure and start the OTBR addon."""

homeassistant/components/homeassistant_hardware/update.py

@@ -26,6 +26,7 @@
     ApplicationType,
     FirmwareInfo,
     ResetTarget,
+    async_firmware_flashing_context,
     async_flash_silabs_firmware,
 )
 
@@ -274,16 +275,18 @@ async def async_install(
         )
 
         try:
-            firmware_info = await async_flash_silabs_firmware(
-                hass=self.hass,
-                device=self._current_device,
-                fw_data=fw_data,
-                expected_installed_firmware_type=self.entity_description.expected_firmware_type,
-                bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
-                application_probe_methods=self.APPLICATION_PROBE_METHODS,
-                progress_callback=self._update_progress,
-                domain=self._config_entry.domain,
-            )
+            async with async_firmware_flashing_context(
+                self.hass, self._current_device, self._config_entry.domain
+            ):
+                firmware_info = await async_flash_silabs_firmware(
+                    hass=self.hass,
+                    device=self._current_device,
+                    fw_data=fw_data,
+                    expected_installed_firmware_type=self.entity_description.expected_firmware_type,
+                    bootloader_reset_methods=self.BOOTLOADER_RESET_METHODS,
+                    application_probe_methods=self.APPLICATION_PROBE_METHODS,
+                    progress_callback=self._update_progress,
+                )
         finally:
             self._attr_in_progress = False
             self.async_write_ha_state()

homeassistant/components/homeassistant_hardware/util.py

@@ -26,7 +26,6 @@
 
 from . import DATA_COMPONENT
 from .const import (
-    DOMAIN,
     OTBR_ADDON_MANAGER_DATA,
     OTBR_ADDON_NAME,
     OTBR_ADDON_SLUG,
@@ -366,6 +365,22 @@ async def probe_silabs_firmware_type(
     return fw_info.firmware_type
 
 
+@asynccontextmanager
+async def async_firmware_flashing_context(
+    hass: HomeAssistant, device: str, source_domain: str
+) -> AsyncIterator[None]:
+    """Register a device as having its firmware being actively interacted with."""
+    async with async_firmware_update_context(hass, device, source_domain):
+        firmware_info = await guess_firmware_info(hass, device)
+        _LOGGER.debug("Guessed firmware info before update: %s", firmware_info)
+
+        async with AsyncExitStack() as stack:
+            for owner in firmware_info.owners:
+                await stack.enter_async_context(owner.temporarily_stop(hass))
+
+            yield
+
+
 async def async_flash_silabs_firmware(
     hass: HomeAssistant,
     device: str,
@@ -374,10 +389,11 @@ async def async_flash_silabs_firmware(
     bootloader_reset_methods: Sequence[ResetTarget],
     application_probe_methods: Sequence[tuple[ApplicationType, int]],
     progress_callback: Callable[[int, int], None] | None = None,
-    *,
-    domain: str = DOMAIN,
 ) -> FirmwareInfo:
-    """Flash firmware to the SiLabs device."""
+    """Flash firmware to the SiLabs device.
+
+    This function is meant to be used within a firmware update context.
+    """
     if not any(
         method == expected_installed_firmware_type
         for method, _ in application_probe_methods
@@ -387,54 +403,44 @@ async def async_flash_silabs_firmware(
             f" not in application probe methods {application_probe_methods!r}"
         )
 
-    async with async_firmware_update_context(hass, device, domain):
-        firmware_info = await guess_firmware_info(hass, device)
-        _LOGGER.debug("Identified firmware info: %s", firmware_info)
-
-        fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
-
-        flasher = Flasher(
-            device=device,
-            probe_methods=tuple(
-                (m.as_flasher_application_type(), baudrate)
-                for m, baudrate in application_probe_methods
-            ),
-            bootloader_reset=tuple(
-                m.as_flasher_reset_target() for m in bootloader_reset_methods
-            ),
-        )
+    fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
 
-        async with AsyncExitStack() as stack:
-            for owner in firmware_info.owners:
-                await stack.enter_async_context(owner.temporarily_stop(hass))
+    flasher = Flasher(
+        device=device,
+        probe_methods=tuple(
+            (m.as_flasher_application_type(), baudrate)
+            for m, baudrate in application_probe_methods
+        ),
+        bootloader_reset=tuple(
+            m.as_flasher_reset_target() for m in bootloader_reset_methods
+        ),
+    )
+
+    try:
+        # Enter the bootloader with indeterminate progress
+        await flasher.enter_bootloader()
+
+        # Flash the firmware, with progress
+        await flasher.flash_firmware(fw_image, progress_callback=progress_callback)
+    except PermissionError as err:
+        raise HomeAssistantError(
+            "Failed to flash firmware: Device is used by another application"
+        ) from err
+    except Exception as err:
+        raise HomeAssistantError("Failed to flash firmware") from err
+
+    probed_firmware_info = await probe_silabs_firmware_info(
+        device,
+        bootloader_reset_methods=bootloader_reset_methods,
+        # Only probe for the expected installed firmware type
+        application_probe_methods=[
+            (method, baudrate)
+            for method, baudrate in application_probe_methods
+            if method == expected_installed_firmware_type
+        ],
+    )
+
+    if probed_firmware_info is None:
+        raise HomeAssistantError("Failed to probe the firmware after flashing")
 
-            try:
-                # Enter the bootloader with indeterminate progress
-                await flasher.enter_bootloader()
-
-                # Flash the firmware, with progress
-                await flasher.flash_firmware(
-                    fw_image, progress_callback=progress_callback
-                )
-            except PermissionError as err:
-                raise HomeAssistantError(
-                    "Failed to flash firmware: Device is used by another application"
-                ) from err
-            except Exception as err:
-                raise HomeAssistantError("Failed to flash firmware") from err
-
-            probed_firmware_info = await probe_silabs_firmware_info(
-                device,
-                bootloader_reset_methods=bootloader_reset_methods,
-                # Only probe for the expected installed firmware type
-                application_probe_methods=[
-                    (method, baudrate)
-                    for method, baudrate in application_probe_methods
-                    if method == expected_installed_firmware_type
-                ],
-            )
-
-        if probed_firmware_info is None:
-            raise HomeAssistantError("Failed to probe the firmware after flashing")
-
-        return probed_firmware_info
+    return probed_firmware_info