Check only_supervisor against the Supervisor user provided by hassio (#184041)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Stefan Agner
2026-10-02 17:39:09 +02:00
committed by GitHub
co-authored by Claude Fable 5.1
parent 8f017efd72
commit a3b4956d74
3 changed files with 57 additions and 6 deletions
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Any
import probatio
from homeassistant.const import HASSIO_USER_NAME
from homeassistant.components.http.const import DATA_SUPERVISOR_USER
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import Unauthorized
from homeassistant.helpers.typing import VolDictType
@@ -117,7 +117,10 @@ def ws_require_user(
output_error("only_inactive_user", "Not allowed as active user")
return None
if only_supervisor and connection.user.name != HASSIO_USER_NAME:
if only_supervisor and (
(supervisor_user := hass.data.get(DATA_SUPERVISOR_USER)) is None
or connection.user.id != supervisor_user.id
):
output_error("only_supervisor", "Only allowed as Supervisor")
return None
@@ -3,10 +3,15 @@
from typing import Any
import probatio
import pytest
from homeassistant.components import http, websocket_api
from homeassistant.const import HASSIO_USER_NAME
from homeassistant.core import HomeAssistant
from tests.common import MockUser
from tests.typing import MockHAClientWebSocket, WebSocketGenerator
async def test_async_response_request_context(
hass: HomeAssistant, websocket_client
@@ -156,8 +161,18 @@ async def test_async_response_request_context(
)
async def test_supervisor_only(hass: HomeAssistant, websocket_client) -> None:
"""Test that only the Supervisor can make requests."""
@pytest.mark.usefixtures("hass_supervisor_user")
async def test_supervisor_only(
hass: HomeAssistant,
websocket_client: MockHAClientWebSocket,
hass_admin_user: MockUser,
) -> None:
"""Test that only the Supervisor can make requests.
With the Supervisor user registered, an admin that is merely named like
the Supervisor user must still be rejected.
"""
await hass.auth.async_update_user(hass_admin_user, name=HASSIO_USER_NAME)
@websocket_api.ws_require_user(only_supervisor=True)
@websocket_api.websocket_command({"type": "test-require-supervisor-user"})
@@ -181,3 +196,29 @@ async def test_supervisor_only(hass: HomeAssistant, websocket_client) -> None:
assert msg["id"] == 5
assert not msg["success"]
assert msg["error"]["code"] == "only_supervisor"
async def test_supervisor_only_allows_supervisor(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
hass_supervisor_access_token: str,
) -> None:
"""Test that the Supervisor user can make Supervisor-only requests."""
@websocket_api.ws_require_user(only_supervisor=True)
@websocket_api.websocket_command({"type": "test-require-supervisor-user"})
def require_supervisor_request(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
connection.send_result(msg["id"])
websocket_api.async_register_command(hass, require_supervisor_request)
client = await hass_ws_client(hass, hass_supervisor_access_token)
await client.send_json({"id": 5, "type": "test-require-supervisor-user"})
msg = await client.receive_json()
assert msg["id"] == 5
assert msg["success"]
+9 -2
View File
@@ -60,6 +60,7 @@ from homeassistant.auth.const import GROUP_ID_ADMIN, GROUP_ID_READ_ONLY
from homeassistant.auth.models import Credentials
from homeassistant.auth.providers import homeassistant
from homeassistant.components.device_tracker.legacy import Device
from homeassistant.components.http.const import DATA_SUPERVISOR_USER
# pylint: disable-next=home-assistant-component-root-import
from homeassistant.components.websocket_api.auth import (
@@ -902,11 +903,17 @@ async def hass_read_only_access_token(
async def hass_supervisor_user(
hass: HomeAssistant, local_auth: homeassistant.HassAuthProvider
) -> MockUser:
"""Return the Home Assistant Supervisor user."""
"""Return the Home Assistant Supervisor user.
The user is published the same way the hassio integration does, so
commands restricted to the Supervisor accept it.
"""
admin_group = await hass.auth.async_get_group(GROUP_ID_ADMIN)
return MockUser(
user = MockUser(
name=HASSIO_USER_NAME, groups=[admin_group], system_generated=True
).add_to_hass(hass)
hass.data[DATA_SUPERVISOR_USER] = user
return user
@pytest.fixture