Add iOS Live Activity webhook handlers to mobile_app

homeassistant/components/mobile_app/__init__.py

@@ -21,6 +21,7 @@
 )
 from homeassistant.helpers.storage import Store
 from homeassistant.helpers.typing import ConfigType
+from homeassistant.util import dt as dt_util
 
 # Pre-import the platforms so they get loaded when the integration
 # is imported as they are almost always going to be loaded and its
@@ -42,10 +43,12 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DATA_PUSH_CHANNEL,
     DATA_STORE,
     DOMAIN,
+    LIVE_ACTIVITY_TOKEN_TTL_SECONDS,
     SENSOR_TYPES,
     STORAGE_KEY,
     STORAGE_VERSION,
@@ -73,14 +76,28 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         app_config, dict
     ):
         app_config = {
-            DATA_CONFIG_ENTRIES: {},
             DATA_DELETED_IDS: [],
+            DATA_LIVE_ACTIVITY_TOKENS: {},
         }
+    elif DATA_LIVE_ACTIVITY_TOKENS not in app_config:
+        app_config[DATA_LIVE_ACTIVITY_TOKENS] = {}
+
+    cutoff = dt_util.utcnow().timestamp() - LIVE_ACTIVITY_TOKEN_TTL_SECONDS
+    live_activity_tokens: dict[str, Any] = {
+        wh_id: {
+            tag: entry
+            for tag, entry in tags.items()
+            if entry.get("stored_at", 0) > cutoff
+        }
+        for wh_id, tags in app_config[DATA_LIVE_ACTIVITY_TOKENS].items()
+        if any(entry.get("stored_at", 0) > cutoff for entry in tags.values())
+    }
 
     hass.data[DOMAIN] = {
         DATA_CONFIG_ENTRIES: {},
         DATA_DELETED_IDS: app_config.get(DATA_DELETED_IDS, []),
         DATA_DEVICES: {},
+        DATA_LIVE_ACTIVITY_TOKENS: live_activity_tokens,
         DATA_PUSH_CHANNEL: {},
         DATA_STORE: store,
         DATA_PENDING_UPDATES: {sensor_type: {} for sensor_type in SENSOR_TYPES},
@@ -231,6 +248,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     webhook_unregister(hass, webhook_id)
     del hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id]
     del hass.data[DOMAIN][DATA_DEVICES][webhook_id]
+    if hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS].pop(webhook_id, None) is not None:
+        await hass.data[DOMAIN][DATA_STORE].async_save(savable_state(hass))
     await hass_notify.async_reload(hass, DOMAIN)
 
     return True

homeassistant/components/mobile_app/const.py

@@ -9,6 +9,8 @@
 STORAGE_KEY = DOMAIN
 STORAGE_VERSION = 1
 
+LIVE_ACTIVITY_TOKEN_TTL_SECONDS = 8 * 3600
+
 CONF_CLOUDHOOK_URL = "cloudhook_url"
 CONF_REMOTE_UI_URL = "remote_ui_url"
 CONF_SECRET = "secret"
@@ -32,6 +34,7 @@
 ATTR_NO_LEGACY_ENCRYPTION = "no_legacy_encryption"
 ATTR_OS_NAME = "os_name"
 ATTR_OS_VERSION = "os_version"
+ATTR_PUSH_TAG = "tag"
 ATTR_PUSH_WEBSOCKET_CHANNEL = "push_websocket_channel"
 ATTR_PUSH_TOKEN = "push_token"
 ATTR_PUSH_URL = "push_url"
@@ -42,6 +45,12 @@
 ATTR_PUSH_RATE_LIMITS_SUCCESSFUL = "successful"
 ATTR_SUPPORTS_ENCRYPTION = "supports_encryption"
 
+ATTR_LIVE_UPDATE = "live_update"
+ATTR_LIVE_ACTIVITY_TOKEN = "live_activity_token"
+ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN = "live_activity_push_to_start_token"
+ATTR_LIVE_ACTIVITY_TAG = "live_activity_tag"
+DATA_LIVE_ACTIVITY_TOKENS = "live_activity_tokens"
+
 ATTR_EVENT_DATA = "event_data"
 ATTR_EVENT_TYPE = "event_type"
 
