music-assistant-server

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 typing import TYPE_CHECKING, Any
13
14from aiohttp import web
15from music_assistant_models.errors import LoginFailed
16
17from music_assistant.helpers.json import json_loads
18from music_assistant.helpers.process import AsyncProcess, check_output
19
20from .constants import CHECK_AUTH_TIMEOUT, CREDENTIALS_FILE
21
22LOGGER = logging.getLogger(__name__)
23PAIRING_LOG_TIMESTAMP = re.compile(r"^\[\d{4}-\d{2}-\d{2}T[^ ]+ ")
24
25LOOPBACK_RESPONSE_HTML = """
26<html>
27<body onload="window.close();">
28    Playback approved, you may now close this window and return to Music Assistant.
29</body>
30</html>
31"""
32
33if TYPE_CHECKING:
34    import aiohttp
35
36
37async def get_librespot_binary() -> str:
38    """Find the correct librespot binary belonging to the platform."""
39
40    async def check_librespot(librespot_path: str) -> str | None:
41        try:
42            returncode, output = await check_output(librespot_path, "--version")
43            if returncode == 0 and b"librespot" in output:
44                return librespot_path
45            return None
46        except OSError:
47            return None
48
49    base_path = os.path.join(os.path.dirname(__file__), "bin")
50    system = platform.system().lower().replace("darwin", "macos")
51    architecture = platform.machine().lower()
52
53    if librespot_binary := await check_librespot(
54        os.path.join(base_path, f"librespot-{system}-{architecture}")
55    ):
56        return librespot_binary
57
58    msg = f"Unable to locate Librespot for {system}/{architecture}"
59    raise RuntimeError(msg)
60
61
62async def librespot_credentials_via_pairing(librespot_bin: str, device_name: str) -> str:
63    """
64    Advertise a Spotify Connect device and return the credential librespot stores once paired.
65
66    Blocks until the user selects the device in the official Spotify app; the caller is expected
67    to bound the wait (the setup flow's step deadline cancels it).
68
69    :param librespot_bin: Path to the librespot binary.
70    :param device_name: Device name to advertise to the Spotify app.
71    """
72    with tempfile.TemporaryDirectory() as cache_dir:
73        args = [
74            librespot_bin,
75            "--cache",
76            cache_dir,
77            "--disable-audio-cache",
78            "--backend",
79            "pipe",
80            "--name",
81            device_name,
82        ]
83        # stdout carries decoded audio once the user hits play; discard it so the pairing
84        # daemon never blocks on a pipe nobody reads
85        async with AsyncProcess(
86            args, stdout=asyncio.subprocess.DEVNULL, stderr=True, name="librespot-pairing"
87        ) as librespot_proc:
88            # librespot advertises over mDNS, which fails silently in host-network-less
89            # containers; without its log the user would just watch the step time out
90            librespot_proc.attach_stderr_reader(
91                asyncio.create_task(_log_pairing_output(librespot_proc))
92            )
93            return await _await_credentials_file(cache_dir)
94
95
96async def librespot_credentials_via_token(librespot_bin: str, access_token: str) -> str:
97    """
98    Exchange a keymaster access token for librespot's reusable stored credential.
99
100    :param librespot_bin: Path to the librespot binary.
101    :param access_token: Spotify access token minted with the keymaster client id.
102    :raises LoginFailed: When librespot could not turn the token into a stored credential.
103    """
104    with tempfile.TemporaryDirectory() as cache_dir:
105        returncode, output = await check_output(
106            librespot_bin,
107            "--cache",
108            cache_dir,
109            "--check-auth",
110            "--access-token",
111            access_token,
112            timeout=CHECK_AUTH_TIMEOUT,
113        )
114        if returncode != 0:
115            raise LoginFailed(
116                f"Librespot rejected the playback authorization: {output.decode().strip()}"
117            )
118        credentials_file = os.path.join(cache_dir, CREDENTIALS_FILE)
119        if not os.path.exists(credentials_file):
120            raise LoginFailed("Librespot did not store a playback credential")
121        return await asyncio.to_thread(_read_credentials_file, credentials_file)
122
123
124async def await_loopback_authorization(port: int, path: str) -> dict[str, str]:
125    """
126    Serve the loopback redirect target and return the OAuth params the browser arrives with.
127
128    Only reachable when the browser runs on the same host as Music Assistant; callers are
129    expected to offer a manual fallback for everyone else.
130
131    :param port: Loopback port to listen on.
132    :param path: Request path the redirect URI points at.
133    :raises OSError: When the port cannot be bound.
134    """
135    received: asyncio.Future[dict[str, str]] = asyncio.get_running_loop().create_future()
136
137    async def handle(request: web.Request) -> web.Response:
138        if not received.done():
139            received.set_result(dict(request.query))
140        return web.Response(text=LOOPBACK_RESPONSE_HTML, content_type="text/html")
141
142    app = web.Application()
143    app.router.add_get(path, handle)
144    runner = web.AppRunner(app)
145    await runner.setup()
146    try:
147        await web.TCPSite(runner, "127.0.0.1", port).start()
148        return await received
149    finally:
150        await runner.cleanup()
151
152
153async def get_spotify_token(
154    http_session: aiohttp.ClientSession,
155    client_id: str,
156    refresh_token: str,
157    session_name: str = "spotify",
158) -> dict[str, Any]:
159    """
160    Refresh Spotify access token using refresh token.
161
162    :param http_session: aiohttp client session.
163    :param client_id: Spotify client ID.
164    :param refresh_token: Spotify refresh token.
165    :param session_name: Name for logging purposes.
166    :return: Auth info dict with access_token, refresh_token, expires_at.
167    :raises LoginFailed: If token refresh fails.
168    """
169    params = {
170        "grant_type": "refresh_token",
171        "refresh_token": refresh_token,
172        "client_id": client_id,
173    }
174    err = "Unknown error"
175    for _ in range(2):
176        async with http_session.post(
177            "https://accounts.spotify.com/api/token", data=params
178        ) as response:
179            if response.status != 200:
180                err = await response.text()
181                # invalid_grant means the refresh token is revoked or expired (Spotify
182                # enforces a 6-month lifetime); retrying won't recover it, so fail now and
183                # let the caller clear the stored token and prompt re-authentication.
184                if "invalid_grant" in err or "revoked" in err:
185                    raise LoginFailed(
186                        f"Refresh token no longer valid for {session_name}: {err}",
187                        translation_key="refresh_token_invalid",
188                        translation_owner="provider.spotify",
189                    )
190                # the token failed to refresh, we allow one retry
191                await asyncio.sleep(2)
192                continue
193            # if we reached this point, the token has been successfully refreshed
194            auth_info: dict[str, Any] = await response.json()
195            auth_info["expires_at"] = int(auth_info["expires_in"] + time.time())
196            # Spotify only returns a refresh_token when it rotates one; when the response
197            # omits it, keep using the existing token (per Spotify's refresh-token docs).
198            auth_info.setdefault("refresh_token", refresh_token)
199            return auth_info
200
201    raise LoginFailed(f"Failed to refresh {session_name} access token: {err}")
202
203
204async def _log_pairing_output(librespot_proc: AsyncProcess) -> None:
205    """Log the pairing daemon's output so a failure to advertise is diagnosable."""
206    reported_warnings: set[str] = set()
207    async for line in librespot_proc.iter_stderr():
208        warning_key = PAIRING_LOG_TIMESTAMP.sub("[", line, count=1)
209        if ("ERROR" in line or "WARN" in line) and warning_key not in reported_warnings:
210            reported_warnings.add(warning_key)
211            LOGGER.warning("[librespot-pairing] %s", line)
212        else:
213            LOGGER.debug("[librespot-pairing] %s", line)
214
215
216async def _await_credentials_file(cache_dir: str) -> str:
217    """Poll librespot's cache directory until it holds a complete credential file."""
218    credentials_file = os.path.join(cache_dir, CREDENTIALS_FILE)
219    while True:
220        if os.path.exists(credentials_file):
221            try:
222                return await asyncio.to_thread(_read_credentials_file, credentials_file)
223            except OSError, ValueError:
224                # the file was caught mid-write; fall through and retry
225                pass
226        await asyncio.sleep(1)
227
228
229def _read_credentials_file(credentials_file: str) -> str:
230    """Read and validate librespot's credential file, returning its raw contents."""
231    with open(credentials_file, encoding="utf-8") as fileobj:
232        contents = fileobj.read()
233    if not json_loads(contents).get("auth_data"):
234        msg = "Incomplete librespot credential file"
235        raise ValueError(msg)
236    return contents
237