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