/
/
/
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 daemon writes all of its logging to stdout and only ever puts
172 # argument-parsing complaints on stderr, so the two are merged into one
173 # captured stream. Capturing is also what makes the redaction below
174 # reachable: an unset stdout is inherited, which would leak the daemon's
175 # output - argv included - straight to the server console.
176 async with AsyncProcess(
177 args,
178 stdout=True,
179 stderr=asyncio.subprocess.STDOUT,
180 name="soloist-pair",
181 ) as pair_proc:
182 log_task = asyncio.create_task(_log_soloist_pairing_output(pair_proc, api_key))
183 try:
184 # watched together: nothing else drains the daemon's stdout, so a
185 # reader that died would leave it blocked on a full pipe until the
186 # setup step expires
187 wait_task = asyncio.ensure_future(pair_proc.wait())
188 await asyncio.wait({wait_task, log_task}, return_when=asyncio.FIRST_COMPLETED)
189 if log_task.done() and not log_task.cancelled() and log_task.exception():
190 wait_task.cancel()
191 raise LoginFailed(
192 "Soloist pairing could not be monitored"
193 ) from log_task.exception()
194 returncode = await wait_task
195 # an exited daemon still has its last (and most telling) lines in
196 # the stream buffer; the shield keeps the reader alive across the
197 # timeout so a pairing failure stays diagnosable
198 with suppress(TimeoutError):
199 await asyncio.wait_for(asyncio.shield(log_task), PAIR_LOG_DRAIN_TIMEOUT)
200 finally:
201 log_task.cancel()
202 with suppress(asyncio.CancelledError, Exception):
203 await log_task
204 if returncode != 0:
205 raise LoginFailed(f"Soloist pairing failed (exit code {returncode})")
206 if not await asyncio.to_thread(soloist_session_present, data_dir):
207 raise LoginFailed("Soloist did not store a paired session")
208
209
210def soloist_session_account(data_dir: Path) -> str | None:
211 """
212 Return the Spotify username a stored soloist session belongs to (blocking).
213
214 Answers None when it cannot be told apart: no session yet, or state for more
215 than one account.
216
217 :param data_dir: The soloist data directory to inspect.
218 """
219 accounts = _soloist_session_accounts(data_dir)
220 return accounts[0] if len(accounts) == 1 else None
221
222
223def soloist_session_present(data_dir: Path) -> bool:
224 """
225 Return whether a soloist data dir holds a stored (paired) session (blocking).
226
227 :param data_dir: The soloist data directory to inspect.
228 """
229 return bool(_soloist_session_accounts(data_dir))
230
231
232async def await_loopback_authorization(port: int, path: str) -> dict[str, str]:
233 """
234 Serve the loopback redirect target and return the OAuth params the browser arrives with.
235
236 Only reachable when the browser runs on the same host as Music Assistant; callers are
237 expected to offer a manual fallback for everyone else.
238
239 :param port: Loopback port to listen on.
240 :param path: Request path the redirect URI points at.
241 :raises OSError: When the port cannot be bound.
242 """
243 received: asyncio.Future[dict[str, str]] = asyncio.get_running_loop().create_future()
244
245 async def handle(request: web.Request) -> web.Response:
246 if not received.done():
247 received.set_result(dict(request.query))
248 return web.Response(text=LOOPBACK_RESPONSE_HTML, content_type="text/html")
249
250 app = web.Application()
251 app.router.add_get(path, handle)
252 runner = web.AppRunner(app)
253 await runner.setup()
254 try:
255 await web.TCPSite(runner, "127.0.0.1", port).start()
256 return await received
257 finally:
258 await runner.cleanup()
259
260
261async def get_spotify_token(
262 http_session: aiohttp.ClientSession,
263 client_id: str,
264 refresh_token: str,
265 session_name: str = "spotify",
266) -> dict[str, Any]:
267 """
268 Refresh Spotify access token using refresh token.
269
270 :param http_session: aiohttp client session.
271 :param client_id: Spotify client ID.
272 :param refresh_token: Spotify refresh token.
273 :param session_name: Name for logging purposes.
274 :return: Auth info dict with access_token, refresh_token, expires_at.
275 :raises LoginFailed: If token refresh fails.
276 """
277 params = {
278 "grant_type": "refresh_token",
279 "refresh_token": refresh_token,
280 "client_id": client_id,
281 }
282 err = "Unknown error"
283 for _ in range(2):
284 async with http_session.post(
285 "https://accounts.spotify.com/api/token", data=params
286 ) as response:
287 if response.status != 200:
288 err = await response.text()
289 # invalid_grant means the refresh token is revoked or expired (Spotify
290 # enforces a 6-month lifetime); retrying won't recover it, so fail now and
291 # let the caller clear the stored token and prompt re-authentication.
292 if "invalid_grant" in err or "revoked" in err:
293 raise LoginFailed(
294 f"Refresh token no longer valid for {session_name}: {err}",
295 translation_key="refresh_token_invalid",
296 translation_owner="provider.spotify",
297 )
298 # the token failed to refresh, we allow one retry
299 await asyncio.sleep(2)
300 continue
301 # if we reached this point, the token has been successfully refreshed
302 auth_info: dict[str, Any] = await response.json()
303 auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
304 # Spotify only returns a refresh_token when it rotates one; when the response
305 # omits it, keep using the existing token (per Spotify's refresh-token docs).
306 auth_info.setdefault("refresh_token", refresh_token)
307 return auth_info
308
309 raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
310
311
312async def _log_soloist_pairing_output(pair_proc: AsyncProcess, api_key: str) -> None:
313 """Log the pairing daemon's output (API key redacted) so failures are diagnosable."""
314 async for line in pair_proc.iter_stdout():
315 # the third-party binary's own output may echo argv (which carries the
316 # api key), so redact it before logging
317 text = line.replace(api_key, "<redacted>") if api_key else line
318 LOGGER.debug("[soloist-pair] %s", text)
319
320
321async def _log_pairing_output(librespot_proc: AsyncProcess) -> None:
322 """Log the pairing daemon's output so a failure to advertise is diagnosable."""
323 reported_warnings: set[str] = set()
324 async for line in librespot_proc.iter_stderr():
325 warning_key = PAIRING_LOG_TIMESTAMP.sub("[", line, count=1)
326 if ("ERROR" in line or "WARN" in line) and warning_key not in reported_warnings:
327 reported_warnings.add(warning_key)
328 LOGGER.warning("[librespot-pairing] %s", line)
329 else:
330 LOGGER.debug("[librespot-pairing] %s", line)
331
332
333async def _await_credentials_file(cache_dir: str) -> str:
334 """Poll librespot's cache directory until it holds a complete credential file."""
335 credentials_file = os.path.join(cache_dir, CREDENTIALS_FILE)
336 while True:
337 if Path(credentials_file).exists():
338 try:
339 return await asyncio.to_thread(_read_credentials_file, credentials_file)
340 except OSError, ValueError:
341 # the file was caught mid-write; fall through and retry
342 pass
343 await asyncio.sleep(1)
344
345
346def _read_credentials_file(credentials_file: str) -> str:
347 """Read and validate librespot's credential file, returning its raw contents."""
348 with open(credentials_file, encoding="utf-8") as fileobj:
349 contents = fileobj.read()
350 if not json_loads(contents).get("auth_data"):
351 msg = "Incomplete librespot credential file"
352 raise ValueError(msg)
353 return contents
354
355
356def _soloist_session_accounts(data_dir: Path) -> list[str]:
357 """Return the Spotify accounts a soloist data dir holds paired state for (blocking)."""
358 # The per-account state under settings/Users/<username>-user is the only thing
359 # a pairing leaves behind: the prefs stores rewritten before every spawn, the
360 # pid and lock files and the WebSocket endpoint all outlive one, so a data
361 # directory that ever ran a daemon never looks empty again.
362 users_dir = data_dir / "settings" / "Users"
363 try:
364 return [
365 entry.name.removesuffix(SOLOIST_USER_DIR_SUFFIX)
366 for entry in users_dir.iterdir()
367 if entry.is_dir() and entry.name.endswith(SOLOIST_USER_DIR_SUFFIX)
368 ]
369 except OSError:
370 return []
371