Add iOS Live Activity webhook handlers to mobile_app

homeassistant/components/mobile_app/__init__.py

@@ -42,6 +42,7 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DATA_PUSH_CHANNEL,
     DATA_STORE,
@@ -81,6 +82,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         DATA_CONFIG_ENTRIES: {},
         DATA_DELETED_IDS: app_config.get(DATA_DELETED_IDS, []),
         DATA_DEVICES: {},
+        DATA_LIVE_ACTIVITY_TOKENS: {},
         DATA_PUSH_CHANNEL: {},
         DATA_STORE: store,
         DATA_PENDING_UPDATES: {sensor_type: {} for sensor_type in SENSOR_TYPES},
@@ -231,6 +233,7 @@ 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]
+    hass.data[DOMAIN][DATA_LIVE_ACTIVITY_TOKENS].pop(webhook_id, None)
     await hass_notify.async_reload(hass, DOMAIN)
 
     return True

homeassistant/components/mobile_app/const.py

@@ -32,6 +32,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 +43,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"
 
@@ -89,9 +96,8 @@
     {
         vol.Inclusive(ATTR_PUSH_TOKEN, "push_cloud"): cv.string,
         vol.Inclusive(ATTR_PUSH_URL, "push_cloud"): cv.url,
-        # 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/notify.py

@@ -35,16 +35,21 @@
     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,
     DOMAIN,
@@ -211,12 +216,40 @@ 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"
             )
 
+    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 tag in device_tokens:
+            return device_tokens[tag][ATTR_PUSH_TOKEN]
+
+        # 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=self._get_live_activity_token(entry, data),
+            )
         except HomeAssistantError as e:
             if e.translation_key == "rate_limit_exceeded_sending_notification":
                 _LOGGER.warning(str(e))
@@ -225,7 +258,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 = {
@@ -236,15 +273,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

@@ -67,10 +67,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,
@@ -99,6 +101,7 @@
     DATA_CONFIG_ENTRIES,
     DATA_DELETED_IDS,
     DATA_DEVICES,
+    DATA_LIVE_ACTIVITY_TOKENS,
     DATA_PENDING_UPDATES,
     DOMAIN,
     ERR_ENCRYPTION_ALREADY_ENABLED,
@@ -798,3 +801,48 @@ 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] = {
+        ATTR_PUSH_TOKEN: data[ATTR_PUSH_TOKEN]
+    }
+
+    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]
+
+    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]