Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00
Code Activity

Don't return remote/cloudhook URLs while registering a local user (#166336)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Timothy
2026-03-25 17:24:24 +01:00
committed by GitHub
parent c8f7d9dd42
commit 8c73dcad91
7 changed files with 227 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
homeassistant/components/mobile_app/__init__.py
View File

@@ -49,7 +49,7 @@ from .const import (
STORAGE_KEY,
STORAGE_VERSION,
)
from .helpers import savable_state
from .helpers import async_is_local_only_user, savable_state
from .http_api import RegistrationsView
from .timers import async_handle_timer_event
from .util import async_create_cloud_hook, supports_push
@@ -107,29 +107,14 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a mobile_app entry."""
registration = entry.data
webhook_id = registration[CONF_WEBHOOK_ID]
hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id] = entry
device_registry = dr.async_get(hass)
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, registration[ATTR_DEVICE_ID])},
manufacturer=registration[ATTR_MANUFACTURER],
model=registration[ATTR_MODEL],
name=registration[ATTR_DEVICE_NAME],
sw_version=registration[ATTR_OS_VERSION],
)
hass.data[DOMAIN][DATA_DEVICES][webhook_id] = device
registration_name = f"Mobile App: {registration[ATTR_DEVICE_NAME]}"
webhook_register(hass, DOMAIN, registration_name, webhook_id, handle_webhook)
async def _async_setup_cloudhook(
hass: HomeAssistant,
entry: ConfigEntry,
user_id: str,
webhook_id: str,
) -> None:
"""Set up cloudhook forwarding for a mobile_app entry."""
local_only = await async_is_local_only_user(hass, user_id)
def clean_cloudhook() -> None:
"""Clean up cloudhook from config entry."""
@@ -138,6 +123,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
data.pop(CONF_CLOUDHOOK_URL)
hass.config_entries.async_update_entry(entry, data=data)
if local_only:
# Local-only user should not have a cloudhook
if cloud.async_is_logged_in(hass) and CONF_CLOUDHOOK_URL in entry.data:
with suppress(cloud.CloudNotAvailable, ValueError):
await cloud.async_delete_cloudhook(hass, webhook_id)
clean_cloudhook()
return
def on_cloudhook_change(cloudhook: dict[str, Any] | None) -> None:
"""Handle cloudhook changes."""
if cloudhook:
@@ -180,6 +173,33 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
entry.async_on_unload(cloud.async_listen_connection_change(hass, manage_cloudhook))
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a mobile_app entry."""
registration = entry.data
webhook_id = registration[CONF_WEBHOOK_ID]
hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id] = entry
device_registry = dr.async_get(hass)
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, registration[ATTR_DEVICE_ID])},
manufacturer=registration[ATTR_MANUFACTURER],
model=registration[ATTR_MODEL],
name=registration[ATTR_DEVICE_NAME],
sw_version=registration[ATTR_OS_VERSION],
)
hass.data[DOMAIN][DATA_DEVICES][webhook_id] = device
registration_name = f"Mobile App: {registration[ATTR_DEVICE_NAME]}"
webhook_register(hass, DOMAIN, registration_name, webhook_id, handle_webhook)
await _async_setup_cloudhook(hass, entry, registration[CONF_USER_ID], webhook_id)
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
if supports_push(hass, webhook_id):

9
homeassistant/components/mobile_app/helpers.py
View File

@@ -114,6 +114,15 @@ def decrypt_payload_legacy(key: str, ciphertext: bytes) -> JsonValueType | None:
)
async def async_is_local_only_user(hass: HomeAssistant, user_id: str) -> bool:
"""Return True if the user is local only."""
user = await hass.auth.async_get_user(user_id)
if user is None:
# Treat unknown/missing users as local-only to avoid exposing cloud URLs
return True
return user.local_only
def registration_context(registration: Mapping[str, Any]) -> Context:
"""Generate a context from a request."""
return Context(user_id=registration[CONF_USER_ID])

11
homeassistant/components/mobile_app/http_api.py
View File