@@ -93,6 +102,7 @@
         # Set to True to indicate that this registration will connect via websocket channel
         # to receive push notifications.
         vol.Optional(ATTR_PUSH_WEBSOCKET_CHANNEL): cv.boolean,
+        vol.Optional(ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN): cv.string,
     },
     extra=vol.ALLOW_EXTRA,
 )

homeassistant/components/mobile_app/helpers.py

@@ -29,6 +29,7 @@
     CONF_SECRET,
     CONF_USER_ID,
     DATA_DELETED_IDS,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DOMAIN,
 )
 
@@ -167,10 +168,11 @@ def safe_registration(registration: dict) -> dict:
 
 def savable_state(hass: HomeAssistant) -> dict:
     """Return a clean object containing things that should be saved."""
+    # pylint: disable-next=hass-use-runtime-data
+    domain_data = hass.data[DOMAIN]
     return {
-        # Uses legacy hass.data[DOMAIN] pattern
-        # pylint: disable-next=hass-use-runtime-data
-        DATA_DELETED_IDS: hass.data[DOMAIN][DATA_DELETED_IDS],
+        DATA_DELETED_IDS: domain_data[DATA_DELETED_IDS],
+        DATA_LIVE_ACTIVITY_TOKENS: domain_data[DATA_LIVE_ACTIVITY_TOKENS],
     }
 
 

homeassistant/components/mobile_app/notify.py

@@ -37,22 +37,29 @@
     ATTR_APP_ID,
     ATTR_APP_VERSION,
     ATTR_DEVICE_NAME,
+    ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN,
+    ATTR_LIVE_ACTIVITY_TOKEN,
+    ATTR_LIVE_UPDATE,
     ATTR_OS_VERSION,
     ATTR_PUSH_RATE_LIMITS,
     ATTR_PUSH_RATE_LIMITS_ERRORS,
     ATTR_PUSH_RATE_LIMITS_MAXIMUM,
     ATTR_PUSH_RATE_LIMITS_RESETS_AT,
     ATTR_PUSH_RATE_LIMITS_SUCCESSFUL,
+    ATTR_PUSH_TAG,
     ATTR_PUSH_TOKEN,
     ATTR_PUSH_URL,
     ATTR_WEBHOOK_ID,
     DATA_CONFIG_ENTRIES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_NOTIFY,
     DATA_PUSH_CHANNEL,
+    DATA_STORE,
     DOMAIN,
+    LIVE_ACTIVITY_TOKEN_TTL_SECONDS,
     SIGNAL_RECORD_NOTIFICATION,
 )
-from .helpers import device_info
+from .helpers import device_info, savable_state
 from .push_notification import PushChannel
 from .util import supports_push
 
@@ -231,12 +238,51 @@ async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
                 f"Device(s) with webhook id(s) {', '.join(failed_targets)} not connected to local push notifications"
             )
 
+    async def _get_live_activity_token(
+        self, entry: ConfigEntry, data: dict[str, Any]
+    ) -> str | None:
+        """Return the Live Activity APNs token for this notification, or None."""
+        notification_data = data.get(ATTR_DATA) or {}
+        if not notification_data.get(ATTR_LIVE_UPDATE):
+            return None
+
+        tag = notification_data.get(ATTR_PUSH_TAG)
+        if not tag:
+            return None
+
+        # Per-activity token — the activity is already running on the device.
+        webhook_id = entry.data[ATTR_WEBHOOK_ID]
+        live_activity_tokens = self.hass.data[DOMAIN].get(DATA_LIVE_ACTIVITY_TOKENS, {})
+        device_tokens = live_activity_tokens.get(webhook_id, {})
+        if stored := device_tokens.get(tag):
+            if (
+                dt_util.utcnow().timestamp() - stored.get("stored_at", 0)
+                < LIVE_ACTIVITY_TOKEN_TTL_SECONDS
+            ):
+                return stored["token"]
+            # Token expired — remove it lazily.
+            device_tokens.pop(tag, None)
+            if not device_tokens:
+                live_activity_tokens.pop(webhook_id, None)
+            await self.hass.data[DOMAIN][DATA_STORE].async_save(
+                savable_state(self.hass)
+            )
+
+        # Push-to-start token — start a new activity remotely (iOS 17.2+).
+        app_data = entry.data[ATTR_APP_DATA]
+        return app_data.get(ATTR_LIVE_ACTIVITY_PUSH_TO_START_TOKEN)
+
     async def _async_send_remote_message_target(
         self, entry: ConfigEntry, data: dict[str, Any]
-    ):
+    ) -> None:
         """Send a message to a target."""
         try:
