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        self._finish_handler = finish_handler
152        self._translation_owner = f"provider.{context.domain}"
153        self._callback_path = f"/setup_flow/callback/{flow_id}"
154        self._input_future: asyncio.Future[dict[str, ConfigValueType]] | None = None
155        self._callback_future: asyncio.Future[dict[str, str]] | None = None
156        self._step_changed = asyncio.Event()
157        self._unregister_callback_route: Callable[[], None] | None = None
158
159    @property
160    def callback_path(self) -> str:
161        """Return the local callback path that resumes this flow's external step."""
162        self._ensure_callback_route()
163        return self._callback_path
164
165    @property
166    def callback_url(self) -> str:
167        """Return the public callback URL that resumes this flow's external step."""
168        return f"{self.mass.webserver.base_url}{self.callback_path}"
169
170    async def form(
171        self,
172        entries: list[ConfigEntry],
173        step_id: str = "user",
174        errors: dict[str, str] | None = None,
175        last_step: bool | None = None,
176        expires_in: float | None = None,
177        translation_params: list[str] | None = None,
178    ) -> dict[str, ConfigValueType]:
179        """
180        Show a form to the user and wait for the submitted (validated) values.
181
182        :param entries: The config entries that make up the form fields.
183        :param step_id: Stable slug identifying this step (also the i18n key segment).
184        :param errors: Optional field-key (or "base") -> error slug to display.
185        :param last_step: Optional hint that this is the flow's final form.
186        :param expires_in: Optional deadline in seconds; when it passes,
187            StepExpiredError is raised here (and the client countdown runs out).
188        :param translation_params: Optional values for placeholders in the step translations.
189        """
190        step = self._build_step(
191            FlowStepType.FORM,
192            step_id,
193            entries=self._prepare_entries(entries),
194            errors=dict(errors) if errors else {},
195            last_step=last_step,
196            translation_params=translation_params,
197            expires_in=expires_in,
198        )
199        self._input_future = asyncio.get_running_loop().create_future()
200        self._publish_step(step)
201        try:
202            return await self._await_with_deadline(self._input_future, expires_in)
203        finally:
204            self._input_future = None
205
206    async def external(
207        self,
208        url: str,
209        step_id: str = "auth",
210        expires_in: float | None = None,
211    ) -> dict[str, str]:
212        """
213        Send the user to an external URL (e.g. OAuth) and wait for the callback params.
214
215        The flow resumes when the external party (or a bounce page) hits this flow's
216        ``callback_url``; GET query and POST body parameters are merged and returned.
217        As soon as the callback lands a generic progress step replaces the external one,
218        so the client stops asking the user for something they already did.
219
220        :param url: The URL the user must open.
221        :param step_id: Stable slug identifying this step (also the i18n key segment).
222        :param expires_in: Optional deadline in seconds; when it passes,
223            StepExpiredError is raised here (and the client countdown runs out).
224        """
225        self._ensure_callback_route()
226        step = self._build_step(FlowStepType.EXTERNAL, step_id, url=url, expires_in=expires_in)
227        self._callback_future = asyncio.get_running_loop().create_future()
228        self._publish_step(step)
229        try:
230            params = await self._await_with_deadline(self._callback_future, expires_in)
231        finally:
232            self._callback_future = None
233        self.progress("working")
234        return params
235
236    async def external_until(
237        self,
238        awaitable: Awaitable[_T],
239        url: str,
240        step_id: str = "auth",
241        expires_in: float | None = None,
242    ) -> _T:
243        """
244        Show an external "Open URL" step that completes when ``awaitable`` resolves.
245
246        Unlike :meth:`external`, which waits for a browser callback, this drives
247        completion from the given awaitable (e.g. a device-code poll) for flows that
248        have no callback to return to. The step renders identically - an Open button for
249        ``url`` plus a waiting spinner - and is dismissed when the awaitable resolves.
250
251        :param awaitable: The work/wait whose completion advances the flow.
252        :param url: The URL the user must open.
253        :param step_id: Stable slug identifying this step (also the i18n key segment).
254        :param expires_in: Optional deadline in seconds; when it passes,
255            StepExpiredError is raised here (and the client countdown runs out).
256        """
257        step = self._build_step(FlowStepType.EXTERNAL, step_id, url=url, expires_in=expires_in)
258        self._publish_step(step)
259        return await self._await_with_deadline(awaitable, expires_in)
260
261    def progress(
262        self,
263        step_id: str,
264        text: str | None = None,
265        pct: float | None = None,
266        image: str | None = None,
267    ) -> None:
268        """
269        Publish a display-only progress step (fire-and-forget, re-emittable).
270
271        :param step_id: Stable slug identifying this step (also the i18n key segment).
272        :param text: Optional progress text (slug, resolved from the translations).
273        :param pct: Optional completion fraction between 0 and 1.
274        :param image: Optional data-URI illustration (e.g. a pairing QR code).
275        """
276        step = self._build_step(
277            FlowStepType.PROGRESS, step_id, progress_text=text, progress=pct, image=image
278        )
279        self._publish_step(step)
280
281    async def progress_until(
282        self,
283        awaitable: Awaitable[_T],
284        step_id: str,
285        text: str | None = None,
286        image: str | None = None,
287        expires_in: float | None = None,
288    ) -> _T:
289        """
290        Publish a progress step and wait for the given awaitable, deadline-enforced.
291
292        Display and enforcement come from the single ``expires_in`` declaration: the
293        step's countdown and the engine deadline cannot drift. On the deadline
294        StepExpiredError is raised here and the awaitable is cancelled - this also
295        cancels a pre-existing Task (cancellation propagates through the await);
296        pass ``join_task(task)`` (helpers.util) if a task must survive the deadline.
297
298        :param awaitable: The work/wait to perform while the progress step shows.
299        :param step_id: Stable slug identifying this step (also the i18n key segment).
300        :param text: Optional progress text (slug, resolved from the translations).
301        :param image: Optional data-URI illustration (e.g. a pairing QR code).
302        :param expires_in: Optional deadline in seconds.
303        """
304        step = self._build_step(
305            FlowStepType.PROGRESS, step_id, progress_text=text, image=image, expires_in=expires_in
306        )
307        self._publish_step(step)
308        return await self._await_with_deadline(awaitable, expires_in)
309
310    async def finish(self, values: dict[str, ConfigValueType]) -> dict[str, str]:
311        """
312        Complete the flow: persist the collected values and create/reload the target.
313
314        Raises SetupFlowError when applying the values failed (e.g. the provider
315        does not load with them); the author may catch it and loop back to a form.
316        On success the FINISH step is published and the result reference
317        (e.g. ``{"instance_id": ...}``) is returned.
318
319        :param values: The values to persist as the target's setup_data.
320        """
321        if self.finished:
322            msg = "Setup flow already finished"
323            raise RuntimeError(msg)
324        result = await self._finish_handler(self, values)
325        self.finished = True
326        step = self._build_step(FlowStepType.FINISH, "finish", result=result)
327        self._publish_step(step)
328        return result
329
330    def retarget(
331        self,
332        *,
333        domain: str,
334        instance_id: str | None,
335        player_id: str,
336        setup_data: dict[str, Any],
337        values: dict[str, ConfigValueType],
338    ) -> None:
339        """
340        Re-point this running flow at a different (child) player.
341
342        Used by a parent player's wrapper flow to delegate to one of its protocol
343        children: after re-pointing, the child's steps localize under the child's
344        provider namespace and ``finish()`` persists to the child's config. The steps
345        published before the retarget (e.g. a "which device" selection form owned by
346        the parent) keep the parent's namespace.
347
348        :param domain: The child provider's domain.
349        :param instance_id: The child provider instance id.
350        :param player_id: The child player id (the new persist/finish target).
351        :param setup_data: The child's decrypted setup_data, for prefill.
352        :param values: The child's existing config values, for prefill.
353        """
354        self.context = replace(
355            self.context,
356            domain=domain,
357            instance_id=instance_id,
358            player_id=player_id,
359            setup_data=setup_data,
360            values=values,
361        )
362        self._translation_owner = f"provider.{domain}"
363
364    # ------------------------------------------------------------------------------
365    # Engine-facing methods below: called by the SetupFlowMixin, never by flow authors
366    # ------------------------------------------------------------------------------
367
368    def handle_submit(self, values: dict[str, ConfigValueType]) -> SetupFlowStep | None:
369        """
370        Validate submitted form values and hand them to the awaiting flow coroutine.
371
372        Returns the updated FORM step when validation failed (the coroutine is not
373        woken), or None when the values were accepted and the coroutine will resume.
374
375        :param values: The raw values as submitted by the client.
376        """
377        step = self.current_step
378        if (
379            step is None
380            or step.type != FlowStepType.FORM
381            or self._input_future is None
382            or self._input_future.done()
383        ):
384            raise ActionUnavailable("The setup flow is not awaiting form input")
385        self.last_activity = time.monotonic()
386        errors: dict[str, str] = {}
387        parsed: dict[str, ConfigValueType] = {}
388        # gates resolve against the submitted values, so flipping one takes effect on the
389        # same submit regardless of where it sits on the form
390        submitted_entries = [
391            replace(entry, value=values.get(entry.key, entry.value)) for entry in step.entries
392        ]
393        for entry in step.entries:
394            if entry.type in UI_ONLY:
395                continue
396            raw_value = values.get(entry.key, entry.value)
397            try:
398                # parse_value also runs the entry's optional validate callback and
399                # stores the parsed value on the entry (echoed on a re-render)
400                # an entry behind an unmet dependency renders disabled, so demanding a
401                # value the user has no way to supply would wedge the flow
402                parsed[entry.key] = entry.parse_value(
403                    raw_value, allow_none=not entry.dependency_met(submitted_entries)
404                )
405            except TypeError, ValueError:
406                errors[entry.key] = "required" if raw_value in (None, "") else "invalid_value"
407                if not isinstance(raw_value, list):
408                    entry.value = raw_value
409            if entry.type == ConfigEntryType.SECURE_STRING:
410                # secure values are only handed to the flow coroutine; they are never
411                # kept on (or echoed back with) the stored step
412                entry.value = None
413        if errors:
414            step.errors = errors
415            self._publish_step(step)
416            return step
417        step.errors = {}
418        # clear the step-changed marker before waking the coroutine, so the submit
419        # caller can await the *next* published step without racing the coroutine
420        self._step_changed.clear()
421        self._input_future.set_result(parsed)
422        return None
423
424    async def handle_callback(self, request: Request) -> Response:
425        """
426        Handle an incoming request on this flow's external-step callback route.
427
428        :param request: The incoming (GET or POST) request; query parameters and any
429            POST body (JSON or form encoded) are merged into the callback params.
430        """
431        params: dict[str, str] = dict(request.query)
432        if request.method == "POST" and request.can_read_body:
433            try:
434                if request.content_type == "application/json":
435                    body = json_loads(await request.read())
436                    if isinstance(body, dict):
437                        # coerce to str so the declared params contract holds for
438                        # json bodies carrying numbers/bools
439                        params.update({str(key): str(val) for key, val in body.items()})
440                    else:
441                        LOGGER.error("Ignoring non-object JSON setup flow callback body")
442                else:
443                    params.update({key: str(val) for key, val in (await request.post()).items()})
444            except Exception as err:
445                LOGGER.error("Failed to parse setup flow callback body: %s", err)
446        if self._callback_future is not None and not self._callback_future.done():
447            # the values carry the authorization code/token, so only log the keys
448            LOGGER.debug(
449                "Setup flow %s resumed by callback with params: %s",
450                self.flow_id,
451                ", ".join(sorted(params)),
452            )
453            # only a callback that resolves a pending external step counts as
454            # activity: the route is unauthenticated, so bare requests must not
455            # be able to keep the flow alive past the idle TTL
456            self.last_activity = time.monotonic()
457            self._callback_future.set_result(params)
458        else:
459            LOGGER.debug(
460                "Received setup flow callback for %s while no external step is pending",
461                self.flow_id,
462            )
463        return Response(body=CALLBACK_RESPONSE_HTML, headers={"content-type": "text/html"})
464
465    async def wait_for_step_change(self, timeout: float) -> None:
466        """
467        Wait (bounded) until the flow publishes a step, returning silently on timeout.
468
469        :param timeout: Maximum time in seconds to wait.
470        """
471        try:
472            async with asyncio.timeout(timeout):
473                await self._step_changed.wait()
474        except TimeoutError:
475            return
476
477    def publish_abort(self, reason: str) -> None:
478        """
479        Publish the terminal ABORT step for this flow.
480
481        :param reason: Slug describing why the flow was aborted; resolved from the
482            translations (setup_flow.abort.<reason>) when the step is served.
483        """
484        step = self._build_step(FlowStepType.ABORT, "abort", reason=reason)
485        self._publish_step(step)
486
487    def close(self) -> None:
488        """Release the session's engine resources (callback route); called on flow end."""
489        if self._unregister_callback_route is not None:
490            self._unregister_callback_route()
491            self._unregister_callback_route = None
492
493    async def _await_with_deadline(self, awaitable: Awaitable[_T], expires_in: float | None) -> _T:
494        """
495        Await the given awaitable, converting the step deadline into StepExpiredError.
496
497        asyncio.timeout cancels the inner await at the deadline and raises TimeoutError
498        in *this* (the flow's) coroutine - exactly the required inject-into-the-author
499        semantic: catch it right at the awaiting call to refresh the step, or let it
500        propagate so the engine converts it into a timed_out ABORT.
501        """
502        if expires_in is None:
503            return await awaitable
504        try:
505            async with asyncio.timeout(expires_in):
506                return await awaitable
507        except TimeoutError as err:
508            raise StepExpiredError from err
509
510    def _build_step(  # noqa: PLR0913 - mirrors the step model's fields
511        self,
512        step_type: FlowStepType,
513        step_id: str,
514        *,
515        entries: list[ConfigEntry] | None = None,
516        errors: dict[str, str] | None = None,
517        last_step: bool | None = None,
518        url: str | None = None,
519        progress_text: str | None = None,
520        progress: float | None = None,
521        image: str | None = None,
522        result: dict[str, str] | None = None,
523        reason: str | None = None,
524        translation_params: list[str] | None = None,
525        expires_in: float | None = None,
526    ) -> SetupFlowStep:
527        """Build a SetupFlowStep for this flow, stamping owner and absolute deadline."""
528        return SetupFlowStep(
529            flow_id=self.flow_id,
530            step_id=step_id,
531            type=step_type,
532            entries=entries or [],
533            errors=errors or {},
534            last_step=last_step,
535            url=url,
536            progress_text=progress_text,
537            progress=progress,
538            image=image,
539            # absolute epoch deadline so the remaining time survives a client re-fetch
540            expires_at=time.time() + expires_in if expires_in is not None else None,
541            result=result,
542            reason=reason,
543            translation_owner=self._translation_owner,
544            translation_params=translation_params,
545        )
546
547    def _prepare_entries(self, entries: list[ConfigEntry]) -> list[ConfigEntry]:
548        """Return copies of the form entries, validated and stamped for this flow."""
549        prepared: list[ConfigEntry] = []
550        for entry in entries:
551            if entry.action or entry.type == ConfigEntryType.ACTION:
552                # actions are the pseudo-flow mechanism of the options surface;
553                # inside real flows they are structurally banned
554                msg = f"Config entry {entry.key} is an action entry, not allowed in setup flows"
555                raise ValueError(msg)
556            # replace() returns a copy so the (often module-level) entry definitions
557            # are never mutated by submit-side value/owner stamping
558            copied = replace(
559                entry, translation_owner=entry.translation_owner or self._translation_owner
560            )
561            if copied.type == ConfigEntryType.SECURE_STRING:
562                # secrets never leave the server: a (prefilled) secure value is
563                # stripped from the step; the user always (re)types it
564                copied.value = None
565            prepared.append(copied)
566        return prepared
567
568    def _publish_step(self, step: SetupFlowStep) -> None:
569        """Store the step as current, mark activity and push it to subscribers."""
570        LOGGER.debug("Setup flow %s published %s step: %s", self.flow_id, step.type, step.step_id)
571        self.current_step = step
572        self.last_activity = time.monotonic()
573        self._step_changed.set()
574        self.mass.signal_event(EventType.SETUP_FLOW_UPDATED, object_id=self.flow_id, data=step)
575
576    def _ensure_callback_route(self) -> None:
577        """Register the flow's dynamic callback route (idempotent)."""
578        if self._unregister_callback_route is not None:
579            return
580        self._unregister_callback_route = self.mass.webserver.register_dynamic_route(
581            self._callback_path, self.handle_callback
582        )
583