@@ -68,8 +68,13 @@ class RegistrationsView(HomeAssistantView):
hass = request.app[KEY_HASS]
webhook_id = secrets.token_hex()
user = request["hass_user"]
if cloud.async_active_subscription(hass) and cloud.async_is_connected(hass):
if (
not user.local_only
and cloud.async_active_subscription(hass)
and cloud.async_is_connected(hass)
):
data[CONF_CLOUDHOOK_URL] = await async_create_cloud_hook(
hass, webhook_id, None
)
@@ -79,7 +84,7 @@ class RegistrationsView(HomeAssistantView):
if data[ATTR_SUPPORTS_ENCRYPTION]:
data[CONF_SECRET] = secrets.token_hex(SecretBox.KEY_SIZE)
data[CONF_USER_ID] = request["hass_user"].id
data[CONF_USER_ID] = user.id
# Fallback to DEVICE_ID if slug is empty.
if not slugify(data[ATTR_DEVICE_NAME], separator=""):
@@ -92,7 +97,7 @@ class RegistrationsView(HomeAssistantView):
)
remote_ui_url = None
if cloud.async_active_subscription(hass):
if not user.local_only and cloud.async_active_subscription(hass):
with suppress(cloud.CloudNotAvailable):
remote_ui_url = cloud.async_remote_ui_url(hass)

6
homeassistant/components/mobile_app/webhook.py
View File

