Complete server examples

'use strict';
/* A complete partner server in Node.js 18+ with no packages and no SDK.
 *
 *   IDP_URL=https://id.superapp.example CLIENT_ID=app_xxx CLIENT_SECRET=xxx \
 *   SHELL_ORIGIN=https://shell.superapp.example node server.js
 *
 * To try it from the docs Playground, also set CORS_ORIGIN to the Playground's origin (for example
 * https://superapp.ideapreneurnepal.com).
 */
const http = require('node:http');
const fs = require('node:fs');
const path = require('node:path');
const crypto = require('node:crypto');

const PORT = Number(process.env.PORT || 9300);
const IDP_URL = (process.env.IDP_URL || 'http://localhost:8000').replace(/\/+$/, '');
const CLIENT_ID = process.env.CLIENT_ID;
const CLIENT_SECRET = process.env.CLIENT_SECRET;
const SHELL_ORIGIN = process.env.SHELL_ORIGIN || 'http://localhost:8000';
// Only for trying the server from the docs Playground: the playground page's origin. Leave it unset in production.
const CORS_ORIGIN = process.env.CORS_ORIGIN || '';
if (!CLIENT_ID || !CLIENT_SECRET) throw new Error('Set CLIENT_ID and CLIENT_SECRET');

const NONCE_TTL_MS = 2 * 60 * 1000;
const SESSION_TTL_MS = 60 * 60 * 1000;
const NONCE_RE = /^[A-Za-z0-9_-]{16,128}$/;
const nonces = new Map();    // nonce -> expiry time     (use Redis if you run several instances)
const sessions = new Map();  // session -> { sub, name, expires }

// 1. A new one-time nonce for a page load.
function newNonce() {
  const nonce = crypto.randomBytes(24).toString('base64url');   // 32 URL-safe characters
  const now = Date.now();
  for (const [n, expires] of nonces) if (expires < now) nonces.delete(n);
  nonces.set(nonce, now + NONCE_TTL_MS);
  return nonce;
}

// 2. Ask the identity provider who the token belongs to.
async function verifyWithSuperapp(token, nonce) {
  const basic = Buffer.from(`${CLIENT_ID}:${CLIENT_SECRET}`).toString('base64');
  try {
    const res = await fetch(`${IDP_URL}/api/launch/verify`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json', Authorization: `Basic ${basic}` },
      body: JSON.stringify({ token, nonce }),
      signal: AbortSignal.timeout(5000),
    });
    return { status: res.status, body: await res.json().catch(() => ({ error: 'idp_error' })) };
  } catch (_) {
    return { status: 503, body: { error: 'idp_unreachable' } };
  }
}

// 3. Exchange a launch token for your own session.
async function login(token, nonce) {
  if (typeof token !== 'string' || !token || typeof nonce !== 'string' || !NONCE_RE.test(nonce)) {
    return { status: 400, body: { error: 'bad_request' } };
  }
  const expires = nonces.get(nonce) || 0;
  nonces.delete(nonce);                                   // burn it first: it can never be used twice
  if (expires < Date.now()) return { status: 400, body: { error: 'unknown_or_expired_nonce' } };

  const { status, body } = await verifyWithSuperapp(token, nonce);
  if (status !== 200) return { status: 401, body: { error: body.error || 'verify_failed' } };

  const session = crypto.randomBytes(32).toString('base64url');
  sessions.set(session, { sub: body.sub, name: body.name || '', expires: Date.now() + SESSION_TTL_MS });
  return { status: 200, body: { session, sub: body.sub, name: body.name || '' } };
}

// 4. Who is calling? Returns { sub, name } or null (answer 401).
function currentUser(authorization) {
  const header = authorization || '';
  const session = header.startsWith('Bearer ') ? header.slice(7) : '';
  const user = sessions.get(session);
  if (user && user.expires > Date.now()) return user;
  sessions.delete(session);
  return null;
}

