music-assistant-server

1"""
2Setup flow for the Yandex Music provider.
3
4The user picks a login method (Yandex Passport OAuth Device Flow or QR) and whether to
5remember the session, then signs in with the native ``PassportClient``: the device code
6(or QR code) is rendered as an inline image and completion is detected by polling, not a
7browser callback the flow UI cannot drive. A shown code that elapses is minted afresh and
8the 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 base64
19from html import escape
20from typing import TYPE_CHECKING
21
22import segno
23from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
24from music_assistant_models.enums import ConfigEntryType
25from ya_passport_auth import ClientConfig, PassportClient
26from ya_passport_auth.exceptions import (
27    DeviceCodeTimeoutError,
28    InvalidCredentialsError,
29    QRTimeoutError,
30    YaPassportError,
31)
32
33from music_assistant.models.setup_flow import AbortFlow, SetupFlowError, StepExpiredError
34
35from .constants import (
36    CONF_REFRESH_TOKEN,
37    CONF_REMEMBER_SESSION,
38    CONF_TOKEN,
39    CONF_X_TOKEN,
40)
41
42if TYPE_CHECKING:
43    from music_assistant_models.config_entries import ConfigValueType
44    from ya_passport_auth import Credentials
45
46    from music_assistant.models.setup_flow import SetupSession
47
48# login method choices on the first form
49CONF_METHOD = "method"
50METHOD_DEVICE = "device"
51METHOD_QR = "qr"
52
53# device name shown on the Yandex confirmation screen during the login flows
54_DEVICE_NAME = "Music Assistant"
55
56
57async def run_setup(session: SetupSession) -> None:
58    """
59    Run the Yandex Music login flow: choose a method, sign in, persist the tokens.
60
61    :param session: The setup session driving the flow.
62    """
63    errors: dict[str, str] | None = None
64    while True:
65        values = await session.form(
66            [
67                ConfigEntry(
68                    key=CONF_METHOD,
69                    type=ConfigEntryType.STRING,
70                    required=True,
71                    default_value=METHOD_DEVICE,
72                    options=[
73                        ConfigValueOption(value=METHOD_DEVICE),
74                        ConfigValueOption(value=METHOD_QR),
75                    ],
76                ),
77                ConfigEntry(
78                    key=CONF_REMEMBER_SESSION,
79                    type=ConfigEntryType.BOOLEAN,
80                    required=False,
81                    default_value=True,
82                ),
83            ],
84            step_id="user",
85            errors=errors,
86        )
87        method = str(values[CONF_METHOD])
88        remember = bool(values[CONF_REMEMBER_SESSION])
89        try:
90            if method == METHOD_QR:
91                creds = await _qr_login(session)
92            else:
93                creds = await _device_login(session)
94        except AbortFlow:
95            raise
96        except YaPassportError as err:
97            errors = {"base": str(err)}
98            continue
99        if creds.music_token is None:
100            errors = {"base": "no_music_token"}
101            continue
102        collected: dict[str, ConfigValueType] = {CONF_TOKEN: creds.music_token.get_secret()}
103        if remember:
104            collected[CONF_X_TOKEN] = creds.x_token.get_secret()
105            collected[CONF_REFRESH_TOKEN] = (
106                creds.refresh_token.get_secret() if creds.refresh_token is not None else None
107            )
108        else:
109            collected[CONF_X_TOKEN] = None
110            collected[CONF_REFRESH_TOKEN] = None
111        try:
112            await session.finish(collected)
113            return
114        except SetupFlowError as err:
115            errors = {"base": err.translation_key or str(err)}
116
117
118async def _qr_login(session: SetupSession) -> Credentials:
119    """
120    Run the native Yandex Passport QR login, refreshing the code on expiry.
121
122    Shows the QR code as an inline image and polls until the user scans and confirms it
123    in the Yandex app; whenever the scan window elapses a fresh code is minted and the
124    progress step re-emitted in place.
125    """
126    ttl = ClientConfig().qr_poll_total_timeout_seconds
127    async with PassportClient.create(config=ClientConfig()) as client:
128        while True:
129            qr = await client.start_qr_login()
130            try:
131                return await session.progress_until(
132                    client.poll_qr_until_confirmed(qr, total_timeout=ttl),
133                    step_id="scan_qr",
134                    text="scan_qr",
135                    image=_qr_image(qr.qr_url),
136                    expires_in=int(ttl),
137                )
138            except StepExpiredError, QRTimeoutError:
139                continue
140
141
142async def _device_login(session: SetupSession) -> Credentials:
143    """
144    Run the native Yandex Passport OAuth Device Flow, refreshing the code on expiry.
145
146    Shows the ``user_code`` + verification URL as an inline image and polls until the user
147    confirms; a code that elapses is minted afresh and the progress step re-emitted in place.
148    """
149    async with PassportClient.create(config=ClientConfig()) as client:
150        while True:
151            device = await client.start_device_login(device_name=_DEVICE_NAME)
152            try:
153                return await session.progress_until(
154                    # give the poll a longer deadline than the step so the step's
155                    # countdown (StepExpiredError -> refresh) always wins the race
156                    client.poll_device_until_confirmed(
157                        device, total_timeout=float(device.expires_in) + 60
158                    ),
159                    step_id="device_login",
160                    text="device_login",
161                    image=_device_image(device.user_code, device.verification_url),
162                    expires_in=float(device.expires_in),
163                )
164            except StepExpiredError, DeviceCodeTimeoutError:
165                continue
166            except InvalidCredentialsError as err:
167                raise AbortFlow("login_denied") from err
168
169
170def _qr_image(qr_url: str) -> str:
171    """Render a QR-login URL as an SVG data URI to display in the flow."""
172    return segno.make(qr_url, error="m").svg_data_uri(scale=4)
173
174
175def _device_image(user_code: str, verification_url: str) -> str:
176    """Render the device ``user_code`` + verification URL as an SVG data URI."""
177    svg = (
178        '<svg xmlns="http://www.w3.org/2000/svg" width="460" height="180" '
179        'viewBox="0 0 460 180" role="img">'
180        '<rect width="460" height="180" rx="16" fill="#ffdb4d"/>'
181        '<text x="230" y="82" font-family="monospace" font-size="46" font-weight="700" '
182        f'text-anchor="middle" fill="#1a1a1a">{escape(user_code)}</text>'
183        '<text x="230" y="130" font-family="sans-serif" font-size="16" '
184        f'text-anchor="middle" fill="#5a4a00">{escape(verification_url)}</text>'
185        "</svg>"
186    )
187    return "data:image/svg+xml;base64," + base64.b64encode(svg.encode("utf-8")).decode("ascii")
188