@@ -94,6 +94,7 @@ from .const import (
CONF_CLOUDHOOK_URL,
CONF_REMOTE_UI_URL,
CONF_SECRET,
CONF_USER_ID,
DATA_CONFIG_ENTRIES,
DATA_DELETED_IDS,
DATA_DEVICES,
@@ -109,6 +110,7 @@ from .const import (
SIGNAL_SENSOR_UPDATE,
)
from .helpers import (
async_is_local_only_user,
decrypt_payload,
decrypt_payload_legacy,
empty_okay_response,
@@ -756,7 +758,9 @@ async def webhook_get_config(
"theme_color": MANIFEST_JSON["theme_color"],
}
if cloud.async_active_subscription(hass):
if cloud.async_active_subscription(hass) and not await async_is_local_only_user(
hass, config_entry.data[CONF_USER_ID]
):
if CONF_CLOUDHOOK_URL in config_entry.data:
resp[CONF_CLOUDHOOK_URL] = config_entry.data[CONF_CLOUDHOOK_URL]
with suppress(cloud.CloudNotAvailable):

45
tests/components/mobile_app/test_http_api.py
View File

@@ -11,6 +11,7 @@ import pytest
from homeassistant.components.mobile_app.const import (
CONF_CLOUDHOOK_URL,
CONF_REMOTE_UI_URL,
CONF_SECRET,
DOMAIN,
)
@@ -161,6 +162,50 @@ async def test_registration_with_cloud(
)
async def test_registration_with_cloud_local_only_user(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_admin_user: MockUser,
) -> None:
"""Test that cloudhook_url and remote_ui_url are not returned for local_only users."""
hass_admin_user.local_only = True
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
api_client = await hass_client()
with (
patch(
"homeassistant.components.mobile_app.http_api.cloud.async_active_subscription",
return_value=True,
),
patch(
"homeassistant.components.mobile_app.http_api.cloud.async_is_connected",
return_value=True,
),
patch(
"homeassistant.components.mobile_app.http_api.async_create_cloud_hook",
return_value="https://hooks.nabu.casa/test123",
) as mock_create_cloud_hook,
patch(
"homeassistant.components.mobile_app.http_api.cloud.async_remote_ui_url",
return_value="https://remote.ui",
),
patch(
"homeassistant.components.person.async_add_user_device_tracker",
spec=True,
),
):
resp = await api_client.post(
"/api/mobile_app/registrations", json=REGISTER_CLEARTEXT
)
assert resp.status == HTTPStatus.CREATED
register_json = await resp.json()
assert register_json.get(CONF_CLOUDHOOK_URL) is None
assert register_json.get(CONF_REMOTE_UI_URL) is None
mock_create_cloud_hook.assert_not_called()
async def test_registration_encryption_legacy(
hass: HomeAssistant, hass_client: ClientSessionGenerator
) -> None:

78
tests/components/mobile_app/test_init.py
View File

@@ -272,6 +272,84 @@ async def test_delete_cloud_hook(
assert (CONF_CLOUDHOOK_URL in config_entry.data) == should_cloudhook_exist
async def test_setup_entry_local_only_user_no_cloudhook(
hass: HomeAssistant,
hass_admin_user: MockUser,
) -> None:
"""Test that cloudhook is not created for local_only users during setup."""
hass_admin_user.local_only = True
config_entry = MockConfigEntry(
data={
**REGISTER_CLEARTEXT,
CONF_WEBHOOK_ID: "test-webhook-id",
ATTR_DEVICE_NAME: "Test",
ATTR_DEVICE_ID: "Test",
CONF_USER_ID: hass_admin_user.id,
},
domain=DOMAIN,
title="Test",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.cloud.async_active_subscription",
return_value=True,
),
patch("homeassistant.components.cloud.async_is_logged_in", return_value=True),
patch("homeassistant.components.cloud.async_is_connected", return_value=True),
patch(
"homeassistant.components.cloud.async_get_or_create_cloudhook",
) as mock_create_cloudhook,
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
# Cloudhook should not be created for local_only user
assert CONF_CLOUDHOOK_URL not in config_entry.data
mock_create_cloudhook.assert_not_called()
async def test_setup_entry_local_only_user_cleans_existing_cloudhook(
hass: HomeAssistant,
hass_admin_user: MockUser,
) -> None:
"""Test that existing cloudhook is cleaned up for local_only users during setup."""
hass_admin_user.local_only = True
webhook_id = "test-webhook-id"
config_entry = MockConfigEntry(
data={
**REGISTER_CLEARTEXT,
CONF_WEBHOOK_ID: webhook_id,
ATTR_DEVICE_NAME: "Test",
ATTR_DEVICE_ID: "Test",
CONF_USER_ID: hass_admin_user.id,
CONF_CLOUDHOOK_URL: "https://hooks.nabu.casa/stale",
},
domain=DOMAIN,
title="Test",
)
config_entry.add_to_hass(hass)
with (
patch("homeassistant.components.cloud.async_is_logged_in", return_value=True),
patch(
"homeassistant.components.cloud.async_delete_cloudhook",
) as delete_cloudhook,
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
# Existing cloudhook should be removed for local_only user
assert CONF_CLOUDHOOK_URL not in config_entry.data
delete_cloudhook.assert_called_once_with(hass, webhook_id)
async def test_remove_entry_on_user_remove(
hass: HomeAssistant,
hass_admin_user: MockUser,

39
tests/components/mobile_app/test_webhook.py
View File

@@ -29,7 +29,7 @@ from homeassistant.setup import async_setup_component
from .const import CALL_SERVICE, FIRE_EVENT, REGISTER_CLEARTEXT, RENDER_TEMPLATE, UPDATE
from tests.common import async_capture_events, async_mock_service
from tests.common import MockUser, async_capture_events, async_mock_service
from tests.components.conversation import MockAgent
@@ -377,6 +377,43 @@ async def test_webhook_handle_get_config_with_cloudhook_no_subscription(
assert "remote_ui_url" not in json_resp
async def test_webhook_handle_get_config_with_cloudhook_local_only_user(
hass: HomeAssistant,
hass_admin_user: MockUser,
create_registrations: tuple[dict[str, Any], dict[str, Any]],
webhook_client: TestClient,
) -> None:
"""Test get_config doesn't return cloudhook_url or remote_ui_url for local_only users."""
hass_admin_user.local_only = True
webhook_id = create_registrations[1]["webhook_id"]
webhook_url = f"/api/webhook/{webhook_id}"
# Get the config entry and add cloudhook_url to it
config_entry = hass.config_entries.async_entries(DOMAIN)[1]
hass.config_entries.async_update_entry(
config_entry,
data={**config_entry.data, "cloudhook_url": "https://hooks.nabu.casa/test"},
)
with (
patch(
"homeassistant.components.cloud.async_active_subscription",
return_value=True,
),
patch(
"homeassistant.components.cloud.async_remote_ui_url",
return_value="https://remote.ui.url",
),
):
resp = await webhook_client.post(webhook_url, json={"type": "get_config"})
assert resp.status == HTTPStatus.OK
json_resp = await resp.json()
assert "cloudhook_url" not in json_resp
assert "remote_ui_url" not in json_resp
async def test_webhook_returns_error_incorrect_json(
create_registrations: tuple[dict[str, Any], dict[str, Any]],
webhook_client: TestClient,