OAuth Apps

import base64, hashlib, os, secrets
from urllib.parse import urlencode

import requests
from flask import Flask, redirect, request, session, url_for

CAMPFIRE_API = "https://api.meetcampfire.com"
CLIENT_ID = os.environ["CAMPFIRE_CLIENT_ID"]
CLIENT_SECRET = os.environ["CAMPFIRE_CLIENT_SECRET"]
REDIRECT_URI = os.environ.get("REDIRECT_URI", "http://localhost:3999/callback")

app = Flask(__name__)
app.secret_key = os.environ["FLASK_SECRET"]


def pkce_pair():
    verifier = secrets.token_urlsafe(48)
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()
    ).rstrip(b"=").decode()
    return verifier, challenge


@app.get("/connect")
def connect():
    verifier, challenge = pkce_pair()
    state = secrets.token_urlsafe(24)
    session["oauth"] = {"state": state, "verifier": verifier}
    query = {
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "response_type": "code",
        "scope": "invoices:read invoice_payments:write vendors:read",
        "state": state,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    }
    return redirect(f"{CAMPFIRE_API}/ca/oauth/authorize?{urlencode(query)}")


@app.get("/callback")
def callback():
    pending = session.pop("oauth", None)
    if request.args.get("error"):
        return f"Campfire returned {request.args['error']}", 400
    if not pending or request.args.get("state") != pending["state"]:
        return "State mismatch", 400

    r = requests.post(
        f"{CAMPFIRE_API}/ca/oauth/token",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={
            "grant_type": "authorization_code",
            "code": request.args["code"],
            "redirect_uri": REDIRECT_URI,
            "code_verifier": pending["verifier"],
        },
        timeout=10,
    )
    r.raise_for_status()
    session["tokens"] = r.json()  # In production, store server-side, encrypted.
    return redirect(url_for("home"))


@app.get("/")
def home():
    tokens = session.get("tokens")
    if not tokens:
        return '<a href="/connect">Connect to Campfire</a>'
    r = requests.get(
        f"{CAMPFIRE_API}/users/api/get-user-permissions",
        headers={"Authorization": f"Bearer {tokens['access_token']}"},
        timeout=10,
    )
    if r.status_code == 401:
        return redirect(url_for("refresh"))
    return f"<pre>{r.text}</pre>"


@app.get("/refresh")
def refresh():
    tokens = session.get("tokens")
    r = requests.post(
        f"{CAMPFIRE_API}/ca/oauth/token",
        auth=(CLIENT_ID, CLIENT_SECRET),
        data={"grant_type": "refresh_token", "refresh_token": tokens["refresh_token"]},
        timeout=10,
    )
    if r.status_code != 200:
        session.pop("tokens", None)  # Connection is gone; the person must reconnect.
        return redirect(url_for("connect"))
    session["tokens"] = r.json()
    return redirect(url_for("home"))