Review fix: owner-account fallback seed for first setup

Review round 4 (Codex gpt-5.6-terra, HIGH): relying on the flow
context alone can leave first setup ownerless. seed_setup_admin gains
an owner_ids tier: when no persisted copy and no context user exist,
every HA owner account is seeded (owners can already do everything in
HA — no privilege escalation). __init__ resolves owners via
hass.auth.async_get_users on both setup paths; conftest's hass gains
the auth surface. Tests: owner fallback seeds only owners, context
user still wins over owners, owner_ids argument validation. Suite:
910 passed.
This commit is contained in:
Joshua King
2026-09-15 17:22:39 -04:00
parent f0fa39660c
commit 068ace5693
4 changed files with 116 additions and 6 deletions
+25
View File
@@ -226,6 +226,28 @@ class NestQuestRuntimeData:
remove_update_listener: Callable[[], Any]
async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
"""Return every Home Assistant owner-account user id.
The last-resort admin seed: when no persisted allowlist copy and no
flow-context user survive to first setup, the HA owner accounts are
seeded so the household is never left with zero admins (an empty
allowlist fails closed, which would lock the household out of its
own integration). Owner accounts can already do everything in HA,
so granting NestQuest admin is not a privilege escalation.
"""
auth = getattr(hass, "auth", None)
if auth is None:
return []
users = await auth.async_get_users()
owners = [
user.id
for user in users
if getattr(user, "is_owner", False) and getattr(user, "id", None)
]
return owners
async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
"""Set up the NestQuest integration. YAML configuration is not used, returns True."""
return True
@@ -262,6 +284,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
db,
stored_admin_ids=stored_admin_ids,
context_user_id=context_user_id,
owner_ids=await _async_owner_user_ids(hass),
)
existing.database = db
entry.runtime_data = existing
@@ -291,10 +314,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if isinstance(context, dict)
else None
)
owner_ids = await _async_owner_user_ids(hass)
await seed_setup_admin(
database,
stored_admin_ids=stored_admin_ids,
context_user_id=context_user_id,
owner_ids=owner_ids,
)
remove_update_listener = entry.add_update_listener(_async_update_listener)
except BaseException:
+16 -5
View File
@@ -96,15 +96,20 @@ async def seed_setup_admin(
*,
stored_admin_ids: list[str] | None = None,
context_user_id: str | None = None,
owner_ids: list[str] | None = None,
) -> list[str]:
"""Seed an empty allowlist so the owner is never locked out.
Only acts when the allowlist is EMPTY. Preference order: the
stored entry-data copy of the list (disaster recovery for a lost
database), then the HA user who completed the config flow
(``context_user_id``). Returns the allowlist after seeding.
Non-empty allowlists are returned untouched — re-running setup
must never resurrect a deliberately narrowed list.
stored persisted copy of the list (the options flow's current
selection, then the config flow's original — disaster recovery for
a lost database), then the HA user who completed the config flow
(``context_user_id``), then — when no caller identity survived at
all — every Home Assistant OWNER account (``owner_ids``), so a
first setup whose flow context lost the user id still seeds a
working admin. Returns the allowlist after seeding. Non-empty
allowlists are returned untouched — re-running setup must never
resurrect a deliberately narrowed list.
"""
dao = AdminUsersDao(database)
async with _connection_lock(database):
@@ -120,6 +125,12 @@ async def seed_setup_admin(
candidates = [_validate_user_id(value) for value in stored_admin_ids]
elif context_user_id is not None:
candidates = [_validate_user_id(context_user_id)]
elif owner_ids:
if not isinstance(owner_ids, list):
raise ValueError(
"owner_ids must be a list of HA user id strings"
)
candidates = [_validate_user_id(value) for value in owner_ids]
for user_id in candidates:
await dao.add(user_id, _now_stamp())
return [row.ha_user_id for row in await dao.list()]
+5 -1
View File
@@ -29,7 +29,7 @@ import tempfile
import weakref
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -281,6 +281,10 @@ def make_hass() -> tuple:
config=MagicMock(),
config_entries=SimpleNamespace(async_reload=None),
)
# hass.auth mirrors the real surface the integration reads: an
# async_get_users list the tests can repoint at will (the setup
# path resolves HA owner accounts as the last-resort admin seed).
hass.auth = SimpleNamespace(async_get_users=AsyncMock(return_value=[]))
def _cleanup_config_dir():
import shutil
+70
View File
@@ -302,3 +302,73 @@ async def test_setup_seed_prefers_options_over_entry_data(
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_owner_fallback_seeds_owner_accounts(
hass, make_entry
) -> None:
"""Last-resort seed: with no persisted copy and no flow-context
user, HA owner accounts are seeded so first setup is never
ownerless (the real flow may lose the context; owners can already
do everything in HA)."""
from unittest.mock import AsyncMock
hass.auth.async_get_users = AsyncMock(
return_value=[
SimpleNamespace(id="owner-1", is_owner=True),
SimpleNamespace(id="member-1", is_owner=False),
SimpleNamespace(id="owner-2", is_owner=True),
]
)
entry = await _setup_entry(hass, make_entry(), hass.registry)
database = entry.runtime_data.database
try:
assert sorted(await list_admin_ids(database)) == ["owner-1", "owner-2"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
async def test_setup_owner_fallback_not_used_when_context_present(
hass, make_entry
) -> None:
"""The context user (flow caller) wins over the owner fallback:
a non-owner parent configuring the integration becomes the admin."""
from unittest.mock import AsyncMock
hass.auth.async_get_users = AsyncMock(
return_value=[SimpleNamespace(id="owner-1", is_owner=True)]
)
entry = await _setup_entry(
hass,
make_entry(context={"user_id": "parent-configured"}),
hass.registry,
)
database = entry.runtime_data.database
try:
assert await list_admin_ids(database) == ["parent-configured"]
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True
def test_seed_owner_ids_argument(tmp_path) -> None:
async def _body(database):
seeded = await seed_setup_admin(
database,
stored_admin_ids=None,
context_user_id=None,
owner_ids=["owner-1"],
)
assert seeded == ["owner-1"]
# Malformed owner lists are rejected like any other candidate.
with pytest.raises(ValueError):
await seed_setup_admin(
database,
stored_admin_ids=None,
context_user_id=None,
owner_ids=[7],
)
return seeded
_with_db(tmp_path, "seed-owner-arg.db")(_body)