<s> """<STR_LIT>""" <EOL> __docformat__ = "<STR_LIT>" <EOL> import inspect <EOL> import logging <EOL> import uuid <EOL> import re <EOL> import threading <EOL> from . etree import ElementTree , element_to_unicode <EOL> from . xmppparser import XMLStreamHandler <EOL> from . error import StreamErrorElement <EOL> from . jid import JID <EOL> from . exceptions import StreamError <EOL> from . exceptions import FatalStreamError , StreamParseError <EOL> from . constants import STREAM_QNP , XML_LANG_QNAME , STREAM_ROOT_TAG <EOL> from . settings import XMPPSettings <EOL> from . xmppserializer import serialize <EOL> from . streamevents import StreamConnectedEvent , GotFeaturesEvent <EOL> from . streamevents import AuthenticatedEvent , StreamRestartedEvent <EOL> from . stanzaprocessor import stanza_factory <EOL> from . interfaces import StreamFeatureHandler <EOL> from . interfaces import StreamFeatureHandled , StreamFeatureNotHandled <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> LANG_SPLIT_RE = re . compile ( r"<STR_LIT>" ) <EOL> ERROR_TAG = STREAM_QNP + "<STR_LIT:error>" <EOL> FEATURES_TAG = STREAM_QNP + "<STR_LIT>" <EOL> IP_RE = re . compile ( r"<STR_LIT>" ) <EOL> class StreamBase ( XMLStreamHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stanza_namespace , stanza_route , handlers , <EOL> settings = None ) : <EOL> """<STR_LIT>""" <EOL> XMLStreamHandler . __init__ ( self ) <EOL> self . lock = threading . RLock ( ) <EOL> if settings is None : <EOL> settings = XMPPSettings ( ) <EOL> self . settings = settings <EOL> self . stanza_namespace = stanza_namespace <EOL> self . _stanza_namespace_p = "<STR_LIT>" . format ( stanza_namespace ) <EOL> self . stanza_route = stanza_route <EOL> self . handlers = handlers <EOL> self . _stream_feature_handlers = [ ] <EOL> for handler in handlers : <EOL> if isinstance ( handler , StreamFeatureHandler ) : <EOL> self . _stream_feature_handlers . append ( handler ) <EOL> self . me = None <EOL> self . peer = None <EOL> self . stream_id = None <EOL> self . initiator = None <EOL> self . features = None <EOL> self . authenticated = False <EOL> self . peer_authenticated = False <EOL> self . tls_established = False <EOL> self . auth_method_used = None <EOL> self . version = None <EOL> self . language = None <EOL> self . peer_language = None <EOL> self . transport = None <EOL> self . _input_state = None <EOL> self . _output_state = None <EOL> self . _element_handlers = { } <EOL> def initiate ( self , transport , to = None ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . initiator = True <EOL> self . transport = transport <EOL> transport . set_target ( self ) <EOL> if to : <EOL> self . peer = JID ( to ) <EOL> else : <EOL> self . peer = None <EOL> if transport . is_connected ( ) : <EOL> self . _initiate ( ) <EOL> def _initiate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _setup_stream_element_handlers ( ) <EOL> self . _send_stream_start ( ) <EOL> def receive ( self , transport , myname ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . transport = transport <EOL> transport . set_target ( self ) <EOL> self . me = JID ( myname ) <EOL> self . initiator = False <EOL> self . _setup_stream_element_handlers ( ) <EOL> def _setup_stream_element_handlers ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . initiator : <EOL> mode = "<STR_LIT>" <EOL> else : <EOL> mode = "<STR_LIT>" <EOL> self . _element_handlers = { } <EOL> for handler in self . handlers : <EOL> if not isinstance ( handler , StreamFeatureHandler ) : <EOL> continue <EOL> for _unused , meth in inspect . getmembers ( handler , callable ) : <EOL> if not hasattr ( meth , "<STR_LIT>" ) : <EOL> continue <EOL> element_handled = meth . _pyxmpp_stream_element_handled <EOL> if element_handled in self . _element_handlers : <EOL> continue <EOL> if meth . _pyxmpp_usage_restriction in ( None , mode ) : <EOL> self . _element_handlers [ element_handled ] = meth <EOL> def disconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . transport . disconnect ( ) <EOL> self . _output_state = "<STR_LIT>" <EOL> def event ( self , event ) : <EOL> """<STR_LIT>""" <EOL> event . stream = self <EOL> logger . debug ( "<STR_LIT>" . format ( event ) ) <EOL> self . settings [ "<STR_LIT>" ] . put ( event ) <EOL> return False <EOL> def transport_connected ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> if self . initiator : <EOL> if self . _output_state is None : <EOL> self . _initiate ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . transport . close ( ) <EOL> def stream_start ( self , element ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> logger . debug ( "<STR_LIT>" + element_to_unicode ( element ) ) <EOL> if not element . tag . startswith ( STREAM_QNP ) : <EOL> self . _send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( "<STR_LIT>" ) <EOL> if element . tag != STREAM_ROOT_TAG : <EOL> self . _send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( "<STR_LIT>" ) <EOL> if self . _input_state == "<STR_LIT>" : <EOL> event = StreamRestartedEvent ( self . peer ) <EOL> else : <EOL> event = StreamConnectedEvent ( self . peer ) <EOL> self . _input_state = "<STR_LIT>" <EOL> version = element . get ( "<STR_LIT:version>" ) <EOL> if version : <EOL> try : <EOL> major , minor = version . split ( "<STR_LIT:.>" , <NUM_LIT:1> ) <EOL> major , minor = int ( major ) , int ( minor ) <EOL> except ValueError : <EOL> self . _send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( "<STR_LIT>" ) <EOL> self . version = ( major , minor ) <EOL> else : <EOL> self . version = ( <NUM_LIT:0> , <NUM_LIT:9> ) <EOL> if self . version [ <NUM_LIT:0> ] != <NUM_LIT:1> and self . version != ( <NUM_LIT:0> , <NUM_LIT:9> ) : <EOL> self . _send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( "<STR_LIT>" ) <EOL> peer_lang = element . get ( XML_LANG_QNAME ) <EOL> self . peer_language = peer_lang <EOL> if not self . initiator : <EOL> lang = None <EOL> languages = self . settings [ "<STR_LIT>" ] <EOL> while peer_lang : <EOL> if peer_lang in languages : <EOL> lang = peer_lang <EOL> break <EOL> match = LANG_SPLIT_RE . match ( peer_lang ) <EOL> if not match : <EOL> break <EOL> peer_lang = match . group ( <NUM_LIT:0> ) <EOL> if lang : <EOL> self . language = lang <EOL> if self . initiator : <EOL> self . stream_id = element . get ( "<STR_LIT:id>" ) <EOL> peer = element . get ( "<STR_LIT>" ) <EOL> if peer : <EOL> peer = JID ( peer ) <EOL> if self . peer : <EOL> if peer and peer != self . peer : <EOL> logger . debug ( "<STR_LIT>" <EOL> . format ( peer , self . peer ) ) <EOL> self . peer = peer <EOL> else : <EOL> to = element . get ( "<STR_LIT:to>" ) <EOL> if to : <EOL> to = self . check_to ( to ) <EOL> if not to : <EOL> self . _send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( '<STR_LIT>' ) <EOL> self . me = JID ( to ) <EOL> peer = element . get ( "<STR_LIT>" ) <EOL> if peer : <EOL> peer = JID ( peer ) <EOL> self . _send_stream_start ( self . generate_id ( ) , stream_to = peer ) <EOL> self . _send_stream_features ( ) <EOL> self . event ( event ) <EOL> def stream_end ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" ) <EOL> with self . lock : <EOL> self . _input_state = "<STR_LIT>" <EOL> self . transport . disconnect ( ) <EOL> self . _output_state = "<STR_LIT>" <EOL> def stream_eof ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stream_end ( ) <EOL> def stream_element ( self , element ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . _process_element ( element ) <EOL> def stream_parse_error ( self , descr ) : <EOL> """<STR_LIT>""" <EOL> self . send_stream_error ( "<STR_LIT>" ) <EOL> raise StreamParseError ( descr ) <EOL> def _send_stream_start ( self , stream_id = None , stream_to = None ) : <EOL> """<STR_LIT>""" <EOL> if self . _output_state in ( "<STR_LIT>" , "<STR_LIT>" ) : <EOL> raise StreamError ( "<STR_LIT>" ) <EOL> if not self . language : <EOL> self . language = self . settings [ "<STR_LIT>" ] <EOL> if stream_to : <EOL> stream_to = str ( stream_to ) <EOL> elif self . peer and self . initiator : <EOL> stream_to = str ( self . peer ) <EOL> stream_from = None <EOL> if self . me and ( self . tls_established or not self . initiator ) : <EOL> stream_from = str ( self . me ) <EOL> if stream_id : <EOL> self . stream_id = stream_id <EOL> else : <EOL> self . stream_id = None <EOL> self . transport . send_stream_head ( self . stanza_namespace , <EOL> stream_from , stream_to , <EOL> self . stream_id , language = self . language ) <EOL> self . _output_state = "<STR_LIT>" <EOL> def send_stream_error ( self , condition ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . _send_stream_error ( condition ) <EOL> def _send_stream_error ( self , condition ) : <EOL> """<STR_LIT>""" <EOL> if self . _output_state is "<STR_LIT>" : <EOL> return <EOL> if self . _output_state in ( None , "<STR_LIT>" ) : <EOL> self . _send_stream_start ( ) <EOL> element = StreamErrorElement ( condition ) . as_xml ( ) <EOL> self . transport . send_element ( element ) <EOL> self . transport . disconnect ( ) <EOL> self . _output_state = "<STR_LIT>" <EOL> def _restart_stream ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _input_state = "<STR_LIT>" <EOL> self . _output_state = "<STR_LIT>" <EOL> self . features = None <EOL> self . transport . restart ( ) <EOL> if self . initiator : <EOL> self . _send_stream_start ( self . stream_id ) <EOL> def _make_stream_features ( self ) : <EOL> """<STR_LIT>""" <EOL> features = ElementTree . Element ( FEATURES_TAG ) <EOL> for handler in self . _stream_feature_handlers : <EOL> handler . make_stream_features ( self , features ) <EOL> return features <EOL> def _send_stream_features ( self ) : <EOL> """<STR_LIT>""" <EOL> self . features = self . _make_stream_features ( ) <EOL> self . _write_element ( self . features ) <EOL> def write_element ( self , element ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . _write_element ( element ) <EOL> def _write_element ( self , element ) : <EOL> """<STR_LIT>""" <EOL> self . transport . send_element ( element ) <EOL> def send ( self , stanza ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> return self . _send ( stanza ) <EOL> def _send ( self , stanza ) : <EOL> """<STR_LIT>""" <EOL> self . fix_out_stanza ( stanza ) <EOL> element = stanza . as_xml ( ) <EOL> self . _write_element ( element ) <EOL> def _process_element ( self , element ) : <EOL> """<STR_LIT>""" <EOL> tag = element . tag <EOL> if tag in self . _element_handlers : <EOL> handler = self . _element_handlers [ tag ] <EOL> logger . debug ( "<STR_LIT>" <EOL> . format ( element , handler ) ) <EOL> handled = handler ( self , element ) <EOL> if handled : <EOL> return <EOL> if tag . startswith ( self . _stanza_namespace_p ) : <EOL> stanza = stanza_factory ( element , self , self . language ) <EOL> self . uplink_receive ( stanza ) <EOL> elif tag == ERROR_TAG : <EOL> error = StreamErrorElement ( element ) <EOL> self . process_stream_error ( error ) <EOL> elif tag == FEATURES_TAG : <EOL> logger . debug ( "<STR_LIT>" . format ( serialize ( element ) ) ) <EOL> self . _got_features ( element ) <EOL> else : <EOL> logger . debug ( "<STR_LIT>" . format ( serialize ( element ) ) ) <EOL> logger . debug ( "<STR_LIT>" . format ( <EOL> self . _element_handlers ) ) <EOL> def uplink_receive ( self , stanza ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> if self . stanza_route : <EOL> self . stanza_route . uplink_receive ( stanza ) <EOL> else : <EOL> logger . debug ( "<STR_LIT>" . format ( stanza ) ) <EOL> def process_stream_error ( self , error ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" <EOL> . format ( error . condition_name , error . serialize ( ) ) ) <EOL> def check_to ( self , to ) : <EOL> """<STR_LIT>""" <EOL> if to != self . me : <EOL> return None <EOL> return to <EOL> def generate_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return str ( uuid . uuid4 ( ) ) <EOL> def _got_features ( self , features ) : <EOL> """<STR_LIT>""" <EOL> self . features = features <EOL> logger . debug ( "<STR_LIT>" ) <EOL> handled = self . event ( GotFeaturesEvent ( self . features ) ) <EOL> logger . debug ( "<STR_LIT>" . format ( handled ) ) <EOL> if not handled : <EOL> mandatory_handled = [ ] <EOL> mandatory_not_handled = [ ] <EOL> logger . debug ( "<STR_LIT>" <EOL> . format ( self . _stream_feature_handlers ) ) <EOL> for handler in self . _stream_feature_handlers : <EOL> ret = handler . handle_stream_features ( self , self . features ) <EOL> if ret is None : <EOL> continue <EOL> elif isinstance ( ret , StreamFeatureHandled ) : <EOL> if ret . mandatory : <EOL> mandatory_handled . append ( str ( ret ) ) <EOL> break <EOL> break <EOL> elif isinstance ( ret , StreamFeatureNotHandled ) : <EOL> if ret . mandatory : <EOL> mandatory_not_handled . append ( str ( ret ) ) <EOL> break <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( ret ) ) <EOL> if mandatory_not_handled and not mandatory_handled : <EOL> self . send_stream_error ( "<STR_LIT>" ) <EOL> raise FatalStreamError ( <EOL> "<STR_LIT>" <EOL> + "<STR_LIT:U+0020>" . join ( mandatory_not_handled ) ) <EOL> def is_connected ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . transport . is_connected ( ) and self . _output_state == "<STR_LIT>" <EOL> def set_peer_authenticated ( self , peer , restart_stream = False ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . peer_authenticated = True <EOL> self . peer = peer <EOL> if restart_stream : <EOL> self . _restart_stream ( ) <EOL> self . event ( AuthenticatedEvent ( self . peer ) ) <EOL> def set_authenticated ( self , me , restart_stream = False ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . authenticated = True <EOL> self . me = me <EOL> if restart_stream : <EOL> self . _restart_stream ( ) <EOL> self . event ( AuthenticatedEvent ( self . me ) ) <EOL> def fix_in_stanza ( self , stanza ) : <EOL> """<STR_LIT>""" <EOL> return stanza <EOL> def fix_out_stanza ( self , stanza ) : <EOL> """<STR_LIT>""" <EOL> return stanza <EOL> def _languages_factory ( settings ) : <EOL> """<STR_LIT>""" <EOL> return [ settings [ "<STR_LIT>" ] ] <EOL> XMPPSettings . add_setting ( "<STR_LIT>" , type = str , default = "<STR_LIT>" , <EOL> cmdline_help = "<STR_LIT>" , <EOL> doc = """<STR_LIT>""" <EOL> ) <EOL> XMPPSettings . add_setting ( "<STR_LIT>" , type = "<STR_LIT>" , <EOL> validator = XMPPSettings . validate_string_list , <EOL> factory = _languages_factory , <EOL> cmdline_help = "<STR_LIT>" , <EOL> doc = """<STR_LIT>""" <EOL> ) <EOL> XMPPSettings . add_setting ( "<STR_LIT>" , type = "<STR_LIT>" , <EOL> default = { } , <EOL> doc = """<STR_LIT>""" <EOL> ) </s>
