From d78de2149d7f35897d3e2f365a2c306ea1c06bec Mon Sep 17 00:00:00 2001
From: Rafael Moraes <50295204+glomatico@users.noreply.github.com>
Date: Tue, 9 Sep 2025 16:15:59 -0300
Subject: [PATCH] Refactor device flow auth to separate module
---
votify/device_flow.py | 119 ++++++++++++++++++++++++++++++++++++++++++
votify/spotify_api.py | 106 ++-----------------------------------
2 files changed, 122 insertions(+), 103 deletions(-)
create mode 100644 votify/device_flow.py
diff --git a/votify/device_flow.py b/votify/device_flow.py
new file mode 100644
index 0000000..e90556f
--- /dev/null
+++ b/votify/device_flow.py
@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import json
+import re
+import time
+from urllib.parse import parse_qs, urlparse
+
+import requests
+
+from .utils import check_response
+
+
+class SpotifyDeviceFlow:
+ DEVICE_AUTH_URL = "https://accounts.spotify.com/oauth2/device/authorize"
+ DEVICE_TOKEN_URL = "https://accounts.spotify.com/api/token"
+ DEVICE_RESOLVE_URL = "https://accounts.spotify.com/pair/api/resolve"
+ DEVICE_CLIENT_ID = "65b708073fc0480ea92a077233ca87bd" # Spotify for Desktop
+ DEVICE_SCOPE = "app-remote-control,playlist-modify,playlist-modify-private,playlist-modify-public,playlist-read,playlist-read-collaborative,playlist-read-private,streaming,transfer-auth-session,ugc-image-upload,user-follow-modify,user-follow-read,user-library-modify,user-library-read,user-modify,user-modify-playback-state,user-modify-private,user-personalized,user-read-birthdate,user-read-currently-playing,user-read-email,user-read-play-history,user-read-playback-position,user-read-playback-state,user-read-private,user-read-recently-played,user-top-read"
+ DEVICE_FLOW_USER_AGENT = "Spotify/126600447 Win32_x86_64/0 (PC laptop)"
+
+ def __init__(self, sp_dc: str) -> None:
+ self.session = requests.Session()
+ self.session.cookies.update({"sp_dc": sp_dc})
+
+ def get_token(self) -> dict:
+ """Get access token via device flow."""
+ auth_data = self._initiate_device_authorization()
+ device_code = auth_data["device_code"]
+ user_code = auth_data["user_code"]
+ verification_url = auth_data["verification_uri_complete"]
+ flow_ctx, csrf_token = self._parse_verification_page(verification_url)
+ self._submit_user_code(user_code, flow_ctx, csrf_token, verification_url)
+ return self._exchange_device_code(device_code)
+
+ def _initiate_device_authorization(self) -> dict:
+ response = requests.post(
+ self.DEVICE_AUTH_URL,
+ data={
+ "client_id": self.DEVICE_CLIENT_ID,
+ "scope": self.DEVICE_SCOPE,
+ },
+ headers={
+ "User-Agent": self.DEVICE_FLOW_USER_AGENT,
+ "Content-Type": "application/x-www-form-urlencoded",
+ },
+ )
+ check_response(response)
+ return response.json()
+
+ def _parse_verification_page(self, verification_url: str) -> tuple[str, str]:
+ response = self.session.get(
+ verification_url,
+ allow_redirects=True,
+ )
+ check_response(response)
+
+ parsed_url = urlparse(response.url)
+ try:
+ flow_ctx_full = parse_qs(parsed_url.query)["flow_ctx"][0]
+ flow_ctx = flow_ctx_full.split(":")[0]
+ except (KeyError, IndexError):
+ raise ValueError("Failed to extract flow_ctx")
+
+ csrf_token = self._extract_csrf_token(response.text)
+
+ return flow_ctx, csrf_token
+
+ def _extract_csrf_token(self, html_content: str) -> str | None:
+ pattern = (
+ r''
+ )
+ match = re.search(pattern, html_content, re.DOTALL)
+ try:
+ json_data = json.loads(match.group(1))
+ return json_data["props"]["initialToken"]
+ except (AttributeError, json.JSONDecodeError, KeyError):
+ raise ValueError("Failed to extract CSRF token")
+
+ def _submit_user_code(
+ self,
+ user_code: str,
+ flow_ctx: str,
+ csrf_token: str,
+ referer_url: str,
+ ) -> None:
+ current_ts = int(time.time())
+ response = self.session.post(
+ self.DEVICE_RESOLVE_URL,
+ params={"flow_ctx": f"{flow_ctx}:{current_ts}"},
+ json={"code": user_code},
+ headers={
+ "x-csrf-token": csrf_token,
+ "referer": referer_url,
+ "origin": "https://accounts.spotify.com",
+ "content-type": "application/json",
+ },
+ )
+ check_response(response)
+ try:
+ submit_result = response.json()
+ assert submit_result.get("result") == "ok"
+ except (json.JSONDecodeError, AssertionError):
+ raise ValueError("Failed to submit user code")
+
+ def _exchange_device_code(self, device_code: str) -> dict:
+ response = requests.post(
+ self.DEVICE_TOKEN_URL,
+ data={
+ "client_id": self.DEVICE_CLIENT_ID,
+ "device_code": device_code,
+ "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+ },
+ headers={
+ "User-Agent": self.DEVICE_FLOW_USER_AGENT,
+ "Content-Type": "application/x-www-form-urlencoded",
+ },
+ )
+ check_response(response)
+ return response.json()
diff --git a/votify/spotify_api.py b/votify/spotify_api.py
index 87eba0c..cb637eb 100644
--- a/votify/spotify_api.py
+++ b/votify/spotify_api.py
@@ -8,11 +8,11 @@ import time
import typing
from http.cookiejar import MozillaCookieJar
from pathlib import Path
-from urllib.parse import parse_qs, urlparse
import base62
import requests
+from .device_flow import SpotifyDeviceFlow
from .totp import TOTP
from .utils import check_response
@@ -43,12 +43,6 @@ class SpotifyApi:
EXTEND_TRACK_COLLECTION_WAIT_TIME = 0.5
SERVER_TIME_URL = "https://open.spotify.com/api/server-time"
SESSION_TOKEN_URL = "https://open.spotify.com/api/token"
- DEVICE_AUTH_URL = "https://accounts.spotify.com/oauth2/device/authorize"
- DEVICE_TOKEN_URL = "https://accounts.spotify.com/api/token"
- DEVICE_RESOLVE_URL = "https://accounts.spotify.com/pair/api/resolve"
- DEVICE_CLIENT_ID = "65b708073fc0480ea92a077233ca87bd" # Spotify for Desktop
- DEVICE_SCOPE = "app-remote-control,playlist-modify,playlist-modify-private,playlist-modify-public,playlist-read,playlist-read-collaborative,playlist-read-private,streaming,transfer-auth-session,ugc-image-upload,user-follow-modify,user-follow-read,user-library-modify,user-library-read,user-modify,user-modify-playback-state,user-modify-private,user-personalized,user-read-birthdate,user-read-currently-playing,user-read-email,user-read-play-history,user-read-playback-position,user-read-playback-state,user-read-private,user-read-recently-played,user-top-read"
- DEVICE_FLOW_USER_AGENT = "Spotify/126600447 Win32_x86_64/0 (PC laptop)"
def __init__(
self,
@@ -150,107 +144,13 @@ class SpotifyApi:
self._setup_authorization_with_totp()
def _setup_authorization_with_device_flow(self) -> None:
- token_data = self._get_token_via_device_flow()
+ device_flow = SpotifyDeviceFlow(self.sp_dc)
+ token_data = device_flow.get_token()
self.set_authorization_header(token_data["access_token"])
self.session_auth_expire_time = (
int(time.time()) + token_data["expires_in"]
) * 1000
- def _get_token_via_device_flow(self) -> dict | None:
- auth_data = self._initiate_device_authorization()
- device_code = auth_data["device_code"]
- user_code = auth_data["user_code"]
- verification_url = auth_data["verification_uri_complete"]
- flow_ctx, csrf_token = self._parse_verification_page(verification_url)
- self._submit_user_code(user_code, flow_ctx, csrf_token, verification_url)
- return self._exchange_device_code(device_code)
-
- def _initiate_device_authorization(self) -> dict:
- response = requests.post(
- self.DEVICE_AUTH_URL,
- data={
- "client_id": self.DEVICE_CLIENT_ID,
- "scope": self.DEVICE_SCOPE,
- },
- headers={
- "User-Agent": self.DEVICE_FLOW_USER_AGENT,
- "Content-Type": "application/x-www-form-urlencoded",
- },
- )
- check_response(response)
- return response.json()
-
- def _parse_verification_page(self, verification_url: str) -> tuple[str, str]:
- response = self.session.get(
- verification_url,
- allow_redirects=True,
- )
- check_response(response)
-
- parsed_url = urlparse(response.url)
- try:
- flow_ctx_full = parse_qs(parsed_url.query)["flow_ctx"][0]
- flow_ctx = flow_ctx_full.split(":")[0]
- except (KeyError, IndexError):
- raise ValueError("Failed to extract flow_ctx")
-
- csrf_token = self._extract_csrf_token(response.text)
-
- return flow_ctx, csrf_token
-
- def _extract_csrf_token(self, html_content: str) -> str | None:
- pattern = (
- r''
- )
- match = re.search(pattern, html_content, re.DOTALL)
- try:
- json_data = json.loads(match.group(1))
- return json_data["props"]["initialToken"]
- except (AttributeError, json.JSONDecodeError, KeyError):
- raise ValueError("Failed to extract CSRF token")
-
- def _submit_user_code(
- self,
- user_code: str,
- flow_ctx: str,
- csrf_token: str,
- referer_url: str,
- ) -> None:
- current_ts = int(time.time())
- response = self.session.post(
- self.DEVICE_RESOLVE_URL,
- params={"flow_ctx": f"{flow_ctx}:{current_ts}"},
- json={"code": user_code},
- headers={
- "x-csrf-token": csrf_token,
- "referer": referer_url,
- "origin": "https://accounts.spotify.com",
- "content-type": "application/json",
- },
- )
- check_response(response)
- try:
- submit_result = response.json()
- assert submit_result.get("result") == "ok"
- except (json.JSONDecodeError, AssertionError):
- raise ValueError("Failed to submit user code")
-
- def _exchange_device_code(self, device_code: str) -> dict:
- response = requests.post(
- self.DEVICE_TOKEN_URL,
- data={
- "client_id": self.DEVICE_CLIENT_ID,
- "device_code": device_code,
- "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
- },
- headers={
- "User-Agent": self.DEVICE_FLOW_USER_AGENT,
- "Content-Type": "application/x-www-form-urlencoded",
- },
- )
- check_response(response)
- return response.json()
-
def _refresh_session_auth(self) -> None:
timestamp_session_expire = int(self.session_auth_expire_time)
timestamp_now = time.time() * 1000