/
/
/
1"""
2Setup flow for the Alexa player provider.
3
4NOTE: Amazon offers no usable OAuth for this integration. Authentication is a
5credential-autofilling reverse proxy (alexapy's AlexaProxy) that routes Amazon's own
6login pages through Music Assistant and captures the resulting session cookies. The
7user still solves any CAPTCHA / 2FA on the real Amazon page. This flow is the closest
8clean expression of that mechanism: one credentials form followed by one external step
9that opens the proxied Amazon login.
10"""
11
12from __future__ import annotations
13
14from typing import TYPE_CHECKING, cast
15
16import aiohttp
17from aiohttp import web
18from alexapy import AlexaLogin, AlexaProxy
19from music_assistant_models.config_entries import ConfigEntry
20from music_assistant_models.enums import ConfigEntryType
21
22from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
23from music_assistant.models.setup_flow import SetupFlowError
24from music_assistant.providers.alexa import (
25 CONF_API_BASIC_AUTH_PASSWORD,
26 CONF_API_BASIC_AUTH_USERNAME,
27 CONF_API_URL,
28 CONF_AUTH_SECRET,
29 CONF_URL,
30 save_cookie,
31)
32
33if TYPE_CHECKING:
34 from music_assistant_models.config_entries import ConfigValueType
35
36 from music_assistant.models.setup_flow import SetupSession
37
38_SUCCESS_HTML = (
39 "<html><body><h2>Login successful!</h2><p>You may now close this window.</p></body></html>"
40)
41
42
43async def run_setup(session: SetupSession) -> None:
44 """
45 Run the Alexa setup flow.
46
47 Collects the Amazon + companion-API credentials, drives the proxied Amazon login and
48 persists the credentials (the captured session cookie is stored out-of-band on disk).
49
50 :param session: The setup session driving the flow.
51 """
52 collected = dict(session.context.setup_data)
53 errors: dict[str, str] | None = None
54 while True:
55 values = await session.form(
56 _credential_entries(collected), step_id="credentials", errors=errors, last_step=False
57 )
58 collected.update(values)
59 login = AlexaLogin(
60 url=str(values[CONF_URL]),
61 email=str(values[CONF_USERNAME]),
62 password=str(values[CONF_PASSWORD]),
63 otp_secret=str(values.get(CONF_AUTH_SECRET) or ""),
64 outputpath=lambda path: path,
65 )
66 if not await _proxy_login(session, login):
67 errors = {"base": "login_failed"}
68 continue
69 # the captured session cookie is pickled to disk (keyed by username) before the
70 # provider is loaded, since loaded_in_mass restores auth from that file
71 await save_cookie(login, str(values[CONF_USERNAME]), session.mass)
72 try:
73 await session.finish(collected)
74 return
75 except SetupFlowError as err:
76 errors = {"base": err.translation_key or str(err)}
77
78
79def _credential_entries(prefill: dict[str, ConfigValueType]) -> list[ConfigEntry]:
80 """Return the credential form entries, prefilling the non-secret fields."""
81 return [
82 ConfigEntry(
83 key=CONF_URL,
84 type=ConfigEntryType.STRING,
85 required=True,
86 default_value="amazon.com",
87 value=prefill.get(CONF_URL),
88 ),
89 ConfigEntry(
90 key=CONF_USERNAME,
91 type=ConfigEntryType.STRING,
92 required=True,
93 value=prefill.get(CONF_USERNAME),
94 ),
95 ConfigEntry(key=CONF_PASSWORD, type=ConfigEntryType.SECURE_STRING, required=True),
96 ConfigEntry(key=CONF_AUTH_SECRET, type=ConfigEntryType.SECURE_STRING, required=False),
97 ConfigEntry(
98 key=CONF_API_URL,
99 type=ConfigEntryType.STRING,
100 required=True,
101 default_value="http://localhost:5000",
102 value=prefill.get(CONF_API_URL),
103 ),
104 ConfigEntry(
105 key=CONF_API_BASIC_AUTH_USERNAME,
106 type=ConfigEntryType.STRING,
107 required=False,
108 value=prefill.get(CONF_API_BASIC_AUTH_USERNAME),
109 ),
110 ConfigEntry(
111 key=CONF_API_BASIC_AUTH_PASSWORD, type=ConfigEntryType.SECURE_STRING, required=False
112 ),
113 ]
114
115
116async def _proxy_login(session: SetupSession, login: AlexaLogin) -> bool:
117 """
118 Open the proxied Amazon login as an external step and report whether it succeeded.
119
120 :param session: The setup session driving the flow.
121 :param login: The AlexaLogin the reverse proxy autofills and captures cookies into.
122 """
123 proxy_path = f"/setup_flow/alexa_proxy/{session.flow_id}/"
124 proxy_url = f"{session.mass.webserver.base_url.rstrip('/')}{proxy_path}"
125 proxy = AlexaProxy(login, proxy_url)
126
127 async def proxy_handler(request: web.Request) -> web.StreamResponse:
128 response = await proxy.all_handler(request)
129 if "Successfully logged in" in getattr(response, "text", ""):
130 # the proxy captured the cookies; poke the flow's callback to resume it
131 async with aiohttp.ClientSession() as http:
132 await http.get(session.callback_url)
133 return web.Response(text=_SUCCESS_HTML, content_type="text/html")
134 return cast("web.StreamResponse", response)
135
136 # one wildcard route (all methods) so every proxied Amazon path under the base
137 # resolves: the login traverses assets and multiple /ap/* pages (mfa, cvf, ...)
138 unregister_proxy_route = session.mass.webserver.register_dynamic_route(
139 f"{proxy_path}*", proxy_handler
140 )
141 try:
142 await session.external(proxy_url, step_id="amazon_login", expires_in=300)
143 return bool(await login.test_loggedin())
144 finally:
145 unregister_proxy_route()
146