Skip to content

Instantly share code, notes, and snippets.

@ib-lundgren
Last active May 7, 2022 06:15
Show Gist options
  • Star 4 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save ib-lundgren/6823954 to your computer and use it in GitHub Desktop.
Save ib-lundgren/6823954 to your computer and use it in GitHub Desktop.
Work in progress example of how to do oauth 2 token refresh in a web application. Requires python modules oauthlib master - pip install -e git+https://github.com/idan/oauthlib#egg=oauthlib requests-oauthlib latest from pip (don't think master is needed) flask & requests
from json import dumps
from flask import Flask, request, redirect, session, url_for
from flask.json import jsonify
from requests_oauthlib import OAuth2Session
app = Flask(__name__)
# Ignore this, only used as basic doc string template creator.
def docstring_templator(f):
from functools import wraps
@wraps(f)
def wrapper(*args, **kwargs):
resp = f(*args, **kwargs)
if not isinstance(resp, (str, unicode)):
resp = "<pre>%s</pre>" % dumps(resp, sort_keys=True, indent=4)
token = dumps(session['oauth_token'], sort_keys=True, indent=4)
return """%s
%s
<h2>Token</h2>
<pre>%s</pre>
<a href="/menu"> Go back to menu</a>
""" % (f.__doc__, resp, token)
return wrapper
# This information is obtained upon registration of a new Google OAuth
# application at https://code.google.com/apis/console
client_id = "<your client key>"
client_secret = "<your client secret>"
redirect_uri = 'http://127.0.0.1:5000/callback'
# Uncomment for detailed oauthlib logs
#import logging
#import sys
#log = logging.getLogger('oauthlib')
#log.addHandler(logging.StreamHandler(sys.stdout))
#log.setLevel(logging.DEBUG)
# OAuth endpoints given in the Google API documentation
authorization_base_url = "https://accounts.google.com/o/oauth2/auth"
token_url = "https://accounts.google.com/o/oauth2/token"
scope = [
"https://www.googleapis.com/auth/userinfo.email",
"https://www.googleapis.com/auth/userinfo.profile",
]
@app.route("/")
def demo():
"""Step 1: User Authorization.
Redirect the user/resource owner to the OAuth provider (i.e. Google)
using an URL with a few key OAuth parameters.
"""
google = OAuth2Session(client_id, scope=scope, redirect_uri=redirect_uri)
authorization_url, state = google.authorization_url(authorization_base_url,
# offline for refresh token
# force to always make user click authorize
access_type="offline", approval_prompt="force")
# State is used to prevent CSRF, keep this for later.
session['oauth_state'] = state
return redirect(authorization_url)
# Step 2: User authorization, this happens on the provider.
@app.route("/callback", methods=["GET"])
def callback():
""" Step 3: Retrieving an access token.
The user has been redirected back from the provider to your registered
callback URL. With this redirection comes an authorization code included
in the redirect URL. We will use that to obtain an access token.
"""
google = OAuth2Session(client_id, redirect_uri=redirect_uri,
state=session['oauth_state'])
token = google.fetch_token(token_url, client_secret=client_secret,
authorization_response=request.url)
# We use the session as a simple DB for this example.
session['oauth_token'] = token
return redirect(url_for('.menu'))
@app.route("/menu", methods=["GET"])
@docstring_templator
def menu():
""""""
return """
<h1>Congratulations, you have obtained an OAuth 2 token!</h1>
<h2>What would you like to do next?</h2>
<ul>
<li><a href="/profile"> Get account profile</a></li>
<li><a href="/automatic_refresh"> Implicitly refresh the token</a></li>
<li><a href="/manual_refresh"> Explicitly refresh the token</a></li>
<li><a href="/validate"> Validate the token</a></li>
</ul>
"""
@app.route("/profile", methods=["GET"])
@docstring_templator
def profile():
"""Fetching a protected resource using an OAuth 2 token.
"""
google = OAuth2Session(client_id, token=session['oauth_token'])
return google.get('https://www.googleapis.com/oauth2/v1/userinfo').json()
@app.route("/automatic_refresh", methods=["GET"])
@docstring_templator
def automatic_refresh():
"""Refreshing an OAuth 2 token using a refresh token.
"""
token = session['oauth_token']
# We force an expiration by setting expired at in the past.
# This will trigger an automatic refresh next time we interact with
# Googles API.
from time import time
token['expires_at'] = time() - 10
extra = {
'client_id': client_id,
'client_secret': client_secret,
}
# True for Google but not necessarily all providers
refresh_url = token_url
def token_updater(token):
session['oauth_token'] = token
google = OAuth2Session(client_id,
token=token,
auto_refresh_kwargs=extra,
auto_refresh_url=refresh_url,
token_updater=token_updater)
# Trigger the automatic refresh
jsonify(google.get('https://www.googleapis.com/oauth2/v1/userinfo').json())
return ''
@app.route("/manual_refresh", methods=["GET"])
@docstring_templator
def manual_refresh():
"""Refreshing an OAuth 2 token using a refresh token.
"""
token = session['oauth_token']
extra = {
'client_id': client_id,
'client_secret': client_secret,
}
# True for Google but not necessarily all providers
refresh_url = token_url
google = OAuth2Session(client_id, token=token)
session['oauth_token'] = google.refresh_token(refresh_url, **extra)
return ''
@app.route("/validate", methods=["GET"])
@docstring_templator
def validate():
"""Validate a token with the OAuth provider Google.
"""
token = session['oauth_token']
# Defined at https://developers.google.com/accounts/docs/OAuth2LoginV1#validatingtoken
validate_url = ('https://www.googleapis.com/oauth2/v1/tokeninfo?'
'access_token=%s' % token['access_token'])
# No OAuth2Session is needed, just a plain GET request
import requests
return requests.get(validate_url).json()
if __name__ == "__main__":
# This allows us to use a plain HTTP callback
import os
os.environ['DEBUG'] = "1"
app.secret_key = os.urandom(24)
app.run(debug=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment