/
/
/
1"""
2Setup flow for the Yandex Music provider.
3
4The user picks a login method (Yandex Passport OAuth Device Flow, QR, or a manually
5supplied music token). Passport flows can remember the session, while manual login stores
6only the supplied music token. The device code (or QR code) is rendered as an inline image
7and completion is detected by polling, not a browser callback the flow UI cannot drive. A
8shown code that elapses is minted afresh and the progress step re-emitted in place.
9
10On success the music token is persisted as setup data; when "remember session" is on the
11long-lived ``x_token`` (and, Device-Flow only, the ``refresh_token``) are stored too so the
12provider can silently refresh expired credentials. The music-token-only path (remember off)
13stores no long-lived secrets, exactly as the old action handlers did.
14"""
15
16from __future__ import annotations
17
18import asyncio
19import base64
20from html import escape
21from typing import TYPE_CHECKING
22
23import segno
24from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
25from music_assistant_models.enums import ConfigEntryType
26from ya_passport_auth import ClientConfig, PassportClient
27from ya_passport_auth.exceptions import (
28 DeviceCodeTimeoutError,
29 InvalidCredentialsError,
30 QRTimeoutError,
31 YaPassportError,
32)
33
34from music_assistant.models.setup_flow import AbortFlow, SetupFlowError, StepExpiredError
35
36from .constants import (
37 CONF_REFRESH_TOKEN,
38 CONF_REMEMBER_SESSION,
39 CONF_TOKEN,
40 CONF_X_TOKEN,
41)
42
43if TYPE_CHECKING:
44 from music_assistant_models.config_entries import ConfigValueType
45 from ya_passport_auth import Credentials
46
47 from music_assistant.models.setup_flow import SetupSession
48
49CONF_METHOD = "method"
50METHOD_DEVICE = "device"
51METHOD_QR = "qr"
52METHOD_TOKEN = "token"
53
54_DEVICE_NAME = "Music Assistant"
55_AUTH_FLOW_TIMEOUT_SECONDS = 15 * 60
56
57
58async def run_setup(session: SetupSession) -> None:
59 """
60 Run the Yandex Music login flow: choose a method, sign in, persist the tokens.
61
62 :param session: The setup session driving the flow.
63 """
64 errors: dict[str, str] | None = None
65 while True:
66 values = await session.form(
67 [
68 ConfigEntry(
69 key=CONF_METHOD,
70 type=ConfigEntryType.STRING,
71 required=True,
72 default_value=METHOD_QR,
73 options=[
74 ConfigValueOption(value=METHOD_QR),
75 ConfigValueOption(value=METHOD_DEVICE),
76 ConfigValueOption(value=METHOD_TOKEN),
77 ],
78 ),
79 ConfigEntry(
80 key=CONF_REMEMBER_SESSION,
81 type=ConfigEntryType.BOOLEAN,
82 required=False,
83 default_value=True,
84 ),
85 ],
86 step_id="user",
87 errors=errors,
88 )
89 method = str(values[CONF_METHOD])
90 if method == METHOD_TOKEN:
91 token_errors: dict[str, str] | None = None
92 while True:
93 token_values = await session.form(
94 [
95 ConfigEntry(
96 key=CONF_TOKEN,
97 type=ConfigEntryType.SECURE_STRING,
98 required=True,
99 )
100 ],
101 step_id="token_login",
102 errors=token_errors,
103 last_step=True,
104 )
105 if token_values[CONF_TOKEN]:
106 break
107 token_errors = {CONF_TOKEN: "required"}
108 collected: dict[str, ConfigValueType] = {
109 CONF_TOKEN: str(token_values[CONF_TOKEN]),
110 CONF_X_TOKEN: None,
111 CONF_REFRESH_TOKEN: None,
112 }
113 else:
114 remember = bool(values[CONF_REMEMBER_SESSION])
115 try:
116 if method == METHOD_QR:
117 creds = await _qr_login(session)
118 else:
119 creds = await _device_login(session)
120 except AbortFlow:
121 raise
122 except YaPassportError as err:
123 errors = {"base": str(err)}
124 continue
125 if creds.music_token is None:
126 errors = {"base": "no_music_token"}
127 continue
128 collected = {CONF_TOKEN: creds.music_token.get_secret()}
129 if remember:
130 collected[CONF_X_TOKEN] = creds.x_token.get_secret()
131 collected[CONF_REFRESH_TOKEN] = (
132 creds.refresh_token.get_secret() if creds.refresh_token is not None else None
133 )
134 else:
135 collected[CONF_X_TOKEN] = None
136 collected[CONF_REFRESH_TOKEN] = None
137 try:
138 await session.finish(collected)
139 return
140 except SetupFlowError as err:
141 errors = {"base": err.translation_key or str(err)}
142
143
144async def _qr_login(session: SetupSession) -> Credentials:
145 """
146 Run the native Yandex Passport QR login, refreshing the code on expiry.
147
148 Shows the QR code as an inline image and polls until the user scans and confirms it
149 in the Yandex app; whenever the scan window elapses a fresh code is minted and the
150 progress step re-emitted in place.
151 """
152 ttl = ClientConfig().qr_poll_total_timeout_seconds
153 loop = asyncio.get_running_loop()
154 deadline = loop.time() + _AUTH_FLOW_TIMEOUT_SECONDS
155 try:
156 async with asyncio.timeout_at(deadline):
157 async with PassportClient.create(config=ClientConfig()) as client:
158 while True:
159 qr = await client.start_qr_login()
160 remaining = deadline - loop.time()
161 if remaining <= 0:
162 raise TimeoutError
163 code_timeout = min(float(ttl), remaining)
164 try:
165 return await session.progress_until(
166 client.poll_qr_until_confirmed(qr, total_timeout=code_timeout),
167 step_id="scan_qr",
168 text="scan_qr",
169 image=_qr_image(qr.qr_url),
170 expires_in=code_timeout,
171 )
172 except StepExpiredError, QRTimeoutError:
173 continue
174 except TimeoutError as err:
175 raise StepExpiredError from err
176
177
178async def _device_login(session: SetupSession) -> Credentials:
179 """
180 Run the native Yandex Passport OAuth Device Flow, refreshing the code on expiry.
181
182 Shows the ``user_code`` + verification URL as an inline image and polls until the user
183 confirms; a code that elapses is minted afresh and the progress step re-emitted in place.
184 """
185 loop = asyncio.get_running_loop()
186 deadline = loop.time() + _AUTH_FLOW_TIMEOUT_SECONDS
187 try:
188 async with asyncio.timeout_at(deadline):
189 async with PassportClient.create(config=ClientConfig()) as client:
190 while True:
191 device = await client.start_device_login(device_name=_DEVICE_NAME)
192 remaining = deadline - loop.time()
193 if remaining <= 0:
194 raise TimeoutError
195 code_timeout = min(float(device.expires_in), remaining)
196 poll_timeout = min(float(device.expires_in) + 60, remaining)
197 try:
198 return await session.progress_until(
199 client.poll_device_until_confirmed(device, total_timeout=poll_timeout),
200 step_id="device_login",
201 text="device_login",
202 image=_device_image(device.user_code, device.verification_url),
203 expires_in=code_timeout,
204 )
205 except StepExpiredError, DeviceCodeTimeoutError:
206 continue
207 except InvalidCredentialsError as err:
208 raise AbortFlow("login_denied") from err
209 except TimeoutError as err:
210 raise StepExpiredError from err
211
212
213def _qr_image(qr_url: str) -> str:
214 """Render a high-contrast QR-login URL as an SVG data URI."""
215 return segno.make(qr_url, error="m").svg_data_uri(
216 scale=4,
217 dark="#000",
218 light="#fff",
219 border=4,
220 )
221
222
223def _device_image(user_code: str, verification_url: str) -> str:
224 """Render the device ``user_code`` + verification URL as an SVG data URI."""
225 display_url = verification_url.removeprefix("https://").removeprefix("http://")
226 svg = (
227 '<svg xmlns="http://www.w3.org/2000/svg" width="420" height="260" '
228 'viewBox="0 0 420 260" role="img">'
229 '<rect width="420" height="260" rx="18" fill="#ffdb4d"/>'
230 '<text x="210" y="40" font-family="sans-serif" font-size="18" font-weight="600" '
231 'text-anchor="middle" fill="#5a4a00">Open this address in a browser</text>'
232 '<rect x="20" y="55" width="380" height="58" rx="10" fill="#fff"/>'
233 '<text x="210" y="93" font-family="sans-serif" font-size="24" font-weight="700" '
234 f'text-anchor="middle" fill="#1a1a1a">{escape(display_url)}</text>'
235 '<text x="210" y="154" font-family="sans-serif" font-size="18" font-weight="600" '
236 'text-anchor="middle" fill="#5a4a00">Then enter this code</text>'
237 '<text x="210" y="220" font-family="monospace" font-size="52" font-weight="700" '
238 f'text-anchor="middle" fill="#1a1a1a">{escape(user_code)}</text>'
239 "</svg>"
240 )
241 return "data:image/svg+xml;base64," + base64.b64encode(svg.encode("utf-8")).decode("ascii")
242