Code examples¶

Working mini-integrations in multiple languages. The Node version is in the quickstart; this page mirrors it in Python.

The flow is identical: register sandbox client → run server → connect via OAuth → make API call → receive a webhook. Same endpoints, same token shape, same signature recipe — only the language differs.

Python (Flask + requests)¶

Setup¶

mkdir revento-python-quickstart && cd revento-python-quickstart
python -m venv .venv && source .venv/bin/activate
pip install flask requests

server.py¶

import os
import hmac
import hashlib
import secrets
import base64
from urllib.parse import urlencode
from time import time

from flask import Flask, redirect, request, jsonify
import requests

app = Flask(__name__)

CLIENT_ID      = os.environ["REVENTO_CLIENT_ID"]
CLIENT_SECRET  = os.environ["REVENTO_CLIENT_SECRET"]
WEBHOOK_SECRET = os.environ["REVENTO_WEBHOOK_SECRET"].encode()
REDIRECT_URI   = os.environ.get("REVENTO_REDIRECT_URI", "http://localhost:3000/callback")
# Must byte-match the redirect URI you registered. If using ngrok or a tunnel,
# set REVENTO_REDIRECT_URI to the tunnel's HTTPS URL.

AUTH_BASE = "https://auth.revento.example"
API_BASE  = "https://api.revento.example/api/v1"

# In-memory state — use a real store in production
state = {
    "code_verifier":  None,
    "access_token":   None,
    "refresh_token":  None,
    "event_id":       None,
}

def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode().rstrip("=")


@app.route("/connect")
def connect():
    # PKCE
    verifier  = b64url(secrets.token_bytes(32))
    challenge = b64url(hashlib.sha256(verifier.encode()).digest())
    state["code_verifier"] = verifier

    params = {
        "response_type":         "code",
        "client_id":             CLIENT_ID,
        "redirect_uri":          REDIRECT_URI,
        "scope":                 "event.read participants.read",
        "event_id":              os.environ["SANDBOX_EVENT_ID"],
        "state":                 "demo-state-123",  # CSRF token in real code
        "code_challenge":        challenge,
        "code_challenge_method": "S256",
    }
    return redirect(f"{AUTH_BASE}/oauth/authorize?{urlencode(params)}")


@app.route("/callback")
def callback():
    if request.args.get("error") == "access_denied":
        return "Consent declined.", 400

    if request.args.get("state") != "demo-state-123":
        return "State mismatch.", 400

    code = request.args["code"]

    # Exchange code for tokens
    res = requests.post(f"{AUTH_BASE}/oauth/token", data={
        "grant_type":    "authorization_code",
        "code":          code,
        "redirect_uri":  REDIRECT_URI,
        "client_id":     CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "code_verifier": state["code_verifier"],
    })
    if not res.ok:
        return f"Token exchange failed: {res.text}", 500

    tokens = res.json()
    state["access_token"]  = tokens["access_token"]
    state["refresh_token"] = tokens["refresh_token"]
    state["event_id"]      = tokens["event_id"]

    # First API call — fetch participants
    parts = requests.get(
        f"{API_BASE}/events/{tokens['event_id']}/participants",
        params={"limit": 10},
        headers={
            "Authorization":          f"Bearer {tokens['access_token']}",
            "Accept":                 "application/vnd.revento.v1+json",
            "X-Revento-Request-Id":   secrets.token_hex(16),
        },
    )
    return jsonify({
        "message":      "Connected!",
        "participants": parts.json(),
    })


@app.route("/webhooks/revento", methods=["POST"])
def webhook():
    timestamp = request.headers.get("X-Revento-Timestamp", "")
    signature = request.headers.get("X-Revento-Signature", "")
    body      = request.get_data()  # raw bytes — DON'T parse-then-restringify

    # 1. Replay protection
    try:
        if abs(time() - int(timestamp)) > 300:
            return "Stale", 401
    except ValueError:
        return "Bad timestamp", 401

    # 2. HMAC verification
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET,
        f"{timestamp}.".encode() + body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        return "Bad signature", 401

    # 3. Process — respond fast (10s budget)
    payload = request.get_json()
    print(f"[{request.headers.get('X-Revento-Event')}] {payload}")
    # heavy work would be queued async here

    return "ok", 200


def refresh_access_token():
    """Call when the access token expires (401 token_expired)."""
    res = requests.post(f"{AUTH_BASE}/oauth/token", data={
        "grant_type":    "refresh_token",
        "refresh_token": state["refresh_token"],
        "client_id":     CLIENT_ID,
        "client_secret": CLIENT_SECRET,
    })
    if not res.ok:
        if res.json().get("error") == "invalid_grant":
            # Family revoked — re-consent required, NOT a transient error
            raise RuntimeError("Refresh failed permanently — re-consent required")
        raise RuntimeError(f"Refresh failed: {res.text}")

    tokens = res.json()
    # Persist FIRST, before using the new access token
    state["access_token"]  = tokens["access_token"]
    state["refresh_token"] = tokens["refresh_token"]   # NEW refresh token — store it
    return tokens["access_token"]


if __name__ == "__main__":
    app.run(port=3000)

Run¶

export REVENTO_CLIENT_ID=...
export REVENTO_CLIENT_SECRET=...
export REVENTO_WEBHOOK_SECRET=...
export SANDBOX_EVENT_ID=evt_test_published

python server.py

Then open http://localhost:3000/connect in a browser.

Notes for Python specifically¶

Use hmac.compare_digest for the signature comparison — never ==. Constant-time, prevents timing attacks.
request.get_data() returns the raw body bytes. Don't use request.json for signature verification — Flask's JSON parsing might subtly change the bytes (whitespace, key order in some configs).
secrets.token_bytes and secrets.token_hex are the canonical way to generate tokens — never use random for security-sensitive randomness.

Other languages¶

The same flow translates directly to:

Go: standard library has everything you need (net/http, crypto/hmac, crypto/sha256).
PHP: curl for HTTP, hash_hmac('sha256', ...) for signatures, hash_equals for constant-time comparison.
Ruby: Net::HTTP + OpenSSL::HMAC + Rack::Utils.secure_compare.
Java/Kotlin: java.net.http.HttpClient + javax.crypto.Mac + MessageDigest.isEqual.
C# / .NET: HttpClient + HMACSHA256 + CryptographicOperations.FixedTimeEquals.

The shapes are identical. If you hit something that feels language-specific, it's almost always the constant-time-comparison or raw-bytes handling — both of which have well-known stdlib functions in every language.

What's NOT in these examples¶

Token persistence: in-memory dictionaries are fine for demos, never for production. Use a database with at-rest encryption.
Concurrent-refresh handling: the simple refresh_access_token above doesn't handle two threads/processes refreshing simultaneously. See refresh tokens — concurrent races.
Robust error handling: production code branches on error codes (not just status), distinguishes token_expired from token_revoked, etc.
Logging and observability: include X-Revento-Request-Id correlation in your logs.
Retries with backoff: the examples don't retry on 5xx or rate limit. Production code should.

These are mini-integrations to get you running. Production-grade code adds the operational concerns above.