Code Samples - Flexslot V2

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.

What these samples do

Each implementation:

Generates PKCE values
Sends the user to /authorize
Validates state and iss on 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/games/magic-the-gathering/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 AUTHORIZE_URL = 'https://flexslot.gg/oauth/authorize'
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(AUTHORIZE_URL)
  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/games/magic-the-gathering/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=flx_client_01HX2K…
export FLEXSLOT_CLIENT_SECRET=…
export SESSION_SECRET=$(openssl rand -hex 32)
node app.js

Open 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"
AUTHORIZE_URL = "https://flexslot.gg/oauth/authorize"
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"{AUTHORIZE_URL}?" + 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/games/magic-the-gathering/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=flx_client_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"
AUTHORIZE_URL="https://flexslot.gg/oauth/authorize"
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="$AUTHORIZE_URL?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. Identify the connected user (for a "Connected as …" UI) ---
echo
echo "Connected user:"
# Returns {caller, user:{id, username}, scopes, tier}. user.id is the
# bare Firebase id and matches author.id on that user's decks. No extra scope.
curl -sS "$AS/api/public/v1/_probe/user" \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  | python3 -m json.tool

# --- 6. Call the API ---
echo
echo "Fetching decks:"
curl -sS "$AS/api/public/v1/games/magic-the-gathering/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.

​What these samples do

​Node.js (Express)

​package.json

​app.js

​Run it

​Python (Flask)

​requirements.txt

​app.py

​Run it

​curl (end-to-end manual flow)

​What every sample gets right

Validates state

Validates iss

Uses S256 PKCE

Rotates refresh tokens

Sends form-encoded body

Refreshes proactively

What these samples do

Node.js (Express)

`package.json`

`app.js`

Run it

Python (Flask)

`requirements.txt`

`app.py`

Run it

curl (end-to-end manual flow)

What every sample gets right