music-assistant-server

1"""
2Session model and exceptions for interactive setup flows.
3
4A setup flow is the interactive, multi-step process (credentials, OAuth, pairing, ...)
5that creates or reconfigures a provider or player. A flow is authored as one plain
6coroutine that awaits the user through a SetupSession:
7
8    # music_assistant/providers/<domain>/setup_flow.py
9    async def run_setup(session: SetupSession) -> None:
10        values = await session.form([...])
11        await session.finish(values)
12
13The engine (config controller's SetupFlowMixin) owns the session lifecycle: it stores
14the current step, pushes SETUP_FLOW_UPDATED events, feeds submitted values/callback
15params back into the awaiting coroutine and persists the collected values on finish.
16"""
17
18from __future__ import annotations
19
20import asyncio
21import logging
22import time
23from dataclasses import dataclass, field, replace
24from typing import TYPE_CHECKING, Any, Literal, TypeVar
25
26from aiohttp.web import Request, Response
27from music_assistant_models.config_entries import UI_ONLY, ConfigEntry, ConfigValueType
28from music_assistant_models.enums import ConfigEntryType, EventType, FlowStepType
29from music_assistant_models.errors import ActionUnavailable
30from music_assistant_models.setup_flow import SetupFlowStep
31
32from music_assistant.helpers.json import json_loads
33
34if TYPE_CHECKING:
35    from collections.abc import Awaitable, Callable
36
37    from music_assistant.mass import MusicAssistant
38
39LOGGER = logging.getLogger(__name__)
40
41_T = TypeVar("_T")
42
43FlowKind = Literal["setup", "reconfigure"]
44FlowReason = Literal["user", "auth", "error"]
45
46CALLBACK_RESPONSE_HTML = """
47<html>
48<body onload="window.close();">
49    Authentication completed, you may now close this window.
50</body>
51</html>
52"""
53
54
55class SetupFlowError(Exception):
56    """
57    Error raised into the flow coroutine when session.finish() failed.
58
59    The author may catch it to re-render a form with the error message and retry,
60    or let it propagate so the engine aborts the flow with the failure message.
61    """
62
63    def __init__(self, message: str, translation_key: str | None = None) -> None:
64        """
65        Initialize the error.
66
67        :param message: Human readable (English) description of the failure.
68        :param translation_key: Optional (bare) translation slug of the underlying error,
69            usable by flow authors as a localizable form error.
70        """
71        super().__init__(message)
72        self.translation_key = translation_key
73
74
75class StepExpiredError(Exception):
76    """
77    Raised into the flow coroutine when the active step's deadline passed.
78
79    The author may catch it to refresh the step (e.g. a new QR code or pairing
80    session) or let it propagate so the engine aborts the flow as timed out.
81    """
82
83
84class AbortFlow(Exception):
85    """
86    Raise anywhere inside a flow coroutine to cleanly abort the flow.
87
88    :param reason: Slug describing why the flow was aborted; resolved from the
89        translations (setup_flow.abort.<reason>) when the ABORT step is served.
90    """
91
92    def __init__(self, reason: str = "aborted") -> None:
93        """Initialize with the abort reason slug."""
94        super().__init__(reason)
95        self.reason = reason
96
97
98@dataclass(kw_only=True)
99class SetupFlowContext:
100    """Static context describing what a running setup flow is (re)configuring."""
101
102    # kind: whether this flow sets up a new target or reconfigures an existing one
103    kind: FlowKind
104    # reason: what triggered the flow (user initiated, auth failure or another error)
105    reason: FlowReason
106    # domain: the provider domain the flow belongs to (for players: the player's provider)
107    domain: str
108    # instance_id: the existing provider instance (reconfigure/player flows)
109    instance_id: str | None = None
110    # player_id: the player being set up (player flows only)
111    player_id: str | None = None
112    # setup_data: decrypted values collected by an earlier flow run, for prefill
113    setup_data: dict[str, Any] = field(default_factory=dict)
114    # values: the target's existing (options) config values, for prefill
115    values: dict[str, ConfigValueType] = field(default_factory=dict)
116
117
118class SetupSession:
119    """
120    Coroutine-facing session handle for a running setup flow.
121
122    Handed to the flow coroutine (``run_setup(session)``); the form/external/progress
123    methods each publish a step to the client(s) and - where applicable - suspend the
124    coroutine until the engine feeds the user's response back in.
125    """
126
127    def __init__(
128        self,
129        mass: MusicAssistant,
130        flow_id: str,
131        context: SetupFlowContext,
132        finish_handler: Callable[
133            [SetupSession, dict[str, ConfigValueType]], Awaitable[dict[str, str]]
134        ],
135    ) -> None:
136        """
137        Initialize the session (done by the flow engine, never by flow authors).
138
139        :param mass: The MusicAssistant instance.
140        :param flow_id: Unique id of this flow.
141        :param context: Static context describing the flow's target.
142        :param finish_handler: Engine callback that persists the collected values
143            and creates/reloads the target when the flow finishes.
144        """
145        self.mass = mass
146        self.flow_id = flow_id
147        self.context = context
148        self.current_step: SetupFlowStep | None = None
149        self.finished = False
150        self.last_activity = time.monotonic()
151        # i18n slug of the terminal FINISH step; the engine swaps in a variant when the
152        # target warrants extra closing copy (e.g. a music provider's initial library import)
153        self.finish_step_id = "finish"
154        self._finish_handler = finish_handler
155        self._translation_owner = f"provider.{context.domain}"
156        self._callback_path = f"/setup_flow/callback/{flow_id}"
157        self._input_future: asyncio.Future[dict[str, ConfigValueType]] | None = None
158        self._callback_future: asyncio.Future[dict[str, str]] | None = None
159        self._step_changed = asyncio.Event()
160        self._unregister_callback_route: Callable[[], None] | None = None
161
162    @property
163    def callback_path(self) -> str:
164        """Return the local callback path that resumes this flow's external step."""
165        self._ensure_callback_route()
166        return self._callback_path
167
168    @property
169    def callback_url(self) -> str:
170        """Return the public callback URL that resumes this flow's external step."""
171        return f"{self.mass.webserver.base_url}{self.callback_path}"
172
173    async def form(
174        self,
175        entries: list[ConfigEntry],
176        step_id: str = "user",
177        errors: dict[str, str] | None = None,
178        last_step: bool | None = None,
179        expires_in: float | None = None,
180        translation_params: list[str] | None = None,
181    ) -> dict[str, ConfigValueType]:
182        """
183        Show a form to the user and wait for the submitted (validated) values.
184
185        :param entries: The config entries that make up the form fields.
186        :param step_id: Stable slug identifying this step (also the i18n key segment).
187        :param errors: Optional field-key (or "base") -> error slug to display.
188        :param last_step: Optional hint that this is the flow's final form.
189        :param expires_in: Optional deadline in seconds; when it passes,
190            StepExpiredError is raised here (and the client countdown runs out).
191        :param translation_params: Optional values for placeholders in the step translations.
192        """
193        step = self._build_step(
194            FlowStepType.FORM,
195            step_id,
196            entries=self._prepare_entries(entries),
197            errors=dict(errors) if errors else {},
198            last_step=last_step,
199            translation_params=translation_params,
200            expires_in=expires_in,
201        )
202        self._input_future = asyncio.get_running_loop().create_future()
203        self._publish_step(step)
204        try:
205            return await self._await_with_deadline(self._input_future, expires_in)
206        finally:
207            self._input_future = None
208
209    async def external(
210        self,
211        url: str,
212        step_id: str = "auth",
213        expires_in: float | None = None,
214    ) -> dict[str, str]:
215        """
216        Send the user to an external URL (e.g. OAuth) and wait for the callback params.
217
218        The flow resumes when the external party (or a bounce page) hits this flow's
219        ``callback_url``; GET query and POST body parameters are merged and returned.
220        As soon as the callback lands a generic progress step replaces the external one,
221        so the client stops asking the user for something they already did.
222
223        :param url: The URL the user must open.
224        :param step_id: Stable slug identifying this step (also the i18n key segment).
225        :param expires_in: Optional deadline in seconds; when it passes,
226            StepExpiredError is raised here (and the client countdown runs out).
227        """
228        self._ensure_callback_route()
229        step = self._build_step(FlowStepType.EXTERNAL, step_id, url=url, expires_in=expires_in)
230        self._callback_future = asyncio.get_running_loop().create_future()
231        self._publish_step(step)
232        try:
233            params = await self._await_with_deadline(self._callback_future, expires_in)
234        finally:
235            self._callback_future = None
236        self.progress("working")
237        return params
238
239    async def external_until(
240        self,
241        awaitable: Awaitable[_T],
242        url: str,
243        step_id: str = "auth",
244        expires_in: float | None = None,
245    ) -> _T:
246        """
247        Show an external "Open URL" step that completes when ``awaitable`` resolves.
248
249        Unlike :meth:`external`, which waits for a browser callback, this drives
250        completion from the given awaitable (e.g. a device-code poll) for flows that
251        have no callback to return to. The step renders identically - an Open button for
252        ``url`` plus a waiting spinner - and is dismissed when the awaitable resolves.
253
254        :param awaitable: The work/wait whose completion advances the flow.
255        :param url: The URL the user must open.
256        :param step_id: Stable slug identifying this step (also the i18n key segment).
257        :param expires_in: Optional deadline in seconds; when it passes,
258            StepExpiredError is raised here (and the client countdown runs out).
259        """
260        step = self._build_step(FlowStepType.EXTERNAL, step_id, url=url, expires_in=expires_in)
261        self._publish_step(step)
262        return await self._await_with_deadline(awaitable, expires_in)
263
264    def progress(
265        self,
266        step_id: str,
267        text: str | None = None,
268        pct: float | None = None,
269        image: str | None = None,
270    ) -> None:
271        """
272        Publish a display-only progress step (fire-and-forget, re-emittable).
273
274        :param step_id: Stable slug identifying this step (also the i18n key segment).
275        :param text: Optional progress text (slug, resolved from the translations).
276        :param pct: Optional completion fraction between 0 and 1.
277        :param image: Optional data-URI illustration (e.g. a pairing QR code).
278        """
279        step = self._build_step(
280            FlowStepType.PROGRESS, step_id, progress_text=text, progress=pct, image=image
281        )
282        self._publish_step(step)
283
284    async def progress_until(
285        self,
286        awaitable: Awaitable[_T],
287        step_id: str,
288        text: str | None = None,
289        image: str | None = None,
290        expires_in: float | None = None,
291    ) -> _T:
292        """
293        Publish a progress step and wait for the given awaitable, deadline-enforced.
294
295        Display and enforcement come from the single ``expires_in`` declaration: the
296        step's countdown and the engine deadline cannot drift. On the deadline
297        StepExpiredError is raised here and the awaitable is cancelled - this also
298        cancels a pre-existing Task (cancellation propagates through the await);
299        pass ``join_task(task)`` (helpers.util) if a task must survive the deadline.
300
301        :param awaitable: The work/wait to perform while the progress step shows.
302        :param step_id: Stable slug identifying this step (also the i18n key segment).
303        :param text: Optional progress text (slug, resolved from the translations).
304        :param image: Optional data-URI illustration (e.g. a pairing QR code).
305        :param expires_in: Optional deadline in seconds.
306        """
307        step = self._build_step(
308            FlowStepType.PROGRESS, step_id, progress_text=text, image=image, expires_in=expires_in
309        )
310        self._publish_step(step)
311        return await self._await_with_deadline(awaitable, expires_in)
312
313    async def finish(self, values: dict[str, ConfigValueType]) -> dict[str, str]:
314        """
315        Complete the flow: persist the collected values and create/reload the target.
316
317        Raises SetupFlowError when applying the values failed (e.g. the provider
318        does not load with them); the author may catch it and loop back to a form.
319        On success the FINISH step is published and the result reference
320        (e.g. ``{"instance_id": ...}``) is returned.
321
322        :param values: The values to persist as the target's setup_data.
323        """
324        if self.finished:
325            msg = "Setup flow already finished"
326            raise RuntimeError(msg)
327        result = await self._finish_handler(self, values)
328        self.finished = True
329        step = self._build_step(FlowStepType.FINISH, self.finish_step_id, result=result)
330        self._publish_step(step)
331        return result
332
333    def retarget(
334        self,
335        *,
336        domain: str,
337        instance_id: str | None,
338        player_id: str,
339        setup_data: dict[str, Any],
340        values: dict[str, ConfigValueType],
341    ) -> None:
342        """
343        Re-point this running flow at a different (child) player.
344
345        Used by a parent player's wrapper flow to delegate to one of its protocol
346        children: after re-pointing, the child's steps localize under the child's
347        provider namespace and ``finish()`` persists to the child's config. The steps
348        published before the retarget (e.g. a "which device" selection form owned by
349        the parent) keep the parent's namespace.
350
351        :param domain: The child provider's domain.
352        :param instance_id: The child provider instance id.
353        :param player_id: The child player id (the new persist/finish target).
354        :param setup_data: The child's decrypted setup_data, for prefill.
355        :param values: The child's existing config values, for prefill.
356        """
357        self.context = replace(
358            self.context,
359            domain=domain,
360            instance_id=instance_id,
361            player_id=player_id,
362            setup_data=setup_data,
363            values=values,
364        )
365        self._translation_owner = f"provider.{domain}"
366
367    # ------------------------------------------------------------------------------
368    # Engine-facing methods below: called by the SetupFlowMixin, never by flow authors
369    # ------------------------------------------------------------------------------
370
371    def handle_submit(self, values: dict[str, ConfigValueType]) -> SetupFlowStep | None:
372        """
373        Validate submitted form values and hand them to the awaiting flow coroutine.
374
375        Returns the updated FORM step when validation failed (the coroutine is not
376        woken), or None when the values were accepted and the coroutine will resume.
377
378        :param values: The raw values as submitted by the client.
379        """
380        step = self.current_step
381        if (
382            step is None
383            or step.type != FlowStepType.FORM
384            or self._input_future is None
385            or self._input_future.done()
386        ):
387            raise ActionUnavailable("The setup flow is not awaiting form input")
388        self.last_activity = time.monotonic()
389        errors: dict[str, str] = {}
390        parsed: dict[str, ConfigValueType] = {}
391        # gates resolve against the submitted values, so flipping one takes effect on the
392        # same submit regardless of where it sits on the form
393        submitted_entries = [
394            replace(entry, value=values.get(entry.key, entry.value)) for entry in step.entries
395        ]
396        for entry in step.entries:
397            if entry.type in UI_ONLY:
398                continue
399            raw_value = values.get(entry.key, entry.value)
400            try:
401                # parse_value also runs the entry's optional validate callback and
402                # stores the parsed value on the entry (echoed on a re-render)
403                # an entry behind an unmet dependency renders disabled, so demanding a
404                # value the user has no way to supply would wedge the flow
405                parsed[entry.key] = entry.parse_value(
406                    raw_value, allow_none=not entry.dependency_met(submitted_entries)
407                )
408            except TypeError, ValueError:
409                errors[entry.key] = "required" if raw_value in (None, "") else "invalid_value"
410                if not isinstance(raw_value, list):
411                    entry.value = raw_value
412            if entry.type == ConfigEntryType.SECURE_STRING:
413                # secure values are only handed to the flow coroutine; they are never
414                # kept on (or echoed back with) the stored step
415                entry.value = None
416        if errors:
417            step.errors = errors
418            self._publish_step(step)
419            return step
420        step.errors = {}
421        # clear the step-changed marker before waking the coroutine, so the submit
422        # caller can await the *next* published step without racing the coroutine
423        self._step_changed.clear()
424        self._input_future.set_result(parsed)
425        return None
426
427    async def handle_callback(self, request: Request) -> Response:
428        """
429        Handle an incoming request on this flow's external-step callback route.
430
431        :param request: The incoming (GET or POST) request; query parameters and any
432            POST body (JSON or form encoded) are merged into the callback params.
433        """
434        params: dict[str, str] = dict(request.query)
435        if request.method == "POST" and request.can_read_body:
436            try:
437                if request.content_type == "application/json":
438                    body = json_loads(await request.read())
439                    if isinstance(body, dict):
440                        # coerce to str so the declared params contract holds for
441                        # json bodies carrying numbers/bools
442                        params.update({str(key): str(val) for key, val in body.items()})
443                    else:
444                        LOGGER.error("Ignoring non-object JSON setup flow callback body")
445                else:
446                    params.update({key: str(val) for key, val in (await request.post()).items()})
447            except Exception as err:
448                LOGGER.error("Failed to parse setup flow callback body: %s", err)
449        if self._callback_future is not None and not self._callback_future.done():
450            # the values carry the authorization code/token, so only log the keys
451            LOGGER.debug(
452                "Setup flow %s resumed by callback with params: %s",
453                self.flow_id,
454                ", ".join(sorted(params)),
455            )
456            # only a callback that resolves a pending external step counts as
457            # activity: the route is unauthenticated, so bare requests must not
458            # be able to keep the flow alive past the idle TTL
459            self.last_activity = time.monotonic()
460            self._callback_future.set_result(params)
461        else:
462            LOGGER.debug(
463                "Received setup flow callback for %s while no external step is pending",
464                self.flow_id,
465            )
466        return Response(body=CALLBACK_RESPONSE_HTML, headers={"content-type": "text/html"})
467
468    async def wait_for_step_change(self, timeout: float) -> None:
469        """
470        Wait (bounded) until the flow publishes a step, returning silently on timeout.
471
472        :param timeout: Maximum time in seconds to wait.
473        """
474        try:
475            async with asyncio.timeout(timeout):
476                await self._step_changed.wait()
477        except TimeoutError:
478            return
479
480    def publish_abort(self, reason: str) -> None:
481        """
482        Publish the terminal ABORT step for this flow.
483
484        :param reason: Slug describing why the flow was aborted; resolved from the
485            translations (setup_flow.abort.<reason>) when the step is served.
486        """
487        step = self._build_step(FlowStepType.ABORT, "abort", reason=reason)
488        self._publish_step(step)
489
490    def close(self) -> None:
491        """Release the session's engine resources (callback route); called on flow end."""
492        if self._unregister_callback_route is not None:
493            self._unregister_callback_route()
494            self._unregister_callback_route = None
495
496    async def _await_with_deadline(self, awaitable: Awaitable[_T], expires_in: float | None) -> _T:
497        """
498        Await the given awaitable, converting the step deadline into StepExpiredError.
499
500        asyncio.timeout cancels the inner await at the deadline and raises TimeoutError
501        in *this* (the flow's) coroutine - exactly the required inject-into-the-author
502        semantic: catch it right at the awaiting call to refresh the step, or let it
503        propagate so the engine converts it into a timed_out ABORT.
504        """
505        if expires_in is None:
506            return await awaitable
507        try:
508            async with asyncio.timeout(expires_in):
509                return await awaitable
510        except TimeoutError as err:
511            raise StepExpiredError from err
512
513    def _build_step(  # noqa: PLR0913 - mirrors the step model's fields
514        self,
515        step_type: FlowStepType,
516        step_id: str,
517        *,
518        entries: list[ConfigEntry] | None = None,
519        errors: dict[str, str] | None = None,
520        last_step: bool | None = None,
521        url: str | None = None,
522        progress_text: str | None = None,
523        progress: float | None = None,
524        image: str | None = None,
525        result: dict[str, str] | None = None,
526        reason: str | None = None,
527        translation_params: list[str] | None = None,
528        expires_in: float | None = None,
529    ) -> SetupFlowStep:
530        """Build a SetupFlowStep for this flow, stamping owner and absolute deadline."""
531        return SetupFlowStep(
532            flow_id=self.flow_id,
533            step_id=step_id,
534            type=step_type,
535            entries=entries or [],
536            errors=errors or {},
537            last_step=last_step,
538            url=url,
539            progress_text=progress_text,
540            progress=progress,
541            image=image,
542            # absolute epoch deadline so the remaining time survives a client re-fetch
543            expires_at=time.time() + expires_in if expires_in is not None else None,
544            result=result,
545            reason=reason,
546            translation_owner=self._translation_owner,
547            translation_params=translation_params,
548        )
549
550    def _prepare_entries(self, entries: list[ConfigEntry]) -> list[ConfigEntry]:
551        """Return copies of the form entries, validated and stamped for this flow."""
552        prepared: list[ConfigEntry] = []
553        for entry in entries:
554            if entry.action or entry.type == ConfigEntryType.ACTION:
555                # actions are the pseudo-flow mechanism of the options surface;
556                # inside real flows they are structurally banned
557                msg = f"Config entry {entry.key} is an action entry, not allowed in setup flows"
558                raise ValueError(msg)
559            # replace() returns a copy so the (often module-level) entry definitions
560            # are never mutated by submit-side value/owner stamping
561            copied = replace(
562                entry, translation_owner=entry.translation_owner or self._translation_owner
563            )
564            if copied.type == ConfigEntryType.SECURE_STRING:
565                # secrets never leave the server: a (prefilled) secure value is
566                # stripped from the step; the user always (re)types it
567                copied.value = None
568            prepared.append(copied)
569        return prepared
570
571    def _publish_step(self, step: SetupFlowStep) -> None:
572        """Store the step as current, mark activity and push it to subscribers."""
573        LOGGER.debug("Setup flow %s published %s step: %s", self.flow_id, step.type, step.step_id)
574        self.current_step = step
575        self.last_activity = time.monotonic()
576        self._step_changed.set()
577        self.mass.signal_event(EventType.SETUP_FLOW_UPDATED, object_id=self.flow_id, data=step)
578
579    def _ensure_callback_route(self) -> None:
580        """Register the flow's dynamic callback route (idempotent)."""
581        if self._unregister_callback_route is not None:
582            return
583        self._unregister_callback_route = self.mass.webserver.register_dynamic_route(
584            self._callback_path, self.handle_callback
585        )
586