from flask import Flask, requestimport requestsimport jwtimport jsonimport osapp = Flask(__name__)# The Application Audience (AUD) tag for your applicationPOLICY_AUD = os.getenv('POLICY_AUD')# Your CF Access team domainTEAM_DOMAIN = os.getenv('TEAM_DOMAIN')CERTS_URL = '{}/cdn-cgi/access/certs'.format(TEAM_DOMAIN)def _get_public_keys():'''Returns:List of RSA public keys usable by PyJWT.'''r = requests.get(CERTS_URL)public_keys = []jwk_set = r.json()for key_dict in jwk_set['keys']:public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key_dict))public_keys.append(public_key)return public_keysdef verify_token(f):'''Decorator that wraps a Flask API call to verify the CF Access JWT'''def wrapper():# Check for the POLICY_AUD environment variableif not POLICY_AUD:return 'missing required audience', 403token = ''if 'CF_Authorization' in request.cookies:token = request.cookies['CF_Authorization']else:return 'missing required cf authorization token', 403keys = _get_public_keys()# Loop through the keys since we can't pass the key set to the decodervalid_token = Falsefor key in keys:try:# decode returns the claims that has the email when neededjwt.decode(token, key=key, audience=POLICY_AUD, algorithms=['RS256'])valid_token = Truebreakexcept:passif not valid_token:return 'invalid token', 403return f()return wrapper@app.route('/')@verify_tokendef hello_world():return 'Hello, World!'if __name__ == '__main__':app.run()

const express = require('express');const jose = require('jose');// The Application Audience (AUD) tag for your applicationconst AUD = process.env.POLICY_AUD;// Your CF Access team domainconst TEAM_DOMAIN = process.env.TEAM_DOMAIN;const CERTS_URL = `${TEAM_DOMAIN}/cdn-cgi/access/certs`;const JWKS = jose.createRemoteJWKSet(new URL(CERTS_URL));// verifyToken is a middleware to verify a CF authorization tokenconst verifyToken = async (req, res, next) => {// Check for the AUD environment variableif (!AUD) {return res.status(403).send({status: false,message: 'missing required audience',});}const token = req.headers['cf-access-jwt-assertion'];// Make sure that the incoming request has our token headerif (!token) {return res.status(403).send({status: false,message: 'missing required cf authorization token',});}try {const result = await jose.jwtVerify(token, JWKS, {issuer: TEAM_DOMAIN,audience: AUD,});req.user = result.payload;next();} catch (err) {return res.status(403).send({status: false,message: 'invalid token',});}};const app = express();app.use(verifyToken);app.get('/', (req, res) => {res.send('Hello World!');});app.listen(3333);

from flask import Flask, requestimport requestsimport jwtimport jsonimport osapp = Flask(__name__)# The Application Audience (AUD) tag for your applicationPOLICY_AUD = os.getenv('POLICY_AUD')# Your CF Access team domainTEAM_DOMAIN = os.getenv('TEAM_DOMAIN')CERTS_URL = '{}/cdn-cgi/access/certs'.format(TEAM_DOMAIN)def _get_public_keys():'''Returns:List of RSA public keys usable by PyJWT.'''r = requests.get(CERTS_URL)public_keys = []jwk_set = r.json()for key_dict in jwk_set['keys']:public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key_dict))public_keys.append(public_key)return public_keysdef verify_token(f):'''Decorator that wraps a Flask API call to verify the CF Access JWT'''def wrapper():# Check for the POLICY_AUD environment variableif not POLICY_AUD:return 'missing required audience', 403token = ''if 'CF_Authorization' in request.cookies:token = request.cookies['CF_Authorization']else:return 'missing required cf authorization token', 403keys = _get_public_keys()# Loop through the keys since we can't pass the key set to the decodervalid_token = Falsefor key in keys:try:# decode returns the claims that has the email when neededjwt.decode(token, key=key, audience=POLICY_AUD, algorithms=['RS256'])valid_token = Truebreakexcept:passif not valid_token:return 'invalid token', 403return f()return wrapper@app.route('/')@verify_tokendef hello_world():return 'Hello, World!'if __name__ == '__main__':app.run()

const express = require('express');const jose = require('jose');// The Application Audience (AUD) tag for your applicationconst AUD = process.env.POLICY_AUD;// Your CF Access team domainconst TEAM_DOMAIN = process.env.TEAM_DOMAIN;const CERTS_URL = `${TEAM_DOMAIN}/cdn-cgi/access/certs`;const JWKS = jose.createRemoteJWKSet(new URL(CERTS_URL));// verifyToken is a middleware to verify a CF authorization tokenconst verifyToken = async (req, res, next) => {// Check for the AUD environment variableif (!AUD) {return res.status(403).send({status: false,message: 'missing required audience',});}const token = req.headers['cf-access-jwt-assertion'];// Make sure that the incoming request has our token headerif (!token) {return res.status(403).send({status: false,message: 'missing required cf authorization token',});}try {const result = await jose.jwtVerify(token, JWKS, {issuer: TEAM_DOMAIN,audience: AUD,});req.user = result.payload;next();} catch (err) {return res.status(403).send({status: false,message: 'invalid token',});}};const app = express();app.use(verifyToken);app.get('/', (req, res) => {res.send('Hello World!');});app.listen(3333);