<s> import sys <EOL> import random <EOL> from suelta . util import hash , bytes , quote <EOL> from suelta . sasl import Mechanism , register_mechanism <EOL> from suelta . exceptions import SASLError , SASLCancelled <EOL> def parse_challenge ( stuff ) : <EOL> """<STR_LIT:U+0020>""" <EOL> ret = { } <EOL> var = b'<STR_LIT>' <EOL> val = b'<STR_LIT>' <EOL> in_var = True <EOL> in_quotes = False <EOL> new = False <EOL> escaped = False <EOL> for c in stuff : <EOL> if sys . version_info >= ( <NUM_LIT:3> , <NUM_LIT:0> ) : <EOL> c = bytes ( [ c ] ) <EOL> if in_var : <EOL> if c . isspace ( ) : <EOL> continue <EOL> if c == b'<STR_LIT:=>' : <EOL> in_var = False <EOL> new = True <EOL> else : <EOL> var += c <EOL> else : <EOL> if new : <EOL> if c == b'<STR_LIT:">' : <EOL> in_quotes = True <EOL> else : <EOL> val += c <EOL> new = False <EOL> elif in_quotes : <EOL> if escaped : <EOL> escaped = False <EOL> val += c <EOL> else : <EOL> if c == b'<STR_LIT:\\>' : <EOL> escaped = True <EOL> elif c == b'<STR_LIT:">' : <EOL> in_quotes = False <EOL> else : <EOL> val += c <EOL> else : <EOL> if c == b'<STR_LIT:U+002C>' : <EOL> if var : <EOL> ret [ var ] = val <EOL> var = b'<STR_LIT>' <EOL> val = b'<STR_LIT>' <EOL> in_var = True <EOL> else : <EOL> val += c <EOL> if var : <EOL> ret [ var ] = val <EOL> return ret <EOL> class DIGEST_MD5 ( Mechanism ) : <EOL> """<STR_LIT:U+0020>""" <EOL> enc_magic = '<STR_LIT>' <EOL> dec_magic = '<STR_LIT>' <EOL> def __init__ ( self , sasl , name ) : <EOL> """<STR_LIT:U+0020>""" <EOL> super ( DIGEST_MD5 , self ) . __init__ ( sasl , name , <NUM_LIT:3> ) <EOL> self . hash = hash ( name [ <NUM_LIT:7> : ] ) <EOL> if self . hash is None : <EOL> raise SASLCancelled ( self . sasl , self ) <EOL> if not self . sasl . tls_active ( ) : <EOL> if not self . sasl . sec_query ( self , '<STR_LIT>' ) : <EOL> raise SASLCancelled ( self . sasl , self ) <EOL> self . _rspauth_okay = False <EOL> self . _digest_uri = None <EOL> self . _a1 = None <EOL> self . _enc_buf = b'<STR_LIT>' <EOL> self . _enc_key = None <EOL> self . _enc_seq = <NUM_LIT:0> <EOL> self . _max_buffer = <NUM_LIT> <EOL> self . _dec_buf = b'<STR_LIT>' <EOL> self . _dec_key = None <EOL> self . _dec_seq = <NUM_LIT:0> <EOL> self . _qops = [ b'<STR_LIT>' ] <EOL> self . _qop = b'<STR_LIT>' <EOL> def MAC ( self , seq , msg , key ) : <EOL> """<STR_LIT:U+0020>""" <EOL> mac = hmac . HMAC ( key = key , digestmod = self . hash ) <EOL> seqnum = num_to_bytes ( seq ) <EOL> mac . update ( seqnum ) <EOL> mac . update ( msg ) <EOL> return mac . digest ( ) [ : <NUM_LIT:10> ] + b'<STR_LIT>' + seqnum <EOL> def encode ( self , text ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _enc_buf += text <EOL> def flush ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> result = b'<STR_LIT>' <EOL> mbuf = self . _max_buffer - <NUM_LIT:10> - <NUM_LIT:2> - <NUM_LIT:4> <EOL> while self . _enc_buf : <EOL> msg = self . _encbuf [ : mbuf ] <EOL> mac = self . MAC ( self . _enc_seq , msg , self . _enc_key , self . hash ) <EOL> self . _enc_seq += <NUM_LIT:1> <EOL> msg += mac <EOL> result += num_to_bytes ( len ( msg ) ) + msg <EOL> self . _enc_buf = self . _enc_buf [ mbuf : ] <EOL> return result <EOL> def decode ( self , text ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _dec_buf += text <EOL> result = b'<STR_LIT>' <EOL> while len ( self . _dec_buf ) > <NUM_LIT:4> : <EOL> num = bytes_to_num ( self . _dec_buf ) <EOL> if len ( self . _dec_buf ) < ( num + <NUM_LIT:4> ) : <EOL> return result <EOL> mac = self . _dec_buf [ <NUM_LIT:4> : <NUM_LIT:4> + num ] <EOL> self . _dec_buf = self . _dec_buf [ <NUM_LIT:4> + num : ] <EOL> msg = mac [ : - <NUM_LIT:16> ] <EOL> mac_conf = self . MAC ( self . _dec_mac , msg , self . _dec_key ) <EOL> if mac [ - <NUM_LIT:16> : ] != mac_conf : <EOL> self . _desc_sec = None <EOL> return result <EOL> self . _dec_seq += <NUM_LIT:1> <EOL> result += msg <EOL> return result <EOL> def response ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> vitals = [ '<STR_LIT:username>' ] <EOL> if not self . has_values ( [ '<STR_LIT>' ] ) : <EOL> vitals . append ( '<STR_LIT:password>' ) <EOL> self . check_values ( vitals ) <EOL> resp = { } <EOL> if '<STR_LIT>' in self . _qops : <EOL> self . _qop = b'<STR_LIT>' <EOL> resp [ '<STR_LIT>' ] = self . _qop <EOL> if '<STR_LIT>' in self . values : <EOL> resp [ '<STR_LIT>' ] = quote ( self . values [ '<STR_LIT>' ] ) <EOL> resp [ '<STR_LIT:username>' ] = quote ( bytes ( self . values [ '<STR_LIT:username>' ] ) ) <EOL> resp [ '<STR_LIT>' ] = quote ( self . values [ '<STR_LIT>' ] ) <EOL> if self . values [ '<STR_LIT>' ] : <EOL> self . _cnonce = self . values [ '<STR_LIT>' ] <EOL> else : <EOL> self . _cnonce = bytes ( '<STR_LIT:%s>' % random . random ( ) ) [ <NUM_LIT:2> : ] <EOL> resp [ '<STR_LIT>' ] = quote ( self . _cnonce ) <EOL> self . values [ '<STR_LIT>' ] += <NUM_LIT:1> <EOL> resp [ '<STR_LIT>' ] = bytes ( '<STR_LIT>' % self . values [ '<STR_LIT>' ] ) <EOL> service = bytes ( self . sasl . service ) <EOL> host = bytes ( self . sasl . host ) <EOL> self . _digest_uri = service + b'<STR_LIT:/>' + host <EOL> resp [ '<STR_LIT>' ] = quote ( self . _digest_uri ) <EOL> a2 = b'<STR_LIT>' + self . _digest_uri <EOL> if self . _qop != b'<STR_LIT>' : <EOL> a2 += b'<STR_LIT>' <EOL> resp [ '<STR_LIT>' ] = b'<STR_LIT>' <EOL> resp [ '<STR_LIT>' ] = self . gen_hash ( a2 ) <EOL> return b'<STR_LIT:U+002C>' . join ( [ bytes ( k ) + b'<STR_LIT:=>' + bytes ( v ) for k , v in resp . items ( ) ] ) <EOL> def gen_hash ( self , a2 ) : <EOL> """<STR_LIT:U+0020>""" <EOL> if not self . has_values ( [ '<STR_LIT>' ] ) : <EOL> key_hash = self . hash ( ) <EOL> user = bytes ( self . values [ '<STR_LIT:username>' ] ) <EOL> password = bytes ( self . values [ '<STR_LIT:password>' ] ) <EOL> realm = bytes ( self . values [ '<STR_LIT>' ] ) <EOL> kh = user + b'<STR_LIT::>' + realm + b'<STR_LIT::>' + password <EOL> key_hash . update ( kh ) <EOL> self . values [ '<STR_LIT>' ] = key_hash . digest ( ) <EOL> a1 = self . hash ( self . values [ '<STR_LIT>' ] ) <EOL> a1h = b'<STR_LIT::>' + self . values [ '<STR_LIT>' ] + b'<STR_LIT::>' + self . _cnonce <EOL> a1 . update ( a1h ) <EOL> response = self . hash ( ) <EOL> self . _a1 = a1 . digest ( ) <EOL> rv = bytes ( a1 . hexdigest ( ) . lower ( ) ) <EOL> rv += b'<STR_LIT::>' + self . values [ '<STR_LIT>' ] <EOL> rv += b'<STR_LIT::>' + bytes ( '<STR_LIT>' % self . values [ '<STR_LIT>' ] ) <EOL> rv += b'<STR_LIT::>' + self . _cnonce <EOL> rv += b'<STR_LIT::>' + self . _qop <EOL> rv += b'<STR_LIT::>' + bytes ( self . hash ( a2 ) . hexdigest ( ) . lower ( ) ) <EOL> response . update ( rv ) <EOL> return bytes ( response . hexdigest ( ) . lower ( ) ) <EOL> def mutual_auth ( self , cmp_hash ) : <EOL> """<STR_LIT:U+0020>""" <EOL> a2 = b'<STR_LIT::>' + self . _digest_uri <EOL> if self . _qop != b'<STR_LIT>' : <EOL> a2 += b'<STR_LIT>' <EOL> if self . gen_hash ( a2 ) == cmp_hash : <EOL> self . _rspauth_okay = True <EOL> def prep ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> if '<STR_LIT:password>' in self . values : <EOL> del self . values [ '<STR_LIT:password>' ] <EOL> self . values [ '<STR_LIT>' ] = self . _cnonce <EOL> def process ( self , challenge = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> if challenge is None : <EOL> if self . has_values ( [ '<STR_LIT:username>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) : <EOL> self . _qops = self . values [ '<STR_LIT>' ] <EOL> return self . response ( ) <EOL> else : <EOL> return None <EOL> d = parse_challenge ( challenge ) <EOL> if b'<STR_LIT>' in d : <EOL> self . mutual_auth ( d [ b'<STR_LIT>' ] ) <EOL> else : <EOL> if b'<STR_LIT>' not in d : <EOL> d [ b'<STR_LIT>' ] = self . sasl . def_realm <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if bytes ( key ) in d : <EOL> self . values [ key ] = d [ bytes ( key ) ] <EOL> self . values [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> self . _qops = [ b'<STR_LIT>' ] <EOL> if b'<STR_LIT>' in d : <EOL> self . _qops = [ x . strip ( ) for x in d [ b'<STR_LIT>' ] . split ( b'<STR_LIT:U+002C>' ) ] <EOL> self . values [ '<STR_LIT>' ] = self . _qops <EOL> if b'<STR_LIT>' in d : <EOL> self . _max_buffer = int ( d [ b'<STR_LIT>' ] ) <EOL> return self . response ( ) <EOL> def okay ( self ) : <EOL> """<STR_LIT:U+0020>""" <EOL> if self . _rspauth_okay and self . _qop == b'<STR_LIT>' : <EOL> self . _enc_key = self . hash ( self . _a1 + self . enc_magic ) . digest ( ) <EOL> self . _dec_key = self . hash ( self . _a1 + self . dec_magic ) . digest ( ) <EOL> self . encoding = True <EOL> return self . _rspauth_okay <EOL> register_mechanism ( '<STR_LIT>' , <NUM_LIT:30> , DIGEST_MD5 ) </s>
