music-assistant-server

1"""
2Shared helpers for Spotify Soloist, Spotify's official headless Connect client for Linux.
3
4Owned by the Spotify Connect provider and deliberately provider-neutral: the
5Spotify music provider reuses these helpers instead of shipping a second
6implementation. It manages a single shared install of the ``soloist`` binary
7under the server's storage dir and offers a typed client for the daemon's local
8WebSocket API (see https://developer.spotify.com/documentation/soloist).
9
10SECURITY NOTE: the soloist daemon takes the user's personal API key on its
11command line. This module never sees or logs that key, and callers that manage
12the daemon process must equally never log its argv.
13"""
14
15from __future__ import annotations
16
17import asyncio
18import hashlib
19import logging
20import platform
21import re
22import shutil
23import tarfile
24import tempfile
25import time
26from collections import deque
27from collections.abc import Awaitable, Callable
28from contextlib import suppress
29from dataclasses import dataclass, field
30from datetime import UTC, datetime
31from http import HTTPStatus
32from pathlib import Path, PurePosixPath
33from typing import TYPE_CHECKING, Any, Final
34
35import aiofiles
36from aiohttp import ClientError, ClientTimeout, ClientWebSocketResponse, ClientWSTimeout, WSMsgType
37from mashumaro import DataClassDictMixin
38from mashumaro.exceptions import InvalidFieldValue, MissingField
39from music_assistant_models.errors import MusicAssistantError
40from yarl import URL
41
42from music_assistant.constants import MASS_LOGGER_NAME
43from music_assistant.helpers.json import json_dumps, json_loads
44from music_assistant.helpers.process import check_output
45
46if TYPE_CHECKING:
47    from music_assistant.mass import MusicAssistant
48
49LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.providers.spotify_connect.soloist")
50
51# Official per-architecture release archives (docs: reference/downloads-and-updates).
52CDN_URL_TEMPLATE: Final[str] = "https://soloist-builds.spotifycdn.com/soloist_release_{arch}.tar.gz"
53
54# The daemon publishes its WebSocket endpoint as two small files in its data dir.
55WS_ADDR_FILE: Final[str] = "ws.addr"
56WS_PORT_FILE: Final[str] = "ws.port"
57
58# soloist exits with this code once its build passed the 90-day expiry.
59EXIT_CODE_BUILD_EXPIRED: Final[int] = 10
60
61# platform.machine() values mapped to the CDN artifact architecture;
62# anything else has no official build and is rejected.
63_MACHINE_TO_ARCH: Final[dict[str, str]] = {
64    "aarch64": "arm64",
65    "arm64": "arm64",
66    # the official arm32 build targets ARMv7; older ARM cores are unsupported
67    "armv7l": "arm32",
68    "armv8l": "arm32",
69    "x86_64": "x86_64",
70    "amd64": "x86_64",
71}
72
73# A download may redirect, but only within Spotify's own infrastructure.
74_TRUSTED_DOWNLOAD_DOMAINS: Final[tuple[str, ...]] = ("spotifycdn.com", "spotify.com")
75_MAX_REDIRECTS: Final[int] = 5
76_REDIRECT_STATUSES: Final[frozenset[int]] = frozenset({301, 302, 303, 307, 308})
77
78# Conservative caps so a misbehaving CDN response cannot fill the disk:
79# real release archives are only a few tens of MB.
80_MAX_ARCHIVE_SIZE: Final[int] = 64 * 1024 * 1024
81_MAX_EXTRACTED_SIZE: Final[int] = 128 * 1024 * 1024
82_DOWNLOAD_CHUNK_SIZE: Final[int] = 64 * 1024
83
84# Builds expire 90 days after their build date; start looking for a replacement
85# 14 days ahead of that so there is a comfortable update window.
86_BUILD_EXPIRY_SECONDS: Final[int] = 90 * 24 * 3600
87_REFRESH_AGE_SECONDS: Final[int] = 76 * 24 * 3600
88
89# ELF identity per architecture: (EI_CLASS, e_machine).
90_ELF_IDENT: Final[dict[str, tuple[int, int]]] = {
91    "arm64": (2, 0xB7),  # 64-bit, EM_AARCH64
92    "arm32": (1, 0x28),  # 32-bit, EM_ARM
93    "x86_64": (2, 0x3E),  # 64-bit, EM_X86_64
94}
95
96# Every SoloistBinaryManager instance manages the same shared install paths, so
97# the single-flight install lock is shared process-wide as well.
98_INSTALL_LOCK: Final = asyncio.Lock()
99
100# One successful ensure_fresh covers all provider instances starting together:
101# skip re-verifying the shared binary (a --version spawn plus a CDN update
102# check) when the last verification completed less than this long ago.
103_VERIFY_CACHE_SECONDS: Final[float] = 60.0
104_last_verified: float | None = None
105
106_VERSION_CMD_TIMEOUT: Final[float] = 10.0
107_DOWNLOAD_TIMEOUT: Final[float] = 300.0
108_HEAD_TIMEOUT: Final[float] = 30.0
109
110# --version output is free-form text (the docs give no schema); fish out a
111# version token and a build timestamp defensively. Observed 1.3.7 output:
112# "soloist 1.3.7.345 build 1787077868 (20260818) (gb24005ef46) (linux/aarch64)"
113# — the build timestamp is a unix epoch; an ISO-like date is kept as fallback.
114_VERSION_TOKEN_RE: Final[re.Pattern[str]] = re.compile(r"\bv?(\d+\.\d+(?:\.\d+)*)\b")
115_BUILD_EPOCH_RE: Final[re.Pattern[str]] = re.compile(r"\bbuild\s+(\d{9,11})\b")
116_TIMESTAMP_RE: Final[re.Pattern[str]] = re.compile(
117    r"\b(\d{4}-\d{2}-\d{2}(?:[T ]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?)\b"
118)
119
120# The docs do not guarantee when the endpoint files appear, so poll for them.
121_ENDPOINT_POLL_INTERVAL: Final[float] = 0.25
122_WS_HEARTBEAT: Final[float] = 30.0
123_COMMAND_RESULT_TIMEOUT: Final[float] = 10.0
124
125
126class SoloistError(MusicAssistantError):
127    """Base error for all soloist helper failures."""
128
129    # the soloist strings are authored once, in the spotify_connect provider's
130    # strings.json, regardless of which provider raised the error
131    translation_owner = "provider.spotify_connect"
132
133
134class ConsentRequiredError(SoloistError):
135    """A download from Spotify's CDN is needed but the user did not consent (yet)."""
136
137    translation_key = "soloist_consent_required"
138
139
140class UnsupportedPlatformError(SoloistError):
141    """No official soloist build exists for this platform/architecture."""
142
143    translation_key = "soloist_unsupported_platform"
144
145
146class DownloadFailedError(SoloistError):
147    """The soloist release archive could not be downloaded."""
148
149    translation_key = "soloist_download_failed"
150
151
152class InvalidArchiveError(SoloistError):
153    """The downloaded release archive or its binary failed validation."""
154
155    translation_key = "soloist_invalid_archive"
156
157
158class BuildExpiredError(SoloistError):
159    """The soloist build passed its 90-day expiry and no replacement is available."""
160
161    translation_key = "soloist_build_expired"
162
163
164@dataclass
165class SoloistEntity(DataClassDictMixin):
166    """A Spotify entity (track/album/playlist/...) as used in item/context fields."""
167
168    uri: str
169    entity_type: str
170    # decorations is an extensible bag (identity.name, playback.duration_ms, ...)
171    decorations: dict[str, Any] = field(default_factory=dict)
172
173
174@dataclass
175class SoloistPosition(DataClassDictMixin):
176    """Playback position reference point, to be interpolated with speed over time."""
177
178    position_ms: int
179    timestamp_ms: int
180    speed: float = 1.0
181
182
183@dataclass
184class SoloistPlaybackOptions(DataClassDictMixin):
185    """Playback options (shuffle/repeat/speed)."""
186
187    shuffle: bool = False
188    repeat: str = "off"
189    playback_speed: float = 1.0
190    modes: dict[str, Any] = field(default_factory=dict)
191
192
193@dataclass
194class SoloistQueueEntry(DataClassDictMixin):
195    """One entry in the previous/upcoming queue listing."""
196
197    uid: str
198    source: str
199    item: SoloistEntity | None = None
200
201
202@dataclass
203class SoloistAuthState(DataClassDictMixin):
204    """Payload of the ``auth_state`` event."""
205
206    logged_in: bool
207    is_active: bool
208    device_name: str | None = None
209
210
211@dataclass
212class SoloistPlaybackState(DataClassDictMixin):
213    """Payload of the ``playback_state`` snapshot and the ``playback_changed`` delta."""
214
215    status: str
216    item: SoloistEntity | None = None
217    context: SoloistEntity | None = None
218    position: SoloistPosition | None = None
219    volume: int | None = None
220    is_active: bool | None = None
221    options: SoloistPlaybackOptions | None = None
222    available_actions: dict[str, Any] = field(default_factory=dict)
223
224
225@dataclass
226class SoloistTrackChanged(DataClassDictMixin):
227    """Payload of the ``track_changed`` event."""
228
229    item: SoloistEntity | None = None
230
231
232@dataclass
233class SoloistPositionSync(DataClassDictMixin):
234    """Payload of the ``position_sync`` event."""
235
236    position: SoloistPosition
237
238
239@dataclass
240class SoloistVolumeChanged(DataClassDictMixin):
241    """Payload of the ``volume_changed`` event."""
242
243    volume: int
244
245
246@dataclass
247class SoloistDeviceChanged(DataClassDictMixin):
248    """Payload of the ``device_changed`` event."""
249
250    is_active: bool
251
252
253@dataclass
254class SoloistContextChanged(DataClassDictMixin):
255    """Payload of the ``context_changed`` event."""
256
257    context: SoloistEntity | None = None
258
259
260@dataclass
261class SoloistOptionsChanged(DataClassDictMixin):
262    """Payload of the ``options_changed`` event."""
263
264    options: SoloistPlaybackOptions
265
266
267@dataclass
268class SoloistQueueChanged(DataClassDictMixin):
269    """Payload of the ``queue_changed`` event."""
270
271    previous: list[SoloistQueueEntry] = field(default_factory=list)
272    upcoming: list[SoloistQueueEntry] = field(default_factory=list)
273
274
275@dataclass
276class SoloistCommandResult(DataClassDictMixin):
277    """Payload of the ``command_result`` acknowledgement event."""
278
279    command: str
280
281
282@dataclass
283class SoloistErrorMessage(DataClassDictMixin):
284    """Payload of the ``error`` event."""
285
286    message: str
287
288
289SoloistEventData = (
290    SoloistAuthState
291    | SoloistPlaybackState
292    | SoloistTrackChanged
293    | SoloistPositionSync
294    | SoloistVolumeChanged
295    | SoloistDeviceChanged
296    | SoloistContextChanged
297    | SoloistOptionsChanged
298    | SoloistQueueChanged
299    | SoloistCommandResult
300    | SoloistErrorMessage
301)
302
303# Maps documented event types to their payload model;
304# unrecognized types are passed through as generic raw events.
305_EVENT_MODELS: Final[dict[str, type[SoloistEventData]]] = {
306    "auth_state": SoloistAuthState,
307    "playback_state": SoloistPlaybackState,
308    "playback_changed": SoloistPlaybackState,
309    "track_changed": SoloistTrackChanged,
310    "position_sync": SoloistPositionSync,
311    "volume_changed": SoloistVolumeChanged,
312    "device_changed": SoloistDeviceChanged,
313    "context_changed": SoloistContextChanged,
314    "options_changed": SoloistOptionsChanged,
315    "queue_changed": SoloistQueueChanged,
316    "command_result": SoloistCommandResult,
317    "error": SoloistErrorMessage,
318}
319
320
321@dataclass
322class SoloistEvent:
323    """A single event received from the daemon, with its decoded payload when recognized."""
324
325    type: str
326    data: SoloistEventData | None
327    raw: dict[str, Any]
328
329
330# Called with the decoded event for every WebSocket event.
331EventCallback = Callable[[SoloistEvent], Awaitable[None]]
332
333
334class SoloistBinaryManager:
335    """
336    Manages the single shared soloist binary install for all consumers.
337
338    The install lives under ``<storage_path>/soloist``; concurrent callers
339    share one download.
340    """
341
342    def __init__(self, mass: MusicAssistant) -> None:
343        """
344        Initialize the binary manager.
345
346        :param mass: The MusicAssistant instance (for its HTTP session and storage path).
347        """
348        self.mass = mass
349        self._install_dir = Path(mass.storage_path) / "soloist"
350        self._binary_path = self._install_dir / "soloist"
351        self._previous_path = self._install_dir / "soloist.prev"
352        self._metadata_path = self._install_dir / "soloist.meta.json"
353
354    @property
355    def binary_path(self) -> Path:
356        """Path where the soloist binary is (or will be) installed."""
357        return self._binary_path
358
359    async def ensure_binary(self, consent: bool) -> Path:
360        """
361        Return the path to a validated soloist binary, downloading it when needed.
362
363        :param consent: Whether the user consented to downloading the binary from
364            Spotify's CDN. An already-installed valid binary is returned without
365            any network access, regardless of this flag.
366        :raises ConsentRequiredError: A download is needed but consent was not given.
367        :raises UnsupportedPlatformError: No soloist build exists for this platform.
368        :raises DownloadFailedError: The release archive could not be downloaded.
369        :raises InvalidArchiveError: The downloaded archive or binary failed validation.
370        :raises BuildExpiredError: The freshly downloaded build has already expired.
371        """
372        arch = _resolve_architecture()
373        async with _INSTALL_LOCK:
374            if await self._installed_returncode() == 0 and not self._installed_expired():
375                return self._binary_path
376            await self._install_with_consent(consent, arch)
377            return self._binary_path
378
379    async def ensure_fresh(self, consent: bool, *, force: bool = False) -> Path:
380        """
381        Return a validated soloist binary, refreshing it when it is (close to) expiry.
382
383        Builds expire 90 days after their build date: an installed build that
384        already expired is replaced immediately, one nearing expiry only when
385        the CDN offers a different build. When the refresh fails (e.g. offline)
386        a still-valid binary is returned with a warning instead.
387
388        :param consent: Whether the user consented to downloading from Spotify's
389            CDN. Without consent a still-valid installed binary is returned
390            as-is (no proactive refresh).
391        :param force: Re-verify even when another caller just did — required
392            when the daemon itself reported the build expired (exit code 10).
393        :raises ConsentRequiredError: A download is needed but consent was not given.
394        :raises UnsupportedPlatformError: No soloist build exists for this platform.
395        :raises DownloadFailedError: No usable binary is installed and the download failed.
396        :raises InvalidArchiveError: The downloaded archive or binary failed validation.
397        :raises BuildExpiredError: The installed build expired and no valid
398            replacement could be obtained.
399        """
400        global _last_verified  # noqa: PLW0603
401        arch = _resolve_architecture()
402        async with _INSTALL_LOCK:
403            # provider instances starting together share one shared install;
404            # skip re-verifying when another caller just did successfully
405            if (
406                not force
407                and _last_verified is not None
408                and time.monotonic() - _last_verified < _VERIFY_CACHE_SECONDS
409                and self._binary_path.is_file()
410            ):
411                return self._binary_path
412            binary = await self._ensure_fresh_locked(consent, arch)
413            _last_verified = time.monotonic()
414            return binary
415
416    def diagnostics(self) -> dict[str, Any]:
417        """
418        Return diagnostic details about the installed binary.
419
420        Contains install/build metadata only; never any key or session material.
421        """
422        metadata = self._read_metadata()
423        if metadata is None:
424            return {"installed": False}
425        return {
426            "installed": True,
427            "sha256": metadata.sha256,
428            "etag": metadata.etag,
429            "version": metadata.version,
430            "version_raw": metadata.version_raw,
431            "installed_at": metadata.installed_at,
432            "build_timestamp": metadata.build_timestamp,
433            # upper-bound estimate when the build timestamp could not be parsed
434            "expires_at": (metadata.build_timestamp or metadata.installed_at)
435            + _BUILD_EXPIRY_SECONDS,
436        }
437
438    async def _ensure_fresh_locked(self, consent: bool, arch: str) -> Path:
439        """Verify/refresh the installed binary (see ensure_fresh; runs under _INSTALL_LOCK)."""
440        returncode = await self._installed_returncode()
441        if returncode not in (0, EXIT_CODE_BUILD_EXPIRED):
442            # missing or broken install: plain (re)install
443            await self._install_with_consent(consent, arch)
444            return self._binary_path
445        expired = returncode == EXIT_CODE_BUILD_EXPIRED or self._installed_expired()
446        if not expired and (not consent or not self._due_for_update()):
447            return self._binary_path
448        if expired and not consent:
449            raise ConsentRequiredError("Updating the expired soloist binary requires user consent")
450        if not expired and not await self._update_available(arch):
451            return self._binary_path
452        try:
453            await self._download_and_install(arch)
454        except SoloistError as err:
455            if expired:
456                if isinstance(err, DownloadFailedError):
457                    raise BuildExpiredError(
458                        "soloist build expired and no replacement could be downloaded"
459                    ) from err
460                raise
461            LOGGER.warning("soloist update failed, keeping the current binary: %s", err)
462        return self._binary_path
463
464    async def _install_with_consent(self, consent: bool, arch: str) -> None:
465        """Download and install the binary, requiring user consent first."""
466        if not consent:
467            raise ConsentRequiredError("Downloading the soloist binary requires user consent")
468        await self._download_and_install(arch)
469
470    async def _installed_returncode(self) -> int | None:
471        """Return the ``--version`` exit code of the installed binary, or None if unrunnable."""
472        if not self._binary_path.is_file():
473            return None
474        try:
475            returncode, _ = await check_output(
476                str(self._binary_path), "--version", timeout=_VERSION_CMD_TIMEOUT
477            )
478        except OSError, TimeoutError:
479            return None
480        return returncode
481
482    def _installed_expired(self) -> bool:
483        """
484        Return whether the installed build's known build timestamp has expired.
485
486        Defense in depth next to the exit-code check: whether ``--version``
487        itself reports expiry is not documented, so an expired-by-timestamp
488        build is refused even when the binary still runs. The install time is
489        the fallback anchor — the build is at least as old as its install.
490        """
491        metadata = self._read_metadata()
492        return metadata is not None and _build_expired(
493            metadata.build_timestamp or metadata.installed_at
494        )
495
496    def _due_for_update(self) -> bool:
497        """Return whether the installed build is old enough to look for an update."""
498        metadata = self._read_metadata()
499        if metadata is None:
500            return True
501        anchor = metadata.build_timestamp or metadata.installed_at
502        return (time.time() - anchor) >= _REFRESH_AGE_SECONDS
503
504    async def _update_available(self, arch: str) -> bool:
505        """Return whether the CDN offers a different build than the installed one."""
506        metadata = self._read_metadata()
507        try:
508            remote_etag = await self._fetch_remote_etag(arch)
509        except (ClientError, TimeoutError, OSError, DownloadFailedError) as err:
510            # any failed update check keeps the current (still valid) build
511            LOGGER.warning("Unable to check for a soloist update: %s", err)
512            return False
513        if remote_etag is None:
514            # CDN unreachable/erroring: keep the current build
515            return False
516        if not remote_etag:
517            # reachable but no validator offered: assume an update exists and
518            # let the validated download path decide
519            return True
520        return metadata is None or metadata.etag != remote_etag
521
522    async def _fetch_remote_etag(self, arch: str) -> str | None:
523        """Return the ETag the CDN currently serves for the given architecture, if any."""
524        current_url = CDN_URL_TEMPLATE.format(arch=arch)
525        # follow redirects manually so every hop passes the same host allowlist
526        # the download path enforces
527        for _ in range(_MAX_REDIRECTS + 1):
528            _validate_download_url(current_url)
529            async with self.mass.http_session.head(
530                current_url, allow_redirects=False, timeout=ClientTimeout(total=_HEAD_TIMEOUT)
531            ) as resp:
532                if resp.status in _REDIRECT_STATUSES:
533                    if not (location := resp.headers.get("Location")):
534                        return None
535                    current_url = str(URL(current_url).join(URL(location)))
536                    continue
537                if resp.status != HTTPStatus.OK:
538                    return None
539                # empty string = reachable but no validator offered
540                return resp.headers.get("ETag", "")
541        return None
542
543    async def _download_and_install(self, arch: str) -> None:
544        """Download, validate and atomically install the binary for the given architecture."""
545        global _last_verified  # noqa: PLW0603
546        # a (re)install invalidates any recent-verification claim about the
547        # previous binary
548        _last_verified = None
549        url = CDN_URL_TEMPLATE.format(arch=arch)
550        try:
551            await asyncio.to_thread(self._install_dir.mkdir, parents=True, exist_ok=True)
552            temp_dir = Path(
553                await asyncio.to_thread(tempfile.mkdtemp, prefix=".soloist-", dir=self._install_dir)
554            )
555        except OSError as err:
556            raise DownloadFailedError(f"cannot prepare the soloist install dir: {err}") from err
557        try:
558            archive_path = temp_dir / "soloist.tar.gz"
559            new_binary = temp_dir / "soloist"
560            sha256, etag = await self._download_archive(url, archive_path)
561            await asyncio.to_thread(_extract_binary_from_archive, archive_path, new_binary)
562            await asyncio.to_thread(_validate_elf_header, new_binary, arch)
563            version_raw = await self._swap_in_and_validate(new_binary)
564            metadata = _BinaryMetadata(
565                sha256=sha256,
566                version_raw=version_raw,
567                installed_at=time.time(),
568                etag=etag,
569                version=_parse_version_token(version_raw),
570                build_timestamp=_parse_build_timestamp(version_raw),
571            )
572            # metadata is advisory (diagnostics + refresh hints): the binary is
573            # already validated and installed, so never fail the install over it
574            try:
575                await asyncio.to_thread(self._write_metadata, metadata)
576            except OSError as err:
577                LOGGER.warning("Unable to persist soloist install metadata: %s", err)
578                # stale metadata must not outlive the swap: it would pair the
579                # fresh binary with the previous build's expiry state
580                with suppress(OSError):
581                    await asyncio.to_thread(self._metadata_path.unlink, missing_ok=True)
582            LOGGER.info("Installed soloist binary %s (%s)", metadata.version or "unknown", arch)
583        finally:
584            await asyncio.to_thread(shutil.rmtree, temp_dir, ignore_errors=True)
585
586    async def _download_archive(self, url: str, dest: Path) -> tuple[str, str | None]:
587        """
588        Stream the release archive to a local file, following (validated) redirects.
589
590        :return: Tuple of the archive's sha256 hexdigest and its ETag header (if any).
591        """
592        hasher = hashlib.sha256()
593        current_url = url
594        try:
595            for _ in range(_MAX_REDIRECTS + 1):
596                _validate_download_url(current_url)
597                async with self.mass.http_session.get(
598                    current_url,
599                    allow_redirects=False,
600                    timeout=ClientTimeout(total=_DOWNLOAD_TIMEOUT),
601                ) as resp:
602                    if resp.status in _REDIRECT_STATUSES:
603                        location = resp.headers.get("Location")
604                        if not location:
605                            raise DownloadFailedError("soloist download redirect has no location")
606                        current_url = str(URL(current_url).join(URL(location)))
607                        continue
608                    if resp.status != HTTPStatus.OK:
609                        raise DownloadFailedError(
610                            f"soloist download failed with HTTP {resp.status}"
611                        )
612                    etag = resp.headers.get("ETag")
613                    size = 0
614                    async with aiofiles.open(dest, "wb") as _file:
615                        async for chunk in resp.content.iter_chunked(_DOWNLOAD_CHUNK_SIZE):
616                            size += len(chunk)
617                            if size > _MAX_ARCHIVE_SIZE:
618                                raise DownloadFailedError("soloist archive exceeds the size limit")
619                            hasher.update(chunk)
620                            await _file.write(chunk)
621                    return (hasher.hexdigest(), etag)
622            raise DownloadFailedError("too many redirects while downloading soloist")
623        except (ClientError, TimeoutError, OSError) as err:
624            raise DownloadFailedError(f"soloist download failed: {err}") from err
625
626    async def _swap_in_and_validate(self, new_binary: Path) -> str:
627        """
628        Atomically install the new binary and verify it actually runs.
629
630        Shielded from cancellation so the install always reaches a consistent
631        end state (committed or rolled back) even when the caller goes away.
632
633        :return: The raw ``--version`` output of the installed binary.
634        """
635        inner = asyncio.ensure_future(self._swap_in_and_validate_inner(new_binary))
636        try:
637            return await asyncio.shield(inner)
638        except asyncio.CancelledError:
639            # wait for the shielded install to reach its consistent end state
640            # (committed or rolled back) so the install lock and temp dir stay
641            # held until the shared install is no longer being mutated
642            while not inner.done():
643                with suppress(asyncio.CancelledError):
644                    await asyncio.shield(inner)
645            _consume_install_result(inner)
646            raise
647
648    async def _swap_in_and_validate_inner(self, new_binary: Path) -> str:
649        """Perform the swap/validate/rollback sequence (see _swap_in_and_validate)."""
650
651        def _swap_in() -> None:
652            new_binary.chmod(0o755)
653            if self._binary_path.exists():
654                self._binary_path.replace(self._previous_path)
655            new_binary.replace(self._binary_path)
656
657        def _rollback() -> None:
658            if self._previous_path.exists():
659                self._previous_path.replace(self._binary_path)
660            else:
661                self._binary_path.unlink(missing_ok=True)
662
663        def _commit() -> None:
664            self._previous_path.unlink(missing_ok=True)
665
666        try:
667            await asyncio.to_thread(_swap_in)
668        except OSError as err:
669            # keep SoloistError semantics so ensure_fresh's keep-the-old-binary
670            # fallback also covers filesystem failures during the swap
671            with suppress(OSError):
672                await asyncio.to_thread(_rollback)
673            raise InvalidArchiveError(f"failed to install the soloist binary: {err}") from err
674        try:
675            returncode, output = await check_output(
676                str(self._binary_path), "--version", timeout=_VERSION_CMD_TIMEOUT
677            )
678        except (OSError, TimeoutError) as err:
679            with suppress(OSError):
680                await asyncio.to_thread(_rollback)
681            raise InvalidArchiveError(f"soloist binary failed to run: {err}") from err
682        if returncode == EXIT_CODE_BUILD_EXPIRED:
683            with suppress(OSError):
684                await asyncio.to_thread(_rollback)
685            raise BuildExpiredError("the downloaded soloist build has already expired")
686        if returncode != 0:
687            with suppress(OSError):
688                await asyncio.to_thread(_rollback)
689            raise InvalidArchiveError(f"soloist --version exited with code {returncode}")
690        version_raw = output.decode("utf-8", errors="replace").strip()
691        if _build_expired(_parse_build_timestamp(version_raw)):
692            with suppress(OSError):
693                await asyncio.to_thread(_rollback)
694            raise BuildExpiredError("the downloaded soloist build has already expired")
695        await asyncio.to_thread(_commit)
696        return version_raw
697
698    def _read_metadata(self) -> _BinaryMetadata | None:
699        """Return the persisted install metadata, if present and readable."""
700        try:
701            raw = json_loads(self._metadata_path.read_text(encoding="utf-8"))
702            if not isinstance(raw, dict):
703                return None
704            return _BinaryMetadata.from_dict(raw)
705        except OSError, ValueError, TypeError, MissingField, InvalidFieldValue:
706            return None
707
708    def _write_metadata(self, metadata: _BinaryMetadata) -> None:
709        """Persist install metadata next to the binary."""
710        self._metadata_path.write_text(json_dumps(metadata.to_dict()), encoding="utf-8")
711
712
713class SoloistClient:
714    """
715    Client for the local WebSocket API of a running soloist daemon.
716
717    The daemon publishes its (loopback) endpoint as ``ws.addr``/``ws.port``
718    files in its data directory; this client discovers it from there. Commands
719    are sent over the events connection, so :meth:`listen_events` must be
720    running for the command senders to work.
721    """
722
723    def __init__(self, mass: MusicAssistant, data_dir: Path, logger: logging.Logger) -> None:
724        """
725        Initialize the client.
726
727        :param mass: The MusicAssistant instance (for its shared HTTP session).
728        :param data_dir: The daemon's data directory (holds the endpoint files).
729        :param logger: Logger to use for diagnostics.
730        """
731        self.mass = mass
732        self.data_dir = data_dir
733        self.logger = logger
734        self._ws: ClientWebSocketResponse | None = None
735        self._pending_results: dict[str, deque[asyncio.Future[None]]] = {}
736
737    @property
738    def connected(self) -> bool:
739        """Whether the events WebSocket is currently connected."""
740        return self._ws is not None and not self._ws.closed
741
742    async def wait_until_ready(self, timeout: float = 30.0) -> bool:
743        """
744        Poll the daemon's data directory until the WebSocket endpoint is published.
745
746        :param timeout: Maximum seconds to wait for the endpoint files.
747        :return: True once the endpoint is known, False if the timeout elapses.
748        """
749        loop = asyncio.get_running_loop()
750        deadline = loop.time() + timeout
751        while True:
752            if await asyncio.to_thread(self._read_endpoint) is not None:
753                return True
754            if loop.time() >= deadline:
755                return False
756            await asyncio.sleep(_ENDPOINT_POLL_INTERVAL)
757
758    async def listen_events(self, on_event: EventCallback) -> None:
759        """
760        Connect to the daemon's WebSocket API and dispatch events until it closes.
761
762        Returns normally when the connection is closed by either side; raises on
763        connection errors so the caller can implement a reconnect loop.
764        Malformed or unrecognized events are tolerated and never interrupt the
765        stream.
766
767        :param on_event: Coroutine called with a :class:`SoloistEvent` per event.
768        :raises SoloistError: The daemon has not published its endpoint (yet).
769        """
770        if self._ws is not None and not self._ws.closed:
771            # pending acks are correlated per connection; a second concurrent
772            # listener would cross-resolve them
773            raise SoloistError("listen_events is already running")
774        if (endpoint := await asyncio.to_thread(self._read_endpoint)) is None:
775            raise SoloistError("soloist has not published its WebSocket endpoint (yet)")
776        addr, port = endpoint
777        host = f"[{addr}]" if ":" in addr else addr
778        ws: ClientWebSocketResponse | None = None
779        try:
780            async with self.mass.http_session.ws_connect(
781                f"ws://{host}:{port}/",
782                heartbeat=_WS_HEARTBEAT,
783                # the daemon is often already gone when this closes (a
784                # single-track run exits at its item's end); the default 10s
785                # close-handshake wait would stall every teardown by that much
786                timeout=ClientWSTimeout(ws_close=1.0),
787            ) as ws:
788                self._ws = ws
789                self.logger.debug("Connected to the soloist websocket at %s:%s", addr, port)
790                async for msg in ws:
791                    if msg.type == WSMsgType.TEXT:
792                        await self._handle_message(msg.data, on_event)
793                    elif msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED):
794                        break
795                    elif msg.type == WSMsgType.ERROR:
796                        raise ws.exception() or ClientError("websocket error")
797        finally:
798            # only tear down our own connection state: a reconnecting caller may
799            # already have a newer listen_events connection registered
800            if ws is not None and self._ws is ws:
801                self._ws = None
802                self._fail_pending_results()
803
804    async def play(self, uri: str | None = None, *, await_result: bool = False) -> None:
805        """
806        Start playback of a Spotify URI/context, or resume the current one.
807
808        :param uri: Spotify URI (track/album/playlist/...); omit to resume.
809        :param await_result: Wait for the daemon's command acknowledgement.
810        """
811        fields: dict[str, Any] = {"uri": uri} if uri is not None else {}
812        await self._send_command("play", await_result=await_result, **fields)
813
814    async def resume(self, *, await_result: bool = False) -> None:
815        """Resume playback of the current context."""
816        await self._send_command("play", await_result=await_result)
817
818    async def pause(self, *, await_result: bool = False) -> None:
819        """Pause playback."""
820        await self._send_command("pause", await_result=await_result)
821
822    async def skip_next(self, *, await_result: bool = False) -> None:
823        """Skip to the next track."""
824        await self._send_command("skip_next", await_result=await_result)
825
826    async def skip_prev(self, *, await_result: bool = False) -> None:
827        """Skip to the previous track (or rewind the current one)."""
828        await self._send_command("skip_prev", await_result=await_result)
829
830    async def seek(self, position_ms: int, *, await_result: bool = False) -> None:
831        """
832        Seek to an absolute position in the current track.
833
834        :param position_ms: Target position in milliseconds.
835        """
836        await self._send_command("seek", await_result=await_result, position_ms=max(0, position_ms))
837
838    async def set_volume(self, volume: int, *, await_result: bool = False) -> None:
839        """
840        Set the absolute playback volume.
841
842        :param volume: Volume from 0 to 100.
843        """
844        await self._send_command(
845            "set_volume", await_result=await_result, volume=max(0, min(100, volume))
846        )
847
848    async def activate(self, *, await_result: bool = False) -> None:
849        """Make this daemon the active Spotify Connect device."""
850        await self._send_command("activate", await_result=await_result)
851
852    async def deactivate(self, *, await_result: bool = False) -> None:
853        """Release this daemon as the active Spotify Connect device."""
854        await self._send_command("deactivate", await_result=await_result)
855
856    async def set_shuffle(self, enabled: bool, *, await_result: bool = False) -> None:
857        """Enable or disable shuffle."""
858        await self._send_command("set_shuffle", await_result=await_result, enabled=enabled)
859
860    async def set_repeat_context(self, enabled: bool, *, await_result: bool = False) -> None:
861        """Enable or disable repeating the current context."""
862        await self._send_command("set_repeat_context", await_result=await_result, enabled=enabled)
863
864    async def set_repeat_track(self, enabled: bool, *, await_result: bool = False) -> None:
865        """Enable or disable repeating the current track."""
866        await self._send_command("set_repeat_track", await_result=await_result, enabled=enabled)
867
868    async def add_to_queue(self, uri: str, *, await_result: bool = False) -> None:
869        """
870        Add a track to the play queue.
871
872        :param uri: Spotify track URI.
873        """
874        await self._send_command("add_to_queue", await_result=await_result, uri=uri)
875
876    async def get_state(self) -> None:
877        """Request a full ``playback_state`` snapshot (answered as an event)."""
878        await self._send_command("get_state")
879
880    async def get_auth_state(self) -> None:
881        """Request the current ``auth_state`` (answered as an event)."""
882        await self._send_command("get_auth_state")
883
884    async def get_queue(self, limit: int = 10) -> None:
885        """
886        Request the play queue (answered as a ``queue_changed`` event).
887
888        :param limit: Maximum number of upcoming tracks to return.
889        """
890        await self._send_command("get_queue", limit=limit)
891
892    async def _send_command(
893        self,
894        command: str,
895        *,
896        await_result: bool = False,
897        timeout: float = _COMMAND_RESULT_TIMEOUT,
898        **fields: Any,
899    ) -> None:
900        """
901        Send a command frame, optionally waiting for its ``command_result`` ack.
902
903        Acks carry only the command name (no request id), so concurrent calls of
904        the same command resolve in FIFO order with no per-call correlation, and
905        an ack for a fire-and-forget send of the same command can resolve an
906        awaited call early — avoid mixing both styles for one command.
907        Never wait for an ack from inside an ``on_event`` callback: the ack is
908        delivered by the same event loop, so the wait can only time out.
909
910        :raises SoloistError: The WebSocket is not connected (or closed while waiting).
911        :raises TimeoutError: The acknowledgement did not arrive within the timeout.
912        """
913        ws = self._ws
914        if ws is None or ws.closed:
915            raise SoloistError("soloist websocket is not connected")
916        future: asyncio.Future[None] | None = None
917        if await_result:
918            future = asyncio.get_running_loop().create_future()
919            self._pending_results.setdefault(command, deque()).append(future)
920        try:
921            await ws.send_json({"type": "command", "command": command, **fields})
922            if future is not None:
923                await asyncio.wait_for(future, timeout)
924        finally:
925            if future is not None:
926                self._discard_pending_result(command, future)
927
928    async def _handle_message(self, raw: str, on_event: EventCallback) -> None:
929        """Decode a single WebSocket text frame and dispatch it to the callback."""
930        try:
931            payload = json_loads(raw)
932        except ValueError:
933            self.logger.debug("Ignoring non-JSON websocket message: %s", raw)
934            return
935        if not isinstance(payload, dict) or not isinstance(event_type := payload.get("type"), str):
936            self.logger.debug("Ignoring websocket message without an event type: %s", raw)
937            return
938        data: SoloistEventData | None = None
939        if (model := _EVENT_MODELS.get(event_type)) is not None:
940            try:
941                data = model.from_dict(payload)
942            except Exception as err:
943                # tolerance to malformed events is a hard requirement: whatever
944                # the decode error, log it and keep the stream alive
945                self.logger.debug("Ignoring malformed %s event: %s (%s)", event_type, raw, err)
946                return
947            if event_type == "command_result" and isinstance(data, SoloistCommandResult):
948                self._resolve_pending_result(data.command)
949        await on_event(SoloistEvent(type=event_type, data=data, raw=payload))
950
951    def _read_endpoint(self) -> tuple[str, int] | None:
952        """Read the WebSocket endpoint published in the daemon's data directory."""
953        try:
954            addr = (self.data_dir / WS_ADDR_FILE).read_text(encoding="utf-8").strip()
955            port = int((self.data_dir / WS_PORT_FILE).read_text(encoding="utf-8").strip())
956        except OSError, ValueError:
957            return None
958        if not addr or not 0 < port <= 65535:
959            return None
960        return (addr, port)
961
962    def _resolve_pending_result(self, command: str) -> None:
963        """Resolve the oldest result waiter for the given command, if any."""
964        if (queue := self._pending_results.get(command)) is None:
965            return
966        while queue:
967            future = queue.popleft()
968            if not future.done():
969                future.set_result(None)
970                break
971        if not queue:
972            self._pending_results.pop(command, None)
973
974    def _discard_pending_result(self, command: str, future: asyncio.Future[None]) -> None:
975        """Drop a result waiter that is no longer interested in its result."""
976        if (queue := self._pending_results.get(command)) is None:
977            return
978        with suppress(ValueError):
979            queue.remove(future)
980        if not queue:
981            self._pending_results.pop(command, None)
982
983    def _fail_pending_results(self) -> None:
984        """Fail all outstanding result waiters (connection lost)."""
985        for queue in self._pending_results.values():
986            for future in queue:
987                if not future.done():
988                    future.set_exception(SoloistError("websocket connection closed"))
989        self._pending_results.clear()
990
991
992def verify_platform_supported() -> None:
993    """
994    Verify an official soloist build exists for this platform.
995
996    :raises UnsupportedPlatformError: soloist has no build for this OS/architecture.
997    """
998    _resolve_architecture()
999
1000
1001@dataclass
1002class _BinaryMetadata(DataClassDictMixin):
1003    """Install metadata persisted next to the soloist binary."""
1004
1005    sha256: str
1006    version_raw: str
1007    installed_at: float
1008    etag: str | None = None
1009    version: str | None = None
1010    build_timestamp: float | None = None
1011
1012
1013def _resolve_architecture() -> str:
1014    """Return the CDN artifact architecture for the current platform."""
1015    system = platform.system()
1016    if system != "Linux":
1017        raise UnsupportedPlatformError(f"soloist is only available for Linux, not {system}")
1018    machine = platform.machine().lower()
1019    if (arch := _MACHINE_TO_ARCH.get(machine)) is None:
1020        raise UnsupportedPlatformError(f"no soloist build is available for {machine}")
1021    return arch
1022
1023
1024def _validate_download_url(url: str) -> None:
1025    """Reject download URLs that are insecure or outside Spotify's infrastructure."""
1026    parsed = URL(url)
1027    host = (parsed.host or "").lower()
1028    if parsed.scheme != "https" or not host:
1029        raise DownloadFailedError(f"refusing insecure soloist download url: {url}")
1030    if not any(
1031        host == domain or host.endswith(f".{domain}") for domain in _TRUSTED_DOWNLOAD_DOMAINS
1032    ):
1033        raise DownloadFailedError(f"refusing soloist download from untrusted host: {host}")
1034
1035
1036def _extract_binary_from_archive(archive_path: Path, dest_path: Path) -> None:
1037    """
1038    Validate the release archive and extract its soloist binary (blocking).
1039
1040    :raises InvalidArchiveError: The archive is corrupt, unsafe, or does not
1041        contain exactly one regular ``soloist`` file.
1042    """
1043    try:
1044        with tarfile.open(archive_path, mode="r:gz") as tar:
1045            binary_member: tarfile.TarInfo | None = None
1046            total_size = 0
1047            for member in tar:
1048                name = PurePosixPath(member.name)
1049                if name.is_absolute() or ".." in name.parts:
1050                    raise InvalidArchiveError(f"unsafe path in soloist archive: {member.name}")
1051                if member.isdir():
1052                    continue
1053                total_size += member.size
1054                if total_size > _MAX_EXTRACTED_SIZE:
1055                    raise InvalidArchiveError("soloist archive exceeds the extracted size limit")
1056                # the release ships docs (CHANGELOG.md, THIRD_PARTY_LICENSES.txt) next to
1057                # the binary, so we skip siblings instead of refusing the whole archive
1058                if name.name != "soloist":
1059                    continue
1060                if not member.isreg():
1061                    raise InvalidArchiveError(
1062                        f"unsupported member type in soloist archive: {member.name}"
1063                    )
1064                if binary_member is not None:
1065                    raise InvalidArchiveError("soloist archive contains multiple binaries")
1066                binary_member = member
1067            if binary_member is None:
1068                raise InvalidArchiveError("soloist archive does not contain the soloist binary")
1069            source = tar.extractfile(binary_member)
1070            if source is None:
1071                raise InvalidArchiveError("soloist archive member is not extractable")
1072            with source, dest_path.open("wb") as dest:
1073                shutil.copyfileobj(source, dest, _DOWNLOAD_CHUNK_SIZE)
1074    except (tarfile.TarError, EOFError, OSError) as err:
1075        raise InvalidArchiveError(f"invalid soloist archive: {err}") from err
1076
1077
1078def _validate_elf_header(binary_path: Path, arch: str) -> None:
1079    """
1080    Verify the extracted binary is a little-endian ELF for the expected architecture.
1081
1082    :raises InvalidArchiveError: The file is not an ELF binary for the given arch.
1083    """
1084    with binary_path.open("rb") as _file:
1085        header = _file.read(20)
1086    if len(header) < 20 or header[:4] != b"\x7fELF":
1087        raise InvalidArchiveError("soloist binary is not an ELF executable")
1088    # all supported targets are little-endian (EI_DATA == 1)
1089    if header[5] != 1:
1090        raise InvalidArchiveError("soloist binary has an unexpected ELF byte order")
1091    ei_class, e_machine = _ELF_IDENT[arch]
1092    if header[4] != ei_class or int.from_bytes(header[18:20], "little") != e_machine:
1093        raise InvalidArchiveError(f"soloist binary does not match architecture {arch}")
1094
1095
1096def _consume_install_result(fut: asyncio.Future[str]) -> None:
1097    """Log the outcome of an install that finished after its caller was cancelled."""
1098    if (exc := fut.exception()) is not None and not isinstance(exc, asyncio.CancelledError):
1099        LOGGER.warning("soloist install finished with an error after cancellation: %s", exc)
1100
1101
1102def _build_expired(build_timestamp: float | None) -> bool:
1103    """Return whether a build timestamp (when known) has passed the 90-day expiry."""
1104    return build_timestamp is not None and build_timestamp + _BUILD_EXPIRY_SECONDS <= time.time()
1105
1106
1107def _parse_version_token(version_output: str) -> str | None:
1108    """Best-effort extraction of a version number from free-form ``--version`` output."""
1109    match = _VERSION_TOKEN_RE.search(version_output)
1110    return match.group(1) if match else None
1111
1112
1113def _parse_build_timestamp(version_output: str) -> float | None:
1114    """Best-effort extraction of the build timestamp from free-form ``--version`` output."""
1115    if epoch_match := _BUILD_EPOCH_RE.search(version_output):
1116        epoch = float(epoch_match.group(1))
1117        # sanity range guard so an unrelated number is not taken for a timestamp
1118        if 1_500_000_000 <= epoch <= 4_100_000_000:
1119            return epoch
1120    for match in _TIMESTAMP_RE.finditer(version_output):
1121        candidate = match.group(1).replace("Z", "+00:00")
1122        try:
1123            parsed = datetime.fromisoformat(candidate)
1124        except ValueError:
1125            continue
1126        if parsed.tzinfo is None:
1127            parsed = parsed.replace(tzinfo=UTC)
1128        return parsed.timestamp()
1129    return None
1130