music-assistant-server

1"""Setup flow for the Spotify provider."""
2
3from __future__ import annotations
4
5from dataclasses import replace
6from typing import TYPE_CHECKING, Any
7from urllib.parse import urlencode
8
9import pkce
10from aiohttp import ClientError
11from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
12from music_assistant_models.enums import ConfigEntryType
13from music_assistant_models.errors import LoginFailed
14
15from music_assistant.helpers.app_vars import app_var
16from music_assistant.helpers.oauth import (
17    HOSTED_CALLBACK_URL,
18    OAUTH_STEP_TIMEOUT,
19    authorization_code_from_params,
20    authorization_code_from_url,
21    hosted_bounce_redirect,
22)
23from music_assistant.models.setup_flow import SetupFlowError, StepExpiredError
24
25from .constants import (
26    CONF_CLIENT_ID,
27    CONF_LIBRESPOT_CREDENTIALS,
28    CONF_REFRESH_TOKEN_DEV,
29    CONF_REFRESH_TOKEN_GLOBAL,
30    KEYMASTER_CLIENT_ID,
31    LIBRESPOT_REDIRECT_PATH,
32    LIBRESPOT_REDIRECT_PORT,
33    LIBRESPOT_REDIRECT_URI,
34    LIBRESPOT_SCOPE,
35    LOOPBACK_WAIT_TIMEOUT,
36    PAIRING_DEVICE_NAME,
37    PAIRING_TIMEOUT,
38    SCOPE,
39)
40from .helpers import (
41    await_loopback_authorization,
42    get_librespot_binary,
43    librespot_credentials_via_pairing,
44    librespot_credentials_via_token,
45)
46
47if TYPE_CHECKING:
48    from music_assistant.models.setup_flow import SetupSession
49
50AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
51TOKEN_URL = "https://accounts.spotify.com/api/token"
52
53# the developer client id is a public OAuth identifier (not a secret), so it is a plain
54# STRING that can be prefilled on reconfigure
55CONF_ENTRY_DEV_CLIENT_ID = ConfigEntry(
56    key=CONF_CLIENT_ID,
57    type=ConfigEntryType.STRING,
58    required=False,
59)
60
61CONF_USE_DEV_KEY = "use_developer_key"
62CONF_ENTRY_USE_DEV_KEY = ConfigEntry(
63    key=CONF_USE_DEV_KEY,
64    type=ConfigEntryType.BOOLEAN,
65    default_value=False,
66    required=False,
67)
68
69CONF_PLAYBACK_AUTH_METHOD = "playback_auth_method"
70PLAYBACK_AUTH_APP = "spotify_app"
71PLAYBACK_AUTH_BROWSER = "browser"
72CONF_ENTRY_PLAYBACK_AUTH_METHOD = ConfigEntry(
73    key=CONF_PLAYBACK_AUTH_METHOD,
74    type=ConfigEntryType.STRING,
75    default_value=PLAYBACK_AUTH_APP,
76    options=[ConfigValueOption(PLAYBACK_AUTH_APP), ConfigValueOption(PLAYBACK_AUTH_BROWSER)],
77)
78
79CONF_PLAYBACK_CALLBACK_URL = "playback_callback_url"
80CONF_ENTRY_PLAYBACK_CALLBACK_URL = ConfigEntry(
81    key=CONF_PLAYBACK_CALLBACK_URL,
82    type=ConfigEntryType.STRING,
83)
84
85
86async def run_setup(session: SetupSession) -> None:
87    """
88    Run the Spotify setup flow.
89
90    Authenticates the (required) global session with Music Assistant's own client id, authorizes
91    playback separately, then optionally a developer session with the user's own client id, and
92    persists the resulting tokens and credentials as setup data.
93
94    :param session: The setup session driving the flow.
95    """
96    setup_data = dict(session.context.setup_data)
97    # the global session always (re)authenticates: a refresh token cannot be reused across a
98    # re-auth and secure values are never prefilled back into the flow
99    setup_data[CONF_REFRESH_TOKEN_GLOBAL] = await _pkce_authenticate(
100        session, app_var("spotify_client_id"), step_id="authenticate"
101    )
102    # playback needs its own credential, minted with Spotify's keymaster client id
103    setup_data[CONF_LIBRESPOT_CREDENTIALS] = await _authorize_playback(session)
104    # everything needed is collected by now; the developer key is a purely optional extra,
105    # so it is offered as an opt-in rather than a field the user has to reason about
106    client_id_default = str(session.context.setup_data.get(CONF_CLIENT_ID) or "")
107    errors: dict[str, str] | None = None
108    while True:
109        optin_values = await session.form(
110            [replace(CONF_ENTRY_USE_DEV_KEY, value=bool(client_id_default))],
111            step_id="developer_optin",
112            errors=errors,
113            last_step=True,
114        )
115        if not optin_values.get(CONF_USE_DEV_KEY):
116            # opted out: clear any previously stored developer session
117            setup_data[CONF_CLIENT_ID] = None
118            setup_data[CONF_REFRESH_TOKEN_DEV] = None
119            try:
120                await session.finish(setup_data)
121                return
122            except SetupFlowError as err:
123                errors = {"base": err.translation_key or str(err)}
124                continue
125        client_id_default, errors = await _authorize_developer_key(
126            session, setup_data, client_id_default
127        )
128        if errors is None:
129            return
130
131
132async def _authorize_developer_key(
133    session: SetupSession, setup_data: dict[str, Any], client_id_default: str
134) -> tuple[str, dict[str, str] | None]:
135    """
136    Collect and authorize the user's own Spotify developer key, then finish the flow.
137
138    Returns the client id to prefill and the errors to show when the attempt failed; the
139    errors are None once the flow has finished.
140
141    :param session: The setup session driving the flow.
142    :param setup_data: The setup data collected so far, updated in place.
143    :param client_id_default: Client id to prefill in the form.
144    """
145    dev_values = await session.form(
146        [replace(CONF_ENTRY_DEV_CLIENT_ID, value=client_id_default)],
147        step_id="developer",
148        last_step=True,
149        translation_params=[HOSTED_CALLBACK_URL],
150    )
151    client_id = str(dev_values.get(CONF_CLIENT_ID) or "").strip()
152    try:
153        if client_id:
154            setup_data[CONF_CLIENT_ID] = client_id
155            setup_data[CONF_REFRESH_TOKEN_DEV] = await _pkce_authenticate(
156                session, client_id, step_id="authenticate_dev"
157            )
158        else:
159            # opted in but left the field empty: keep using the shared key
160            setup_data[CONF_CLIENT_ID] = None
161            setup_data[CONF_REFRESH_TOKEN_DEV] = None
162        await session.finish(setup_data)
163    except SetupFlowError as err:
164        return client_id, {"base": err.translation_key or str(err)}
165    return client_id, None
166
167
168async def _authorize_playback(session: SetupSession) -> str:
169    """
170    Obtain librespot's playback credential and return it as stored-credential JSON.
171
172    Offers pairing through the Spotify app first and falls back to a browser sign-in for
173    setups where the Spotify app cannot discover Music Assistant.
174
175    :param session: The setup session driving the flow.
176    """
177    try:
178        librespot_bin = await get_librespot_binary()
179    except RuntimeError as err:
180        raise SetupFlowError(str(err), translation_key="librespot_unavailable") from err
181    errors: dict[str, str] | None = None
182    while True:
183        method_values = await session.form(
184            [CONF_ENTRY_PLAYBACK_AUTH_METHOD],
185            step_id="playback_auth",
186            errors=errors,
187        )
188        method = str(method_values.get(CONF_PLAYBACK_AUTH_METHOD) or PLAYBACK_AUTH_APP)
189        # every failure loops back to this form: the account is already authorized by now, so
190        # aborting the flow would throw that away over a retryable mistake
191        try:
192            if method == PLAYBACK_AUTH_APP:
193                return await session.progress_until(
194                    librespot_credentials_via_pairing(librespot_bin, PAIRING_DEVICE_NAME),
195                    step_id="playback_pairing",
196                    text="pairing_instructions",
197                    expires_in=PAIRING_TIMEOUT,
198                )
199            return await _authorize_playback_via_browser(session, librespot_bin)
200        except StepExpiredError:
201            errors = {
202                "base": "pairing_not_completed"
203                if method == PLAYBACK_AUTH_APP
204                else "playback_not_completed"
205            }
206        except SetupFlowError as err:
207            errors = {"base": err.translation_key or "playback_auth_failed"}
208        except LoginFailed, ClientError, KeyError:
209            # librespot refusing the token, a transport failure, or a token response without a
210            # token; LoginFailed's own default key is too generic to show here
211            errors = {"base": "playback_auth_failed"}
212
213
214async def _authorize_playback_via_browser(session: SetupSession, librespot_bin: str) -> str:
215    """
216    Run the keymaster sign-in and return librespot's stored credential.
217
218    Spotify only accepts a loopback redirect for this client id, so the browser cannot report
219    back to Music Assistant: the user copies the URL their browser ended up on instead.
220
221    :param session: The setup session driving the flow.
222    :param librespot_bin: Path to the librespot binary.
223    """
224    code_verifier, code_challenge = pkce.generate_pkce_pair()
225    params = {
226        "response_type": "code",
227        "client_id": KEYMASTER_CLIENT_ID,
228        "scope": " ".join(LIBRESPOT_SCOPE),
229        "code_challenge_method": "S256",
230        "code_challenge": code_challenge,
231        "redirect_uri": LIBRESPOT_REDIRECT_URI,
232    }
233    authorize_url = f"{AUTHORIZE_URL}?{urlencode(params)}"
234    try:
235        # the loopback target is only reachable when the browser runs on this host, in which
236        # case the step completes on its own; everyone else falls through to the paste form
237        callback_params = await session.external_until(
238            await_loopback_authorization(LIBRESPOT_REDIRECT_PORT, LIBRESPOT_REDIRECT_PATH),
239            authorize_url,
240            step_id="playback_browser_open",
241            expires_in=LOOPBACK_WAIT_TIMEOUT,
242        )
243        code = authorization_code_from_params(callback_params)
244    except StepExpiredError, OSError:
245        values = await session.form(
246            [CONF_ENTRY_PLAYBACK_CALLBACK_URL],
247            step_id="playback_browser",
248            expires_in=OAUTH_STEP_TIMEOUT,
249            translation_params=[authorize_url],
250        )
251        code = authorization_code_from_url(str(values.get(CONF_PLAYBACK_CALLBACK_URL) or ""))
252    token_params = {
253        "grant_type": "authorization_code",
254        "code": code,
255        "redirect_uri": LIBRESPOT_REDIRECT_URI,
256        "client_id": KEYMASTER_CLIENT_ID,
257        "code_verifier": code_verifier,
258    }
259    async with session.mass.http_session.post(TOKEN_URL, data=token_params) as response:
260        if response.status != 200:
261            raise SetupFlowError(
262                f"Failed to get access token: {await response.text()}",
263                translation_key="playback_code_invalid",
264            )
265        token_result = await response.json()
266    return await librespot_credentials_via_token(librespot_bin, token_result["access_token"])
267
268
269async def _pkce_authenticate(session: SetupSession, client_id: str, step_id: str) -> str:
270    """
271    Run the Spotify PKCE auth flow via the setup session and return a refresh token.
272
273    :param session: The setup session driving the flow.
274    :param client_id: The Spotify client id to authenticate with.
275    :param step_id: The external step id (also the i18n key segment).
276    """
277    code_verifier, code_challenge = pkce.generate_pkce_pair()
278    redirect_uri, state = hosted_bounce_redirect(session.callback_url)
279    params = {
280        "response_type": "code",
281        "client_id": client_id,
282        "scope": " ".join(SCOPE),
283        "code_challenge_method": "S256",
284        "code_challenge": code_challenge,
285        "redirect_uri": redirect_uri,
286        "state": state,
287    }
288    callback_params = await session.external(
289        f"{AUTHORIZE_URL}?{urlencode(params)}", step_id=step_id, expires_in=OAUTH_STEP_TIMEOUT
290    )
291    code = authorization_code_from_params(callback_params)
292    token_params = {
293        "grant_type": "authorization_code",
294        "code": code,
295        "redirect_uri": redirect_uri,
296        "client_id": client_id,
297        "code_verifier": code_verifier,
298    }
299    async with session.mass.http_session.post(TOKEN_URL, data=token_params) as response:
300        if response.status != 200:
301            raise SetupFlowError(f"Failed to get access token: {await response.text()}")
302        token_result = await response.json()
303    if not (refresh_token := token_result.get("refresh_token")):
304        raise SetupFlowError("No refresh token in the token response")
305    return str(refresh_token)
306