<s> from __future__ import absolute_import <EOL> import json <EOL> import functools <EOL> import re <EOL> try : <EOL> from urllib . parse import urlencode <EOL> except ImportError : <EOL> from urllib import urlencode <EOL> import tornado . web <EOL> import tornado . auth <EOL> from tornado import httpclient <EOL> from tornado . options import options <EOL> from celery . utils . imports import instantiate <EOL> from . . views import BaseHandler <EOL> class GoogleAuth2LoginHandler ( BaseHandler , tornado . auth . GoogleOAuth2Mixin ) : <EOL> _OAUTH_SETTINGS_KEY = '<STR_LIT>' <EOL> @ tornado . web . asynchronous <EOL> def get ( self ) : <EOL> redirect_uri = self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT>' ] <EOL> if self . get_argument ( '<STR_LIT:code>' , False ) : <EOL> self . get_authenticated_user ( <EOL> redirect_uri = redirect_uri , <EOL> code = self . get_argument ( '<STR_LIT:code>' ) , <EOL> callback = self . _on_auth , <EOL> ) <EOL> else : <EOL> self . authorize_redirect ( <EOL> redirect_uri = redirect_uri , <EOL> client_id = self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT:key>' ] , <EOL> scope = [ '<STR_LIT>' , '<STR_LIT:email>' ] , <EOL> response_type = '<STR_LIT:code>' , <EOL> extra_params = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def _on_auth ( self , user ) : <EOL> if not user : <EOL> raise tornado . web . HTTPError ( <NUM_LIT> , '<STR_LIT>' ) <EOL> access_token = user [ '<STR_LIT>' ] <EOL> try : <EOL> response = httpclient . HTTPClient ( ) . fetch ( <EOL> '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' % access_token } ) <EOL> except Exception as e : <EOL> raise tornado . web . HTTPError ( <NUM_LIT> , '<STR_LIT>' % e ) <EOL> email = json . loads ( response . body . decode ( '<STR_LIT:utf-8>' ) ) [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:value>' ] <EOL> if not re . match ( self . application . options . auth , email ) : <EOL> message = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) . format ( email = email ) <EOL> raise tornado . web . HTTPError ( <NUM_LIT> , message ) <EOL> self . set_secure_cookie ( "<STR_LIT:user>" , str ( email ) ) <EOL> next = self . get_argument ( '<STR_LIT>' , '<STR_LIT:/>' ) <EOL> self . redirect ( next ) <EOL> class LoginHandler ( BaseHandler ) : <EOL> def __new__ ( cls , * args , ** kwargs ) : <EOL> return instantiate ( options . auth_provider , * args , ** kwargs ) <EOL> class GithubLoginHandler ( BaseHandler , tornado . auth . OAuth2Mixin ) : <EOL> _OAUTH_AUTHORIZE_URL = "<STR_LIT>" <EOL> _OAUTH_ACCESS_TOKEN_URL = "<STR_LIT>" <EOL> _OAUTH_NO_CALLBACKS = False <EOL> _OAUTH_SETTINGS_KEY = '<STR_LIT>' <EOL> @ tornado . auth . _auth_return_future <EOL> def get_authenticated_user ( self , redirect_uri , code , callback ) : <EOL> http = self . get_auth_http_client ( ) <EOL> body = urlencode ( { <EOL> "<STR_LIT>" : redirect_uri , <EOL> "<STR_LIT:code>" : code , <EOL> "<STR_LIT>" : self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT:key>' ] , <EOL> "<STR_LIT>" : self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT>' ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ) <EOL> http . fetch ( <EOL> self . _OAUTH_ACCESS_TOKEN_URL , <EOL> functools . partial ( self . _on_access_token , callback ) , <EOL> method = "<STR_LIT:POST>" , <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:application/json>' } , body = body ) <EOL> @ tornado . web . asynchronous <EOL> def _on_access_token ( self , future , response ) : <EOL> if response . error : <EOL> future . set_exception ( tornado . auth . AuthError ( <EOL> '<STR_LIT>' % str ( response ) ) ) <EOL> return <EOL> future . set_result ( json . loads ( response . body ) ) <EOL> def get_auth_http_client ( self ) : <EOL> return httpclient . AsyncHTTPClient ( ) <EOL> @ tornado . web . asynchronous <EOL> def get ( self ) : <EOL> redirect_uri = self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT>' ] <EOL> if self . get_argument ( '<STR_LIT:code>' , False ) : <EOL> self . get_authenticated_user ( <EOL> redirect_uri = redirect_uri , <EOL> code = self . get_argument ( '<STR_LIT:code>' ) , <EOL> callback = self . _on_auth , <EOL> ) <EOL> else : <EOL> self . authorize_redirect ( <EOL> redirect_uri = redirect_uri , <EOL> client_id = self . settings [ self . _OAUTH_SETTINGS_KEY ] [ '<STR_LIT:key>' ] , <EOL> scope = [ '<STR_LIT>' ] , <EOL> response_type = '<STR_LIT:code>' , <EOL> extra_params = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> @ tornado . web . asynchronous <EOL> def _on_auth ( self , user ) : <EOL> if not user : <EOL> raise tornado . web . HTTPError ( <NUM_LIT> , '<STR_LIT>' ) <EOL> access_token = user [ '<STR_LIT>' ] <EOL> req = httpclient . HTTPRequest ( <EOL> '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' + access_token , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> response = httpclient . HTTPClient ( ) . fetch ( req ) <EOL> emails = [ email [ '<STR_LIT:email>' ] . lower ( ) for email in json . loads ( response . body . decode ( '<STR_LIT:utf-8>' ) ) <EOL> if email [ '<STR_LIT>' ] and re . match ( self . application . options . auth , email [ '<STR_LIT:email>' ] ) ] <EOL> if not emails : <EOL> message = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> raise tornado . web . HTTPError ( <NUM_LIT> , message ) <EOL> self . set_secure_cookie ( "<STR_LIT:user>" , str ( emails . pop ( ) ) ) <EOL> next_ = self . get_argument ( '<STR_LIT>' , '<STR_LIT:/>' ) <EOL> self . redirect ( next_ ) <EOL> class LogoutHandler ( BaseHandler ) : <EOL> def get ( self ) : <EOL> self . clear_cookie ( '<STR_LIT:user>' ) <EOL> self . render ( '<STR_LIT>' , message = '<STR_LIT>' ) </s>
