/
/
/
1"""Setup flow for the Spotify provider."""
2
3from __future__ import annotations
4
5from dataclasses import replace
6from typing import TYPE_CHECKING
7from urllib.parse import urlencode
8
9import pkce
10from music_assistant_models.config_entries import ConfigEntry
11from music_assistant_models.enums import ConfigEntryType
12
13from music_assistant.helpers.app_vars import app_var
14from music_assistant.helpers.oauth import (
15 HOSTED_CALLBACK_URL,
16 OAUTH_STEP_TIMEOUT,
17 authorization_code_from_params,
18 hosted_bounce_redirect,
19)
20from music_assistant.models.setup_flow import SetupFlowError
21
22from .constants import (
23 CONF_CLIENT_ID,
24 CONF_REFRESH_TOKEN_DEV,
25 CONF_REFRESH_TOKEN_GLOBAL,
26 SCOPE,
27)
28
29if TYPE_CHECKING:
30 from music_assistant.models.setup_flow import SetupSession
31
32AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
33TOKEN_URL = "https://accounts.spotify.com/api/token"
34
35# the developer client id is a public OAuth identifier (not a secret), so it is a plain
36# STRING that can be prefilled on reconfigure
37CONF_ENTRY_DEV_CLIENT_ID = ConfigEntry(
38 key=CONF_CLIENT_ID,
39 type=ConfigEntryType.STRING,
40 required=False,
41)
42
43
44async def run_setup(session: SetupSession) -> None:
45 """
46 Run the Spotify setup flow.
47
48 Authenticates the (required) global session with Music Assistant's own client id, then
49 optionally a developer session with the user's own client id, and persists the resulting
50 refresh tokens as setup data.
51
52 :param session: The setup session driving the flow.
53 """
54 setup_data = dict(session.context.setup_data)
55 # the global session always (re)authenticates: a refresh token cannot be reused across a
56 # re-auth and secure values are never prefilled back into the flow
57 setup_data[CONF_REFRESH_TOKEN_GLOBAL] = await _pkce_authenticate(
58 session, app_var("spotify_client_id"), step_id="authenticate"
59 )
60 # optional developer session using the user's own Spotify client id
61 client_id_default = str(session.context.setup_data.get(CONF_CLIENT_ID) or "")
62 errors: dict[str, str] | None = None
63 while True:
64 dev_values = await session.form(
65 [replace(CONF_ENTRY_DEV_CLIENT_ID, value=client_id_default)],
66 step_id="developer",
67 errors=errors,
68 last_step=True,
69 translation_params=[HOSTED_CALLBACK_URL],
70 )
71 client_id = str(dev_values.get(CONF_CLIENT_ID) or "").strip()
72 try:
73 if client_id:
74 setup_data[CONF_CLIENT_ID] = client_id
75 setup_data[CONF_REFRESH_TOKEN_DEV] = await _pkce_authenticate(
76 session, client_id, step_id="authenticate_dev"
77 )
78 else:
79 # no developer client id: clear any previously stored developer session
80 setup_data[CONF_CLIENT_ID] = None
81 setup_data[CONF_REFRESH_TOKEN_DEV] = None
82 await session.finish(setup_data)
83 return
84 except SetupFlowError as err:
85 errors = {"base": err.translation_key or str(err)}
86 client_id_default = client_id
87
88
89async def _pkce_authenticate(session: SetupSession, client_id: str, step_id: str) -> str:
90 """
91 Run the Spotify PKCE auth flow via the setup session and return a refresh token.
92
93 :param session: The setup session driving the flow.
94 :param client_id: The Spotify client id to authenticate with.
95 :param step_id: The external step id (also the i18n key segment).
96 """
97 code_verifier, code_challenge = pkce.generate_pkce_pair()
98 redirect_uri, state = hosted_bounce_redirect(session.callback_url)
99 params = {
100 "response_type": "code",
101 "client_id": client_id,
102 "scope": " ".join(SCOPE),
103 "code_challenge_method": "S256",
104 "code_challenge": code_challenge,
105 "redirect_uri": redirect_uri,
106 "state": state,
107 }
108 callback_params = await session.external(
109 f"{AUTHORIZE_URL}?{urlencode(params)}", step_id=step_id, expires_in=OAUTH_STEP_TIMEOUT
110 )
111 code = authorization_code_from_params(callback_params)
112 token_params = {
113 "grant_type": "authorization_code",
114 "code": code,
115 "redirect_uri": redirect_uri,
116 "client_id": client_id,
117 "code_verifier": code_verifier,
118 }
119 async with session.mass.http_session.post(TOKEN_URL, data=token_params) as response:
120 if response.status != 200:
121 raise SetupFlowError(f"Failed to get access token: {await response.text()}")
122 token_result = await response.json()
123 if not (refresh_token := token_result.get("refresh_token")):
124 raise SetupFlowError("No refresh token in the token response")
125 return str(refresh_token)
126