tornado-dropbox-oauth2.py

import tornado.ioloop
import tornado.web
import tornado.auth
import tornado.gen

from tornado import escape, httpclient
from tornado.options import parse_command_line, define, options

try:
    import urllib.parse as urllib_parse  # py3
except ImportError:
    import urllib as urllib_parse  # py2

define("port", default=8888)
define("debug", default=False)


class DropboxMixin(tornado.auth.OAuth2Mixin):
    """Dropbox authentication using OAuth2.

    https://www.dropbox.com/developers/core/docs

    """
    _OAUTH_AUTHORIZE_URL = "https://www.dropbox.com/1/oauth2/authorize"
    _OAUTH_ACCESS_TOKEN_URL = "https://api.dropbox.com/1/oauth2/token"
    _OAUTH_SETTINGS_KEY = 'dropbox_oauth'

    @property
    def oauth_settings(self):
        return self.settings[self._OAUTH_SETTINGS_KEY]

    @tornado.auth._auth_return_future
    def get_authenticated_user(self, code, callback):
        """Handles the login for the Dropbox user, returning a user object."""
        http = self.get_auth_http_client()
        body = urllib_parse.urlencode({
            "redirect_uri": self.oauth_settings["redirect"],
            "code": code,
            "client_id": self.oauth_settings['key'],
            "client_secret": self.oauth_settings['secret'],
            "grant_type": "authorization_code",
        })

        http.fetch(
            self._OAUTH_ACCESS_TOKEN_URL,
            self.async_callback(self._on_access_token, callback),
            method="POST",
            headers={'Content-Type': 'application/x-www-form-urlencoded'},
            body=body
        )

    def _on_access_token(self, future, response):
        """Callback function for the exchange to the access token."""
        if response.error:
            msg = 'Dropbox auth error: {}'.format(str(response))
            future.set_exception(tornado.auth.AuthError(msg))
            return

        args = escape.json_decode(response.body)
        future.set_result(args)

    def get_auth_http_client(self):
        return httpclient.AsyncHTTPClient()

    def authorize_redirect(self, callback=None):
        kwargs = {
            "redirect_uri": self.oauth_settings["redirect"],
            "client_id": self.oauth_settings["key"],
            "callback": callback,
            "extra_params": {"response_type": "code"}
        }

        return super(DropboxMixin, self).authorize_redirect(**kwargs)


class AuthHandler(tornado.web.RequestHandler, DropboxMixin):
    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self):
        code = self.get_argument("code", None)

        if code:
            user = yield self.get_authenticated_user(code=code)

            self.set_secure_cookie("oauth_user", user.get("uid", ""))
            self.set_secure_cookie("oauth_token", user.get("access_token", ""))

            self.redirect("/")
        else:
            yield self.authorize_redirect()


class MainHandler(tornado.web.RequestHandler):
    def get_current_user(self):
        return self.get_secure_cookie("oauth_user")

    def get(self):
        user = self.get_current_user()
        if not user:
            raise tornado.web.HTTPError(401, "Access denied")

        self.write("Welcome back, {}".format(user))

handlers = [
    (r"/auth", AuthHandler),
    (r"/", MainHandler),
]

if __name__ == "__main__":
    parse_command_line()

    opts = {
        "debug": options.debug,
        "cookie_secret": "II*^WvQiOkv)QihQwQZ<JH!YY/q)v%TY",

        #https://www.dropbox.com/developers
        "dropbox_oauth": {
            "redirect": "http://localhost:{}/auth".format(options.port),

            "key": "DROPBOX APP KEY",
            "secret": "DROPBOX APP SECRET"
        }
    }

    application = tornado.web.Application(handlers, **opts)
    application.listen(options.port)

    tornado.ioloop.IOLoop.instance().start()