Proposal: Make authentication method configurable

pyproject.toml

@@ -15,7 +15,7 @@ classifiers = [
 ]
 dynamic = ["version"]
 dependencies = [
-    "mopidy >= 4.0.0a14",
+    "mopidy >= 4.0.0a15",
     "pykka >= 4.1",
     "requests >= 2.32",
 ]

src/mopidy_spotify/__init__.py

@@ -1,4 +1,5 @@
 import pathlib
+from enum import Enum
 from importlib.metadata import version
 
 from mopidy import config, ext
@@ -11,6 +12,10 @@ class Extension(ext.Extension):
     ext_name = "spotify"
     version = __version__
 
+    class Provider(Enum):
+        MOPIDY_PROXY = "mopidy_proxy"
+        SPOTIFY_DIRECT = "spotify_direct"
+
     def get_default_config(self):
         return config.read(pathlib.Path(__file__).parent / "ext.conf")
 
@@ -20,8 +25,13 @@ def get_config_schema(self):
         schema["username"] = config.Deprecated()  # since 5.0
         schema["password"] = config.Deprecated()  # since 5.0
 
+        schema["authentication_provider"] = config.String(
+            choices=[p.value for p in self.Provider]
+        )
         schema["client_id"] = config.String()
         schema["client_secret"] = config.Secret()
+        schema["refresh_url"] = config.String()
+        schema["cache_credentials_path"] = config.String(optional=True)
 
         schema["bitrate"] = config.Integer(choices=(96, 160, 320))
         schema["volume_normalization"] = config.Boolean()

src/mopidy_spotify/authenticate.py

@@ -0,0 +1,139 @@
+import base64
+import json
+import secrets
+import urllib.parse
+import urllib.request
+from abc import ABC, abstractmethod
+from pathlib import Path
+
+
+class Authentication(ABC):
+    @abstractmethod
+    def fetch_token(self) -> tuple[dict[str, str] | None, int | None]:
+        """Fetch an OAuth token."""
+
+
+class SpotifyDirectAuthentication(Authentication):
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        refresh_url: str,
+        cache_credentials_path: str | None = None,
+    ):
+        self._client_id: str = client_id
+        self._client_secret: str = client_secret
+        self._refresh_url: str = refresh_url
+        self._credentials: dict[str, str] | None = None
+        self._cache_credentials_path: str | None = cache_credentials_path
+
+    def _save_token(self, token: dict[str, str]) -> None:
+        self._credentials = token
+        if self._cache_credentials_path:
+            with Path(self._cache_credentials_path).open("w") as f:
+                json.dump(token, f)
+
+    def _load_token(self) -> dict[str, str] | None:
+        if self._cache_credentials_path and Path(self._cache_credentials_path).exists():
+            with Path(self._cache_credentials_path).open() as f:
+                return json.load(f)
+        return None
+
+    def _spotify_token_request(self, payload: dict[str, str]) -> tuple[dict[str, str], int]:
+        url = "https://accounts.spotify.com/api/token"
+
+        auth_str = f"{self._client_id}:{self._client_secret}"
+        auth_header = base64.b64encode(auth_str.encode()).decode()
+
+        encoded_data = urllib.parse.urlencode(payload).encode()
+
+        req = urllib.request.Request(url, data=encoded_data, method="POST")
+        req.add_header("Authorization", f"Basic {auth_header}")
+        req.add_header("Content-Type", "application/x-www-form-urlencoded")
+
+        with urllib.request.urlopen(req) as response:
+            return json.loads(response.read().decode()), response.status
+
+    def _refresh_access_token(self, refresh_token: str) -> tuple[dict[str, str], int]:
+        data = {
+            "grant_type": "refresh_token",
+            "refresh_token": refresh_token,
+        }
+        new_token, status = self._spotify_token_request(data)
+        if "refresh_token" not in new_token:
+            new_token["refresh_token"] = refresh_token
+        return new_token, status
+
+    def _exchange_code_for_token(self, code: str) -> tuple[dict[str, str], int]:
+        data = {
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": self._refresh_url,
+        }
+        return self._spotify_token_request(data)
+
+    def _authenticate(self) -> tuple[dict[str, str], int]:
+        state = secrets.token_urlsafe(16)
+
+        auth_params = {
+            "client_id": self._client_id,
+            "response_type": "code",
+            "redirect_uri": self._refresh_url,
+            "scope": "playlist-read-private streaming",
+            "state": state,
+        }
+        authorization_url = f"https://accounts.spotify.com/authorize?{urllib.parse.urlencode(auth_params)}"
+        response_url = input(f"""
+Please go here and authorize: {authorization_url}
+    Paste the full redirect URL here:
+                             """)
+
+        parsed_url = urllib.parse.urlparse(response_url)
+        query_params = urllib.parse.parse_qs(parsed_url.query)
+
+        returned_state = query_params.get("state", [None])[0]
+        code = query_params.get("code", [None])[0]
+
+        if returned_state != state:
+            msg = "STATE MISMATCH: Possible CSRF attack detected."
+            raise RuntimeError(msg)
+
+        if not code:
+            msg = "Authorization failed: No code returned."
+            raise RuntimeError(msg)
+
+        return self._exchange_code_for_token(code)
+
+    def fetch_token(self) -> tuple[dict[str, str] | None, int | None]:
+        token = self._credentials or self._load_token()
+        status = None
+
+        if not token:
+            token, status = self._authenticate()
+        else:
+            token, status = self._refresh_access_token(token["refresh_token"])
+
+        self._save_token(token)
+        return token, status
+
+
+class MopidyProxyAuthentication(Authentication):
+    def __init__(
+        self,
+        request,
+        auth: tuple[str, str] | None,
+        refresh_url: str,
+    ):
+        self._request = request
+        self._auth: tuple[str, str] | None = auth
+        self._refresh_url: str = refresh_url
+
+    def _request_with_retries(self, method, url, *args, **kwargs) -> tuple[dict[str, str] | None, int | None]:
+        result, status = self._request.execute(method, url, *args, **kwargs)
+        return result, status
+
+    def fetch_token(self) -> tuple[dict[str, str] | None, int | None]:
+        data = {"grant_type": "client_credentials"}
+        return self._request_with_retries(
+            "POST", self._refresh_url, auth=self._auth, data=data
+        )

