<s> import base64 <EOL> import hashlib <EOL> import hmac <EOL> import struct <EOL> import sys <EOL> from puresasl import SASLError , SASLProtocolException , QOP <EOL> try : <EOL> import kerberos <EOL> _have_kerberos = True <EOL> except ImportError : <EOL> _have_kerberos = False <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> <EOL> if PY3 : <EOL> def _b ( s ) : <EOL> return s . encode ( "<STR_LIT:utf-8>" ) <EOL> else : <EOL> def _b ( s ) : <EOL> return s <EOL> class Mechanism ( object ) : <EOL> """<STR_LIT>""" <EOL> name = None <EOL> """<STR_LIT>""" <EOL> score = <NUM_LIT:0> <EOL> """<STR_LIT>""" <EOL> complete = False <EOL> """<STR_LIT>""" <EOL> has_initial_response = False <EOL> allows_anonymous = True <EOL> """<STR_LIT>""" <EOL> uses_plaintext = True <EOL> """<STR_LIT>""" <EOL> active_safe = False <EOL> """<STR_LIT>""" <EOL> dictionary_safe = False <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sasl , ** props ) : <EOL> self . sasl = sasl <EOL> def process ( self , challenge = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def wrap ( self , outgoing ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def unwrap ( self , incoming ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def dispose ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def _fetch_properties ( self , * properties ) : <EOL> """<STR_LIT>""" <EOL> needed = [ p for p in properties if getattr ( self , p , None ) is None ] <EOL> if needed and not self . sasl . callback : <EOL> raise SASLError ( '<STR_LIT>' <EOL> '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( needed ) ) <EOL> for prop in needed : <EOL> setattr ( self , prop , self . sasl . callback ( prop ) ) <EOL> def _pick_qop ( self , server_qop_set ) : <EOL> """<STR_LIT>""" <EOL> configured_qops = set ( _b ( qop ) if isinstance ( qop , str ) else qop for qop in self . sasl . qops ) <EOL> available_qops = configured_qops & server_qop_set <EOL> if not available_qops : <EOL> configured = b'<STR_LIT:U+002CU+0020>' . join ( configured_qops ) . decode ( '<STR_LIT:ascii>' ) <EOL> offered = b'<STR_LIT:U+002CU+0020>' . join ( server_qop_set ) . decode ( '<STR_LIT:ascii>' ) <EOL> raise SASLProtocolException ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( configured , offered ) ) <EOL> else : <EOL> self . qops = available_qops <EOL> for qop in ( QOP . AUTH_CONF , QOP . AUTH_INT , QOP . AUTH ) : <EOL> if qop in self . qops : <EOL> self . qop = qop <EOL> break <EOL> class AnonymousMechanism ( Mechanism ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> score = <NUM_LIT:0> <EOL> uses_plaintext = False <EOL> def process ( self , challenge = None ) : <EOL> self . complete = True <EOL> return b'<STR_LIT>' <EOL> class PlainMechanism ( Mechanism ) : <EOL> """<STR_LIT>""" <EOL> name = '<STR_LIT>' <EOL> score = <NUM_LIT:1> <EOL> allows_anonymous = False <EOL> def wrap ( self , outgoing ) : <EOL> return outgoing <EOL> def unwrap ( self , incoming ) : <EOL> return incoming <EOL> def __init__ ( self , sasl , username = None , password = None , identity = '<STR_LIT>' , ** props ) : <EOL> Mechanism . __init__ ( self , sasl ) <EOL> self . identity = identity <EOL> self . username = username <EOL> self . password = password <EOL> def process ( self , challenge = None ) : <EOL> self . _fetch_properties ( '<STR_LIT:username>' , '<STR_LIT:password>' ) <EOL> self . complete = True <EOL> return b'<STR_LIT>' . join ( ( _b ( self . identity ) , b'<STR_LIT:\x00>' , _b ( self . username ) , b'<STR_LIT:\x00>' , _b ( self . password ) ) ) <EOL> def dispose ( self ) : <EOL> self . password = None <EOL> class CramMD5Mechanism ( PlainMechanism ) : <EOL> name = "<STR_LIT>" <EOL> score = <NUM_LIT:20> <EOL> allows_anonymous = False <EOL> uses_plaintext = False <EOL> def __init__ ( self , sasl , username = None , password = None , ** props ) : <EOL> Mechanism . __init__ ( self , sasl ) <EOL> self . username = username <EOL> self . password = password <EOL> def process ( self , challenge = None ) : <EOL> if challenge is None : <EOL> return None <EOL> self . _fetch_properties ( '<STR_LIT:username>' , '<STR_LIT:password>' ) <EOL> mac = hmac . HMAC ( key = _b ( self . password ) , digestmod = hashlib . md5 ) <EOL> mac . update ( challenge ) <EOL> return b'<STR_LIT>' . join ( ( _b ( self . username ) , b'<STR_LIT:U+0020>' , _b ( mac . hexdigest ( ) ) ) ) <EOL> def dispose ( self ) : <EOL> self . password = None <EOL> class DigestMD5Mechanism ( Mechanism ) : <EOL> name = "<STR_LIT>" <EOL> score = <NUM_LIT:30> <EOL> allows_anonymous = False <EOL> uses_plaintext = False <EOL> enc_magic = '<STR_LIT>' <EOL> dec_magic = '<STR_LIT>' <EOL> def __init__ ( self , sasl , username = None , password = None , ** props ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> class GSSAPIMechanism ( Mechanism ) : <EOL> name = '<STR_LIT>' <EOL> score = <NUM_LIT:100> <EOL> allows_anonymous = False <EOL> uses_plaintext = False <EOL> active_safe = True <EOL> def __init__ ( self , sasl , principal = None , ** props ) : <EOL> Mechanism . __init__ ( self , sasl ) <EOL> self . user = None <EOL> self . _have_negotiated_details = False <EOL> self . host = self . sasl . host <EOL> self . service = self . sasl . service <EOL> self . principal = principal <EOL> self . _fetch_properties ( '<STR_LIT:host>' , '<STR_LIT>' ) <EOL> krb_service = '<STR_LIT:@>' . join ( ( self . service , self . host ) ) <EOL> try : <EOL> _ , self . context = kerberos . authGSSClientInit ( service = krb_service , <EOL> principal = self . principal ) <EOL> except TypeError : <EOL> if self . principal is not None : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> _ , self . context = kerberos . authGSSClientInit ( service = krb_service ) <EOL> def process ( self , challenge = None ) : <EOL> if not self . _have_negotiated_details : <EOL> kerberos . authGSSClientStep ( self . context , '<STR_LIT>' ) <EOL> _negotiated_details = kerberos . authGSSClientResponse ( self . context ) <EOL> self . _have_negotiated_details = True <EOL> return base64 . b64decode ( _negotiated_details ) <EOL> challenge = base64 . b64encode ( challenge ) . decode ( '<STR_LIT:ascii>' ) <EOL> if self . user is None : <EOL> ret = kerberos . authGSSClientStep ( self . context , challenge ) <EOL> if ret == kerberos . AUTH_GSS_COMPLETE : <EOL> self . user = kerberos . authGSSClientUserName ( self . context ) <EOL> return b'<STR_LIT>' <EOL> else : <EOL> response = kerberos . authGSSClientResponse ( self . context ) <EOL> if response : <EOL> response = base64 . b64decode ( response ) <EOL> else : <EOL> response = b'<STR_LIT>' <EOL> return response <EOL> kerberos . authGSSClientUnwrap ( self . context , challenge ) <EOL> data = kerberos . authGSSClientResponse ( self . context ) <EOL> plaintext_data = base64 . b64decode ( data ) <EOL> if len ( plaintext_data ) != <NUM_LIT:4> : <EOL> raise SASLProtocolException ( "<STR_LIT>" ) <EOL> word , = struct . unpack ( '<STR_LIT>' , plaintext_data ) <EOL> qop_bits = word >> <NUM_LIT> <EOL> max_length = word & <NUM_LIT> <EOL> server_offered_qops = QOP . names_from_bitmask ( qop_bits ) <EOL> self . _pick_qop ( server_offered_qops ) <EOL> self . max_buffer = min ( self . sasl . max_buffer , max_length ) <EOL> """<STR_LIT>""" <EOL> l = len ( self . user ) <EOL> fmt = '<STR_LIT>' + str ( l ) + '<STR_LIT:s>' <EOL> word = QOP . flag_from_name ( self . qop ) << <NUM_LIT> | self . max_buffer <EOL> out = struct . pack ( fmt , word , _b ( self . user ) , ) <EOL> encoded = base64 . b64encode ( out ) . decode ( '<STR_LIT:ascii>' ) <EOL> kerberos . authGSSClientWrap ( self . context , encoded ) <EOL> response = kerberos . authGSSClientResponse ( self . context ) <EOL> self . complete = True <EOL> return base64 . b64decode ( response ) <EOL> def wrap ( self , outgoing ) : <EOL> if self . qop != '<STR_LIT>' : <EOL> outgoing = base64 . b64encode ( outgoing ) <EOL> if self . qop == '<STR_LIT>' : <EOL> protect = <NUM_LIT:1> <EOL> else : <EOL> protect = <NUM_LIT:0> <EOL> kerberos . authGSSClientWrap ( self . context , outgoing , None , protect ) <EOL> return base64 . b64decode ( kerberos . authGSSClientResponse ( self . context ) ) <EOL> else : <EOL> return outgoing <EOL> def unwrap ( self , incoming ) : <EOL> if self . qop != '<STR_LIT>' : <EOL> incoming = base64 . b64encode ( incoming ) <EOL> kerberos . authGSSClientUnwrap ( self . context , incoming ) <EOL> conf = kerberos . authGSSClientResponseConf ( self . context ) <EOL> if <NUM_LIT:0> == conf and self . qop == '<STR_LIT>' : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> return base64 . b64decode ( kerberos . authGSSClientResponse ( self . context ) ) <EOL> else : <EOL> return incoming <EOL> def dispose ( self ) : <EOL> kerberos . authGSSClientClean ( self . context ) <EOL> mechanisms = dict ( ( m . name , m ) for m in ( <EOL> AnonymousMechanism , <EOL> PlainMechanism , <EOL> CramMD5Mechanism , <EOL> DigestMD5Mechanism ) ) <EOL> if _have_kerberos : <EOL> mechanisms [ GSSAPIMechanism . name ] = GSSAPIMechanism </s>
