74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""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_generation: int = 0
|
|
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."""
|
|
now = time.monotonic()
|
|
expired_states = (
|
|
pending_state
|
|
for pending_state, (_, created_at) in self._entries.items()
|
|
if now - created_at >= PENDING_LOGIN_LIFETIME_SECONDS
|
|
)
|
|
for expired_state in tuple(expired_states):
|
|
del self._entries[expired_state]
|
|
self._entries[state] = (code_verifier, now)
|
|
|
|
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
|