/
/
/
1"""Setup flow for the Apple Music provider."""
2
3from __future__ import annotations
4
5import json
6import pathlib
7import re
8from typing import TYPE_CHECKING
9
10from aiohttp import ClientError, ClientTimeout, web
11from music_assistant_models.config_entries import ConfigEntry
12from music_assistant_models.enums import ConfigEntryType
13
14from music_assistant.constants import CONF_ENTRY_UNOFFICIAL_PROVIDER
15from music_assistant.models.setup_flow import AbortFlow, SetupFlowError
16
17from .constants import (
18 CONF_MUSIC_APP_TOKEN,
19 CONF_MUSIC_USER_MANUAL_TOKEN,
20 CONF_MUSIC_USER_TOKEN,
21 CONF_MUSIC_USER_TOKEN_TIMESTAMP,
22 MUSIC_APP_TOKEN,
23)
24
25if TYPE_CHECKING:
26 from music_assistant_models.config_entries import ConfigValueType
27
28 from music_assistant import MusicAssistant
29 from music_assistant.models.setup_flow import SetupSession
30
31# the MusicKit page self-closes and posts its (possibly empty) token this many seconds
32# before the server-side external-step deadline, so the flow always resumes with params
33MUSICKIT_FLOW_TIMEOUT = 600
34
35
36async def run_setup(session: SetupSession) -> None:
37 """
38 Run the Apple Music setup flow.
39
40 Resolves the Apple Music developer (app) token - preferring the bundled one and
41 prompting for a manual token only when the bundle is empty/expired - then obtains a
42 music user token via the MusicKit browser sign-in (or an advanced manual override)
43 and persists them as setup data.
44
45 :param session: The setup session driving the flow.
46 """
47 mass = session.mass
48 collected: dict[str, ConfigValueType] = {}
49 # prefer the bundled developer token; only prompt for (and store) a manual one when
50 # the bundle ships an empty/expired token, e.g. on development builds
51 app_token = MUSIC_APP_TOKEN
52 # a throttled /v1/test says nothing about validity, so only an explicit rejection counts
53 if await _app_token_accepted(mass, app_token) is False:
54 app_token_errors: dict[str, str] | None = None
55 while True:
56 values = await session.form(
57 [
58 ConfigEntry(
59 key=CONF_MUSIC_APP_TOKEN,
60 type=ConfigEntryType.SECURE_STRING,
61 required=True,
62 )
63 ],
64 step_id="app_token",
65 errors=app_token_errors,
66 )
67 app_token = str(values.get(CONF_MUSIC_APP_TOKEN) or "")
68 # a typed token must be positively accepted, not merely un-rejected
69 if await _app_token_accepted(mass, app_token):
70 break
71 app_token_errors = {CONF_MUSIC_APP_TOKEN: "invalid_value"}
72 collected[CONF_MUSIC_APP_TOKEN] = app_token
73
74 errors: dict[str, str] | None = None
75 while True:
76 user_values = await session.form(
77 [
78 CONF_ENTRY_UNOFFICIAL_PROVIDER,
79 ConfigEntry(
80 key=CONF_MUSIC_USER_MANUAL_TOKEN,
81 type=ConfigEntryType.SECURE_STRING,
82 required=False,
83 advanced=True,
84 help_link="https://www.music-assistant.io/music-providers/apple-music/",
85 ),
86 ],
87 step_id="user",
88 errors=errors,
89 )
90 manual_token = str(user_values.get(CONF_MUSIC_USER_MANUAL_TOKEN) or "").strip()
91 attempt = dict(collected)
92 if manual_token:
93 # advanced escape hatch: a manual user token skips the browser sign-in
94 # (e.g. child accounts, where MusicKit authorize() is unavailable)
95 attempt[CONF_MUSIC_USER_MANUAL_TOKEN] = manual_token
96 else:
97 params = await _musickit_authenticate(session, app_token)
98 token = params.get("music-user-token")
99 if not token:
100 # the page closed or timed out without returning a token
101 raise AbortFlow("auth_cancelled")
102 attempt[CONF_MUSIC_USER_TOKEN] = token
103 # the callback stringifies posted values, so coerce the timestamp back to an
104 # int; CONF_MUSIC_USER_TOKEN_TIMESTAMP is stored as INTEGER, not encrypted
105 attempt[CONF_MUSIC_USER_TOKEN_TIMESTAMP] = int(
106 params.get("music-user-token-timestamp", 0)
107 )
108 try:
109 await session.finish(attempt)
110 return
111 except SetupFlowError as err:
112 errors = {"base": err.translation_key or str(err)}
113
114
115async def _app_token_accepted(mass: MusicAssistant, app_token: str) -> bool | None:
116 """
117 Return whether the API accepted the given Apple Music developer (app) token.
118
119 True when accepted, False when rejected, None when inconclusive (throttled/unreachable).
120
121 :param mass: The MusicAssistant instance.
122 :param app_token: The developer (app) token to validate.
123 """
124 if not app_token:
125 return False
126 try:
127 async with mass.http_session.get(
128 "https://api.music.apple.com/v1/test",
129 headers={"Authorization": f"Bearer {app_token}"},
130 ssl=True,
131 timeout=ClientTimeout(total=10),
132 ) as response:
133 if response.status == 200:
134 return True
135 return False if response.status in (401, 403) else None
136 except ClientError, TimeoutError:
137 return None
138
139
140def _validate_user_token(token: ConfigValueType) -> bool:
141 """
142 Return whether the given value looks like a (base64) Apple Music user token.
143
144 :param token: The candidate music user token to check.
145 """
146 if not isinstance(token, str):
147 return False
148 return bool(re.findall(r"[a-zA-Z0-9=/+]{32,}==$", token))
149
150
151async def _musickit_authenticate(session: SetupSession, app_token: str) -> dict[str, str]:
152 """
153 Serve the MusicKit JS sign-in page and return the params posted back to the flow.
154
155 Registers the (flow-scoped) page/style/glue routes, sends the user to the page via an
156 external step and always unregisters the routes afterwards. The page must stay
157 MA-hosted: MusicKit's authorize() popup and postMessage need a real HTTP origin that
158 can reach the local http callback.
159
160 :param session: The setup session driving the flow.
161 :param app_token: The (validated) developer token the MusicKit page configures with.
162 """
163 mass = session.mass
164 asset_dir = pathlib.Path(__file__).parent.joinpath("musickit_auth")
165 prefill = session.context.setup_data
166 prefill_token = prefill.get(CONF_MUSIC_USER_TOKEN)
167 user_token = prefill_token if _validate_user_token(prefill_token) else ""
168 user_token_timestamp = prefill.get(CONF_MUSIC_USER_TOKEN_TIMESTAMP) or 0
169
170 async def serve_mk_auth_page(request: web.Request) -> web.FileResponse: # noqa: ARG001
171 return web.FileResponse(
172 asset_dir.joinpath("musickit_wrapper.html"),
173 headers={"content-type": "text/html"},
174 )
175
176 async def serve_mk_auth_css(request: web.Request) -> web.FileResponse: # noqa: ARG001
177 return web.FileResponse(
178 asset_dir.joinpath("musickit_wrapper.css"),
179 headers={"content-type": "text/css"},
180 )
181
182 def _js_str(value: object) -> str:
183 # json.dumps yields a quoted, escaped JS string literal; escaping "<" also
184 # neutralizes a "</script>" inside user-supplied values (manual tokens)
185 return json.dumps(str(value)).replace("<", "\\u003c")
186
187 async def serve_mk_glue(request: web.Request) -> web.Response: # noqa: ARG001
188 glue = f"""
189 const return_url={_js_str(session.callback_url)};
190 const app_token={_js_str(app_token)};
191 const callback_method='POST';
192 const user_token={_js_str(user_token)};
193 const user_token_timestamp={_js_str(user_token_timestamp)};
194 const flow_timeout={max(MUSICKIT_FLOW_TIMEOUT - 10, 60)};
195 const mass_version={_js_str(mass.version)};
196 """
197 return web.Response(body=glue, headers={"content-type": "text/javascript"})
198
199 base_path = f"/apple_music_auth/{session.flow_id}/"
200 unregister = [
201 mass.webserver.register_dynamic_route(f"{base_path}index.html", serve_mk_auth_page),
202 mass.webserver.register_dynamic_route(f"{base_path}index.css", serve_mk_auth_css),
203 mass.webserver.register_dynamic_route(f"{base_path}index.js", serve_mk_glue),
204 ]
205 try:
206 return await session.external(
207 f"{mass.webserver.base_url}{base_path}index.html",
208 step_id="auth",
209 expires_in=MUSICKIT_FLOW_TIMEOUT,
210 )
211 finally:
212 for remove in unregister:
213 remove()
214