Drop-in implementations for the three most common stacks. Every example here is real, runnable code — copy it, change the constants at the top, and ship.Documentation Index
Fetch the complete documentation index at: https://docs.flexslot.gg/llms.txt
Use this file to discover all available pages before exploring further.
What these samples do
Each implementation:- Generates PKCE values
- Sends the user to
/authorize - Validates
stateandisson the callback - Exchanges the code at
/token - Stores the access + refresh tokens (in a session — your storage layer is up to you)
- Calls
/api/public/v1/decks/with the access token - Refreshes when the token expires, rotating the refresh token
Node.js (Express)
A complete server-side OAuth client.package.json
{
"name": "flexslot-oauth-example",
"type": "module",
"dependencies": {
"express": "^4.21.0",
"express-session": "^1.18.0",
"undici": "^6.21.0"
}
}
app.js
import express from 'express'
import session from 'express-session'
import crypto from 'node:crypto'
import { request } from 'undici'
const CLIENT_ID = process.env.FLEXSLOT_CLIENT_ID
const CLIENT_SECRET = process.env.FLEXSLOT_CLIENT_SECRET
const REDIRECT_URI = 'http://localhost:3000/oauth/callback'
const AS = 'https://api.flexslot.gg'
const SCOPES = 'decks:read sideboards:read'
const app = express()
app.use(session({
secret: process.env.SESSION_SECRET ?? 'dev-only-change-me',
resave: false,
saveUninitialized: false,
cookie: { httpOnly: true, sameSite: 'lax', secure: false },
}))
function b64url(buf) {
return buf.toString('base64')
.replace(/\+/g, '-').replace(/\//g, '_').replace(/=/g, '')
}
function generatePkce() {
const verifier = b64url(crypto.randomBytes(32))
const challenge = b64url(crypto.createHash('sha256').update(verifier).digest())
return { verifier, challenge }
}
app.get('/login', (req, res) => {
const { verifier, challenge } = generatePkce()
const state = b64url(crypto.randomBytes(16))
req.session.pkceVerifier = verifier
req.session.oauthState = state
const url = new URL(`${AS}/api/public/v1/oauth/authorize`)
url.searchParams.set('response_type', 'code')
url.searchParams.set('client_id', CLIENT_ID)
url.searchParams.set('redirect_uri', REDIRECT_URI)
url.searchParams.set('scope', SCOPES)
url.searchParams.set('state', state)
url.searchParams.set('code_challenge', challenge)
url.searchParams.set('code_challenge_method', 'S256')
res.redirect(url.toString())
})
app.get('/oauth/callback', async (req, res, next) => {
try {
const { code, state, iss, error, error_description } = req.query
if (error) {
return res.status(400).send(`${error}: ${error_description ?? ''}`)
}
if (!state || state !== req.session.oauthState) {
return res.status(400).send('state mismatch — possible CSRF')
}
if (iss !== AS) {
return res.status(400).send('issuer mismatch — possible mix-up attack')
}
const body = new URLSearchParams({
grant_type: 'authorization_code',
code,
redirect_uri: REDIRECT_URI,
code_verifier: req.session.pkceVerifier,
})
const basic = Buffer.from(`${CLIENT_ID}:${CLIENT_SECRET}`).toString('base64')
const { statusCode, body: respBody } = await request(`${AS}/api/public/v1/oauth/token`, {
method: 'POST',
headers: {
'Authorization': `Basic ${basic}`,
'Content-Type': 'application/x-www-form-urlencoded',
},
body: body.toString(),
})
const tokens = await respBody.json()
if (statusCode !== 200) {
return res.status(statusCode).json(tokens)
}
req.session.accessToken = tokens.access_token
req.session.refreshToken = tokens.refresh_token
req.session.expiresAt = Date.now() + tokens.expires_in * 1000
delete req.session.pkceVerifier
delete req.session.oauthState
res.redirect('/me')
} catch (err) {
next(err)
}
})
async function refreshIfNeeded(req) {
if (Date.now() < req.session.expiresAt - 5 * 60 * 1000) return
const basic = Buffer.from(`${CLIENT_ID}:${CLIENT_SECRET}`).toString('base64')
const { statusCode, body } = await request(`${AS}/api/public/v1/oauth/token`, {
method: 'POST',
headers: {
'Authorization': `Basic ${basic}`,
'Content-Type': 'application/x-www-form-urlencoded',
},
body: new URLSearchParams({
grant_type: 'refresh_token',
refresh_token: req.session.refreshToken,
}).toString(),
})
const tokens = await body.json()
if (statusCode !== 200) {
throw new Error(`refresh failed: ${tokens.error}`)
}
req.session.accessToken = tokens.access_token
req.session.refreshToken = tokens.refresh_token
req.session.expiresAt = Date.now() + tokens.expires_in * 1000
}
app.get('/me', async (req, res, next) => {
try {
if (!req.session.accessToken) return res.redirect('/login')
await refreshIfNeeded(req)
const { body } = await request(`${AS}/api/public/v1/decks/`, {
headers: { 'Authorization': `Bearer ${req.session.accessToken}` },
})
const data = await body.json()
res.json(data)
} catch (err) {
next(err)
}
})
app.listen(3000, () => console.log('listening on http://localhost:3000'))
Run it
export FLEXSLOT_CLIENT_ID=cli_01HX2K…
export FLEXSLOT_CLIENT_SECRET=…
export SESSION_SECRET=$(openssl rand -hex 32)
node app.js
http://localhost:3000/login and the flow takes over.
Python (Flask)
Same flow, idiomatic Python.requirements.txt
Flask==3.0.3
requests==2.32.3
app.py
import base64
import hashlib
import os
import secrets
import time
import requests
from flask import Flask, redirect, request, session, url_for, jsonify
CLIENT_ID = os.environ["FLEXSLOT_CLIENT_ID"]
CLIENT_SECRET = os.environ["FLEXSLOT_CLIENT_SECRET"]
REDIRECT_URI = "http://localhost:3000/oauth/callback"
AS = "https://api.flexslot.gg"
SCOPES = "decks:read sideboards:read"
app = Flask(__name__)
app.secret_key = os.environ.get("SESSION_SECRET", "dev-only-change-me")
def b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def generate_pkce() -> tuple[str, str]:
verifier = b64url(secrets.token_bytes(32))
challenge = b64url(hashlib.sha256(verifier.encode("ascii")).digest())
return verifier, challenge
@app.route("/login")
def login():
verifier, challenge = generate_pkce()
state = b64url(secrets.token_bytes(16))
session["pkce_verifier"] = verifier
session["oauth_state"] = state
params = {
"response_type": "code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": SCOPES,
"state": state,
"code_challenge": challenge,
"code_challenge_method": "S256",
}
return redirect(f"{AS}/api/public/v1/oauth/authorize?" + requests.compat.urlencode(params))
@app.route("/oauth/callback")
def callback():
error = request.args.get("error")
if error:
return jsonify({"error": error, "description": request.args.get("error_description")}), 400
if request.args.get("state") != session.get("oauth_state"):
return "state mismatch — possible CSRF", 400
if request.args.get("iss") != AS:
return "issuer mismatch — possible mix-up attack", 400
code = request.args["code"]
resp = requests.post(
f"{AS}/api/public/v1/oauth/token",
auth=(CLIENT_ID, CLIENT_SECRET),
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI,
"code_verifier": session["pkce_verifier"],
},
timeout=10,
)
tokens = resp.json()
if resp.status_code != 200:
return jsonify(tokens), resp.status_code
session["access_token"] = tokens["access_token"]
session["refresh_token"] = tokens["refresh_token"]
session["expires_at"] = time.time() + tokens["expires_in"]
session.pop("pkce_verifier", None)
session.pop("oauth_state", None)
return redirect(url_for("me"))
def refresh_if_needed():
if time.time() < session["expires_at"] - 5 * 60:
return
resp = requests.post(
f"{AS}/api/public/v1/oauth/token",
auth=(CLIENT_ID, CLIENT_SECRET),
data={
"grant_type": "refresh_token",
"refresh_token": session["refresh_token"],
},
timeout=10,
)
tokens = resp.json()
if resp.status_code != 200:
raise RuntimeError(f"refresh failed: {tokens.get('error')}")
session["access_token"] = tokens["access_token"]
session["refresh_token"] = tokens["refresh_token"]
session["expires_at"] = time.time() + tokens["expires_in"]
@app.route("/me")
def me():
if "access_token" not in session:
return redirect(url_for("login"))
refresh_if_needed()
resp = requests.get(
f"{AS}/api/public/v1/decks/",
headers={"Authorization": f"Bearer {session['access_token']}"},
timeout=10,
)
return jsonify(resp.json()), resp.status_code
if __name__ == "__main__":
app.run(host="0.0.0.0", port=3000, debug=True)
Run it
export FLEXSLOT_CLIENT_ID=cli_01HX2K…
export FLEXSLOT_CLIENT_SECRET=…
export SESSION_SECRET=$(openssl rand -hex 32)
python app.py
curl (end-to-end manual flow)
When you want to see every byte on the wire. Useful for debugging.#!/usr/bin/env bash
set -euo pipefail
CLIENT_ID="${FLEXSLOT_CLIENT_ID:?}"
CLIENT_SECRET="${FLEXSLOT_CLIENT_SECRET:?}"
REDIRECT_URI="http://localhost:8765/callback"
AS="https://api.flexslot.gg"
SCOPES="decks:read"
# --- 1. PKCE ---
VERIFIER=$(openssl rand 32 | base64 | tr '+/' '-_' | tr -d '=\n')
CHALLENGE=$(printf '%s' "$VERIFIER" \
| openssl dgst -sha256 -binary \
| base64 | tr '+/' '-_' | tr -d '=\n')
STATE=$(openssl rand -hex 16)
echo "verifier=$VERIFIER"
echo "challenge=$CHALLENGE"
echo "state=$STATE"
# --- 2. Authorization URL ---
AUTH_URL="$AS/api/public/v1/oauth/authorize?response_type=code"
AUTH_URL+="&client_id=$CLIENT_ID"
AUTH_URL+="&redirect_uri=$(python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=""))' "$REDIRECT_URI")"
AUTH_URL+="&scope=$(python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=""))' "$SCOPES")"
AUTH_URL+="&state=$STATE"
AUTH_URL+="&code_challenge=$CHALLENGE"
AUTH_URL+="&code_challenge_method=S256"
echo
echo "Open in browser:"
echo "$AUTH_URL"
echo
# --- 3. Capture the code ---
read -r -p "Paste the code from the callback URL: " CODE
read -r -p "Paste the state from the callback URL: " CB_STATE
read -r -p "Paste the iss from the callback URL: " CB_ISS
[[ "$CB_STATE" == "$STATE" ]] || { echo "state mismatch — abort"; exit 1; }
[[ "$CB_ISS" == "$AS" ]] || { echo "iss mismatch — abort"; exit 1; }
# --- 4. Exchange ---
TOKENS=$(curl -sS -X POST "$AS/api/public/v1/oauth/token" \
-u "$CLIENT_ID:$CLIENT_SECRET" \
-d "grant_type=authorization_code" \
-d "code=$CODE" \
-d "redirect_uri=$REDIRECT_URI" \
-d "code_verifier=$VERIFIER")
echo "$TOKENS" | python3 -m json.tool
ACCESS_TOKEN=$(echo "$TOKENS" | python3 -c 'import sys,json; print(json.load(sys.stdin)["access_token"])')
# --- 5. Call the API ---
echo
echo "Fetching decks:"
curl -sS "$AS/api/public/v1/decks/" \
-H "Authorization: Bearer $ACCESS_TOKEN" \
| python3 -m json.tool
What every sample gets right
Validates state
Compared against the value stored at flow start. Mismatch = abort.
Validates iss
Defends against mix-up attacks (RFC 9207).
Uses S256 PKCE
plain is forbidden. SHA-256 only.Rotates refresh tokens
Every refresh stores the new token. The old one is dead.
Sends form-encoded body
Content-Type: application/x-www-form-urlencoded, not JSON.Refreshes proactively
5 minutes before expiry, not after the API returns 401.
Treat these as starting points. Production code should add retry/backoff, structured logging that never logs tokens, and a real token store (database or encrypted Redis) instead of session storage.