<s> """<STR_LIT>""" <EOL> import os <EOL> import time <EOL> import hashlib <EOL> import logging <EOL> from base64 import b64encode <EOL> from . compat import urlparse , str <EOL> from . utils import parse_dict_header <EOL> log = logging . getLogger ( __name__ ) <EOL> CONTENT_TYPE_FORM_URLENCODED = '<STR_LIT>' <EOL> CONTENT_TYPE_MULTI_PART = '<STR_LIT>' <EOL> def _basic_auth_str ( username , password ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' + b64encode ( ( '<STR_LIT>' % ( username , password ) ) . encode ( '<STR_LIT>' ) ) . strip ( ) . decode ( '<STR_LIT>' ) <EOL> class AuthBase ( object ) : <EOL> """<STR_LIT>""" <EOL> def __call__ ( self , r ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> class HTTPBasicAuth ( AuthBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , username , password ) : <EOL> self . username = username <EOL> self . password = password <EOL> def __call__ ( self , r ) : <EOL> r . headers [ '<STR_LIT>' ] = _basic_auth_str ( self . username , self . password ) <EOL> return r <EOL> class HTTPProxyAuth ( HTTPBasicAuth ) : <EOL> """<STR_LIT>""" <EOL> def __call__ ( self , r ) : <EOL> r . headers [ '<STR_LIT>' ] = _basic_auth_str ( self . username , self . password ) <EOL> return r <EOL> class HTTPDigestAuth ( AuthBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , username , password ) : <EOL> self . username = username <EOL> self . password = password <EOL> self . last_nonce = '<STR_LIT>' <EOL> self . nonce_count = <NUM_LIT:0> <EOL> self . chal = { } <EOL> def build_digest_header ( self , method , url ) : <EOL> realm = self . chal [ '<STR_LIT>' ] <EOL> nonce = self . chal [ '<STR_LIT>' ] <EOL> qop = self . chal . get ( '<STR_LIT>' ) <EOL> algorithm = self . chal . get ( '<STR_LIT>' ) <EOL> opaque = self . chal . get ( '<STR_LIT>' ) <EOL> if algorithm is None : <EOL> _algorithm = '<STR_LIT>' <EOL> else : <EOL> _algorithm = algorithm . upper ( ) <EOL> if _algorithm == '<STR_LIT>' : <EOL> def md5_utf8 ( x ) : <EOL> if isinstance ( x , str ) : <EOL> x = x . encode ( '<STR_LIT:utf-8>' ) <EOL> return hashlib . md5 ( x ) . hexdigest ( ) <EOL> hash_utf8 = md5_utf8 <EOL> elif _algorithm == '<STR_LIT>' : <EOL> def sha_utf8 ( x ) : <EOL> if isinstance ( x , str ) : <EOL> x = x . encode ( '<STR_LIT:utf-8>' ) <EOL> return hashlib . sha1 ( x ) . hexdigest ( ) <EOL> hash_utf8 = sha_utf8 <EOL> KD = lambda s , d : hash_utf8 ( "<STR_LIT>" % ( s , d ) ) <EOL> if hash_utf8 is None : <EOL> return None <EOL> entdig = None <EOL> p_parsed = urlparse ( url ) <EOL> path = p_parsed . path <EOL> if p_parsed . query : <EOL> path += '<STR_LIT:?>' + p_parsed . query <EOL> A1 = '<STR_LIT>' % ( self . username , realm , self . password ) <EOL> A2 = '<STR_LIT>' % ( method , path ) <EOL> if qop == '<STR_LIT>' : <EOL> if nonce == self . last_nonce : <EOL> self . nonce_count += <NUM_LIT:1> <EOL> else : <EOL> self . nonce_count = <NUM_LIT:1> <EOL> ncvalue = '<STR_LIT>' % self . nonce_count <EOL> s = str ( self . nonce_count ) . encode ( '<STR_LIT:utf-8>' ) <EOL> s += nonce . encode ( '<STR_LIT:utf-8>' ) <EOL> s += time . ctime ( ) . encode ( '<STR_LIT:utf-8>' ) <EOL> s += os . urandom ( <NUM_LIT:8> ) <EOL> cnonce = ( hashlib . sha1 ( s ) . hexdigest ( ) [ : <NUM_LIT:16> ] ) <EOL> noncebit = "<STR_LIT>" % ( nonce , ncvalue , cnonce , qop , hash_utf8 ( A2 ) ) <EOL> respdig = KD ( hash_utf8 ( A1 ) , noncebit ) <EOL> elif qop is None : <EOL> respdig = KD ( hash_utf8 ( A1 ) , "<STR_LIT>" % ( nonce , hash_utf8 ( A2 ) ) ) <EOL> else : <EOL> return None <EOL> self . last_nonce = nonce <EOL> base = '<STR_LIT>' '<STR_LIT>' % ( self . username , realm , nonce , path , respdig ) <EOL> if opaque : <EOL> base += '<STR_LIT>' % opaque <EOL> if algorithm : <EOL> base += '<STR_LIT>' % algorithm <EOL> if entdig : <EOL> base += '<STR_LIT>' % entdig <EOL> if qop : <EOL> base += '<STR_LIT>' % ( ncvalue , cnonce ) <EOL> return '<STR_LIT>' % ( base ) <EOL> def handle_401 ( self , r , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> num_401_calls = getattr ( self , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> s_auth = r . headers . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if '<STR_LIT>' in s_auth . lower ( ) and num_401_calls < <NUM_LIT:2> : <EOL> setattr ( self , '<STR_LIT>' , num_401_calls + <NUM_LIT:1> ) <EOL> self . chal = parse_dict_header ( s_auth . replace ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> r . content <EOL> r . raw . release_conn ( ) <EOL> r . request . headers [ '<STR_LIT>' ] = self . build_digest_header ( r . request . method , r . request . url ) <EOL> _r = r . connection . send ( r . request , ** kwargs ) <EOL> _r . history . append ( r ) <EOL> return _r <EOL> setattr ( self , '<STR_LIT>' , <NUM_LIT:1> ) <EOL> return r <EOL> def __call__ ( self , r ) : <EOL> if self . last_nonce : <EOL> r . headers [ '<STR_LIT>' ] = self . build_digest_header ( r . method , r . url ) <EOL> r . register_hook ( '<STR_LIT>' , self . handle_401 ) <EOL> return r </s>
