<s> """<STR_LIT>""" <EOL> try : import simplejson as json <EOL> except ImportError : import json <EOL> from base64 import b64encode <EOL> from django . http import HttpResponse <EOL> from django . contrib . auth import authenticate <EOL> from django . views . decorators . csrf import csrf_exempt <EOL> from . exceptions import OAuth2Exception <EOL> from . consts import ACCESS_TOKEN_EXPIRATION , REFRESH_TOKEN_LENGTH <EOL> from . consts import AUTHENTICATION_METHOD , MAC , BEARER , MAC_KEY_LENGTH <EOL> from . consts import REFRESHABLE <EOL> from . lib . uri import normalize <EOL> from . models import Client , AccessRange , Code , AccessToken , TimestampGenerator <EOL> from . models import KeyGenerator <EOL> class AccessTokenException ( OAuth2Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class UnvalidatedRequest ( OAuth2Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InvalidRequest ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> class InvalidClient ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> class UnauthorizedClient ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> class InvalidGrant ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> class UnsupportedGrantType ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> class InvalidScope ( AccessTokenException ) : <EOL> """<STR_LIT>""" <EOL> error = '<STR_LIT>' <EOL> @ csrf_exempt <EOL> def handler ( request ) : <EOL> """<STR_LIT>""" <EOL> return TokenGenerator ( ) ( request ) <EOL> class TokenGenerator ( object ) : <EOL> """<STR_LIT>""" <EOL> valid = False <EOL> code = None <EOL> client = None <EOL> access_token = None <EOL> user = None <EOL> error = None <EOL> request = None <EOL> def __init__ ( <EOL> self , <EOL> scope = None , <EOL> authentication_method = AUTHENTICATION_METHOD , <EOL> refreshable = REFRESHABLE ) : <EOL> self . refreshable = refreshable <EOL> if authentication_method not in [ BEARER , MAC ] : <EOL> raise OAuth2Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . authentication_method = authentication_method <EOL> if scope is None : <EOL> self . authorized_scope = None <EOL> elif isinstance ( scope , AccessRange ) : <EOL> self . authorized_scope = set ( [ scope . key ] ) <EOL> else : <EOL> self . authorized_scope = set ( [ x . key for x in scope ] ) <EOL> @ csrf_exempt <EOL> def __call__ ( self , request ) : <EOL> """<STR_LIT>""" <EOL> self . grant_type = request . REQUEST . get ( '<STR_LIT>' ) <EOL> self . client_id = request . REQUEST . get ( '<STR_LIT>' ) <EOL> self . client_secret = request . POST . get ( '<STR_LIT>' ) <EOL> self . scope = request . REQUEST . get ( '<STR_LIT>' ) <EOL> if self . scope is not None : <EOL> self . scope = set ( self . scope . split ( ) ) <EOL> self . code_key = request . REQUEST . get ( '<STR_LIT:code>' ) <EOL> self . redirect_uri = request . REQUEST . get ( '<STR_LIT>' ) <EOL> self . refresh_token = request . REQUEST . get ( '<STR_LIT>' ) <EOL> self . email = request . REQUEST . get ( '<STR_LIT:email>' ) <EOL> self . username = request . REQUEST . get ( '<STR_LIT:username>' ) <EOL> self . password = request . REQUEST . get ( '<STR_LIT:password>' ) <EOL> self . callback = request . REQUEST . get ( '<STR_LIT>' ) <EOL> self . request = request <EOL> try : <EOL> self . validate ( ) <EOL> except AccessTokenException , e : <EOL> return self . error_response ( ) <EOL> return self . grant_response ( ) <EOL> def validate ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _validate ( ) <EOL> except AccessTokenException , e : <EOL> self . error = e <EOL> raise e <EOL> self . valid = True <EOL> def _validate ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . grant_type is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> if self . grant_type not in [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:password>" , <EOL> "<STR_LIT>" ] : <EOL> raise UnsupportedGrantType ( '<STR_LIT>' % self . grant_type ) <EOL> if self . client_id is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> try : <EOL> self . client = Client . objects . get ( key = self . client_id ) <EOL> except Client . DoesNotExist : <EOL> raise InvalidClient ( "<STR_LIT>" % self . client_id ) <EOL> if self . scope is not None : <EOL> access_ranges = AccessRange . objects . filter ( key__in = self . scope ) <EOL> access_ranges = set ( access_ranges . values_list ( '<STR_LIT:key>' , flat = True ) ) <EOL> difference = access_ranges . symmetric_difference ( self . scope ) <EOL> if len ( difference ) != <NUM_LIT:0> : <EOL> raise InvalidScope ( "<STR_LIT>" <EOL> "<STR_LIT:%s>" % '<STR_LIT:U+002CU+0020>' . join ( difference ) ) <EOL> if self . grant_type == "<STR_LIT>" : <EOL> self . _validate_authorization_code ( ) <EOL> elif self . grant_type == "<STR_LIT>" : <EOL> self . _validate_refresh_token ( ) <EOL> elif self . grant_type == "<STR_LIT:password>" : <EOL> self . _validate_password ( ) <EOL> elif self . grant_type == "<STR_LIT>" : <EOL> self . _validate_client_credentials ( ) <EOL> else : <EOL> raise UnsupportedGrantType ( '<STR_LIT>' ) <EOL> def _validate_access_credentials ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . client_secret is None and "<STR_LIT>" in self . request . META : <EOL> authorization = self . request . META [ "<STR_LIT>" ] <EOL> auth_type , auth_value = authorization . split ( ) [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> if auth_type . lower ( ) == "<STR_LIT>" : <EOL> credentials = "<STR_LIT>" % ( self . client . key , self . client . secret ) <EOL> if auth_value != b64encode ( credentials ) : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> else : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> elif self . client_secret != self . client . secret : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> def _validate_client_credentials ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_access_credentials ( ) <EOL> def _validate_authorization_code ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . code_key is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> self . _validate_access_credentials ( ) <EOL> try : <EOL> self . code = Code . objects . get ( key = self . code_key ) <EOL> except Code . DoesNotExist : <EOL> raise InvalidRequest ( '<STR_LIT>' % self . code_key ) <EOL> now = TimestampGenerator ( ) ( ) <EOL> if self . code . expire < now : <EOL> raise InvalidGrant ( "<STR_LIT>" ) <EOL> self . scope = set ( [ x . key for x in self . code . scope . all ( ) ] ) <EOL> if self . redirect_uri is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> if normalize ( self . redirect_uri ) != normalize ( self . code . redirect_uri ) : <EOL> raise InvalidRequest ( "<STR_LIT>" ) <EOL> def _validate_password ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . username is None and self . email is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> if self . password is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> if self . scope is not None : <EOL> access_ranges = AccessRange . objects . filter ( key__in = self . scope ) <EOL> access_ranges = set ( access_ranges . values_list ( '<STR_LIT:key>' , flat = True ) ) <EOL> difference = access_ranges . symmetric_difference ( self . scope ) <EOL> if len ( difference ) != <NUM_LIT:0> : <EOL> raise InvalidScope ( """<STR_LIT>""" % '<STR_LIT:U+002CU+0020>' . join ( difference ) ) <EOL> if self . authorized_scope is not None : <EOL> new_scope = self . scope - self . authorized_scope <EOL> if len ( new_scope ) > <NUM_LIT:0> : <EOL> raise InvalidScope ( <EOL> "<STR_LIT>" % '<STR_LIT:U+002CU+0020>' . join ( new_scope ) ) <EOL> if "<STR_LIT>" in self . request . META : <EOL> authorization = self . request . META [ "<STR_LIT>" ] <EOL> auth_type , auth_value = authorization . split ( ) [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> if auth_type . lower ( ) == "<STR_LIT>" : <EOL> credentials = "<STR_LIT>" % ( self . client . key , self . client . secret ) <EOL> if auth_value != b64encode ( credentials ) : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> else : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> else : <EOL> raise InvalidClient ( '<STR_LIT>' ) <EOL> if self . username is not None : <EOL> user = authenticate ( username = self . username , password = self . password ) <EOL> else : <EOL> user = authenticate ( email = self . email , password = self . password ) <EOL> if user is not None : <EOL> if not user . is_active : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> else : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> self . user = user <EOL> def _validate_refresh_token ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . refresh_token is None : <EOL> raise InvalidRequest ( '<STR_LIT>' ) <EOL> try : <EOL> self . access_token = AccessToken . objects . get ( <EOL> refresh_token = self . refresh_token ) <EOL> except AccessToken . DoesNotExist : <EOL> raise InvalidRequest ( <EOL> '<STR_LIT>' % self . refresh_token ) <EOL> self . _validate_access_credentials ( ) <EOL> if not self . access_token . refreshable : <EOL> raise InvalidGrant ( "<STR_LIT>" ) <EOL> if self . scope is not None : <EOL> access_ranges = set ( [ x . key for x in self . access_token . scope . all ( ) ] ) <EOL> new_scope = self . scope - access_ranges <EOL> if len ( new_scope ) > <NUM_LIT:0> : <EOL> raise InvalidScope ( "<STR_LIT>" <EOL> "<STR_LIT>" % new_scope ) <EOL> def error_response ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . error is not None : <EOL> e = self . error <EOL> else : <EOL> e = InvalidRequest ( "<STR_LIT>" ) <EOL> data = { '<STR_LIT:error>' : e . error , '<STR_LIT>' : u'<STR_LIT:%s>' % e . message } <EOL> json_data = json . dumps ( data ) <EOL> if self . callback is not None : <EOL> json_data = "<STR_LIT>" % ( self . callback , json_data ) <EOL> return HttpResponse ( <EOL> content = json_data , <EOL> content_type = '<STR_LIT:application/json>' ) <EOL> else : <EOL> response = HttpResponse ( <EOL> content = json_data , <EOL> content_type = '<STR_LIT:application/json>' ) <EOL> if isinstance ( self . error , InvalidClient ) : <EOL> response . status_code = <NUM_LIT> <EOL> else : <EOL> response . status_code = <NUM_LIT> <EOL> return response <EOL> def grant_response ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . valid : <EOL> raise UnvalidatedRequest ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if self . grant_type == "<STR_LIT>" : <EOL> access_token = self . _get_authorization_code_token ( ) <EOL> elif self . grant_type == "<STR_LIT>" : <EOL> access_token = self . _get_refresh_token ( ) <EOL> elif self . grant_type == "<STR_LIT:password>" : <EOL> access_token = self . _get_password_token ( ) <EOL> elif self . grant_type == "<STR_LIT>" : <EOL> access_token = self . _get_client_credentials_token ( ) <EOL> data = { <EOL> '<STR_LIT>' : access_token . token , <EOL> '<STR_LIT>' : ACCESS_TOKEN_EXPIRATION } <EOL> if self . authentication_method == MAC : <EOL> data [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> data [ "<STR_LIT>" ] = access_token . mac_key <EOL> data [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> elif self . authentication_method == BEARER : <EOL> data [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> if access_token . refreshable : <EOL> data [ '<STR_LIT>' ] = access_token . refresh_token <EOL> if self . scope is not None : <EOL> data [ '<STR_LIT>' ] = '<STR_LIT:U+0020>' . join ( self . scope ) <EOL> json_data = json . dumps ( data ) <EOL> if self . callback is not None : <EOL> json_data = "<STR_LIT>" % ( self . callback , json_data ) <EOL> response = HttpResponse ( <EOL> content = json_data , <EOL> content_type = '<STR_LIT:application/json>' ) <EOL> response [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return response <EOL> def _get_authorization_code_token ( self ) : <EOL> """<STR_LIT>""" <EOL> access_token = AccessToken . objects . create ( <EOL> user = self . code . user , <EOL> client = self . client , <EOL> refreshable = self . refreshable ) <EOL> if self . authentication_method == MAC : <EOL> access_token . mac_key = KeyGenerator ( MAC_KEY_LENGTH ) ( ) <EOL> access_ranges = AccessRange . objects . filter ( key__in = self . scope ) if self . scope else [ ] <EOL> access_token . scope = access_ranges <EOL> access_token . save ( ) <EOL> self . code . delete ( ) <EOL> return access_token <EOL> def _get_password_token ( self ) : <EOL> """<STR_LIT>""" <EOL> access_token = AccessToken . objects . create ( <EOL> user = self . user , <EOL> client = self . client , <EOL> refreshable = self . refreshable ) <EOL> if self . authentication_method == MAC : <EOL> access_token . mac_key = KeyGenerator ( MAC_KEY_LENGTH ) ( ) <EOL> access_ranges = AccessRange . objects . filter ( key__in = self . scope ) if self . scope else [ ] <EOL> access_token . scope = access_ranges <EOL> access_token . save ( ) <EOL> return access_token <EOL> def _get_refresh_token ( self ) : <EOL> """<STR_LIT>""" <EOL> self . access_token . refresh_token = KeyGenerator ( REFRESH_TOKEN_LENGTH ) ( ) <EOL> self . access_token . expire = TimestampGenerator ( ACCESS_TOKEN_EXPIRATION ) ( ) <EOL> if self . scope : <EOL> self . access_token . scope = AccessRange . objects . filter ( key__in = self . scope ) <EOL> elif not self . access_token . scope . exists ( ) : <EOL> self . access_token . scope = [ ] <EOL> self . access_token . save ( ) <EOL> return self . access_token <EOL> def _get_client_credentials_token ( self ) : <EOL> """<STR_LIT>""" <EOL> access_token = AccessToken . objects . create ( <EOL> user = self . client . user , <EOL> client = self . client , <EOL> refreshable = self . refreshable ) <EOL> if self . authentication_method == MAC : <EOL> access_token . mac_key = KeyGenerator ( MAC_KEY_LENGTH ) ( ) <EOL> access_ranges = AccessRange . objects . filter ( key__in = self . scope ) if self . scope else [ ] <EOL> access_token . scope = access_ranges <EOL> access_token . save ( ) <EOL> return access_token </s>
