/
/
/
1"""Tests for websocket API command authorization."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.api import CommandMessage, ErrorResultMessage
10from music_assistant_models.auth import Scope, User, UserRole
11from music_assistant_models.errors import InsufficientPermissions
12
13from music_assistant.controllers.webserver.helpers.auth_middleware import (
14 get_current_client_id,
15 get_current_token,
16 get_current_user,
17 set_current_client_id,
18 set_current_token,
19 set_current_user,
20)
21from music_assistant.controllers.webserver.websocket_client import WebsocketClientHandler
22from music_assistant.helpers.api import APICommandHandler
23
24
25async def _noop_command() -> None:
26 """Test command target."""
27
28
29def _create_client(user_role: UserRole | None, handler: APICommandHandler) -> Any:
30 """Create a minimally wired websocket client handler for dispatch tests."""
31 client: Any = WebsocketClientHandler.__new__(WebsocketClientHandler)
32 client._logger = MagicMock()
33 client.mass = MagicMock()
34 client.mass.command_handlers = {handler.command: handler}
35 # close the coroutine passed to create_task to avoid "never awaited" warnings
36 client.mass.create_task = MagicMock(side_effect=lambda coro, *_: coro.close())
37 client._authenticated_user = (
38 User(user_id="user_1", username="tester", role=user_role) if user_role else None
39 )
40 client._current_token = "token" if user_role else None
41 client._sendspin_player_id = None
42 client.client_id = "test_client"
43 client._send_message = AsyncMock()
44 return client
45
46
47def _command_handler(
48 required_scope: Scope | None = None,
49 authenticated: bool = True,
50) -> APICommandHandler:
51 """Create an API command handler with the given auth requirements."""
52 return APICommandHandler.parse(
53 "test/protected",
54 _noop_command,
55 authenticated=authenticated,
56 required_scope=required_scope,
57 )
58
59
60def _sent_error_code(client: Any) -> str | None:
61 """Return the error code of the last sent error message, if any."""
62 for call in client._send_message.await_args_list:
63 message = call.args[0]
64 if isinstance(message, ErrorResultMessage):
65 return str(message.error_code)
66 return None
67
68
69PERMISSION_DENIED = str(InsufficientPermissions.error_code)
70
71
72@pytest.mark.asyncio
73async def test_user_scoped_command_rejects_guest() -> None:
74 """A guest must not be able to invoke commands gated on a user-only scope."""
75 client = _create_client(UserRole.GUEST, _command_handler(required_scope=Scope.USERS_INVITE))
76
77 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
78
79 assert _sent_error_code(client) == PERMISSION_DENIED
80 client.mass.create_task.assert_not_called()
81
82
83@pytest.mark.asyncio
84@pytest.mark.parametrize("role", [UserRole.USER, UserRole.ADMIN])
85async def test_user_scoped_command_allows_user_and_admin(role: UserRole) -> None:
86 """Users and admins hold the USERS_INVITE scope used by host commands."""
87 client = _create_client(role, _command_handler(required_scope=Scope.USERS_INVITE))
88
89 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
90
91 assert _sent_error_code(client) is None
92 client.mass.create_task.assert_called_once()
93
94
95@pytest.mark.asyncio
96@pytest.mark.parametrize("role", [UserRole.USER, UserRole.GUEST])
97async def test_admin_scoped_command_rejects_non_admin(role: UserRole) -> None:
98 """Only admins hold admin-only scopes such as USERS_MANAGE."""
99 client = _create_client(role, _command_handler(required_scope=Scope.USERS_MANAGE))
100
101 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
102
103 assert _sent_error_code(client) == PERMISSION_DENIED
104 client.mass.create_task.assert_not_called()
105
106
107@pytest.mark.asyncio
108@pytest.mark.parametrize("role", [UserRole.USER, UserRole.GUEST])
109async def test_users_read_command_rejects_non_service(role: UserRole) -> None:
110 """Reading user accounts is off limits for regular users and guests."""
111 client = _create_client(role, _command_handler(required_scope=Scope.USERS_READ))
112
113 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
114
115 assert _sent_error_code(client) == PERMISSION_DENIED
116 client.mass.create_task.assert_not_called()
117
118
119@pytest.mark.asyncio
120@pytest.mark.parametrize("role", [UserRole.SERVICE, UserRole.ADMIN])
121async def test_users_read_command_allows_service_and_admin(role: UserRole) -> None:
122 """The Home Assistant integration runs as a service account and may read user accounts."""
123 client = _create_client(role, _command_handler(required_scope=Scope.USERS_READ))
124
125 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
126
127 assert _sent_error_code(client) is None
128 client.mass.create_task.assert_called_once()
129
130
131@pytest.mark.asyncio
132async def test_authenticated_command_without_scope_allows_guest() -> None:
133 """Guests may invoke authenticated commands that require no specific scope."""
134 client = _create_client(UserRole.GUEST, _command_handler(required_scope=None))
135
136 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
137
138 assert _sent_error_code(client) is None
139 client.mass.create_task.assert_called_once()
140
141
142@pytest.mark.asyncio
143async def test_scoped_command_rejects_unauthenticated_socket() -> None:
144 """Without authentication a scoped command must not run at all."""
145 client = _create_client(None, _command_handler(required_scope=Scope.USERS_INVITE))
146
147 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
148
149 assert _sent_error_code(client) is not None
150 client.mass.create_task.assert_not_called()
151
152
153@pytest.mark.asyncio
154@pytest.mark.parametrize("authenticated", [True, False])
155async def test_command_sets_client_id_in_context(authenticated: bool) -> None:
156 """
157 Every dispatched command exposes the connection's client id, authenticated or not.
158
159 Unauthenticated handlers such as the join code exchange throttle per connection and
160 would otherwise see the id of whatever ran on this connection before them.
161
162 :param authenticated: Whether the dispatched command requires authentication.
163 """
164 set_current_client_id("stale_client")
165 role = UserRole.GUEST if authenticated else None
166 client = _create_client(role, _command_handler(authenticated=authenticated))
167
168 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
169
170 assert _sent_error_code(client) is None
171 assert get_current_client_id() == "test_client"
172
173
174@pytest.mark.asyncio
175async def test_unauthenticated_command_does_not_inherit_a_user() -> None:
176 """An unauthenticated command must not see the user of a command that ran before it."""
177 set_current_user(User(user_id="user_1", username="tester", role=UserRole.ADMIN))
178 set_current_token("stale_token")
179 client = _create_client(None, _command_handler(authenticated=False))
180
181 await client._handle_command(CommandMessage(message_id="1", command="test/protected"))
182
183 assert _sent_error_code(client) is None
184 assert get_current_user() is None
185 assert get_current_token() is None
186