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