<s> import json <EOL> import urlparse <EOL> from django . http import HttpResponse <EOL> from django . http import HttpResponseRedirect , QueryDict <EOL> from django . utils . translation import ugettext as _ <EOL> from django . views . generic . base import TemplateView <EOL> from django . core . exceptions import ObjectDoesNotExist <EOL> from oauth2 . models import Client <EOL> from . import constants , scope <EOL> class OAuthError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class OAuthView ( TemplateView ) : <EOL> """<STR_LIT>""" <EOL> def dispatch ( self , request , * args , ** kwargs ) : <EOL> response = super ( OAuthView , self ) . dispatch ( request , * args , ** kwargs ) <EOL> response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return response <EOL> class Mixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def get_data ( self , request , key = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return request . session . get ( '<STR_LIT>' % ( constants . SESSION_KEY , key ) ) <EOL> def cache_data ( self , request , data , key = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> request . session [ '<STR_LIT>' % ( constants . SESSION_KEY , key ) ] = data <EOL> def clear_data ( self , request ) : <EOL> """<STR_LIT>""" <EOL> for key in request . session . keys ( ) : <EOL> if key . startswith ( constants . SESSION_KEY ) : <EOL> del request . session [ key ] <EOL> def authenticate ( self , request ) : <EOL> """<STR_LIT>""" <EOL> for backend in self . authentication : <EOL> client = backend ( ) . authenticate ( request ) <EOL> if client is not None : <EOL> return client <EOL> return None <EOL> class Capture ( OAuthView , Mixin ) : <EOL> """<STR_LIT>""" <EOL> template_name = '<STR_LIT>' <EOL> def get_redirect_url ( self , request ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def handle ( self , request , data ) : <EOL> self . cache_data ( request , data ) <EOL> if constants . ENFORCE_SECURE and not request . is_secure ( ) : <EOL> return self . render_to_response ( { '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) , <EOL> '<STR_LIT>' : None } , <EOL> status = <NUM_LIT> ) <EOL> return HttpResponseRedirect ( self . get_redirect_url ( request ) ) <EOL> def get ( self , request ) : <EOL> return self . handle ( request , request . GET ) <EOL> def post ( self , request ) : <EOL> return self . handle ( request , request . POST ) <EOL> class Authorize ( OAuthView , Mixin ) : <EOL> """<STR_LIT>""" <EOL> template_name = '<STR_LIT>' <EOL> def get_redirect_url ( self , request ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_request_form ( self , client , data ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_authorization_form ( self , request , client , data , client_data ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_client ( self , client_id ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def save_authorization ( self , request , client , form , client_data ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def _validate_client ( self , request , data ) : <EOL> """<STR_LIT>""" <EOL> client = self . get_client ( data . get ( '<STR_LIT>' ) ) <EOL> if client is None : <EOL> raise OAuthError ( { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> } ) <EOL> form = self . get_request_form ( client , data ) <EOL> if not form . is_valid ( ) : <EOL> raise OAuthError ( form . errors ) <EOL> return client , form . cleaned_data <EOL> def error_response ( self , request , error , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> ctx = { } <EOL> ctx . update ( error ) <EOL> if error [ '<STR_LIT:error>' ] in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> ctx . update ( next = '<STR_LIT:/>' ) <EOL> return self . render_to_response ( ctx , ** kwargs ) <EOL> ctx . update ( next = self . get_redirect_url ( request ) ) <EOL> return self . render_to_response ( ctx , ** kwargs ) <EOL> def handle ( self , request , post_data = None ) : <EOL> data = self . get_data ( request ) <EOL> if data is None : <EOL> return self . error_response ( request , { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) } ) <EOL> try : <EOL> client , data = self . _validate_client ( request , data ) <EOL> except OAuthError , e : <EOL> return self . error_response ( request , e . args [ <NUM_LIT:0> ] , status = <NUM_LIT> ) <EOL> authorization_form = self . get_authorization_form ( request , client , <EOL> post_data , data ) <EOL> if not authorization_form . is_bound or not authorization_form . is_valid ( ) : <EOL> return self . render_to_response ( { <EOL> '<STR_LIT>' : client , <EOL> '<STR_LIT>' : authorization_form , <EOL> '<STR_LIT>' : data , } ) <EOL> code = self . save_authorization ( request , client , <EOL> authorization_form , data ) <EOL> self . cache_data ( request , data ) <EOL> self . cache_data ( request , code , "<STR_LIT:code>" ) <EOL> self . cache_data ( request , client . serialize ( ) , "<STR_LIT>" ) <EOL> return HttpResponseRedirect ( self . get_redirect_url ( request ) ) <EOL> def get ( self , request ) : <EOL> return self . handle ( request , None ) <EOL> def post ( self , request ) : <EOL> return self . handle ( request , request . POST ) <EOL> class Redirect ( OAuthView , Mixin ) : <EOL> """<STR_LIT>""" <EOL> def error_response ( self , error , mimetype = '<STR_LIT:application/json>' , status = <NUM_LIT> , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return HttpResponse ( json . dumps ( error ) , mimetype = mimetype , <EOL> status = status , ** kwargs ) <EOL> def get ( self , request ) : <EOL> data = self . get_data ( request ) <EOL> code = self . get_data ( request , "<STR_LIT:code>" ) <EOL> error = self . get_data ( request , "<STR_LIT:error>" ) <EOL> client = self . get_data ( request , "<STR_LIT>" ) <EOL> client = Client . deserialize ( client ) <EOL> if data is None or client is None : <EOL> return self . error_response ( { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( '<STR_LIT>' ) } ) <EOL> redirect_uri = data . get ( '<STR_LIT>' , None ) or client . redirect_uri <EOL> parsed = urlparse . urlparse ( redirect_uri ) <EOL> query = QueryDict ( '<STR_LIT>' , mutable = True ) <EOL> if '<STR_LIT:state>' in data : <EOL> query [ '<STR_LIT:state>' ] = data [ '<STR_LIT:state>' ] <EOL> if error is not None : <EOL> query . update ( error ) <EOL> elif code is None : <EOL> query [ '<STR_LIT:error>' ] = '<STR_LIT>' <EOL> else : <EOL> query [ '<STR_LIT:code>' ] = code <EOL> parsed = parsed [ : <NUM_LIT:4> ] + ( query . urlencode ( ) , '<STR_LIT>' ) <EOL> redirect_uri = urlparse . ParseResult ( * parsed ) . geturl ( ) <EOL> self . clear_data ( request ) <EOL> return HttpResponseRedirect ( redirect_uri ) <EOL> class AccessToken ( OAuthView , Mixin ) : <EOL> """<STR_LIT>""" <EOL> authentication = ( ) <EOL> """<STR_LIT>""" <EOL> grant_types = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:password>' ] <EOL> """<STR_LIT>""" <EOL> def get_authorization_code_grant ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_refresh_token_grant ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_password_grant ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_access_token ( self , request , user , scope , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def create_access_token ( self , request , user , scope , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def create_refresh_token ( self , request , user , scope , access_token , client ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def invalidate_grant ( self , grant ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def invalidate_refresh_token ( self , refresh_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def invalidate_access_token ( self , access_token ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def error_response ( self , error , mimetype = '<STR_LIT:application/json>' , status = <NUM_LIT> , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return HttpResponse ( json . dumps ( error ) , mimetype = mimetype , <EOL> status = status , ** kwargs ) <EOL> def access_token_response ( self , access_token ) : <EOL> """<STR_LIT>""" <EOL> response_data = { <EOL> '<STR_LIT>' : access_token . token , <EOL> '<STR_LIT>' : constants . TOKEN_TYPE , <EOL> '<STR_LIT>' : access_token . get_expire_delta ( ) , <EOL> '<STR_LIT>' : '<STR_LIT:U+0020>' . join ( scope . names ( access_token . scope ) ) , <EOL> } <EOL> try : <EOL> rt = access_token . refresh_token <EOL> response_data [ '<STR_LIT>' ] = rt . token <EOL> except ObjectDoesNotExist : <EOL> pass <EOL> return HttpResponse ( <EOL> json . dumps ( response_data ) , mimetype = '<STR_LIT:application/json>' <EOL> ) <EOL> def authorization_code ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> grant = self . get_authorization_code_grant ( request , request . POST , <EOL> client ) <EOL> if constants . SINGLE_ACCESS_TOKEN : <EOL> at = self . get_access_token ( request , grant . user , grant . scope , client ) <EOL> else : <EOL> at = self . create_access_token ( request , grant . user , grant . scope , client ) <EOL> rt = self . create_refresh_token ( request , grant . user , grant . scope , at , <EOL> client ) <EOL> self . invalidate_grant ( grant ) <EOL> return self . access_token_response ( at ) <EOL> def refresh_token ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> rt = self . get_refresh_token_grant ( request , data , client ) <EOL> self . invalidate_refresh_token ( rt ) <EOL> self . invalidate_access_token ( rt . access_token ) <EOL> at = self . create_access_token ( request , rt . user , rt . access_token . scope , <EOL> client ) <EOL> rt = self . create_refresh_token ( request , at . user , at . scope , at , client ) <EOL> return self . access_token_response ( at ) <EOL> def password ( self , request , data , client ) : <EOL> """<STR_LIT>""" <EOL> data = self . get_password_grant ( request , data , client ) <EOL> user = data . get ( '<STR_LIT:user>' ) <EOL> scope = data . get ( '<STR_LIT>' ) <EOL> if constants . SINGLE_ACCESS_TOKEN : <EOL> at = self . get_access_token ( request , user , scope , client ) <EOL> else : <EOL> at = self . create_access_token ( request , user , scope , client ) <EOL> if client . client_type != <NUM_LIT:1> : <EOL> rt = self . create_refresh_token ( request , user , scope , at , client ) <EOL> return self . access_token_response ( at ) <EOL> def get_handler ( self , grant_type ) : <EOL> """<STR_LIT>""" <EOL> if grant_type == '<STR_LIT>' : <EOL> return self . authorization_code <EOL> elif grant_type == '<STR_LIT>' : <EOL> return self . refresh_token <EOL> elif grant_type == '<STR_LIT:password>' : <EOL> return self . password <EOL> return None <EOL> def get ( self , request ) : <EOL> """<STR_LIT>""" <EOL> return self . error_response ( { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) } ) <EOL> def post ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if constants . ENFORCE_SECURE and not request . is_secure ( ) : <EOL> return self . error_response ( { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" ) } ) <EOL> if not '<STR_LIT>' in request . POST : <EOL> return self . error_response ( { <EOL> '<STR_LIT:error>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) } ) <EOL> grant_type = request . POST [ '<STR_LIT>' ] <EOL> if grant_type not in self . grant_types : <EOL> return self . error_response ( { '<STR_LIT:error>' : '<STR_LIT>' } ) <EOL> client = self . authenticate ( request ) <EOL> if client is None : <EOL> return self . error_response ( { '<STR_LIT:error>' : '<STR_LIT>' } ) <EOL> handler = self . get_handler ( grant_type ) <EOL> try : <EOL> return handler ( request , request . POST , client ) <EOL> except OAuthError , e : <EOL> return self . error_response ( e . args [ <NUM_LIT:0> ] ) </s>
