spotify-cli/spotify_cli/auth.py

import hashlib
import base64
import secrets
import webbrowser
import time
import json
from urllib.parse import urlencode
import requests

from .config import load_config, save_token, load_token, clear_token

SPOTIFY_AUTH_URL = "https://accounts.spotify.com/authorize"
SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
SCOPES = [
    "user-read-playback-state",
    "user-modify-playback-state",
    "user-read-currently-playing",
    "playlist-read-private",
    "playlist-read-collaborative",
    "user-library-read",
    "user-read-recently-played",
    "user-top-read",
    "streaming",
]

def _generate_pkce():
    code_verifier = secrets.token_urlsafe(64)[:128]
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return code_verifier, code_challenge

class SpotifyAuth:
    def __init__(self):
        self.cfg = load_config()
        self.client_id = self.cfg.get("client_id", "")
        self.token_data = load_token()

    def _is_token_valid(self):
        if not self.token_data:
            return False
        return time.time() < self.token_data.get("expires_at", 0) - 60

    def get_token(self):
        if self._is_token_valid():
            return self.token_data["access_token"]

        if self.token_data and self.token_data.get("refresh_token"):
            refreshed = self._refresh(self.token_data["refresh_token"])
            if refreshed:
                return self.token_data["access_token"]

        return self._full_auth()

    def _refresh(self, refresh_token):
        resp = requests.post(SPOTIFY_TOKEN_URL, data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": self.client_id,
        })
        if resp.status_code == 200:
            data = resp.json()
            self.token_data = {
                "access_token": data["access_token"],
                "refresh_token": data.get("refresh_token", refresh_token),
                "expires_at": time.time() + data["expires_in"],
            }
            save_token(self.token_data)
            return True
        clear_token()
        self.token_data = None
        return False

    def _full_auth(self):
        if not self.client_id:
            self.client_id = input("Spotify Client ID: ").strip()
            self.cfg["client_id"] = self.client_id
            from .config import save_config
            save_config(self.cfg)

        code_verifier, code_challenge = _generate_pkce()
        state = secrets.token_urlsafe(16)

        params = {
            "client_id": self.client_id,
            "response_type": "code",
            "redirect_uri": "http://localhost:8888/callback",
            "code_challenge_method": "S256",
            "code_challenge": code_challenge,
            "state": state,
            "scope": " ".join(SCOPES),
        }

        auth_url = f"{SPOTIFY_AUTH_URL}?{urlencode(params)}"
        print(f"\nOeffne Browser fuer Spotify Login...")
        print(f"Falls der Browser nicht öffnet, nutze diese URL:\n{auth_url}\n")
        webbrowser.open(auth_url)

        code = self._wait_for_callback()
        if not code:
            print("Authentifizierung fehlgeschlagen.")
            return None

        return self._exchange_code(code, code_verifier)

    def _wait_for_callback(self, timeout=300):
        from http.server import HTTPServer, BaseHTTPRequestHandler
        import threading

        received_code = {"code": None, "state": None}

        class Handler(BaseHTTPRequestHandler):
            def do_GET(self):
                if "/callback" in self.path:
                    from urllib.parse import urlparse, parse_qs
                    query = urlparse(self.path).query
                    params = parse_qs(query)
                    received_code["code"] = params.get("code", [None])[0]
                    received_code["state"] = params.get("state", [None])[0]
                    self.send_response(200)
                    self.send_header("Content-Type", "text/html")
                    self.end_headers()
                    self.wfile.write(b"<h1>Authentifizierung erfolgreich! Du kannst das Fenster schliessen.</h1>")
                else:
                    self.send_response(404)
                    self.end_headers()

            def log_message(self, format, *args):
                pass

        server = HTTPServer(("localhost", 8888), Handler)
        server.timeout = timeout

        start = time.time()
        while time.time() - start < timeout:
            server.handle_request()
            if received_code["code"]:
                break

        server.server_close()
        return received_code["code"]

    def _exchange_code(self, code, code_verifier):
        resp = requests.post(SPOTIFY_TOKEN_URL, data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": "http://localhost:8888/callback",
            "client_id": self.client_id,
            "code_verifier": code_verifier,
        })

        if resp.status_code == 200:
            data = resp.json()
            self.token_data = {
                "access_token": data["access_token"],
                "refresh_token": data["refresh_token"],
                "expires_at": time.time() + data["expires_in"],
            }
            save_token(self.token_data)
            return data["access_token"]

        print(f"Token-Austausch fehlgeschlagen: {resp.status_code}")
        print(resp.text)
        return None

    def logout(self):
        clear_token()
        self.token_data = None
        print("Ausgelogged.")