feat: add Spotify authentication and sessions
This commit is contained in:
parent
000b435b9a
commit
af3b77a7d0
5 changed files with 294 additions and 0 deletions
65
backend/app/adapters/spotify/session.py
Normal file
65
backend/app/adapters/spotify/session.py
Normal file
|
|
@ -0,0 +1,65 @@
|
|||
"""In-memory Spotify login and session state."""
|
||||
|
||||
import asyncio
|
||||
import secrets
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from app.adapters.spotify.auth import TokenSet
|
||||
|
||||
PENDING_LOGIN_LIFETIME_SECONDS = 600.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SpotifySession:
|
||||
"""Authenticated Spotify identity and refresh coordination state."""
|
||||
|
||||
tokens: TokenSet
|
||||
account_id: str
|
||||
display_name: str
|
||||
refresh_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
||||
|
||||
|
||||
class SessionStore:
|
||||
"""Store Spotify sessions behind opaque cookie-safe identifiers."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Create an empty session store."""
|
||||
self._sessions: dict[str, SpotifySession] = {}
|
||||
|
||||
def create(self, session: SpotifySession) -> str:
|
||||
"""Store a session and return its opaque identifier."""
|
||||
session_id = secrets.token_urlsafe(32)
|
||||
self._sessions[session_id] = session
|
||||
return session_id
|
||||
|
||||
def get(self, session_id: str) -> SpotifySession | None:
|
||||
"""Return a session by identifier when present."""
|
||||
return self._sessions.get(session_id)
|
||||
|
||||
def remove(self, session_id: str) -> None:
|
||||
"""Remove a session if it exists."""
|
||||
self._sessions.pop(session_id, None)
|
||||
|
||||
|
||||
class PendingLogins:
|
||||
"""Store short-lived, single-use PKCE verifiers by OAuth state."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Create an empty pending-login store."""
|
||||
self._entries: dict[str, tuple[str, float]] = {}
|
||||
|
||||
def add(self, state: str, code_verifier: str) -> None:
|
||||
"""Store a PKCE verifier for a newly issued OAuth state."""
|
||||
self._entries[state] = (code_verifier, time.monotonic())
|
||||
|
||||
def pop(self, state: str) -> str | None:
|
||||
"""Consume a verifier unless its OAuth state is unknown or expired."""
|
||||
entry = self._entries.pop(state, None)
|
||||
if entry is None:
|
||||
return None
|
||||
|
||||
code_verifier, created_at = entry
|
||||
if time.monotonic() - created_at >= PENDING_LOGIN_LIFETIME_SECONDS:
|
||||
return None
|
||||
return code_verifier
|
||||
Loading…
Add table
Add a link
Reference in a new issue