File: real_world_example_with_refresh.rst

package info (click to toggle)
python-requests-oauthlib 1.3.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 472 kB
  • sloc: python: 2,000; makefile: 159
file content (181 lines) | stat: -rw-r--r-- 6,555 bytes parent folder | download | duplicates (4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
.. _token_refresh:

Refreshing tokens in OAuth 2
============================

OAuth 2 providers may allow you to refresh access tokens using refresh tokens.
Commonly, only clients that authenticate may refresh tokens, e.g. web applications
but not javascript clients. The provider will mention whether they allow token
refresh in their API documentation and if you see a "refresh_token" in your
token response you are good to go.

This example shows how a simple web application (using the `Flask web framework
<http://flask.pocoo.org/>`_) can refresh Google OAuth 2 tokens. It should be
trivial to transfer to any other web framework and provider.

.. code-block:: python

    from pprint import pformat
    from time import time

    from flask import Flask, request, redirect, session, url_for
    from flask.json import jsonify
    import requests
    from requests_oauthlib import OAuth2Session

    app = Flask(__name__)

    # 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 = 'https://your.registered/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"
    refresh_url = token_url # True for Google but not all providers.
    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", prompt="select_account")

        # 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"])
    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>

        <pre>
        %s
        </pre>
        """ % pformat(session['oauth_token'], indent=4)


    @app.route("/profile", methods=["GET"])
    def profile():
        """Fetching a protected resource using an OAuth 2 token.
        """
        google = OAuth2Session(client_id, token=session['oauth_token'])
        return jsonify(google.get('https://www.googleapis.com/oauth2/v1/userinfo').json())


    @app.route("/automatic_refresh", methods=["GET"])
    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.
        token['expires_at'] = time() - 10

        extra = {
            'client_id': client_id,
            'client_secret': client_secret,
        }

        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 jsonify(session['oauth_token'])


    @app.route("/manual_refresh", methods=["GET"])
    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,
        }

        google = OAuth2Session(client_id, token=token)
        session['oauth_token'] = google.refresh_token(refresh_url, **extra)
        return jsonify(session['oauth_token'])

    @app.route("/validate", methods=["GET"])
    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
        return jsonify(requests.get(validate_url).json())


    if __name__ == "__main__":
        # This allows us to use a plain HTTP callback
        import os
        os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = "1"

        app.secret_key = os.urandom(24)
        app.run(debug=True)