137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
"""Spotify Authorization Code with PKCE helpers and token exchange."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx2
|
|
|
|
from app.adapters.spotify.errors import SpotifyAuthenticationError
|
|
|
|
AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
|
|
TOKEN_URL = "https://accounts.spotify.com/api/token"
|
|
SCOPES = (
|
|
"user-top-read user-library-read user-read-recently-played "
|
|
"playlist-modify-public playlist-modify-private "
|
|
"user-read-playback-state user-modify-playback-state"
|
|
)
|
|
TOKEN_EXPIRY_SKEW_SECONDS = 60.0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TokenSet:
|
|
"""An access token and its refresh credentials."""
|
|
|
|
access_token: str
|
|
refresh_token: str
|
|
expires_at: float
|
|
|
|
@property
|
|
def is_expired(self) -> bool:
|
|
"""Return whether the access token is expired within the safety skew."""
|
|
return time.monotonic() >= self.expires_at - TOKEN_EXPIRY_SKEW_SECONDS
|
|
|
|
|
|
def generate_code_verifier() -> str:
|
|
"""Create a high-entropy PKCE verifier within the allowed length."""
|
|
return secrets.token_urlsafe(64)
|
|
|
|
|
|
def derive_code_challenge(verifier: str) -> str:
|
|
"""Derive an unpadded base64url S256 challenge from a PKCE verifier."""
|
|
digest = hashlib.sha256(verifier.encode("ascii")).digest()
|
|
return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def build_authorize_url(
|
|
client_id: str,
|
|
redirect_uri: str,
|
|
state: str,
|
|
code_challenge: str,
|
|
) -> str:
|
|
"""Build the Spotify consent URL for the required scopes."""
|
|
query = urlencode(
|
|
{
|
|
"client_id": client_id,
|
|
"response_type": "code",
|
|
"redirect_uri": redirect_uri,
|
|
"state": state,
|
|
"scope": SCOPES,
|
|
"code_challenge_method": "S256",
|
|
"code_challenge": code_challenge,
|
|
}
|
|
)
|
|
return f"{AUTHORIZE_URL}?{query}"
|
|
|
|
|
|
async def exchange_authorization_code(
|
|
http: httpx2.AsyncClient,
|
|
*,
|
|
client_id: str,
|
|
redirect_uri: str,
|
|
code: str,
|
|
code_verifier: str,
|
|
) -> TokenSet:
|
|
"""Exchange a Spotify authorization code for an initial token set."""
|
|
response = await http.post(
|
|
TOKEN_URL,
|
|
data={
|
|
"client_id": client_id,
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
"code_verifier": code_verifier,
|
|
},
|
|
)
|
|
return _parse_token_response(response, existing_refresh_token=None)
|
|
|
|
|
|
async def refresh_access_token(
|
|
http: httpx2.AsyncClient,
|
|
*,
|
|
client_id: str,
|
|
tokens: TokenSet,
|
|
) -> TokenSet:
|
|
"""Refresh an access token while retaining an omitted refresh token."""
|
|
response = await http.post(
|
|
TOKEN_URL,
|
|
data={
|
|
"client_id": client_id,
|
|
"grant_type": "refresh_token",
|
|
"refresh_token": tokens.refresh_token,
|
|
},
|
|
)
|
|
return _parse_token_response(response, existing_refresh_token=tokens.refresh_token)
|
|
|
|
|
|
def _parse_token_response(
|
|
response: httpx2.Response,
|
|
existing_refresh_token: str | None,
|
|
) -> TokenSet:
|
|
if not response.is_success:
|
|
raise SpotifyAuthenticationError(
|
|
f"Spotify token request failed with status {response.status_code}"
|
|
)
|
|
|
|
try:
|
|
payload = response.json()
|
|
access_token = payload["access_token"]
|
|
expires_in = payload["expires_in"]
|
|
refresh_token = payload.get("refresh_token", existing_refresh_token)
|
|
if (
|
|
not isinstance(access_token, str)
|
|
or not isinstance(expires_in, int | float)
|
|
or not isinstance(refresh_token, str)
|
|
):
|
|
raise TypeError
|
|
except (TypeError, KeyError, ValueError) as error:
|
|
raise SpotifyAuthenticationError("Spotify returned an invalid token response") from error
|
|
|
|
return TokenSet(
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
expires_at=time.monotonic() + float(expires_in),
|
|
)
|