music-assistant-server

1"""Tests for the URL scheme, settings alerts and verify action driven by the SSL state."""
2
3from __future__ import annotations
4
5from datetime import timedelta
6from typing import TYPE_CHECKING, Any, cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from cryptography import x509
11from cryptography.hazmat.primitives import hashes, serialization
12from cryptography.hazmat.primitives.asymmetric import ec
13from cryptography.x509.oid import NameOID
14from music_assistant_models.config_entries import ConfigActionResult
15from music_assistant_models.enums import ConfigEntryType
16from music_assistant_models.errors import InvalidDataError
17
18from music_assistant.constants import WILDCARD_BIND_IPS
19from music_assistant.controllers.webserver.controller import (
20    CONF_ACTION_VERIFY_SSL,
21    WebserverController,
22)
23from music_assistant.helpers.datetime import utc
24
25if TYPE_CHECKING:
26    from pathlib import Path
27
28    from music_assistant_models.config_entries import CoreConfig
29
30
31@pytest.fixture(scope="module")
32def self_signed_cert() -> tuple[str, str]:
33    """Return a self-signed certificate and its private key, as PEM content."""
34    private_key = ec.generate_private_key(ec.SECP256R1())
35    subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
36    now = utc()
37    cert = (
38        x509.CertificateBuilder()
39        .subject_name(subject)
40        .issuer_name(subject)
41        .public_key(private_key.public_key())
42        .serial_number(x509.random_serial_number())
43        .not_valid_before(now - timedelta(days=1))
44        .not_valid_after(now + timedelta(days=1))
45        .sign(private_key, hashes.SHA256())
46    )
47    return (
48        cert.public_bytes(serialization.Encoding.PEM).decode(),
49        private_key.private_bytes(
50            encoding=serialization.Encoding.PEM,
51            format=serialization.PrivateFormat.PKCS8,
52            encryption_algorithm=serialization.NoEncryption(),
53        ).decode(),
54    )
55
56
57@pytest.fixture
58def mock_mass() -> MagicMock:
59    """Create a mock Music Assistant instance."""
60    # deliberate override of the package fixture: the base URL these tests exercise
61    # is also derived from the addon and provider state
62    mass = MagicMock()
63    mass.config.get_raw_core_config_value.return_value = "GLOBAL"
64    mass.running_as_hass_addon = False
65    mass.providers = []
66    return mass
67
68
69async def test_valid_certificate_advertises_tls(
70    mock_mass: MagicMock, tmp_path: Path, self_signed_cert: tuple[str, str]
71) -> None:
72    """Verify a usable certificate results in https URLs."""
73    certificate, private_key = self_signed_cert
74    webserver, server = await _setup_webserver(
75        mock_mass, tmp_path, certificate=certificate, private_key=private_key
76    )
77
78    assert webserver.base_url == "https://192.168.1.5:8095"
79    assert webserver.internal_base_url == "https://127.0.0.1:8095"
80    assert server.setup.await_args.kwargs["ssl_context"] is not None
81
82
83async def test_missing_certificate_advertises_plain_http(
84    mock_mass: MagicMock, tmp_path: Path
85) -> None:
86    """Verify the URLs follow the plain HTTP fallback when no certificate is configured."""
87    webserver, server = await _setup_webserver(mock_mass, tmp_path, certificate="", private_key="")
88
89    assert webserver.base_url == "http://192.168.1.5:8095"
90    assert webserver.internal_base_url == "http://127.0.0.1:8095"
91    assert server.setup.await_args.kwargs["ssl_context"] is None
92
93
94async def test_invalid_certificate_advertises_plain_http(
95    mock_mass: MagicMock, tmp_path: Path
96) -> None:
97    """Verify the URLs follow the plain HTTP fallback when the certificate is unusable."""
98    webserver, server = await _setup_webserver(
99        mock_mass, tmp_path, certificate="not a certificate", private_key="not a private key"
100    )
101
102    assert webserver.base_url == "http://192.168.1.5:8095"
103    assert webserver.internal_base_url == "http://127.0.0.1:8095"
104    assert server.setup.await_args.kwargs["ssl_context"] is None
105
106
107async def test_valid_certificate_shows_no_alert(
108    mock_mass: MagicMock, tmp_path: Path, self_signed_cert: tuple[str, str]
109) -> None:
110    """Verify a served HTTPS webserver is not flagged as unencrypted."""
111    certificate, private_key = self_signed_cert
112    webserver, _ = await _setup_webserver(
113        mock_mass, tmp_path, certificate=certificate, private_key=private_key
114    )
115
116    assert await _visible_alerts(webserver) == set()
117
118
119async def test_invalid_certificate_alerts_ssl_did_not_take_effect(
120    mock_mass: MagicMock, tmp_path: Path
121) -> None:
122    """Verify enabling SSL with an unusable certificate is reported as such."""
123    webserver, _ = await _setup_webserver(
124        mock_mass, tmp_path, certificate="not a certificate", private_key="not a private key"
125    )
126
127    assert await _visible_alerts(webserver) == {"ssl_inactive_warn"}
128
129
130async def test_ssl_disabled_alerts_the_webserver_is_unencrypted(
131    mock_mass: MagicMock, tmp_path: Path
132) -> None:
133    """Verify the generic warning is the only alert when SSL was never switched on."""
134    webserver, _ = await _setup_webserver(
135        mock_mass, tmp_path, certificate="", private_key="", enable_ssl=False
136    )
137
138    assert await _visible_alerts(webserver) == {"webserver_warn"}
139
140
141async def test_switching_ssl_off_drops_the_certificate_alert(
142    mock_mass: MagicMock, tmp_path: Path
143) -> None:
144    """Verify a reload onto the same controller reports the SSL state it was reloaded with."""
145    webserver, _ = await _setup_webserver(
146        mock_mass, tmp_path, certificate="not a certificate", private_key="not a private key"
147    )
148    assert await _visible_alerts(webserver) == {"ssl_inactive_warn"}
149
150    await _run_setup(webserver, tmp_path, certificate="", private_key="", enable_ssl=False)
151
152    assert await _visible_alerts(webserver) == {"webserver_warn"}
153
154
155async def test_verify_ssl_action_reports_the_certificate_info(
156    mock_mass: MagicMock, tmp_path: Path, self_signed_cert: tuple[str, str]
157) -> None:
158    """The verify action reports the certificate details as its outcome message."""
159    certificate, private_key = self_signed_cert
160    webserver, _ = await _setup_webserver(
161        mock_mass, tmp_path, certificate=certificate, private_key=private_key
162    )
163
164    result = await webserver.handle_config_action(CONF_ACTION_VERIFY_SSL)
165
166    assert isinstance(result, ConfigActionResult)
167    assert result.message is not None
168    assert "Certificate verification:" in result.message
169    assert "Key type: ECDSA" in result.message
170    assert "Subject: CN=localhost" in result.message
171    # computed details, so there is nothing to look up in a strings.json
172    assert result.translation_key is None
173
174
175async def test_verify_ssl_action_raises_for_an_unusable_certificate(
176    mock_mass: MagicMock, tmp_path: Path, self_signed_cert: tuple[str, str]
177) -> None:
178    """A certificate that cannot be verified is reported as a failure, not an outcome."""
179    certificate, _ = self_signed_cert
180    webserver, _ = await _setup_webserver(
181        mock_mass, tmp_path, certificate=certificate, private_key="not-a-key"
182    )
183
184    with pytest.raises(InvalidDataError) as err:
185        await webserver.handle_config_action(CONF_ACTION_VERIFY_SSL)
186
187    assert err.value.translation_key == "ssl_verification_failed"
188    assert err.value.translation_owner == "core.webserver"
189    # the reason fills the {0} placeholder so the user learns what is wrong
190    assert len(err.value.translation_args) == 1
191    assert err.value.translation_args[0]
192
193
194async def test_config_entries_hold_no_verify_result_label(
195    mock_mass: MagicMock, tmp_path: Path, self_signed_cert: tuple[str, str]
196) -> None:
197    """The config form carries the verify action itself, with no label to render a result into."""
198    certificate, private_key = self_signed_cert
199    webserver, _ = await _setup_webserver(
200        mock_mass, tmp_path, certificate=certificate, private_key=private_key
201    )
202
203    with patch(
204        "music_assistant.controllers.webserver.controller.get_ip_addresses",
205        AsyncMock(return_value=("192.168.1.5",)),
206    ):
207        entries = await webserver._build_config_entries()
208
209    keys = {entry.key for entry in entries}
210    assert CONF_ACTION_VERIFY_SSL in keys
211    assert "ssl_verify_result" not in keys
212
213
214def _make_server_mock() -> MagicMock:
215    """Create a Webserver double that adopts the address it is set up with."""
216    server = MagicMock()
217
218    async def _adopt_setup_args(**kwargs: Any) -> None:
219        server.port = kwargs["bind_port"]
220        bind_ip = kwargs["bind_ip"]
221        server.bind_ip = None if bind_ip in WILDCARD_BIND_IPS else bind_ip
222
223    server.setup = AsyncMock(side_effect=_adopt_setup_args)
224    return server
225
226
227async def _visible_alerts(webserver: WebserverController) -> set[str]:
228    """
229    Return the keys of the alert entries the settings UI renders for a controller.
230
231    :param webserver: The controller to read the config entries from.
232    """
233    with patch(
234        "music_assistant.controllers.webserver.controller.get_ip_addresses",
235        AsyncMock(return_value=("192.168.1.5",)),
236    ):
237        entries = await webserver._build_config_entries()
238    return {
239        entry.key for entry in entries if entry.type == ConfigEntryType.ALERT and not entry.hidden
240    }
241
242
243async def _setup_webserver(
244    mock_mass: MagicMock,
245    tmp_path: Path,
246    *,
247    certificate: str,
248    private_key: str,
249    enable_ssl: bool = True,
250) -> tuple[WebserverController, MagicMock]:
251    """
252    Run the real setup of a WebserverController.
253
254    :param mock_mass: Mocked Music Assistant instance to build the controller on.
255    :param tmp_path: Directory to serve as the frontend, in place of the bundled one.
256    :param certificate: Value for the ssl_certificate config option.
257    :param private_key: Value for the ssl_private_key config option.
258    :param enable_ssl: Value for the enable_ssl config option.
259    :return: The controller and the Webserver double it was set up against.
260    """
261    webserver = WebserverController(mock_mass)
262    server = _make_server_mock()
263    webserver._server = server
264    webserver.auth = MagicMock(setup=AsyncMock())
265    webserver.remote_access = MagicMock(setup=AsyncMock())
266    await _run_setup(
267        webserver,
268        tmp_path,
269        certificate=certificate,
270        private_key=private_key,
271        enable_ssl=enable_ssl,
272    )
273    return webserver, server
274
275
276async def _run_setup(
277    webserver: WebserverController,
278    tmp_path: Path,
279    *,
280    certificate: str,
281    private_key: str,
282    enable_ssl: bool,
283) -> None:
284    """
285    Set up a prepared controller against the given SSL config, as a (re)load does.
286
287    :param webserver: The prepared controller to set up.
288    :param tmp_path: Directory to serve as the frontend, in place of the bundled one.
289    :param certificate: Value for the ssl_certificate config option.
290    :param private_key: Value for the ssl_private_key config option.
291    :param enable_ssl: Value for the enable_ssl config option.
292    """
293    config_values: dict[str, Any] = {
294        "bind_port": 8095,
295        "bind_ip": None,
296        "enable_ssl": enable_ssl,
297        "ssl_certificate": certificate,
298        "ssl_private_key": private_key,
299    }
300    config = MagicMock()
301    config.get_value.side_effect = lambda key, default=None: config_values.get(key, default)
302
303    with (
304        patch(
305            "music_assistant.controllers.webserver.controller.get_publish_ip_candidates",
306            AsyncMock(return_value=("192.168.1.5",)),
307        ),
308        patch(
309            "music_assistant.controllers.webserver.controller.locate_frontend",
310            return_value=str(tmp_path),
311        ),
312    ):
313        await webserver.setup(cast("CoreConfig", config))
314