-            await _send_message(async_get_clientsession(self.hass), entry, data)
+            await _send_message(
+                async_get_clientsession(self.hass),
+                entry,
+                data,
+                live_activity_token=await self._get_live_activity_token(entry, data),
+            )
         except HomeAssistantError as e:
             if e.translation_key == "rate_limit_exceeded_sending_notification":
                 _LOGGER.warning(str(e))
@@ -245,7 +291,11 @@ async def _async_send_remote_message_target(
 
 
 async def _send_message(
-    session: ClientSession, entry: ConfigEntry, data: dict[str, Any]
+    session: ClientSession,
+    entry: ConfigEntry,
+    data: dict[str, Any],
+    *,
+    live_activity_token: str | None = None,
 ) -> None:
     """Shared internal helper to send messages via cloud push notification services."""
     reg_info = {
@@ -256,15 +306,19 @@ async def _send_message(
     if ATTR_OS_VERSION in entry.data:
         reg_info[ATTR_OS_VERSION] = entry.data[ATTR_OS_VERSION]
 
+    payload: dict[str, Any] = {
+        **data,
+        ATTR_PUSH_TOKEN: entry.data[ATTR_APP_DATA][ATTR_PUSH_TOKEN],
+        "registration_info": reg_info,
+    }
+    if live_activity_token:
+        payload[ATTR_LIVE_ACTIVITY_TOKEN] = live_activity_token
+
     try:
         async with asyncio.timeout(10):
             response = await session.post(
                 entry.data[ATTR_APP_DATA][ATTR_PUSH_URL],
-                json={
-                    **data,
-                    ATTR_PUSH_TOKEN: entry.data[ATTR_APP_DATA][ATTR_PUSH_TOKEN],
-                    "registration_info": reg_info,
-                },
+                json=payload,
             )
             result: dict[str, Any] = await response.json()
 

homeassistant/components/mobile_app/webhook.py

@@ -48,6 +48,7 @@
     template,
 )
 from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.util import dt as dt_util
 from homeassistant.util.decorator import Registry
 
 from .const import (
@@ -57,10 +58,12 @@
     ATTR_DEVICE_NAME,
     ATTR_EVENT_DATA,
     ATTR_EVENT_TYPE,
+    ATTR_LIVE_ACTIVITY_TAG,
     ATTR_MANUFACTURER,
     ATTR_MODEL,
     ATTR_NO_LEGACY_ENCRYPTION,
     ATTR_OS_VERSION,
+    ATTR_PUSH_TOKEN,
     ATTR_SENSOR_ATTRIBUTES,
     ATTR_SENSOR_DEVICE_CLASS,
     ATTR_SENSOR_DISABLED,
@@ -87,7 +90,9 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
+    DATA_STORE,
     DOMAIN,
     ERR_ENCRYPTION_ALREADY_ENABLED,
     ERR_ENCRYPTION_REQUIRED,
@@ -107,6 +112,7 @@
     error_response,
     registration_context,
     safe_registration,
+    savable_state,
     webhook_response,
 )
 
@@ -771,3 +777,51 @@ async def webhook_scan_tag(
         registration_context(config_entry.data),
     )
     return empty_okay_response()
+
+
+@WEBHOOK_COMMANDS.register("live_activity_token")
+@validate_schema(
+    {
+        vol.Required(ATTR_LIVE_ACTIVITY_TAG): cv.string,
+        vol.Required(ATTR_PUSH_TOKEN): cv.string,
+    }
+)
+async def webhook_update_live_activity_token(
+    hass: HomeAssistant, config_entry: ConfigEntry, data: dict[str, Any]
+) -> Response:
+    """Store a Live Activity APNs token sent by the iOS app."""
+    webhook_id = config_entry.data[CONF_WEBHOOK_ID]
+    activity_tag = data[ATTR_LIVE_ACTIVITY_TAG]
+
+    live_activity_tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    live_activity_tokens.setdefault(webhook_id, {})[activity_tag] = {
+        "token": data[ATTR_PUSH_TOKEN],
+        "stored_at": dt_util.utcnow().timestamp(),
+    }
+    await hass.data[DOMAIN][DATA_STORE].async_save(savable_state(hass))
+
+    return empty_okay_response()
+
+
+@WEBHOOK_COMMANDS.register("live_activity_dismissed")
+@validate_schema(
+    {
+        vol.Required(ATTR_LIVE_ACTIVITY_TAG): cv.string,
+    }
+)
+async def webhook_live_activity_dismissed(
+    hass: HomeAssistant, config_entry: ConfigEntry, data: dict[str, str]
+) -> Response:
+    """Remove a stored Live Activity token when the activity ends on device."""
+    webhook_id = config_entry.data[CONF_WEBHOOK_ID]
+    activity_tag = data[ATTR_LIVE_ACTIVITY_TAG]
+
+    live_activity_tokens = hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+    if webhook_id in live_activity_tokens:
+        live_activity_tokens[webhook_id].pop(activity_tag, None)
+        # Clean up the device key if no activities remain.
+        if not live_activity_tokens[webhook_id]:
+            del live_activity_tokens[webhook_id]
+        await hass.data[DOMAIN][DATA_STORE].async_save(savable_state(hass))
+
+    return empty_okay_response()

tests/components/mobile_app/test_init.py

@@ -1,9 +1,11 @@
 """Tests for the mobile app integration."""
 
 from collections.abc import Awaitable, Callable
+from http import HTTPStatus
 from typing import Any
 from unittest.mock import Mock, patch
 
+from aiohttp.test_utils import TestClient
 import pytest
 
 from homeassistant.components.cloud import CloudNotAvailable
@@ -12,6 +14,7 @@
     CONF_CLOUDHOOK_URL,
     CONF_USER_ID,
     DATA_DELETED_IDS,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DOMAIN,
 )
 from homeassistant.config_entries import ConfigEntry, ConfigEntryState
@@ -615,3 +618,33 @@ def mock_listen_cloudhook_change(hass_instance, wh_id: str, callback):
 
         # URL should remain the same
         assert config_entry.data[CONF_CLOUDHOOK_URL] == new_url
+
+
+@pytest.mark.usefixtures("create_registrations")
+async def test_unload_removes_live_activity_tokens(
+    hass: HomeAssistant, webhook_client: TestClient
+) -> None:
+    """Test that live activity tokens are removed from hass.data when entry is unloaded."""
+    # Use the cleartext (non-encrypted) entry
+    config_entry = hass.config_entries.async_entries("mobile_app")[1]
+    webhook_id = config_entry.data["webhook_id"]
+
+    # Store a live activity token via the webhook
+    resp = await webhook_client.post(
+        f"/api/webhook/{webhook_id}",
+        json={
+            "type": "live_activity_token",
+            "data": {
+                "live_activity_tag": "washer_cycle",
+                "push_token": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4e5f6a1b2",
+            },
+        },
+    )
+    assert resp.status == HTTPStatus.OK
+    assert webhook_id in hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]
+
+    # Unload the config entry
+    await hass.config_entries.async_unload(config_entry.entry_id)
+
+    # Verify the token is removed so stale tokens cannot be used after reloads/unloads
+    assert webhook_id not in hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS]