/
/
1"""Helpers/utils for the Spotify musicprovider."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import os
8import platform
9import re
10import tempfile
11import time
12from contextlib import suppress
13from pathlib import Path
14from typing import TYPE_CHECKING, Any
15
16from aiohttp import web
17from music_assistant_models.errors import LoginFailed
18
19from music_assistant.helpers.json import json_loads
20from music_assistant.helpers.process import AsyncProcess, check_output
21from music_assistant.providers.spotify_connect.soloist import SoloistBinaryManager
22
23from .constants import (
24 CHECK_AUTH_TIMEOUT,
25 CREDENTIALS_FILE,
26 PAIRING_DEVICE_NAME,
27 SOLOIST_USER_DIR_SUFFIX,
28)
29
30# how long the pairing daemon's log reader is given to drain after it exits
31PAIR_LOG_DRAIN_TIMEOUT = 2.0
32
33LOGGER = logging.getLogger(__name__)
34PAIRING_LOG_TIMESTAMP = re.compile(r"^\[\d{4}-\d{2}-\d{2}T[^ ]+ ")
35
36LOOPBACK_RESPONSE_HTML = """
37<html>
38<body onload="window.close();">
39 Playback approved, you may now close this window and return to Music Assistant.
40</body>
41</html>
42"""
43
44if TYPE_CHECKING:
45 import aiohttp
46
47 from music_assistant.mass import MusicAssistant
48
49
50async def get_librespot_binary() -> str:
51 """Find the correct librespot binary belonging to the platform."""
52
53 async def check_librespot(librespot_path: str) -> str | None:
54 try:
55 returncode, output = await check_output(librespot_path, "--version")
56 if returncode == 0 and b"librespot" in output:
57 return librespot_path
58 return None
59 except OSError:
60 return None
61
62 base_path = os.path.join(os.path.dirname(__file__), "bin")
63 system = platform.system().lower().replace("darwin", "macos")
64 architecture = platform.machine().lower()
65
66 if librespot_binary := await check_librespot(
67 os.path.join(base_path, f"librespot-{system}-{architecture}")
68 ):
69 return librespot_binary
70
71 msg = f"Unable to locate Librespot for {system}/{architecture}"
72 raise RuntimeError(msg)
73
74
75async def librespot_credentials_via_pairing(librespot_bin: str, device_name: str) -> str:
76 """
77 Advertise a Spotify Connect device and return the credential librespot stores once paired.
78
79 Blocks until the user selects the device in the official Spotify app; the caller is expected
80 to bound the wait (the setup flow's step deadline cancels it).
81
82 :param librespot_bin: Path to the librespot binary.
83 :param device_name: Device name to advertise to the Spotify app.
84 """
85 with tempfile.TemporaryDirectory() as cache_dir:
86 args = [
87 librespot_bin,
88 "--cache",
89 cache_dir,
90 "--disable-audio-cache",
91 "--backend",
92 "pipe",
93 "--name",
94 device_name,
95 ]
96 # stdout carries decoded audio once the user hits play; discard it so the pairing
97 # daemon never blocks on a pipe nobody reads
98 async with AsyncProcess(
99 args, stdout=asyncio.subprocess.DEVNULL, stderr=True, name="librespot-pairing"
100 ) as librespot_proc:
101 # librespot advertises over mDNS, which fails silently in host-network-less
102 # containers; without its log the user would just watch the step time out
103 librespot_proc.attach_stderr_reader(
104 asyncio.create_task(_log_pairing_output(librespot_proc))
105 )
106 return await _await_credentials_file(cache_dir)
107
108
109async def librespot_credentials_via_token(librespot_bin: str, access_token: str) -> str:
110 """
111 Exchange a keymaster access token for librespot's reusable stored credential.
112
113 :param librespot_bin: Path to the librespot binary.
114 :param access_token: Spotify access token minted with the keymaster client id.
115 :raises LoginFailed: When librespot could not turn the token into a stored credential.
116 """
117 with tempfile.TemporaryDirectory() as cache_dir:
118 returncode, output = await check_output(
119 librespot_bin,
120 "--cache",
121 cache_dir,
122 "--check-auth",
123 "--access-token",
124 access_token,
125 timeout=CHECK_AUTH_TIMEOUT,
126 )
127 if returncode != 0:
128 raise LoginFailed(
129 f"Librespot rejected the playback authorization: {output.decode().strip()}"
130 )
131 credentials_file = os.path.join(cache_dir, CREDENTIALS_FILE)
132 if not Path(credentials_file).exists():
133 raise LoginFailed("Librespot did not store a playback credential")
134 return await asyncio.to_thread(_read_credentials_file, credentials_file)
135
136
137async def pair_soloist_session(mass: MusicAssistant, api_key: str, data_dir: Path) -> None:
138 """
139 Pair a Spotify account with soloist and store the session in the given data dir.
140
141 Advertises a Spotify Connect device and blocks until the user selects it in the
142 official Spotify app; the caller is expected to bound the wait (the setup flow's
143 step deadline cancels it).
144
145 :param mass: The MusicAssistant instance.
146 :param api_key: The user's personal Soloist API key (secret, kept out of all logs).
147 :param data_dir: Directory the paired session is stored in.
148 :raises LoginFailed: When pairing did not complete with a stored session.
149 """
150 binary = await SoloistBinaryManager(mass).ensure_fresh(consent=True)
151
152 def _prepare() -> None:
153 data_dir.mkdir(parents=True, exist_ok=True)
154 # the paired session holds the Spotify device identity and login session
155 data_dir.chmod(0o700)
156
157 await asyncio.to_thread(_prepare)
158 with tempfile.TemporaryDirectory() as cache_dir:
159 args = [
160 str(binary),
161 "--pair",
162 "--device-name",
163 PAIRING_DEVICE_NAME,
164 "--api-key",
165 api_key,
166 "--data-dir",
167 str(data_dir),
168 "--cache-dir",
169 cache_dir,
170 ]
171 # the explicit process name keeps AsyncProcess logging free of the argv
172 # (which carries the API key)
173 # the daemon writes all of its logging to stdout and only ever puts
174 # argument-parsing complaints on stderr, so the two are merged into one
175 # captured stream. Capturing is also what makes the redaction below
176 # reachable: an unset stdout is inherited, which would leak the daemon's
177 # output - argv included - straight to the server console.
178 async with AsyncProcess(
179 args,
180 stdout=True,
181 stderr=asyncio.subprocess.STDOUT,
182 name="soloist-pair",
183 ) as pair_proc:
184 log_task = asyncio.create_task(_log_soloist_pairing_output(pair_proc, api_key))
185 try:
186 # watched together: nothing else drains the daemon's stdout, so a
187 # reader that died would leave it blocked on a full pipe until the
188 # setup step expires
189 wait_task = asyncio.ensure_future(pair_proc.wait())
190 await asyncio.wait({wait_task, log_task}, return_when=asyncio.FIRST_COMPLETED)
191 if log_task.done() and not log_task.cancelled() and log_task.exception():
192 wait_task.cancel()
193 raise LoginFailed(
194 "Soloist pairing could not be monitored"
195 ) from log_task.exception()
196 returncode = await wait_task
197 # an exited daemon still has its last (and most telling) lines in
198 # the stream buffer; the shield keeps the reader alive across the
199 # timeout so a pairing failure stays diagnosable
200 with suppress(TimeoutError):
201 await asyncio.wait_for(asyncio.shield(log_task), PAIR_LOG_DRAIN_TIMEOUT)
202 finally:
203 log_task.cancel()
204 with suppress(asyncio.CancelledError, Exception):
205 await log_task
206 if returncode != 0:
207 raise LoginFailed(f"Soloist pairing failed (exit code {returncode})")
208 if not await asyncio.to_thread(soloist_session_present, data_dir):
209 raise LoginFailed("Soloist did not store a paired session")
210
211
212def soloist_session_account(data_dir: Path) -> str | None:
213 """
214 Return the Spotify username a stored soloist session belongs to (blocking).
215
216 Answers None when it cannot be told apart: no session yet, or state for more
217 than one account.
218
219 :param data_dir: The soloist data directory to inspect.
220 """
221 accounts = _soloist_session_accounts(data_dir)
222 return accounts[0] if len(accounts) == 1 else None
223
224
225def soloist_session_present(data_dir: Path) -> bool:
226 """
227 Return whether a soloist data dir holds a stored (paired) session (blocking).
228
229 :param data_dir: The soloist data directory to inspect.
230 """
231 return bool(_soloist_session_accounts(data_dir))
232
233
234async def await_loopback_authorization(port: int, path: str) -> dict[str, str]:
235 """
236 Serve the loopback redirect target and return the OAuth params the browser arrives with.
237
238 Only reachable when the browser runs on the same host as Music Assistant; callers are
239 expected to offer a manual fallback for everyone else.
240
241 :param port: Loopback port to listen on.
242 :param path: Request path the redirect URI points at.
243 :raises OSError: When the port cannot be bound.
244 """
245 received: asyncio.Future[dict[str, str]] = asyncio.get_running_loop().create_future()
246
247 async def handle(request: web.Request) -> web.Response:
248 if not received.done():
249 received.set_result(dict(request.query))
250 return web.Response(text=LOOPBACK_RESPONSE_HTML, content_type="text/html")
251
252 app = web.Application()
253 app.router.add_get(path, handle)
254 runner = web.AppRunner(app)
255 await runner.setup()
256 try:
257 await web.TCPSite(runner, "127.0.0.1", port).start()
258 return await received
259 finally:
260 await runner.cleanup()
261
262
263async def get_spotify_token(
264 http_session: aiohttp.ClientSession,
265 client_id: str,
266 refresh_token: str,
267 session_name: str = "spotify",
268) -> dict[str, Any]:
269 """
270 Refresh Spotify access token using refresh token.
271
272 :param http_session: aiohttp client session.
273 :param client_id: Spotify client ID.
274 :param refresh_token: Spotify refresh token.
275 :param session_name: Name for logging purposes.
276 :return: Auth info dict with access_token, refresh_token, expires_at.
277 :raises LoginFailed: If token refresh fails.
278 """
279 params = {
280 "grant_type": "refresh_token",
281 "refresh_token": refresh_token,
282 "client_id": client_id,
283 }
284 err = "Unknown error"
285 for _ in range(2):
286 async with http_session.post(
287 "https://accounts.spotify.com/api/token", data=params
288 ) as response:
289 if response.status != 200:
290 err = await response.text()
291 # invalid_grant means the refresh token is revoked or expired (Spotify
292 # enforces a 6-month lifetime); retrying won't recover it, so fail now and
293 # let the caller clear the stored token and prompt re-authentication.
294 if "invalid_grant" in err or "revoked" in err:
295 raise LoginFailed(
296 f"Refresh token no longer valid for {session_name}: {err}",
297 translation_key="refresh_token_invalid",
298 translation_owner="provider.spotify",
299 )
300 # the token failed to refresh, we allow one retry
301 await asyncio.sleep(2)
302 continue
303 # if we reached this point, the token has been successfully refreshed
304 auth_info: dict[str, Any] = await response.json()
305 auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
306 # Spotify only returns a refresh_token when it rotates one; when the response
307 # omits it, keep using the existing token (per Spotify's refresh-token docs).
308 auth_info.setdefault("refresh_token", refresh_token)
309 return auth_info
310
311 raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
312
313
314async def _log_soloist_pairing_output(pair_proc: AsyncProcess, api_key: str) -> None:
315 """Log the pairing daemon's output (API key redacted) so failures are diagnosable."""
316 async for line in pair_proc.iter_stdout():
317 # the third-party binary's own output may echo argv (which carries the
318 # api key), so redact it before logging
319 text = line.replace(api_key, "<redacted>") if api_key else line
320 LOGGER.debug("[soloist-pair] %s", text)
321
322
323async def _log_pairing_output(librespot_proc: AsyncProcess) -> None:
324 """Log the pairing daemon's output so a failure to advertise is diagnosable."""
325 reported_warnings: set[str] = set()
326 async for line in librespot_proc.iter_stderr():
327 warning_key = PAIRING_LOG_TIMESTAMP.sub("[", line, count=1)
328 if ("ERROR" in line or "WARN" in line) and warning_key not in reported_warnings:
329 reported_warnings.add(warning_key)
330 LOGGER.warning("[librespot-pairing] %s", line)
331 else:
332 LOGGER.debug("[librespot-pairing] %s", line)
333
334
335async def _await_credentials_file(cache_dir: str) -> str:
336 """Poll librespot's cache directory until it holds a complete credential file."""
337 credentials_file = os.path.join(cache_dir, CREDENTIALS_FILE)
338 while True:
339 if Path(credentials_file).exists():
340 try:
341 return await asyncio.to_thread(_read_credentials_file, credentials_file)
342 except OSError, ValueError:
343 # the file was caught mid-write; fall through and retry
344 pass
345 await asyncio.sleep(1)
346
347
348def _read_credentials_file(credentials_file: str) -> str:
349 """Read and validate librespot's credential file, returning its raw contents."""
350 with open(credentials_file, encoding="utf-8") as fileobj:
351 contents = fileobj.read()
352 if not json_loads(contents).get("auth_data"):
353 msg = "Incomplete librespot credential file"
354 raise ValueError(msg)
355 return contents
356
357
358def _soloist_session_accounts(data_dir: Path) -> list[str]:
359 """Return the Spotify accounts a soloist data dir holds paired state for (blocking)."""
360 # The per-account state under settings/Users/<username>-user is the only thing
361 # a pairing leaves behind: the prefs stores rewritten before every spawn, the
362 # pid and lock files and the WebSocket endpoint all outlive one, so a data
363 # directory that ever ran a daemon never looks empty again.
364 users_dir = data_dir / "settings" / "Users"
365 try:
366 return [
367 entry.name.removesuffix(SOLOIST_USER_DIR_SUFFIX)
368 for entry in users_dir.iterdir()
369 if entry.is_dir() and entry.name.endswith(SOLOIST_USER_DIR_SUFFIX)
370 ]
371 except OSError:
372 return []
373