mirror of
https://github.com/home-assistant/core.git
synced 2026-09-26 01:11:51 -04:00
Prevent removing the owner user (#182398)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5.1
parent
a795ea4117
commit
1d13e1604a
@@ -344,6 +344,8 @@ class AuthManager:
|
||||
|
||||
async def async_remove_user(self, user: models.User) -> None:
|
||||
"""Remove a user."""
|
||||
if user.is_owner:
|
||||
raise ValueError("Unable to remove the owner")
|
||||
tasks = [
|
||||
self.async_remove_credentials(credentials)
|
||||
for credentials in user.credentials
|
||||
|
||||
@@ -58,6 +58,24 @@ async def websocket_delete(
|
||||
)
|
||||
return
|
||||
|
||||
if user.system_generated:
|
||||
connection.send_message(
|
||||
websocket_api.error_message(
|
||||
msg["id"],
|
||||
"cannot_modify_system_generated",
|
||||
"Unable to delete system generated users.",
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
if user.is_owner:
|
||||
connection.send_message(
|
||||
websocket_api.error_message(
|
||||
msg["id"], "cannot_delete_owner", "Unable to delete the owner"
|
||||
)
|
||||
)
|
||||
return
|
||||
|
||||
await hass.auth.async_remove_user(user)
|
||||
|
||||
connection.send_message(websocket_api.result_message(msg["id"]))
|
||||
|
||||
@@ -83,6 +83,15 @@ async def websocket_delete(
|
||||
# if not new, an existing credential exists.
|
||||
# Removing the credential will also remove the auth.
|
||||
if not credentials.is_new:
|
||||
user = await hass.auth.async_get_user_by_credentials(credentials)
|
||||
if user is not None and user.is_owner:
|
||||
connection.send_error(
|
||||
msg["id"],
|
||||
"cannot_delete_owner_credentials",
|
||||
"Unable to delete the credentials of the owner",
|
||||
)
|
||||
return
|
||||
|
||||
await hass.auth.async_remove_credentials(credentials)
|
||||
|
||||
connection.send_result(msg["id"])
|
||||
|
||||
@@ -578,6 +578,17 @@ async def test_cannot_deactive_owner(mock_hass) -> None:
|
||||
await manager.async_deactivate_user(owner)
|
||||
|
||||
|
||||
async def test_cannot_remove_owner(mock_hass: HomeAssistant) -> None:
|
||||
"""Test that we cannot remove the owner."""
|
||||
manager = await auth.auth_manager_from_config(mock_hass, [], [])
|
||||
owner = MockUser(is_owner=True).add_to_auth_manager(manager)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
await manager.async_remove_user(owner)
|
||||
|
||||
assert await manager.async_get_user(owner.id) is owner
|
||||
|
||||
|
||||
async def test_deactivate_user_removes_refresh_tokens(hass: HomeAssistant) -> None:
|
||||
"""Test that deactivating a user removes their refresh tokens."""
|
||||
manager = await auth.auth_manager_from_config(hass, [], [])
|
||||
|
||||
@@ -161,6 +161,36 @@ async def test_delete_unknown_user(
|
||||
assert result["error"]["code"] == "not_found"
|
||||
|
||||
|
||||
async def test_delete_owner(
|
||||
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
|
||||
) -> None:
|
||||
"""Test that the owner cannot be deleted."""
|
||||
owner = MockUser(id="abc", name="Test Owner", is_owner=True).add_to_hass(hass)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json({"id": 5, "type": "config/auth/delete", "user_id": owner.id})
|
||||
|
||||
result = await client.receive_json()
|
||||
assert not result["success"], result
|
||||
assert result["error"]["code"] == "cannot_delete_owner"
|
||||
assert await hass.auth.async_get_user(owner.id) is owner
|
||||
|
||||
|
||||
async def test_delete_system_generated(
|
||||
hass: HomeAssistant, hass_ws_client: WebSocketGenerator
|
||||
) -> None:
|
||||
"""Test that system generated users cannot be deleted."""
|
||||
user = MockUser(id="abc", name="System", system_generated=True).add_to_hass(hass)
|
||||
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json({"id": 5, "type": "config/auth/delete", "user_id": user.id})
|
||||
|
||||
result = await client.receive_json()
|
||||
assert not result["success"], result
|
||||
assert result["error"]["code"] == "cannot_modify_system_generated"
|
||||
assert await hass.auth.async_get_user(user.id) is user
|
||||
|
||||
|
||||
async def test_delete(
|
||||
hass: HomeAssistant, hass_ws_client: WebSocketGenerator, hass_access_token: str
|
||||
) -> None:
|
||||
|
||||
@@ -221,34 +221,66 @@ async def test_delete_removes_just_auth(
|
||||
async def test_delete_removes_credential(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
hass_storage: dict[str, Any],
|
||||
auth_provider: prov_ha.HassAuthProvider,
|
||||
) -> None:
|
||||
"""Test deleting auth that is connected to a user."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
user = MockUser().add_to_hass(hass)
|
||||
hass_storage[prov_ha.STORAGE_KEY] = {
|
||||
"version": 1,
|
||||
"data": {"users": [{"username": "test-user"}]},
|
||||
}
|
||||
|
||||
user.credentials.append(
|
||||
await hass.auth.auth_providers[0].async_get_or_create_credentials(
|
||||
{"username": "test-user"}
|
||||
)
|
||||
await hass.async_add_executor_job(
|
||||
auth_provider.data.add_auth, "other-user", "other-pass"
|
||||
)
|
||||
credential = await auth_provider.async_get_or_create_credentials(
|
||||
{"username": "other-user"}
|
||||
)
|
||||
await hass.auth.async_link_user(user, credential)
|
||||
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 5,
|
||||
"type": "config/auth_provider/homeassistant/delete",
|
||||
"username": "test-user",
|
||||
"username": "other-user",
|
||||
}
|
||||
)
|
||||
|
||||
result = await client.receive_json()
|
||||
assert result["success"], result
|
||||
assert len(hass_storage[prov_ha.STORAGE_KEY]["data"]["users"]) == 0
|
||||
assert user.credentials == []
|
||||
assert not any(
|
||||
entry["username"] == "other-user" for entry in auth_provider.data.users
|
||||
)
|
||||
|
||||
|
||||
async def test_delete_owner_credentials(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
auth_provider: prov_ha.HassAuthProvider,
|
||||
) -> None:
|
||||
"""Test deleting auth that is connected to the owner is refused."""
|
||||
client = await hass_ws_client(hass)
|
||||
|
||||
owner = MockUser(is_owner=True).add_to_hass(hass)
|
||||
await hass.async_add_executor_job(
|
||||
auth_provider.data.add_auth, "owner-user", "owner-pass"
|
||||
)
|
||||
credential = await auth_provider.async_get_or_create_credentials(
|
||||
{"username": "owner-user"}
|
||||
)
|
||||
await hass.auth.async_link_user(owner, credential)
|
||||
|
||||
await client.send_json(
|
||||
{
|
||||
"id": 5,
|
||||
"type": "config/auth_provider/homeassistant/delete",
|
||||
"username": "owner-user",
|
||||
}
|
||||
)
|
||||
|
||||
result = await client.receive_json()
|
||||
assert not result["success"], result
|
||||
assert result["error"]["code"] == "cannot_delete_owner_credentials"
|
||||
assert owner.credentials == [credential]
|
||||
assert any(entry["username"] == "owner-user" for entry in auth_provider.data.users)
|
||||
|
||||
|
||||
async def test_delete_requires_admin(
|
||||
|
||||
Reference in New Issue
Block a user