test-auth.py

import json
import os
from urllib.parse import parse_qs

from tornado.httpclient import AsyncHTTPClient, HTTPRequest
from tornado.httpserver import HTTPServer
from tornado.httputil import url_concat
from tornado.ioloop import IOLoop
from tornado.log import enable_pretty_logging, app_log
from tornado.options import define, options, parse_command_line
from tornado.web import Application, HTTPError, RequestHandler
from tornado import gen

client_id = os.environ['GITHUB_CLIENT_ID']
client_secret = os.environ['GITHUB_CLIENT_SECRET']


class OAuthCallbackHandler(RequestHandler):
    @gen.coroutine
    def get(self):
        self.set_header("Content-Type", "application/json")
        app_log.info("callback url: %s", self.request.full_url())
        http_client = AsyncHTTPClient()
        error = self.get_argument("error", None)
        if error:
            qs = parse_qs(self.request.query)
            app_log.error("Error in OAuth: %s", qs)
            raise HTTPError(500, qs.get("error_description", [str(qs)])[0])
        code = self.get_argument("code")
        params = dict(client_id=client_id, client_secret=client_secret, code=code)

        url = url_concat("https://github.com/login/oauth/access_token", params)
        app_log.info("Requesting %s", url)
        # pause here to make request with curl, etc.
        # input('proceed? ')

        req = HTTPRequest(
            url,
            method="POST",
            headers={"Accept": "application/json"},
            body='',  # Body is required for a POST...
        )
        resp = yield http_client.fetch(req)
        resp_json = json.loads(resp.body.decode('utf8', 'replace'))
        app_log.info("resp: %s", resp_json)
        self.write(
            json.dumps({'url': self.request.full_url(), 'authorization': resp_json})
        )


class OAuthRedirectHandler(RequestHandler):
    def get(self):
        self.set_header("Content-Type", "text/html")
        self.write(
            """
            <form action="/" method="post">
            <input type="submit" value="Login with GitHub">
            </form>
            """
        )

    def post(self):
        redirect_uri = f"{self.request.protocol}://{self.request.host}/oauth_callback"
        target = url_concat(
            "https://github.com/login/oauth/authorize",
            {
                'authorization_type': 'code',
                'client_id': client_id,
                'redirect_uri': redirect_uri,
            },
        )
        app_log.info(f"Redirecting to {target}")
        self.redirect(target)


def main():
    define("port", default=8765, help="run on the given port", type=int)
    enable_pretty_logging()
    parse_command_line()
    application = Application(
        [(r"/", OAuthRedirectHandler, {}), (r"/oauth_callback", OAuthCallbackHandler)],
        debug=True,
    )
    http_server = HTTPServer(application, xheaders=True)
    http_server.listen(options.port)
    IOLoop.current().start()


if __name__ == '__main__':
    main()