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
|
#!/usr/bin/env python
import cherrypy
import os
import sys
import threading
import traceback
import webbrowser
from base64 import b64encode
from fitbit.api import Fitbit
from oauthlib.oauth2.rfc6749.errors import MismatchingStateError, MissingTokenError
class OAuth2Server:
def __init__(self, client_id, client_secret,
redirect_uri='http://127.0.0.1:8080/'):
""" Initialize the FitbitOauth2Client """
self.success_html = """
<h1>You are now authorized to access the Fitbit API!</h1>
<br/><h3>You can close this window</h3>"""
self.failure_html = """
<h1>ERROR: %s</h1><br/><h3>You can close this window</h3>%s"""
self.fitbit = Fitbit(
client_id,
client_secret,
redirect_uri=redirect_uri,
timeout=10,
)
def browser_authorize(self):
"""
Open a browser to the authorization url and spool up a CherryPy
server to accept the response
"""
url, _ = self.fitbit.client.authorize_token_url()
# Open the web browser in a new thread for command-line browser support
threading.Timer(1, webbrowser.open, args=(url,)).start()
cherrypy.quickstart(self)
@cherrypy.expose
def index(self, state, code=None, error=None):
"""
Receive a Fitbit response containing a verification code. Use the code
to fetch the access_token.
"""
error = None
if code:
try:
self.fitbit.client.fetch_access_token(code)
except MissingTokenError:
error = self._fmt_failure(
'Missing access token parameter.</br>Please check that '
'you are using the correct client_secret')
except MismatchingStateError:
error = self._fmt_failure('CSRF Warning! Mismatching state')
else:
error = self._fmt_failure('Unknown error while authenticating')
# Use a thread to shutdown cherrypy so we can return HTML first
self._shutdown_cherrypy()
return error if error else self.success_html
def _fmt_failure(self, message):
tb = traceback.format_tb(sys.exc_info()[2])
tb_html = '<pre>%s</pre>' % ('\n'.join(tb)) if tb else ''
return self.failure_html % (message, tb_html)
def _shutdown_cherrypy(self):
""" Shutdown cherrypy in one second, if it's running """
if cherrypy.engine.state == cherrypy.engine.states.STARTED:
threading.Timer(1, cherrypy.engine.exit).start()
if __name__ == '__main__':
if not (len(sys.argv) == 3):
print("Arguments: client_id and client_secret")
sys.exit(1)
server = OAuth2Server(*sys.argv[1:])
server.browser_authorize()
profile = server.fitbit.user_profile_get()
print('You are authorized to access data for the user: {}'.format(
profile['user']['fullName']))
print('TOKEN\n=====\n')
for key, value in server.fitbit.client.session.token.items():
print('{} = {}'.format(key, value))
|
