/
/
/
1"""Helper(s) to deal with authentication for (music) providers."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from types import TracebackType
8from typing import TYPE_CHECKING
9
10from aiohttp.web import Request, Response
11from music_assistant_models.enums import EventType
12from music_assistant_models.errors import LoginFailed
13
14from music_assistant.helpers.json import json_loads
15
16if TYPE_CHECKING:
17 from music_assistant.mass import MusicAssistant
18
19LOGGER = logging.getLogger(__name__)
20
21
22class AuthenticationHelper:
23 """Context manager helper class for authentication with a forward and redirect URL."""
24
25 def __init__(self, mass: MusicAssistant, session_id: str, method: str = "GET") -> None:
26 """
27 Initialize the Authentication Helper.
28
29 Params:
30 - session_id: a unique id for this auth session.
31 - method: the HTTP request method to expect, either "GET" or "POST" (default: GET).
32 """
33 self.mass = mass
34 self.session_id = session_id
35 self._cb_path = f"/callback/{self.session_id}"
36 self._callback_response: asyncio.Queue[dict[str, str]] = asyncio.Queue(1)
37 self._method = method
38
39 @property
40 def callback_url(self) -> str:
41 """Return the callback URL."""
42 return f"{self.mass.webserver.base_url}{self._cb_path}"
43
44 async def __aenter__(self) -> AuthenticationHelper:
45 """Enter context manager."""
46 self.mass.webserver.register_dynamic_route(
47 self._cb_path, self._handle_callback, self._method
48 )
49 return self
50
51 async def __aexit__(
52 self,
53 exc_type: type[BaseException] | None,
54 exc_val: BaseException | None,
55 exc_tb: TracebackType | None,
56 ) -> bool | None:
57 """Exit context manager."""
58 self.mass.webserver.unregister_dynamic_route(self._cb_path, self._method)
59 return None
60
61 async def authenticate(self, auth_url: str, timeout: int = 60) -> dict[str, str]:
62 """
63 Start the auth process and return any query params if received on the callback.
64
65 Params:
66 - url: The URL the user needs to open for authentication.
67 - timeout: duration in seconds helpers waits for callback (default: 60).
68 """
69 self.send_url(auth_url)
70 LOGGER.debug("Waiting for authentication callback on %s", self.callback_url)
71 return await self.wait_for_callback(timeout)
72
73 def send_url(self, auth_url: str) -> None:
74 """Send the user to the given URL to authenticate (or fill in a code)."""
75 # redirect the user in the frontend to the auth url
76 self.mass.signal_event(EventType.AUTH_SESSION, self.session_id, auth_url)
77
78 async def wait_for_callback(self, timeout: int = 60) -> dict[str, str]:
79 """Wait for the external party to call the callback and return any query strings."""
80 try:
81 async with asyncio.timeout(timeout):
82 return await self._callback_response.get()
83 except TimeoutError as err:
84 raise LoginFailed("Timeout while waiting for authentication callback") from err
85
86 async def _handle_callback(self, request: Request) -> Response:
87 """Handle callback response."""
88 params = dict(request.query)
89 if request.method == "POST" and request.can_read_body:
90 try:
91 raw_data = await request.read()
92 data = json_loads(raw_data)
93 params.update(data)
94 except Exception as err:
95 LOGGER.error("Failed to parse POST data: %s", err)
96
97 await self._callback_response.put(params)
98 LOGGER.debug("Received callback with params: %s", params)
99 return_html = """
100 <html>
101 <body onload="window.close();">
102 Authentication completed, you may now close this window.
103 </body>
104 </html>
105 """
106 return Response(body=return_html, headers={"content-type": "text/html"})
107