src/mopidy_spotify/backend.py

@@ -23,8 +23,17 @@ def __init__(self, config, audio):
 
     def on_start(self):
         self._web_client = web.SpotifyOAuthClient(
-            client_id=self._config["spotify"]["client_id"],
-            client_secret=self._config["spotify"]["client_secret"],
+            authentication_config=web.SpotifyAuthenticationConfig(
+                provider=Extension.Provider(
+                    self._config["spotify"]["authentication_provider"]
+                ),
+                client_id=self._config["spotify"]["client_id"],
+                client_secret=self._config["spotify"]["client_secret"],
+                refresh_url=self._config["spotify"]["refresh_url"],
+                cache_credentials_path=self._config["spotify"][
+                    "cache_credentials_path"
+                ],
+            ),
             proxy_config=self._config["proxy"],
         )
         self._web_client.login()

src/mopidy_spotify/ext.conf

@@ -1,7 +1,10 @@
 [spotify]
 enabled = true
+authentication_provider = mopidy_proxy
 client_id =
 client_secret =
+refresh_url = https://auth.mopidy.com/spotify/token
+cache_credentials_path = 
 bitrate = 160
 volume_normalization = true
 timeout = 10

src/mopidy_spotify/search.py

@@ -90,9 +90,9 @@ def search(  # noqa: PLR0913
     tracks = (
         [
             translator.web_to_track(web_track)
-            for web_track in result["tracks"]["items"][: config["search_track_count"]]
+            for web_track in result["items"]["items"][: config["search_track_count"]]
         ]
-        if "tracks" in result
+        if "items" in result
         else []
     )
     tracks = [x for x in tracks if x]

src/mopidy_spotify/translator.py

@@ -94,7 +94,7 @@ def web_to_track_refs(web_tracks, *, check_playable=True):
     for web_track in web_tracks:
         ref = web_to_track_ref(
             # The extra level here is to also support "saved track objects".
-            web_track.get("track", web_track),
+            web_track.get("item", web_track),
             check_playable=check_playable,
         )
         if ref is not None:
@@ -113,15 +113,15 @@ def to_playlist(
     if ref is None or as_ref:
         return ref
 
-    web_tracks = web_playlist.get("tracks", {}).get("items") or []
+    web_tracks = web_playlist.get("items", {}).get("items") or []
     if as_items and not isinstance(web_tracks, list):
         return None
 
     if as_items:
         return list(web_to_track_refs(web_tracks))
 
     tracks = [
-        web_to_track(web_track.get("track", {}), bitrate=bitrate)
+        web_to_track(web_track.get("item", {}), bitrate=bitrate)
         for web_track in web_tracks
     ]
     tracks = [t for t in tracks if t]

src/mopidy_spotify/utils.py

@@ -2,7 +2,9 @@
 import itertools
 import logging
 import operator
+import os
 import time
+import urllib.parse
 
 import requests
 from mopidy import httpclient
@@ -54,3 +56,23 @@ def batched(iterable, n):
     it = iter(iterable)
     while batch := tuple(itertools.islice(it, n)):
         yield batch
+
+
+def prepare_url(base_url, url, *args, **kwargs):
+    b = urllib.parse.urlsplit(base_url)
+    u = urllib.parse.urlsplit(url.format(*args))
+
+    if u.scheme or u.netloc:
+        scheme, netloc, path = u.scheme, u.netloc, u.path
+        query = urllib.parse.parse_qsl(u.query, keep_blank_values=True)
+    else:
+        scheme, netloc = b.scheme, b.netloc
+        path = os.path.normpath(os.path.join(b.path, u.path))  # noqa: PTH118
+        query = urllib.parse.parse_qsl(b.query, keep_blank_values=True)
+        query.extend(urllib.parse.parse_qsl(u.query, keep_blank_values=True))
+
+    for key, value in kwargs.items():
+        query.append((key, value))
+
+    encoded_query = urllib.parse.urlencode(dict(query))
+    return urllib.parse.urlunsplit((scheme, netloc, path, encoded_query, ""))