"""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)