This implementation demonstrates how to redirect users to an FI authorization endpoint, handle the callback, exchange the authorization code for tokens, and validate the ID token using JWKS.
from flask import Flask, redirect, request, session, url_for
from jwcrypto import jwk, jwt
import secrets
import base64
import requests
import json
app = Flask(__name__)
app.secret_key = secrets.token_hex(32)
# Configuration
CLIENT_ID = "your_client_id"
CLIENT_SECRET = "your_client_secret"
REDIRECT_URI = "https://yourapp.com/callback"
TOKEN_ENDPOINT = "https://api.stage.candescent.com/oauth2/v1/token"
ISSUER = "https://www.digitalinsight.com"
# Load JWKS from file provided by your PM
with open("jwks.json") as f:
JWKS = jwk.JWKSet.from_json(f.read())
@app.route("/login")
def login():
state = secrets.token_urlsafe(32)
nonce = secrets.token_urlsafe(32)
session["oauth_state"] = state
session["oauth_nonce"] = nonce
# FI-specific authorization endpoint
auth_endpoint = "https://online.acmebank.com/usr-engage-olb/beta/v1/tpv-sso-authorize"
params = {
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"scope": "openid profile email",
"state": state,
"nonce": nonce
}
return redirect(f"{auth_endpoint}?{'&'.join(f'{k}={v}' for k, v in params.items())}")
@app.route("/callback")
def callback():
if request.args.get("state") != session.get("oauth_state"):
return "State mismatch", 400
code = request.args.get("code")
if not code:
return "Missing authorization code", 400
credentials = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
response = requests.post(
TOKEN_ENDPOINT,
headers={
"Authorization": f"Basic {credentials}",
"Content-Type": "application/x-www-form-urlencoded"
},
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI
}
)
if not response.ok:
return f"Token exchange failed: {response.text}", 400
tokens = response.json()
id_token = tokens["id_token"]
try:
token = jwt.JWT(key=JWKS, jwt=id_token)
claims = json.loads(token.claims)
if claims["iss"] != ISSUER:
return "Invalid issuer", 400
if claims["nonce"] != session.get("oauth_nonce"):
return "Invalid nonce", 400
session["user"] = {
"sub": claims["sub"],
"name": f"{claims.get('given_name', '')} {claims.get('family_name', '')}",
"email": claims.get("email")
}
return redirect(url_for("dashboard"))
except Exception as e:
return f"Token validation failed: {e}", 400
@app.route("/dashboard")
def dashboard():
if "user" not in session:
return redirect(url_for("login"))
return f"Welcome, {session['user']['name']}!"
# Required packages: pip install flask requests jwcrypto