music-assistant-server

1"""Setup flow for the Spotify provider."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import shutil
8from dataclasses import replace
9from pathlib import Path
10from typing import TYPE_CHECKING, Any
11from urllib.parse import unquote, urlencode
12
13import pkce
14from aiohttp import ClientError, ClientTimeout
15from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
16from music_assistant_models.enums import ConfigEntryType
17from music_assistant_models.errors import LoginFailed
18
19from music_assistant.helpers.app_vars import app_var
20from music_assistant.helpers.json import json_loads
21from music_assistant.helpers.oauth import (
22    HOSTED_CALLBACK_URL,
23    OAUTH_STEP_TIMEOUT,
24    authorization_code_from_params,
25    authorization_code_from_url,
26    hosted_bounce_redirect,
27)
28from music_assistant.models.setup_flow import AbortFlow, SetupFlowError, StepExpiredError
29from music_assistant.providers.spotify_connect.soloist import (
30    SoloistError,
31    UnsupportedPlatformError,
32    verify_platform_supported,
33)
34
35from .constants import (
36    BACKEND_LIBRESPOT,
37    BACKEND_SOLOIST,
38    CONF_ACCOUNT_ID,
39    CONF_CLIENT_ID,
40    CONF_LIBRESPOT_CREDENTIALS,
41    CONF_PLAYBACK_BACKEND,
42    CONF_REFRESH_TOKEN_DEV,
43    CONF_REFRESH_TOKEN_GLOBAL,
44    CONF_SOLOIST_API_KEY,
45    CONF_SOLOIST_CONSENT,
46    CONF_SOLOIST_SESSION_DIR,
47    KEYMASTER_CLIENT_ID,
48    LIBRESPOT_REDIRECT_PATH,
49    LIBRESPOT_REDIRECT_PORT,
50    LIBRESPOT_REDIRECT_URI,
51    LIBRESPOT_SCOPE,
52    LOOPBACK_WAIT_TIMEOUT,
53    PAIRING_DEVICE_NAME,
54    PAIRING_TIMEOUT,
55    SCOPE,
56    SOLOIST_DATA_DIR_NAME,
57    SOLOIST_PAIRING_DIR,
58)
59from .helpers import (
60    await_loopback_authorization,
61    get_librespot_binary,
62    librespot_credentials_via_pairing,
63    librespot_credentials_via_token,
64    pair_soloist_session,
65    soloist_session_account,
66    soloist_session_present,
67)
68from .provider import SpotifyProvider
69
70if TYPE_CHECKING:
71    from music_assistant.models.setup_flow import SetupSession
72
73LOGGER = logging.getLogger(__name__)
74
75# seconds to wait for the account lookup that gates the setup
76ACCOUNT_LOOKUP_TIMEOUT = 30
77
78AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
79TOKEN_URL = "https://accounts.spotify.com/api/token"
80
81# the developer client id is a public OAuth identifier (not a secret), so it is a plain
82# STRING that can be prefilled on reconfigure
83CONF_ENTRY_DEV_CLIENT_ID = ConfigEntry(
84    key=CONF_CLIENT_ID,
85    type=ConfigEntryType.STRING,
86    required=False,
87)
88
89CONF_USE_DEV_KEY = "use_developer_key"
90CONF_ENTRY_USE_DEV_KEY = ConfigEntry(
91    key=CONF_USE_DEV_KEY,
92    type=ConfigEntryType.BOOLEAN,
93    default_value=False,
94    required=False,
95)
96
97CONF_PLAYBACK_AUTH_METHOD = "playback_auth_method"
98PLAYBACK_AUTH_APP = "spotify_app"
99PLAYBACK_AUTH_BROWSER = "browser"
100CONF_ENTRY_PLAYBACK_AUTH_METHOD = ConfigEntry(
101    key=CONF_PLAYBACK_AUTH_METHOD,
102    type=ConfigEntryType.STRING,
103    default_value=PLAYBACK_AUTH_APP,
104    options=[ConfigValueOption(PLAYBACK_AUTH_APP), ConfigValueOption(PLAYBACK_AUTH_BROWSER)],
105)
106
107CONF_SOLOIST_REPAIR = "soloist_repair"
108
109# Minimum plausible length of a pasted Soloist API key: anything shorter is a
110# partial paste. No further format rules are applied locally — Spotify rejects
111# an invalid key when soloist authenticates.
112MIN_API_KEY_LENGTH = 16
113
114CONF_PLAYBACK_CALLBACK_URL = "playback_callback_url"
115CONF_ENTRY_PLAYBACK_CALLBACK_URL = ConfigEntry(
116    key=CONF_PLAYBACK_CALLBACK_URL,
117    type=ConfigEntryType.STRING,
118)
119
120
121async def run_setup(session: SetupSession) -> None:
122    """
123    Run the Spotify setup flow.
124
125    Authenticates the (required) global session with Music Assistant's own client id, authorizes
126    playback separately, then optionally a developer session with the user's own client id, and
127    persists the resulting tokens and credentials as setup data.
128
129    :param session: The setup session driving the flow.
130    """
131    setup_data = dict(session.context.setup_data)
132    try:
133        # the global session always (re)authenticates: a refresh token cannot be reused across a
134        # re-auth and secure values are never prefilled back into the flow
135        token_result = await _pkce_authenticate(
136            session, app_var("spotify_client_id"), step_id="authenticate"
137        )
138        setup_data[CONF_REFRESH_TOKEN_GLOBAL] = str(token_result["refresh_token"])
139        # an account that cannot work is turned away before the playback authorization
140        account_id = await _verify_account(session, str(token_result["access_token"]))
141        setup_data[CONF_ACCOUNT_ID] = account_id
142        # playback authorization is separate from the Web API tokens and depends on
143        # the explicitly chosen playback backend
144        await _setup_playback(session, setup_data, account_id)
145        # everything needed is collected by now; the developer key is a purely optional extra,
146        # so it is offered as an opt-in rather than a field the user has to reason about
147        client_id_default = str(session.context.setup_data.get(CONF_CLIENT_ID) or "")
148        errors: dict[str, str] | None = None
149        while True:
150            optin_values = await session.form(
151                [replace(CONF_ENTRY_USE_DEV_KEY, value=bool(client_id_default))],
152                step_id="developer_optin",
153                errors=errors,
154                last_step=True,
155            )
156            if not optin_values.get(CONF_USE_DEV_KEY):
157                # opted out: clear any previously stored developer session
158                setup_data[CONF_CLIENT_ID] = None
159                setup_data[CONF_REFRESH_TOKEN_DEV] = None
160                try:
161                    await session.finish(setup_data)
162                    return
163                except SetupFlowError as err:
164                    errors = {"base": err.translation_key or str(err)}
165                    continue
166            client_id_default, errors = await _authorize_developer_key(
167                session, setup_data, client_id_default
168            )
169            if errors is None:
170                return
171    finally:
172        # a soloist session paired by this flow holds reusable login material;
173        # adoption copies it into the instance's own storage during finish, so
174        # the flow-private copy is always discarded once the flow is over
175        await _discard_pairing_dir(session)
176
177
178async def _authorize_developer_key(
179    session: SetupSession, setup_data: dict[str, Any], client_id_default: str
180) -> tuple[str, dict[str, str] | None]:
181    """
182    Collect and authorize the user's own Spotify developer key, then finish the flow.
183
184    Returns the client id to prefill and the errors to show when the attempt failed; the
185    errors are None once the flow has finished.
186
187    :param session: The setup session driving the flow.
188    :param setup_data: The setup data collected so far, updated in place.
189    :param client_id_default: Client id to prefill in the form.
190    """
191    dev_values = await session.form(
192        [replace(CONF_ENTRY_DEV_CLIENT_ID, value=client_id_default)],
193        step_id="developer",
194        last_step=True,
195        translation_params=[HOSTED_CALLBACK_URL],
196    )
197    client_id = str(dev_values.get(CONF_CLIENT_ID) or "").strip()
198    try:
199        if client_id:
200            setup_data[CONF_CLIENT_ID] = client_id
201            dev_token_result = await _pkce_authenticate(
202                session, client_id, step_id="authenticate_dev"
203            )
204            setup_data[CONF_REFRESH_TOKEN_DEV] = str(dev_token_result["refresh_token"])
205        else:
206            # opted in but left the field empty: keep using the shared key
207            setup_data[CONF_CLIENT_ID] = None
208            setup_data[CONF_REFRESH_TOKEN_DEV] = None
209        await session.finish(setup_data)
210    except SetupFlowError as err:
211        return client_id, {"base": err.translation_key or str(err)}
212    return client_id, None
213
214
215async def _verify_account(session: SetupSession, access_token: str) -> str | None:
216    """
217    Check the just-authenticated Spotify account and return its id.
218
219    Turns the user away when the account has no Spotify Premium (neither playback
220    backend can stream for a free account) or when it is already set up on another
221    provider instance. A lookup Spotify does not answer is
222    not held against the user: the setup simply continues and None is returned.
223
224    :param session: The setup session driving the flow.
225    :param access_token: The access token from the just-completed sign-in. Reusing
226        it is deliberate — minting a fresh one rotates the refresh token, which
227        revokes the one just stored as setup data.
228    :raises AbortFlow: When the account is non-Premium or already configured.
229    """
230    try:
231        async with session.mass.http_session.get(
232            "https://api.spotify.com/v1/me",
233            headers={"Authorization": f"Bearer {access_token}"},
234            timeout=ClientTimeout(total=ACCOUNT_LOOKUP_TIMEOUT),
235        ) as response:
236            if response.status != 200:
237                LOGGER.warning("Account check skipped: Spotify replied HTTP %s", response.status)
238                return None
239            # a malformed body raises ValueError, which is not a ClientError
240            userinfo = await response.json()
241    except (ClientError, TimeoutError, ValueError) as err:
242        # a bare TimeoutError stringifies to nothing, so log the type too
243        LOGGER.warning("Account check skipped: %s %s", type(err).__name__, err)
244        return None
245    if not isinstance(userinfo, dict):
246        LOGGER.warning("Account check skipped: Spotify returned an unexpected profile")
247        return None
248    product = str(userinfo.get("product") or "")
249    if product and product != "premium":
250        raise AbortFlow("premium_required")
251    if not (account_id := str(userinfo.get("id") or "")):
252        return None
253    if await _account_in_use(session, account_id):
254        raise AbortFlow("account_already_configured")
255    return account_id
256
257
258async def _account_in_use(session: SetupSession, account_id: str) -> bool:
259    """
260    Return whether another Spotify provider instance is already set up for this account.
261
262    Compares the account id stored with each instance's configuration, so an instance
263    that is disabled or failed to load still holds its account. Configurations
264    predating that stored value fall back to the running instance, which fills the
265    value in on its next successful load. The instance being reconfigured is of
266    course allowed to keep its own account.
267
268    :param session: The setup session driving the flow.
269    :param account_id: The Spotify user id that just signed in.
270    """
271    mass = session.mass
272    for config in await mass.config.get_provider_configs(provider_domain="spotify"):
273        if config.instance_id == session.context.instance_id:
274            continue
275        if stored := mass.config.get_provider_setup_value(config.instance_id, CONF_ACCOUNT_ID):
276            if str(stored) == account_id:
277                return True
278            continue
279        provider = mass.get_provider(config.instance_id, return_unavailable=True)
280        if isinstance(provider, SpotifyProvider) and provider.account_id == account_id:
281            return True
282    return False
283
284
285async def _setup_playback(
286    session: SetupSession, setup_data: dict[str, Any], account_id: str | None
287) -> None:
288    """
289    Choose the playback backend and run its authorization branch.
290
291    :param session: The setup session driving the flow.
292    :param setup_data: The setup data collected so far, updated in place.
293    :param account_id: The signed-in Spotify user id, when known.
294    """
295    # The stored choice wins; everything else preselects librespot. It is the
296    # short path (no consent step, no API key, no pairing), and an instance
297    # predating the choice runs it already, so a routine reconfigure cannot nudge
298    # anyone onto another playback path. An account librespot cannot serve
299    # (created since late 2024) has to switch, which the choice step explains.
300    preselect = str(setup_data.get(CONF_PLAYBACK_BACKEND) or "") or BACKEND_LIBRESPOT
301    errors: dict[str, str] | None = None
302    while True:
303        selected = await _choose_playback_backend(session, preselect, errors)
304        errors = None
305        if selected == BACKEND_SOLOIST:
306            if not await _authorize_soloist(session, setup_data, account_id):
307                # consent refused: back to the backend choice with a clear error
308                preselect = BACKEND_SOLOIST
309                errors = {"base": "soloist_consent_required"}
310                continue
311            # the librespot credential is of no further use
312            setup_data[CONF_LIBRESPOT_CREDENTIALS] = None
313        else:
314            setup_data[CONF_LIBRESPOT_CREDENTIALS] = await _authorize_playback(session, account_id)
315            # switching away from soloist: overwrite the soloist secrets; they
316            # only reach the stored setup_data when finish() succeeds, so an
317            # aborted or failed switch keeps them intact
318            setup_data[CONF_SOLOIST_API_KEY] = None
319            setup_data[CONF_SOLOIST_CONSENT] = False
320            setup_data[CONF_SOLOIST_SESSION_DIR] = None
321        setup_data[CONF_PLAYBACK_BACKEND] = selected
322        return
323
324
325async def _choose_playback_backend(
326    session: SetupSession, preselect: str, errors: dict[str, str] | None
327) -> str:
328    """
329    Show the playback backend choice step until a usable backend is selected.
330
331    :param session: The setup session driving the flow.
332    :param preselect: Backend to preselect (the stored or previously chosen one).
333    :param errors: Optional errors to display on the first render.
334    """
335    while True:
336        values = await session.form(
337            [
338                ConfigEntry(
339                    key=CONF_PLAYBACK_BACKEND,
340                    type=ConfigEntryType.STRING,
341                    required=True,
342                    default_value=BACKEND_LIBRESPOT,
343                    value=preselect,
344                    options=[
345                        ConfigValueOption(BACKEND_SOLOIST),
346                        ConfigValueOption(BACKEND_LIBRESPOT),
347                    ],
348                    expanded_options=True,
349                ),
350            ],
351            step_id="playback_backend",
352            errors=errors,
353        )
354        selected = str(values[CONF_PLAYBACK_BACKEND])
355        if selected == BACKEND_SOLOIST:
356            try:
357                verify_platform_supported()
358            except UnsupportedPlatformError:
359                errors = {"base": "soloist_unsupported_platform"}
360                preselect = BACKEND_LIBRESPOT
361                continue
362        return selected
363
364
365async def _authorize_soloist(
366    session: SetupSession, setup_data: dict[str, Any], account_id: str | None
367) -> bool:
368    """
369    Run the soloist branch: consent, API key and account pairing.
370
371    :param session: The setup session driving the flow.
372    :param setup_data: The setup data collected so far, updated in place.
373    :param account_id: The signed-in Spotify user id, to pair against.
374    :return: True when the branch completed, False when consent was refused.
375    """
376    if not await _ask_soloist_consent(session, bool(setup_data.get(CONF_SOLOIST_CONSENT))):
377        return False
378    setup_data[CONF_SOLOIST_CONSENT] = True
379    # an existing paired session can be kept on reconfigure; the API key can
380    # still be updated either way
381    keep_session = await _has_existing_soloist_session(session) and not await _ask_soloist_repair(
382        session
383    )
384    errors: dict[str, str] | None = None
385    while True:
386        await _ask_soloist_api_key(session, setup_data, errors)
387        if keep_session:
388            if await _paired_account_differs(_instance_data_dir(session), account_id):
389                # the kept pairing belongs to another Spotify account, so it
390                # cannot be kept: fall through to pairing again
391                keep_session = False
392                errors = {"base": "soloist_account_mismatch"}
393                continue
394            setup_data[CONF_SOLOIST_SESSION_DIR] = None
395            return True
396        try:
397            await _pair_soloist(session, setup_data)
398        except StepExpiredError:
399            errors = {"base": "soloist_pairing_not_completed"}
400            continue
401        except SoloistError as err:
402            # download/refresh problems carry their own translation keys
403            errors = {"base": err.translation_key or "soloist_pairing_failed"}
404            continue
405        except LoginFailed:
406            # a rejected key is the most likely cause; re-show the key step
407            errors = {"base": "soloist_pairing_failed"}
408            continue
409        pairing_dir = Path(session.mass.storage_path) / SOLOIST_PAIRING_DIR / session.flow_id
410        if await _paired_account_differs(pairing_dir, account_id):
411            # the user picked the device from a Spotify app signed in as someone
412            # else; discard it so the retry starts from a clean directory
413            setup_data[CONF_SOLOIST_SESSION_DIR] = None
414            await _discard_pairing_dir(session)
415            errors = {"base": "soloist_account_mismatch"}
416            continue
417        return True
418
419
420def _instance_data_dir(session: SetupSession) -> Path:
421    """Return the soloist data dir of the instance being reconfigured."""
422    return (
423        Path(session.mass.storage_path)
424        / "spotify"
425        / str(session.context.instance_id)
426        / SOLOIST_DATA_DIR_NAME
427    )
428
429
430async def _paired_account_differs(data_dir: Path, account_id: str | None) -> bool:
431    """
432    Return whether a paired session belongs to a different account than the sign-in.
433
434    Answers False whenever either side is unknown, so a session whose account
435    cannot be read never blocks a setup that is otherwise fine.
436
437    :param data_dir: The soloist data dir holding the paired session.
438    :param account_id: The signed-in Spotify user id, when known.
439    """
440    if not account_id:
441        return False
442    paired = await asyncio.to_thread(soloist_session_account, data_dir)
443    # the engine records Spotify's canonical username, which is the signed-in
444    # id lowercased. It is stored percent-encoded when it contains non-ASCII
445    # characters (e.g. legacy usernames with accented letters), so decode it
446    # before comparing.
447    if not paired or unquote(paired).casefold() == account_id.casefold():
448        return False
449    LOGGER.warning("Soloist is paired with %s instead of %s", paired, account_id)
450    return True
451
452
453async def _ask_soloist_consent(session: SetupSession, prefill: bool) -> bool:
454    """
455    Show the soloist warning/consent step and return whether consent was given.
456
457    :param session: The setup session driving the flow.
458    :param prefill: Whether consent was already given on an earlier run.
459    """
460    values = await session.form(
461        [
462            ConfigEntry(
463                key=CONF_SOLOIST_CONSENT,
464                type=ConfigEntryType.BOOLEAN,
465                required=False,
466                default_value=False,
467                value=prefill,
468            ),
469        ],
470        step_id="soloist_terms",
471    )
472    return bool(values.get(CONF_SOLOIST_CONSENT))
473
474
475async def _ask_soloist_api_key(
476    session: SetupSession, setup_data: dict[str, Any], errors: dict[str, str] | None = None
477) -> None:
478    """
479    Collect the Soloist API key.
480
481    An already stored key (reconfigure) is kept when the field is left empty;
482    it is never shown back to the user.
483
484    :param session: The setup session driving the flow.
485    :param setup_data: The setup data collected so far, updated in place.
486    :param errors: Optional errors to display on the first render.
487    """
488    has_stored_key = bool(setup_data.get(CONF_SOLOIST_API_KEY))
489    while True:
490        entries = [
491            ConfigEntry(
492                key=CONF_SOLOIST_API_KEY,
493                type=ConfigEntryType.SECURE_STRING,
494                required=not has_stored_key,
495            ),
496        ]
497        if has_stored_key:
498            entries.insert(0, ConfigEntry(key="soloist_api_key_hint", type=ConfigEntryType.LABEL))
499        values = await session.form(entries, step_id="soloist_api_key", errors=errors)
500        api_key = str(values.get(CONF_SOLOIST_API_KEY) or "").strip()
501        if api_key or not has_stored_key:
502            if len(api_key) < MIN_API_KEY_LENGTH:
503                errors = {CONF_SOLOIST_API_KEY: "soloist_api_key_invalid"}
504                continue
505            setup_data[CONF_SOLOIST_API_KEY] = api_key
506        return
507
508
509async def _has_existing_soloist_session(session: SetupSession) -> bool:
510    """Return whether the instance being reconfigured already has a paired session."""
511    if not session.context.instance_id:
512        return False
513    return await asyncio.to_thread(soloist_session_present, _instance_data_dir(session))
514
515
516async def _ask_soloist_repair(session: SetupSession) -> bool:
517    """Ask whether the existing paired session should be replaced by a new pairing."""
518    values = await session.form(
519        [
520            ConfigEntry(
521                key=CONF_SOLOIST_REPAIR,
522                type=ConfigEntryType.BOOLEAN,
523                required=False,
524                default_value=False,
525            ),
526        ],
527        step_id="soloist_repair",
528    )
529    return bool(values.get(CONF_SOLOIST_REPAIR))
530
531
532async def _pair_soloist(session: SetupSession, setup_data: dict[str, Any]) -> None:
533    """
534    Pair the Spotify account through the Spotify app and record the session dir.
535
536    The session is paired into a flow-private directory (this flow may be setting
537    up a brand new instance that has no instance id yet); the provider adopts it
538    into its per-instance data dir on the next load.
539
540    :param session: The setup session driving the flow.
541    :param setup_data: The setup data collected so far, updated in place.
542    """
543    pairing_dir = f"{SOLOIST_PAIRING_DIR}/{session.flow_id}"
544    api_key = str(setup_data.get(CONF_SOLOIST_API_KEY) or "")
545    await session.progress_until(
546        pair_soloist_session(session.mass, api_key, Path(session.mass.storage_path) / pairing_dir),
547        step_id="soloist_pairing",
548        text="soloist_pairing_instructions",
549        expires_in=PAIRING_TIMEOUT,
550    )
551    setup_data[CONF_SOLOIST_SESSION_DIR] = pairing_dir
552
553
554async def _discard_pairing_dir(session: SetupSession) -> None:
555    """Remove this flow's private pairing directory, if it created one."""
556    pairing_dir = Path(session.mass.storage_path) / SOLOIST_PAIRING_DIR / session.flow_id
557    # the directory holds a reusable Spotify login, so a failure to remove it is
558    # logged rather than swallowed - only "it was never there" is uninteresting
559    await asyncio.to_thread(
560        shutil.rmtree,
561        pairing_dir,
562        onexc=lambda _func, path, err: (
563            LOGGER.warning("Failed to remove the Soloist pairing directory %s: %s", path, err)
564            if not isinstance(err, FileNotFoundError)
565            else None
566        ),
567    )
568
569
570async def _authorize_playback(session: SetupSession, account_id: str | None) -> str:
571    """
572    Obtain librespot's playback credential and return it as stored-credential JSON.
573
574    Lets the user pick between pairing through the Spotify app (the default) and a
575    browser sign-in, for setups where the Spotify app cannot discover Music Assistant.
576    The credential has to belong to the account that signed in: authorizing playback
577    from a Spotify app logged in as someone else would leave the library and the audio
578    on different accounts.
579
580    :param session: The setup session driving the flow.
581    :param account_id: The signed-in Spotify user id to match the credential against;
582        the check is skipped when it (or the credential's own account) is unknown.
583    """
584    try:
585        librespot_bin = await get_librespot_binary()
586    except RuntimeError as err:
587        raise SetupFlowError(str(err), translation_key="librespot_unavailable") from err
588    errors: dict[str, str] | None = None
589    while True:
590        method_values = await session.form(
591            [CONF_ENTRY_PLAYBACK_AUTH_METHOD],
592            step_id="playback_auth",
593            errors=errors,
594        )
595        method = str(method_values.get(CONF_PLAYBACK_AUTH_METHOD) or PLAYBACK_AUTH_APP)
596        # every failure loops back to this form: the account is already authorized by now, so
597        # aborting the flow would throw that away over a retryable mistake
598        try:
599            if method == PLAYBACK_AUTH_APP:
600                credentials = await session.progress_until(
601                    librespot_credentials_via_pairing(librespot_bin, PAIRING_DEVICE_NAME),
602                    step_id="playback_pairing",
603                    text="pairing_instructions",
604                    expires_in=PAIRING_TIMEOUT,
605                )
606            else:
607                credentials = await _authorize_playback_via_browser(session, librespot_bin)
608            if _credential_account_differs(credentials, account_id):
609                errors = {"base": "playback_account_mismatch"}
610                continue
611            return credentials
612        except StepExpiredError:
613            errors = {
614                "base": "pairing_not_completed"
615                if method == PLAYBACK_AUTH_APP
616                else "playback_not_completed"
617            }
618        except SetupFlowError as err:
619            errors = {"base": err.translation_key or "playback_auth_failed"}
620        except LoginFailed, ClientError, KeyError:
621            # librespot refusing the token, a transport failure, or a token response without a
622            # token; LoginFailed's own default key is too generic to show here
623            errors = {"base": "playback_auth_failed"}
624
625
626def _credential_account_differs(credentials: str, account_id: str | None) -> bool:
627    """
628    Return whether a playback credential belongs to a different account than the sign-in.
629
630    Answers False whenever either side is unknown, so an unreadable credential never
631    blocks a setup that is otherwise fine.
632
633    :param credentials: librespot's stored-credential JSON.
634    :param account_id: The signed-in Spotify user id, when known.
635    """
636    if not account_id:
637        return False
638    try:
639        stored = json_loads(credentials)
640    except ValueError:
641        return False
642    if not isinstance(stored, dict):
643        return False
644    # librespot stores Spotify's canonical username, which is the signed-in id
645    # lowercased. It is stored percent-encoded when it contains non-ASCII
646    # characters (e.g. legacy usernames with accented letters), so decode it
647    # before comparing.
648    username = str(stored.get("username") or "")
649    if not username or unquote(username).casefold() == account_id.casefold():
650        return False
651    LOGGER.warning("Playback was authorized for %s instead of %s", username, account_id)
652    return True
653
654
655async def _authorize_playback_via_browser(session: SetupSession, librespot_bin: str) -> str:
656    """
657    Run the keymaster sign-in and return librespot's stored credential.
658
659    Spotify only accepts a loopback redirect for this client id, which Music Assistant
660    serves itself: the step completes on its own when the browser runs on this host, and
661    everyone else pastes back the URL their browser ended up on.
662
663    :param session: The setup session driving the flow.
664    :param librespot_bin: Path to the librespot binary.
665    """
666    code_verifier, code_challenge = pkce.generate_pkce_pair()
667    params = {
668        "response_type": "code",
669        "client_id": KEYMASTER_CLIENT_ID,
670        "scope": " ".join(LIBRESPOT_SCOPE),
671        "code_challenge_method": "S256",
672        "code_challenge": code_challenge,
673        "redirect_uri": LIBRESPOT_REDIRECT_URI,
674    }
675    authorize_url = f"{AUTHORIZE_URL}?{urlencode(params)}"
676    try:
677        # the loopback target is only reachable when the browser runs on this host, in which
678        # case the step completes on its own; everyone else falls through to the paste form
679        callback_params = await session.external_until(
680            await_loopback_authorization(LIBRESPOT_REDIRECT_PORT, LIBRESPOT_REDIRECT_PATH),
681            authorize_url,
682            step_id="playback_browser_open",
683            expires_in=LOOPBACK_WAIT_TIMEOUT,
684        )
685        code = authorization_code_from_params(callback_params)
686    except StepExpiredError, OSError:
687        values = await session.form(
688            [CONF_ENTRY_PLAYBACK_CALLBACK_URL],
689            step_id="playback_browser",
690            expires_in=OAUTH_STEP_TIMEOUT,
691            translation_params=[authorize_url],
692        )
693        code = authorization_code_from_url(str(values.get(CONF_PLAYBACK_CALLBACK_URL) or ""))
694    token_params = {
695        "grant_type": "authorization_code",
696        "code": code,
697        "redirect_uri": LIBRESPOT_REDIRECT_URI,
698        "client_id": KEYMASTER_CLIENT_ID,
699        "code_verifier": code_verifier,
700    }
701    async with session.mass.http_session.post(TOKEN_URL, data=token_params) as response:
702        if response.status != 200:
703            raise SetupFlowError(
704                f"Failed to get access token: {await response.text()}",
705                translation_key="playback_code_invalid",
706            )
707        token_result = await response.json()
708    return await librespot_credentials_via_token(librespot_bin, token_result["access_token"])
709
710
711async def _pkce_authenticate(session: SetupSession, client_id: str, step_id: str) -> dict[str, Any]:
712    """
713    Run the Spotify PKCE auth flow and return the token result (refresh + access token).
714
715    :param session: The setup session driving the flow.
716    :param client_id: The Spotify client id to authenticate with.
717    :param step_id: The external step id (also the i18n key segment).
718    """
719    code_verifier, code_challenge = pkce.generate_pkce_pair()
720    redirect_uri, state = hosted_bounce_redirect(session.callback_url)
721    params = {
722        "response_type": "code",
723        "client_id": client_id,
724        "scope": " ".join(SCOPE),
725        "code_challenge_method": "S256",
726        "code_challenge": code_challenge,
727        "redirect_uri": redirect_uri,
728        "state": state,
729    }
730    callback_params = await session.external(
731        f"{AUTHORIZE_URL}?{urlencode(params)}", step_id=step_id, expires_in=OAUTH_STEP_TIMEOUT
732    )
733    code = authorization_code_from_params(callback_params)
734    token_params = {
735        "grant_type": "authorization_code",
736        "code": code,
737        "redirect_uri": redirect_uri,
738        "client_id": client_id,
739        "code_verifier": code_verifier,
740    }
741    async with session.mass.http_session.post(TOKEN_URL, data=token_params) as response:
742        if response.status != 200:
743            raise SetupFlowError(f"Failed to get access token: {await response.text()}")
744        token_result: dict[str, Any] = await response.json()
745    if not token_result.get("refresh_token"):
746        raise SetupFlowError("No refresh token in the token response")
747    return token_result
748