<s> from __future__ import unicode_literals <EOL> import six <EOL> import base64 <EOL> import binascii <EOL> import logging <EOL> from datetime import timedelta <EOL> from django . utils import timezone <EOL> from django . conf import settings <EOL> from django . contrib . auth import authenticate <EOL> from django . core . exceptions import ObjectDoesNotExist <EOL> from oauthlib . oauth2 import RequestValidator <EOL> from . compat import unquote_plus <EOL> from . models import Grant , AccessToken , RefreshToken , get_application_model , AbstractApplication <EOL> from . settings import oauth2_settings <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> GRANT_TYPE_MAPPING = { <EOL> '<STR_LIT>' : ( AbstractApplication . GRANT_AUTHORIZATION_CODE , ) , <EOL> '<STR_LIT:password>' : ( AbstractApplication . GRANT_PASSWORD , ) , <EOL> '<STR_LIT>' : ( AbstractApplication . GRANT_CLIENT_CREDENTIALS , ) , <EOL> '<STR_LIT>' : ( AbstractApplication . GRANT_AUTHORIZATION_CODE , AbstractApplication . GRANT_PASSWORD , <EOL> AbstractApplication . GRANT_CLIENT_CREDENTIALS ) <EOL> } <EOL> class OAuth2Validator ( RequestValidator ) : <EOL> def _extract_basic_auth ( self , request ) : <EOL> """<STR_LIT>""" <EOL> auth = request . headers . get ( '<STR_LIT>' , None ) <EOL> if not auth : <EOL> return None <EOL> splitted = auth . split ( '<STR_LIT:U+0020>' , <NUM_LIT:1> ) <EOL> if len ( splitted ) != <NUM_LIT:2> : <EOL> return None <EOL> auth_type , auth_string = splitted <EOL> if auth_type != "<STR_LIT>" : <EOL> return None <EOL> return auth_string <EOL> def _authenticate_basic_auth ( self , request ) : <EOL> """<STR_LIT>""" <EOL> auth_string = self . _extract_basic_auth ( request ) <EOL> if not auth_string : <EOL> return False <EOL> try : <EOL> encoding = request . encoding or settings . DEFAULT_CHARSET or '<STR_LIT:utf-8>' <EOL> except AttributeError : <EOL> encoding = '<STR_LIT:utf-8>' <EOL> if isinstance ( auth_string , six . string_types ) : <EOL> auth_string = auth_string . encode ( encoding ) <EOL> try : <EOL> b64_decoded = base64 . b64decode ( auth_string ) <EOL> except ( TypeError , binascii . Error ) : <EOL> log . debug ( "<STR_LIT>" , auth_string ) <EOL> return False <EOL> try : <EOL> auth_string_decoded = b64_decoded . decode ( encoding ) <EOL> except UnicodeDecodeError : <EOL> log . debug ( "<STR_LIT>" , <EOL> auth_string , <EOL> encoding ) <EOL> return False <EOL> client_id , client_secret = map ( unquote_plus , auth_string_decoded . split ( '<STR_LIT::>' , <NUM_LIT:1> ) ) <EOL> if self . _load_application ( client_id , request ) is None : <EOL> log . debug ( "<STR_LIT>" % client_id ) <EOL> return False <EOL> elif request . client . client_secret != client_secret : <EOL> log . debug ( "<STR_LIT>" % client_secret ) <EOL> return False <EOL> else : <EOL> return True <EOL> def _authenticate_request_body ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> client_id = request . client_id <EOL> client_secret = request . client_secret <EOL> except AttributeError : <EOL> return False <EOL> if self . _load_application ( client_id , request ) is None : <EOL> log . debug ( "<STR_LIT>" % client_id ) <EOL> return False <EOL> elif request . client . client_secret != client_secret : <EOL> log . debug ( "<STR_LIT>" % client_secret ) <EOL> return False <EOL> else : <EOL> return True <EOL> def _load_application ( self , client_id , request ) : <EOL> """<STR_LIT>""" <EOL> assert hasattr ( request , "<STR_LIT>" ) , "<STR_LIT>" <EOL> Application = get_application_model ( ) <EOL> try : <EOL> request . client = request . client or Application . objects . get ( client_id = client_id ) <EOL> return request . client <EOL> except Application . DoesNotExist : <EOL> log . debug ( "<STR_LIT>" % client_id ) <EOL> return None <EOL> def client_authentication_required ( self , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . _extract_basic_auth ( request ) : <EOL> return True <EOL> try : <EOL> if request . client_id and request . client_secret : <EOL> return True <EOL> except AttributeError : <EOL> log . debug ( "<STR_LIT>" ) <EOL> pass <EOL> self . _load_application ( request . client_id , request ) <EOL> if request . client : <EOL> return request . client . client_type == AbstractApplication . CLIENT_CONFIDENTIAL <EOL> return super ( OAuth2Validator , self ) . client_authentication_required ( request , <EOL> * args , ** kwargs ) <EOL> def authenticate_client ( self , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> authenticated = self . _authenticate_basic_auth ( request ) <EOL> if not authenticated : <EOL> authenticated = self . _authenticate_request_body ( request ) <EOL> return authenticated <EOL> def authenticate_client_id ( self , client_id , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . _load_application ( client_id , request ) is not None : <EOL> log . debug ( "<STR_LIT>" % ( client_id , request . client . client_type ) ) <EOL> return request . client . client_type != AbstractApplication . CLIENT_CONFIDENTIAL <EOL> return False <EOL> def confirm_redirect_uri ( self , client_id , code , redirect_uri , client , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> grant = Grant . objects . get ( code = code , application = client ) <EOL> return grant . redirect_uri_allowed ( redirect_uri ) <EOL> def invalidate_authorization_code ( self , client_id , code , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> grant = Grant . objects . get ( code = code , application = request . client ) <EOL> grant . delete ( ) <EOL> def validate_client_id ( self , client_id , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _load_application ( client_id , request ) is not None <EOL> def get_default_redirect_uri ( self , client_id , request , * args , ** kwargs ) : <EOL> return request . client . default_redirect_uri <EOL> def validate_bearer_token ( self , token , scopes , request ) : <EOL> """<STR_LIT>""" <EOL> if not token : <EOL> return False <EOL> try : <EOL> access_token = AccessToken . objects . select_related ( "<STR_LIT>" , "<STR_LIT:user>" ) . get ( <EOL> token = token ) <EOL> if access_token . is_valid ( scopes ) : <EOL> request . client = access_token . application <EOL> request . user = access_token . user <EOL> request . scopes = scopes <EOL> request . access_token = access_token <EOL> return True <EOL> return False <EOL> except AccessToken . DoesNotExist : <EOL> return False <EOL> def validate_code ( self , client_id , code , client , request , * args , ** kwargs ) : <EOL> try : <EOL> grant = Grant . objects . get ( code = code , application = client ) <EOL> if not grant . is_expired ( ) : <EOL> request . scopes = grant . scope . split ( '<STR_LIT:U+0020>' ) <EOL> request . user = grant . user <EOL> return True <EOL> return False <EOL> except Grant . DoesNotExist : <EOL> return False <EOL> def validate_grant_type ( self , client_id , grant_type , client , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> assert ( grant_type in GRANT_TYPE_MAPPING ) <EOL> return request . client . authorization_grant_type in GRANT_TYPE_MAPPING [ grant_type ] <EOL> def validate_response_type ( self , client_id , response_type , client , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if response_type == '<STR_LIT:code>' : <EOL> return client . authorization_grant_type == AbstractApplication . GRANT_AUTHORIZATION_CODE <EOL> elif response_type == '<STR_LIT>' : <EOL> return client . authorization_grant_type == AbstractApplication . GRANT_IMPLICIT <EOL> else : <EOL> return False <EOL> def validate_scopes ( self , client_id , scopes , client , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return set ( scopes ) . issubset ( set ( oauth2_settings . _SCOPES ) ) <EOL> def get_default_scopes ( self , client_id , request , * args , ** kwargs ) : <EOL> return oauth2_settings . _DEFAULT_SCOPES <EOL> def validate_redirect_uri ( self , client_id , redirect_uri , request , * args , ** kwargs ) : <EOL> return request . client . redirect_uri_allowed ( redirect_uri ) <EOL> def save_authorization_code ( self , client_id , code , request , * args , ** kwargs ) : <EOL> expires = timezone . now ( ) + timedelta ( <EOL> seconds = oauth2_settings . AUTHORIZATION_CODE_EXPIRE_SECONDS ) <EOL> g = Grant ( application = request . client , user = request . user , code = code [ '<STR_LIT:code>' ] , <EOL> expires = expires , redirect_uri = request . redirect_uri , <EOL> scope = '<STR_LIT:U+0020>' . join ( request . scopes ) ) <EOL> g . save ( ) <EOL> def save_bearer_token ( self , token , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if request . refresh_token : <EOL> try : <EOL> RefreshToken . objects . get ( token = request . refresh_token ) . revoke ( ) <EOL> except RefreshToken . DoesNotExist : <EOL> assert ( ) <EOL> expires = timezone . now ( ) + timedelta ( seconds = oauth2_settings . ACCESS_TOKEN_EXPIRE_SECONDS ) <EOL> if request . grant_type == '<STR_LIT>' : <EOL> request . user = None <EOL> access_token = AccessToken ( <EOL> user = request . user , <EOL> scope = token [ '<STR_LIT>' ] , <EOL> expires = expires , <EOL> token = token [ '<STR_LIT>' ] , <EOL> application = request . client ) <EOL> access_token . save ( ) <EOL> if '<STR_LIT>' in token : <EOL> refresh_token = RefreshToken ( <EOL> user = request . user , <EOL> token = token [ '<STR_LIT>' ] , <EOL> application = request . client , <EOL> access_token = access_token <EOL> ) <EOL> refresh_token . save ( ) <EOL> token [ '<STR_LIT>' ] = oauth2_settings . ACCESS_TOKEN_EXPIRE_SECONDS <EOL> def revoke_token ( self , token , token_type_hint , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if token_type_hint not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> token_type_hint = None <EOL> token_types = { <EOL> '<STR_LIT>' : AccessToken , <EOL> '<STR_LIT>' : RefreshToken , <EOL> } <EOL> token_type = token_types . get ( token_type_hint , AccessToken ) <EOL> try : <EOL> token_type . objects . get ( token = token ) . revoke ( ) <EOL> except ObjectDoesNotExist : <EOL> for other_type in [ _t for _t in token_types . values ( ) if _t != token_type ] : <EOL> list ( map ( lambda t : t . revoke ( ) , other_type . objects . filter ( token = token ) ) ) <EOL> def validate_user ( self , username , password , client , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> u = authenticate ( username = username , password = password ) <EOL> if u is not None and u . is_active : <EOL> request . user = u <EOL> return True <EOL> return False <EOL> def get_original_scopes ( self , refresh_token , request , * args , ** kwargs ) : <EOL> rt = request . refresh_token_instance <EOL> return rt . access_token . scope <EOL> def validate_refresh_token ( self , refresh_token , client , request , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> rt = RefreshToken . objects . get ( token = refresh_token ) <EOL> request . user = rt . user <EOL> request . refresh_token = rt . token <EOL> request . refresh_token_instance = rt <EOL> return rt . application == client <EOL> except RefreshToken . DoesNotExist : <EOL> return False </s>
