129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
"""HTTP routes for Spotify login and session management."""
|
|
|
|
import secrets
|
|
from typing import cast
|
|
|
|
import httpx2
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import JSONResponse, RedirectResponse, Response
|
|
|
|
from app.adapters.spotify.auth import (
|
|
build_authorize_url,
|
|
derive_code_challenge,
|
|
exchange_authorization_code,
|
|
generate_code_verifier,
|
|
)
|
|
from app.adapters.spotify.client import SpotifyClient
|
|
from app.adapters.spotify.errors import SpotifyError
|
|
from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession
|
|
from app.config import Settings
|
|
|
|
SESSION_COOKIE_NAME = "discovery_session"
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/api/auth/login")
|
|
def login(request: Request) -> RedirectResponse:
|
|
"""Start Spotify Authorization Code with PKCE login."""
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
pending_logins = cast(PendingLogins, request.app.state.pending_logins)
|
|
state = secrets.token_urlsafe(32)
|
|
code_verifier = generate_code_verifier()
|
|
pending_logins.add(state, code_verifier)
|
|
authorize_url = build_authorize_url(
|
|
application_settings.spotify_client_id,
|
|
application_settings.spotify_redirect_uri,
|
|
state,
|
|
derive_code_challenge(code_verifier),
|
|
)
|
|
return RedirectResponse(authorize_url, status_code=307)
|
|
|
|
|
|
@router.get("/callback")
|
|
async def callback(
|
|
request: Request,
|
|
code: str | None = None,
|
|
state: str | None = None,
|
|
error: str | None = None,
|
|
) -> RedirectResponse:
|
|
"""Complete Spotify login and establish an opaque cookie session."""
|
|
if error is not None or code is None or state is None:
|
|
return _login_error_redirect()
|
|
|
|
pending_logins = cast(PendingLogins, request.app.state.pending_logins)
|
|
code_verifier = pending_logins.pop(state)
|
|
if code_verifier is None:
|
|
return _login_error_redirect()
|
|
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
http = cast(httpx2.AsyncClient, request.app.state.http)
|
|
try:
|
|
tokens = await exchange_authorization_code(
|
|
http,
|
|
client_id=application_settings.spotify_client_id,
|
|
redirect_uri=application_settings.spotify_redirect_uri,
|
|
code=code,
|
|
code_verifier=code_verifier,
|
|
)
|
|
bootstrap_session = SpotifySession(tokens=tokens, account_id="", display_name="")
|
|
current_user = await SpotifyClient(
|
|
http,
|
|
bootstrap_session,
|
|
application_settings,
|
|
).fetch_current_user()
|
|
except (SpotifyError, ValueError):
|
|
return _login_error_redirect()
|
|
|
|
session = SpotifySession(
|
|
tokens=bootstrap_session.tokens,
|
|
account_id=current_user.account_id,
|
|
display_name=current_user.display_name,
|
|
)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
session_id = session_store.create(session)
|
|
response = RedirectResponse("/", status_code=307)
|
|
response.set_cookie(
|
|
SESSION_COOKIE_NAME,
|
|
session_id,
|
|
httponly=True,
|
|
samesite="lax",
|
|
path="/",
|
|
secure=application_settings.session_cookie_secure,
|
|
)
|
|
return response
|
|
|
|
|
|
@router.get("/api/auth/me")
|
|
def current_session(request: Request) -> JSONResponse:
|
|
"""Return the display name for a valid application session."""
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
session = session_store.get(session_id) if session_id is not None else None
|
|
if session is None:
|
|
return JSONResponse({"detail": "Not authenticated"}, status_code=401)
|
|
return JSONResponse({"display_name": session.display_name})
|
|
|
|
|
|
@router.post("/api/auth/logout", status_code=204)
|
|
def logout(request: Request) -> Response:
|
|
"""Remove the current application session and clear its cookie."""
|
|
session_id = request.cookies.get(SESSION_COOKIE_NAME)
|
|
session_store = cast(SessionStore, request.app.state.session_store)
|
|
if session_id is not None:
|
|
session_store.remove(session_id)
|
|
|
|
application_settings = cast(Settings, request.app.state.settings)
|
|
response = Response(status_code=204)
|
|
response.delete_cookie(
|
|
SESSION_COOKIE_NAME,
|
|
path="/",
|
|
secure=application_settings.session_cookie_secure,
|
|
httponly=True,
|
|
samesite="lax",
|
|
)
|
|
return response
|
|
|
|
|
|
def _login_error_redirect() -> RedirectResponse:
|
|
return RedirectResponse("/?login=error", status_code=307)
|