<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals <EOL> import collections <EOL> import datetime <EOL> import logging <EOL> import random <EOL> import re <EOL> import sys <EOL> import time <EOL> try : <EOL> from urllib import quote as _quote <EOL> from urllib import unquote as _unquote <EOL> from urllib import urlencode as _urlencode <EOL> except ImportError : <EOL> from urllib . parse import quote as _quote <EOL> from urllib . parse import unquote as _unquote <EOL> from urllib . parse import urlencode as _urlencode <EOL> try : <EOL> import urlparse <EOL> except ImportError : <EOL> import urllib . parse as urlparse <EOL> UNICODE_ASCII_CHARACTER_SET = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> CLIENT_ID_CHARACTER_SET = ( r'<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> SANITIZE_PATTERN = re . compile ( r'<STR_LIT>' , re . IGNORECASE ) <EOL> INVALID_HEX_PATTERN = re . compile ( r'<STR_LIT>' ) <EOL> always_safe = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' '<STR_LIT>' ) <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> <EOL> if PY3 : <EOL> unicode_type = str <EOL> bytes_type = bytes <EOL> else : <EOL> unicode_type = unicode <EOL> bytes_type = str <EOL> def quote ( s , safe = b'<STR_LIT:/>' ) : <EOL> s = s . encode ( '<STR_LIT:utf-8>' ) if isinstance ( s , unicode_type ) else s <EOL> s = _quote ( s , safe ) <EOL> if isinstance ( s , bytes_type ) : <EOL> s = s . decode ( '<STR_LIT:utf-8>' ) <EOL> return s <EOL> def unquote ( s ) : <EOL> s = _unquote ( s ) <EOL> if isinstance ( s , bytes_type ) : <EOL> s = s . decode ( '<STR_LIT:utf-8>' ) <EOL> return s <EOL> def urlencode ( params ) : <EOL> utf8_params = encode_params_utf8 ( params ) <EOL> urlencoded = _urlencode ( utf8_params ) <EOL> if isinstance ( urlencoded , unicode_type ) : <EOL> return urlencoded <EOL> else : <EOL> return urlencoded . decode ( "<STR_LIT:utf-8>" ) <EOL> def encode_params_utf8 ( params ) : <EOL> """<STR_LIT>""" <EOL> encoded = [ ] <EOL> for k , v in params : <EOL> encoded . append ( ( <EOL> k . encode ( '<STR_LIT:utf-8>' ) if isinstance ( k , unicode_type ) else k , <EOL> v . encode ( '<STR_LIT:utf-8>' ) if isinstance ( v , unicode_type ) else v ) ) <EOL> return encoded <EOL> def decode_params_utf8 ( params ) : <EOL> """<STR_LIT>""" <EOL> decoded = [ ] <EOL> for k , v in params : <EOL> decoded . append ( ( <EOL> k . decode ( '<STR_LIT:utf-8>' ) if isinstance ( k , bytes_type ) else k , <EOL> v . decode ( '<STR_LIT:utf-8>' ) if isinstance ( v , bytes_type ) else v ) ) <EOL> return decoded <EOL> urlencoded = set ( always_safe ) | set ( '<STR_LIT>' ) <EOL> def urldecode ( query ) : <EOL> """<STR_LIT>""" <EOL> if query and not set ( query ) <= urlencoded : <EOL> error = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> raise ValueError ( error % ( set ( query ) - urlencoded , query ) ) <EOL> if INVALID_HEX_PATTERN . search ( query ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> query = query . encode ( <EOL> '<STR_LIT:utf-8>' ) if not PY3 and isinstance ( query , unicode_type ) else query <EOL> params = urlparse . parse_qsl ( query , keep_blank_values = True ) <EOL> return decode_params_utf8 ( params ) <EOL> def extract_params ( raw ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( raw , bytes_type ) or isinstance ( raw , unicode_type ) : <EOL> try : <EOL> params = urldecode ( raw ) <EOL> except ValueError : <EOL> params = None <EOL> elif hasattr ( raw , '<STR_LIT>' ) : <EOL> try : <EOL> dict ( raw ) <EOL> except ValueError : <EOL> params = None <EOL> except TypeError : <EOL> params = None <EOL> else : <EOL> params = list ( raw . items ( ) if isinstance ( raw , dict ) else raw ) <EOL> params = decode_params_utf8 ( params ) <EOL> else : <EOL> params = None <EOL> return params <EOL> def generate_nonce ( ) : <EOL> """<STR_LIT>""" <EOL> return unicode_type ( unicode_type ( random . getrandbits ( <NUM_LIT:64> ) ) + generate_timestamp ( ) ) <EOL> def generate_timestamp ( ) : <EOL> """<STR_LIT>""" <EOL> return unicode_type ( int ( time . time ( ) ) ) <EOL> def generate_token ( length = <NUM_LIT:30> , chars = UNICODE_ASCII_CHARACTER_SET ) : <EOL> """<STR_LIT>""" <EOL> rand = random . SystemRandom ( ) <EOL> return '<STR_LIT>' . join ( rand . choice ( chars ) for x in range ( length ) ) <EOL> def generate_signed_token ( private_pem , request ) : <EOL> import jwt <EOL> now = datetime . datetime . utcnow ( ) <EOL> claims = { <EOL> '<STR_LIT>' : request . scope , <EOL> '<STR_LIT>' : now + datetime . timedelta ( seconds = request . expires_in ) <EOL> } <EOL> claims . update ( request . claims ) <EOL> token = jwt . encode ( claims , private_pem , '<STR_LIT>' ) <EOL> token = to_unicode ( token , "<STR_LIT>" ) <EOL> return token <EOL> def verify_signed_token ( public_pem , token ) : <EOL> import jwt <EOL> return jwt . decode ( token , public_pem , algorithms = [ '<STR_LIT>' ] ) <EOL> def generate_client_id ( length = <NUM_LIT:30> , chars = CLIENT_ID_CHARACTER_SET ) : <EOL> """<STR_LIT>""" <EOL> return generate_token ( length , chars ) <EOL> def add_params_to_qs ( query , params ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( params , dict ) : <EOL> params = params . items ( ) <EOL> queryparams = urlparse . parse_qsl ( query , keep_blank_values = True ) <EOL> queryparams . extend ( params ) <EOL> return urlencode ( queryparams ) <EOL> def add_params_to_uri ( uri , params , fragment = False ) : <EOL> """<STR_LIT>""" <EOL> sch , net , path , par , query , fra = urlparse . urlparse ( uri ) <EOL> if fragment : <EOL> fra = add_params_to_qs ( fra , params ) <EOL> else : <EOL> query = add_params_to_qs ( query , params ) <EOL> return urlparse . urlunparse ( ( sch , net , path , par , query , fra ) ) <EOL> def safe_string_equals ( a , b ) : <EOL> """<STR_LIT>""" <EOL> if len ( a ) != len ( b ) : <EOL> return False <EOL> result = <NUM_LIT:0> <EOL> for x , y in zip ( a , b ) : <EOL> result |= ord ( x ) ^ ord ( y ) <EOL> return result == <NUM_LIT:0> <EOL> def to_unicode ( data , encoding = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , unicode_type ) : <EOL> return data <EOL> if isinstance ( data , bytes_type ) : <EOL> return unicode_type ( data , encoding = encoding ) <EOL> if hasattr ( data , '<STR_LIT>' ) : <EOL> try : <EOL> dict ( data ) <EOL> except TypeError : <EOL> pass <EOL> except ValueError : <EOL> return ( to_unicode ( i , encoding ) for i in data ) <EOL> else : <EOL> if hasattr ( data , '<STR_LIT>' ) : <EOL> data = data . items ( ) <EOL> return dict ( ( ( to_unicode ( k , encoding ) , to_unicode ( v , encoding ) ) for k , v in data ) ) <EOL> return data <EOL> class CaseInsensitiveDict ( dict ) : <EOL> """<STR_LIT>""" <EOL> proxy = { } <EOL> def __init__ ( self , data ) : <EOL> self . proxy = dict ( ( k . lower ( ) , k ) for k in data ) <EOL> for k in data : <EOL> self [ k ] = data [ k ] <EOL> def __contains__ ( self , k ) : <EOL> return k . lower ( ) in self . proxy <EOL> def __delitem__ ( self , k ) : <EOL> key = self . proxy [ k . lower ( ) ] <EOL> super ( CaseInsensitiveDict , self ) . __delitem__ ( key ) <EOL> del self . proxy [ k . lower ( ) ] <EOL> def __getitem__ ( self , k ) : <EOL> key = self . proxy [ k . lower ( ) ] <EOL> return super ( CaseInsensitiveDict , self ) . __getitem__ ( key ) <EOL> def get ( self , k , default = None ) : <EOL> return self [ k ] if k in self else default <EOL> def __setitem__ ( self , k , v ) : <EOL> super ( CaseInsensitiveDict , self ) . __setitem__ ( k , v ) <EOL> self . proxy [ k . lower ( ) ] = k <EOL> class Request ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uri , http_method = '<STR_LIT:GET>' , body = None , headers = None , <EOL> encoding = '<STR_LIT:utf-8>' ) : <EOL> encode = lambda x : to_unicode ( x , encoding ) if encoding else x <EOL> self . uri = encode ( uri ) <EOL> self . http_method = encode ( http_method ) <EOL> self . headers = CaseInsensitiveDict ( encode ( headers or { } ) ) <EOL> self . body = encode ( body ) <EOL> self . decoded_body = extract_params ( self . body ) <EOL> self . oauth_params = [ ] <EOL> self . validator_log = { } <EOL> self . _params = { <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:code>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:state>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT:user>" : None , <EOL> "<STR_LIT>" : None , <EOL> } <EOL> self . _params . update ( dict ( urldecode ( self . uri_query ) ) ) <EOL> self . _params . update ( dict ( self . decoded_body or [ ] ) ) <EOL> self . _params . update ( self . headers ) <EOL> def __getattr__ ( self , name ) : <EOL> if name in self . _params : <EOL> return self . _params [ name ] <EOL> else : <EOL> raise AttributeError ( name ) <EOL> def __repr__ ( self ) : <EOL> body = self . body <EOL> headers = self . headers . copy ( ) <EOL> if body : <EOL> body = SANITIZE_PATTERN . sub ( '<STR_LIT>' , body ) <EOL> if '<STR_LIT>' in headers : <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> return '<STR_LIT>' % ( <EOL> self . uri , self . http_method , headers , body ) <EOL> @ property <EOL> def uri_query ( self ) : <EOL> return urlparse . urlparse ( self . uri ) . query <EOL> @ property <EOL> def uri_query_params ( self ) : <EOL> if not self . uri_query : <EOL> return [ ] <EOL> return urlparse . parse_qsl ( self . uri_query , keep_blank_values = True , <EOL> strict_parsing = True ) <EOL> @ property <EOL> def duplicate_params ( self ) : <EOL> seen_keys = collections . defaultdict ( int ) <EOL> all_keys = ( p [ <NUM_LIT:0> ] <EOL> for p in ( self . decoded_body or [ ] ) + self . uri_query_params ) <EOL> for k in all_keys : <EOL> seen_keys [ k ] += <NUM_LIT:1> <EOL> return [ k for k , c in seen_keys . items ( ) if c > <NUM_LIT:1> ] </s>