const json = (res, status, body) => {
  res.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' });
  res.end(JSON.stringify(body));
};
const readJson = (req) => new Promise((resolve) => {
  let raw = '';
  req.on('data', (c) => { raw += c; if (raw.length > 16384) req.destroy(); });
  req.on('end', () => { try { resolve(JSON.parse(raw || '{}')); } catch (_) { resolve({}); } });
});

http.createServer(async (req, res) => {
  const { pathname } = new URL(req.url, 'http://x');

  if (CORS_ORIGIN) {  // lets the Playground page call this server from the browser
    res.setHeader('Access-Control-Allow-Origin', CORS_ORIGIN);
    res.setHeader('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
    res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization');
    res.setHeader('Access-Control-Allow-Private-Network', 'true');
    if (req.method === 'OPTIONS') { res.writeHead(204); return res.end(); }
  }

  // the sign-in routes
  if (req.method === 'GET' && pathname === '/api/config') return json(res, 200, { shell_origin: SHELL_ORIGIN });
  if (req.method === 'POST' && pathname === '/api/nonce') return json(res, 200, { nonce: newNonce() });
  if (req.method === 'POST' && pathname === '/api/session') {
    const { token, nonce } = await readJson(req);
    const { status, body } = await login(token, nonce);
    return json(res, status, body);
  }

  // everything else needs a session
  if (pathname.startsWith('/api/')) {
    const user = currentUser(req.headers.authorization);
    if (!user) return json(res, 401, { error: 'no_session' });
    if (req.method === 'GET' && pathname === '/api/me') return json(res, 200, { id: user.sub, name: user.name });
    return json(res, 404, { error: 'not_found' });
  }

  // the page, with the header that lets only the SuperPreneur web shell frame it
  if (req.method === 'GET' && pathname === '/') {
    res.writeHead(200, {
      'Content-Type': 'text/html; charset=utf-8',
      'Content-Security-Policy': `frame-ancestors 'self' ${SHELL_ORIGIN}`,
    });
    return res.end(fs.readFileSync(path.join(__dirname, '../public/index.html')));
  }
  json(res, 404, { error: 'not_found' });
}).listen(PORT, () => console.log(`Hello partner (Node) on http://localhost:${PORT}`));

"""A complete partner server in Python 3.8+ with no packages and no SDK.

    IDP_URL=https://id.superapp.example CLIENT_ID=app_xxx CLIENT_SECRET=xxx \
    SHELL_ORIGIN=https://shell.superapp.example python3 server.py

To try it from the docs Playground, also set CORS_ORIGIN to the Playground's origin (for example
https://superapp.ideapreneurnepal.com).
"""

import base64
import json
import os
import re
import secrets
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

PORT = int(os.environ.get("PORT", "9301"))
IDP_URL = os.environ.get("IDP_URL", "http://localhost:8000").rstrip("/")
CLIENT_ID = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
SHELL_ORIGIN = os.environ.get("SHELL_ORIGIN", "http://localhost:8000")
# Only for trying the server from the docs Playground: the playground page's origin. Leave it unset in production.
CORS_ORIGIN = os.environ.get("CORS_ORIGIN", "")

NONCE_TTL = 120          # seconds
SESSION_TTL = 3600
NONCE_RE = re.compile(r"^[A-Za-z0-9_-]{16,128}$")
lock = threading.Lock()
nonces = {}              # nonce -> expiry time      (use Redis if you run several instances)
sessions = {}            # session -> {"sub", "name", "expires"}


def new_nonce():
    """1. A new one-time nonce for a page load."""
    nonce = secrets.token_urlsafe(24)                   # 32 URL-safe characters
    with lock:
        now = time.time()
        for n in [n for n, exp in nonces.items() if exp < now]:
            del nonces[n]
        nonces[nonce] = now + NONCE_TTL
    return nonce


def verify_with_superapp(token, nonce):
    """2. Ask the identity provider who the token belongs to."""
    basic = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
    request = urllib.request.Request(
        f"{IDP_URL}/api/launch/verify",
        data=json.dumps({"token": token, "nonce": nonce}).encode(),
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Basic {basic}",
            "User-Agent": "my-partner-app/1.0",  # the hosted service refuses the default Python agent
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=5) as res:
            return res.status, json.loads(res.read())
    except urllib.error.HTTPError as err:
        try:
            return err.code, json.loads(err.read())
        except ValueError:
            return err.code, {"error": "idp_error"}
    except (urllib.error.URLError, TimeoutError):
        return 503, {"error": "idp_unreachable"}


def login(token, nonce):
    """3. Exchange a launch token for your own session."""
    if not isinstance(token, str) or not token or not isinstance(nonce, str) or not NONCE_RE.match(nonce):
        return 400, {"error": "bad_request"}
    with lock:                                          # burn it first: it can never be used twice
        expires = nonces.pop(nonce, 0)
    if expires < time.time():
        return 400, {"error": "unknown_or_expired_nonce"}

    status, body = verify_with_superapp(token, nonce)
    if status != 200:
        return 401, {"error": body.get("error", "verify_failed")}

    session = secrets.token_urlsafe(32)
    with lock:
        sessions[session] = {"sub": body["sub"], "name": body.get("name", ""), "expires": time.time() + SESSION_TTL}
    return 200, {"session": session, "sub": body["sub"], "name": body.get("name", "")}


def current_user(authorization):
    """4. Who is calling? Returns {"sub", "name"} or None (answer 401)."""
    header = authorization or ""
    session = header[7:] if header.startswith("Bearer ") else ""
    with lock:
        user = sessions.get(session)
        if user and user["expires"] > time.time():
            return user
        sessions.pop(session, None)
    return None


class Handler(BaseHTTPRequestHandler):
    def end_headers(self):
        if CORS_ORIGIN:  # lets the Playground page call this server from the browser
            self.send_header("Access-Control-Allow-Origin", CORS_ORIGIN)
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
            self.send_header("Access-Control-Allow-Private-Network", "true")
        super().end_headers()

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Content-Length", "0")
        self.end_headers()

    def send_json(self, status, body):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def read_json(self):
        try:
            length = min(int(self.headers.get("Content-Length", "0")), 16384)
            body = json.loads(self.rfile.read(length) or b"{}")
            return body if isinstance(body, dict) else {}
        except ValueError:
            return {}

    def do_GET(self):
        if self.path == "/api/config":
            return self.send_json(200, {"shell_origin": SHELL_ORIGIN})
        if self.path.startswith("/api/"):
            user = current_user(self.headers.get("Authorization"))
            if not user:
                return self.send_json(401, {"error": "no_session"})
            if self.path == "/api/me":
                return self.send_json(200, {"id": user["sub"], "name": user["name"]})
            return self.send_json(404, {"error": "not_found"})
        if self.path == "/":
            data = (Path(__file__).parent.parent / "public" / "index.html").read_bytes()
            self.send_response(200)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(data)))
            # only the SuperPreneur web shell may frame this page
            self.send_header("Content-Security-Policy", f"frame-ancestors 'self' {SHELL_ORIGIN}")
            self.end_headers()
            return self.wfile.write(data)
        self.send_json(404, {"error": "not_found"})

    def do_POST(self):
        if self.path == "/api/nonce":
            return self.send_json(200, {"nonce": new_nonce()})
        if self.path == "/api/session":
            body = self.read_json()
            status, result = login(body.get("token"), body.get("nonce"))
            return self.send_json(status, result)
        self.send_json(404, {"error": "not_found"})


if __name__ == "__main__":
    print(f"Hello partner (Python) on http://localhost:{PORT}")
    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()

IDP_URL=https://superapp.ideapreneurnepal.com/sandbox CLIENT_ID=sandbox_app CLIENT_SECRET=sandbox_secret node partner-examples/node/server.js
IDP_URL=https://superapp.ideapreneurnepal.com/sandbox CLIENT_ID=sandbox_app CLIENT_SECRET=sandbox_secret python3 partner-examples/python/server.py