/
/
/
1"""
2OAuth glue between Music Assistant and the Google Drive API library.
3
4The google_drive_api library just needs one thing from us: a method that
5returns a currently-valid access token. This class provides that, refreshing
6the token with Google when it has expired.
7"""
8
9from __future__ import annotations
10
11import time
12from typing import TYPE_CHECKING
13from urllib.parse import urlencode
14
15from aiohttp import ClientError
16from google_drive_api.auth import AbstractAuth
17from music_assistant_models.errors import LoginFailed, ProviderUnavailableError
18
19from music_assistant.helpers.oauth import (
20 OAUTH_STEP_TIMEOUT,
21 authorization_code_from_params,
22 hosted_bounce_redirect,
23)
24from music_assistant.models.setup_flow import SetupFlowError
25
26from .constants import OAUTH_AUTHORIZE_URL, OAUTH_SCOPE, OAUTH_TOKEN_URL
27
28if TYPE_CHECKING:
29 from music_assistant.mass import MusicAssistant
30 from music_assistant.models.setup_flow import SetupSession
31
32
33async def authorize(session: SetupSession, client_id: str, client_secret: str) -> str:
34 """
35 Run the Google OAuth consent flow via the setup session and return the refresh token.
36
37 :param session: The setup session driving the flow.
38 :param client_id: The user's Google OAuth client ID.
39 :param client_secret: The user's Google OAuth client secret.
40 """
41 # Google only allows pre-registered redirect URIs, so send the user through the fixed
42 # MA callback page which forwards to the local callback URL smuggled along in `state`
43 redirect_uri, state = hosted_bounce_redirect(session.callback_url)
44 params = {
45 "response_type": "code",
46 "client_id": client_id,
47 "scope": OAUTH_SCOPE,
48 "redirect_uri": redirect_uri,
49 "state": state,
50 # offline access + forced consent so Google always returns a refresh token
51 "access_type": "offline",
52 "prompt": "consent",
53 }
54 result = await session.external(
55 f"{OAUTH_AUTHORIZE_URL}?{urlencode(params)}",
56 step_id="authenticate",
57 expires_in=OAUTH_STEP_TIMEOUT,
58 )
59 code = authorization_code_from_params(result)
60 data = {
61 "grant_type": "authorization_code",
62 "code": code,
63 "client_id": client_id,
64 "client_secret": client_secret,
65 "redirect_uri": redirect_uri,
66 }
67 try:
68 async with session.mass.http_session.post(OAUTH_TOKEN_URL, data=data) as resp:
69 if resp.status != 200:
70 raise SetupFlowError(f"Failed to exchange authorization code: {await resp.text()}")
71 token_info = await resp.json()
72 except ClientError as err:
73 raise SetupFlowError(f"Failed to exchange authorization code: {err}") from err
74 if not (refresh_token := token_info.get("refresh_token")):
75 raise SetupFlowError(
76 "Google did not return a refresh token, please retry the authorization"
77 )
78 return str(refresh_token)
79
80
81class MAGoogleDriveAuth(AbstractAuth):
82 """Provide Google Drive access tokens using a stored refresh token."""
83
84 def __init__(
85 self,
86 mass: MusicAssistant,
87 client_id: str,
88 client_secret: str,
89 refresh_token: str,
90 ) -> None:
91 """Initialise the auth helper."""
92 super().__init__(mass.http_session)
93 self.mass = mass
94 self._client_id = client_id
95 self._client_secret = client_secret
96 self._refresh_token = refresh_token
97 self._access_token: str | None = None
98 self._expires_at: float = 0.0
99
100 async def async_get_access_token(self) -> str:
101 """Return a valid access token, refreshing it if needed."""
102 # refresh 60s early so a token never expires mid-request
103 if self._access_token and time.time() < self._expires_at - 60:
104 return self._access_token
105 return await self._refresh()
106
107 async def _refresh(self) -> str:
108 """Exchange the refresh token for a new access token."""
109 data = {
110 "client_id": self._client_id,
111 "client_secret": self._client_secret,
112 "refresh_token": self._refresh_token,
113 "grant_type": "refresh_token",
114 }
115 try:
116 async with self.mass.http_session.post(OAUTH_TOKEN_URL, data=data) as resp:
117 if resp.status in (400, 401):
118 # invalid_grant and friends: the refresh token was revoked or expired
119 raise LoginFailed(f"Google token refresh failed: {await resp.text()}")
120 resp.raise_for_status()
121 payload = await resp.json()
122 except ClientError as err:
123 # 5xx or network blip: transient, so don't report it as an auth
124 # problem that sends the user back through the OAuth flow
125 raise ProviderUnavailableError(f"Google token refresh failed: {err}") from err
126 self._access_token = str(payload["access_token"])
127 self._expires_at = time.time() + float(payload.get("expires_in", 3600))
128 return self._access_token
129