<s> """<STR_LIT>""" <EOL> from base64 import b64decode , b64encode <EOL> import re <EOL> from twisted . internet import defer <EOL> from twisted . words . protocols . jabber import sasl_mechanisms , xmlstream <EOL> from twisted . words . xish import domish <EOL> NS_XMPP_SASL = '<STR_LIT>' <EOL> def get_mechanisms ( xs ) : <EOL> """<STR_LIT>""" <EOL> mechanisms = [ ] <EOL> for element in xs . features [ ( NS_XMPP_SASL , '<STR_LIT>' ) ] . elements ( ) : <EOL> if element . name == '<STR_LIT>' : <EOL> mechanisms . append ( str ( element ) ) <EOL> return mechanisms <EOL> class SASLError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class SASLNoAcceptableMechanism ( SASLError ) : <EOL> """<STR_LIT>""" <EOL> class SASLAuthError ( SASLError ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , condition = None ) : <EOL> self . condition = condition <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % self . condition <EOL> class SASLIncorrectEncodingError ( SASLError ) : <EOL> """<STR_LIT>""" <EOL> base64Pattern = re . compile ( "<STR_LIT>" ) <EOL> def fromBase64 ( s ) : <EOL> """<STR_LIT>""" <EOL> if base64Pattern . match ( s ) is None : <EOL> raise SASLIncorrectEncodingError ( ) <EOL> try : <EOL> return b64decode ( s ) <EOL> except Exception , e : <EOL> raise SASLIncorrectEncodingError ( str ( e ) ) <EOL> class SASLInitiatingInitializer ( xmlstream . BaseFeatureInitiatingInitializer ) : <EOL> """<STR_LIT>""" <EOL> feature = ( NS_XMPP_SASL , '<STR_LIT>' ) <EOL> _deferred = None <EOL> def setMechanism ( self ) : <EOL> """<STR_LIT>""" <EOL> jid = self . xmlstream . authenticator . jid <EOL> password = self . xmlstream . authenticator . password <EOL> mechanisms = get_mechanisms ( self . xmlstream ) <EOL> if jid . user is not None : <EOL> if '<STR_LIT>' in mechanisms : <EOL> self . mechanism = sasl_mechanisms . DigestMD5 ( '<STR_LIT>' , jid . host , None , <EOL> jid . user , password ) <EOL> elif '<STR_LIT>' in mechanisms : <EOL> self . mechanism = sasl_mechanisms . Plain ( None , jid . user , password ) <EOL> else : <EOL> raise SASLNoAcceptableMechanism ( ) <EOL> else : <EOL> if '<STR_LIT>' in mechanisms : <EOL> self . mechanism = sasl_mechanisms . Anonymous ( ) <EOL> else : <EOL> raise SASLNoAcceptableMechanism ( ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . setMechanism ( ) <EOL> self . _deferred = defer . Deferred ( ) <EOL> self . xmlstream . addObserver ( '<STR_LIT>' , self . onChallenge ) <EOL> self . xmlstream . addOnetimeObserver ( '<STR_LIT>' , self . onSuccess ) <EOL> self . xmlstream . addOnetimeObserver ( '<STR_LIT>' , self . onFailure ) <EOL> self . sendAuth ( self . mechanism . getInitialResponse ( ) ) <EOL> return self . _deferred <EOL> def sendAuth ( self , data = None ) : <EOL> """<STR_LIT>""" <EOL> auth = domish . Element ( ( NS_XMPP_SASL , '<STR_LIT>' ) ) <EOL> auth [ '<STR_LIT>' ] = self . mechanism . name <EOL> if data is not None : <EOL> auth . addContent ( b64encode ( data ) or '<STR_LIT:=>' ) <EOL> self . xmlstream . send ( auth ) <EOL> def sendResponse ( self , data = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> response = domish . Element ( ( NS_XMPP_SASL , '<STR_LIT>' ) ) <EOL> if data : <EOL> response . addContent ( b64encode ( data ) ) <EOL> self . xmlstream . send ( response ) <EOL> def onChallenge ( self , element ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> challenge = fromBase64 ( str ( element ) ) <EOL> except SASLIncorrectEncodingError : <EOL> self . _deferred . errback ( ) <EOL> else : <EOL> self . sendResponse ( self . mechanism . getResponse ( challenge ) ) <EOL> def onSuccess ( self , success ) : <EOL> """<STR_LIT>""" <EOL> self . xmlstream . removeObserver ( '<STR_LIT>' , self . onChallenge ) <EOL> self . xmlstream . removeObserver ( '<STR_LIT>' , self . onFailure ) <EOL> self . xmlstream . reset ( ) <EOL> self . xmlstream . sendHeader ( ) <EOL> self . _deferred . callback ( xmlstream . Reset ) <EOL> def onFailure ( self , failure ) : <EOL> """<STR_LIT>""" <EOL> self . xmlstream . removeObserver ( '<STR_LIT>' , self . onChallenge ) <EOL> self . xmlstream . removeObserver ( '<STR_LIT>' , self . onSuccess ) <EOL> try : <EOL> condition = failure . firstChildElement ( ) . name <EOL> except AttributeError : <EOL> condition = None <EOL> self . _deferred . errback ( SASLAuthError ( condition ) ) </s>
