# slixmpp.clientxmpp# ~~~~~~~~~~~~~~~~~~~~# This module provides XMPP functionality that# is specific to external server component connections.# Part of Slixmpp: The Slick XMPP Library# :copyright: (c) 2011 Nathanael C. Fritz# :license: MIT, see LICENSE for more detailsimportloggingimporthashlibfromtypingimportOptionalfromslixmppimportMessage,Iq,Presencefromslixmpp.basexmppimportBaseXMPPfromslixmpp.stanzaimportHandshakefromslixmpp.stanza.errorimportErrorfromslixmpp.xmlstreamimportXMLStreamfromslixmpp.xmlstream.matcherimportMatchXPathfromslixmpp.xmlstream.handlerimportCallbackfromslixmpp.xmlstream.stanzabaseimportregister_stanza_pluginlog=logging.getLogger(__name__)
[docs]classComponentXMPP(BaseXMPP):""" Slixmpp's basic XMPP server component. Use only for good, not for evil. :param jid: The JID of the component. :param secret: The secret or password for the component. :param host: The server accepting the component. :param port: The port used to connect to the server. :param plugin_config: A dictionary of plugin configurations. :param plugin_whitelist: A list of approved plugins that will be loaded when calling :meth:`~slixmpp.basexmpp.BaseXMPP.register_plugins()`. :param use_jc_ns: Indicates if the ``'jabber:client'`` namespace should be used instead of the standard ``'jabber:component:accept'`` namespace. Defaults to ``False``. :param fix_error_ns: Fix the namespace of error stanzas. If you use ``use_jc_ns`` namespace, you probably want that, but it can be a problem if you use both a ClientXMPP and a ComponentXMPP in the same interpreter. This is ``False`` by default for backwards compatibility. """def__init__(self,jid,secret,host=None,port=None,plugin_config=None,plugin_whitelist=None,use_jc_ns=False,fix_error_ns=False):ifnotplugin_whitelist:plugin_whitelist=[]ifnotplugin_config:plugin_config={}ifuse_jc_ns:default_ns='jabber:client'else:default_ns='jabber:component:accept'BaseXMPP.__init__(self,jid,default_ns)iffix_error_ns:self._fix_error_ns()self.auto_authorize=Noneself.stream_header='<stream:stream %s%s to="%s">'%('xmlns="jabber:component:accept"','xmlns:stream="%s"'%self.stream_ns,jid)self.stream_footer="</stream:stream>"self.server_host=hostself.server_port=portself.secret=secretself.plugin_config=plugin_configself.plugin_whitelist=plugin_whitelistself.is_component=Trueself.sessionstarted=Falseself.register_handler(Callback('Handshake',MatchXPath('{jabber:component:accept}handshake'),self._handle_handshake))self.add_event_handler('presence_probe',self._handle_probe)def_fix_error_ns(self):Error.namespace=self.default_nsforstinMessage,Iq,Presence:register_stanza_plugin(st,Error)
[docs]defconnect(self,host:Optional[str]=None,port:int=0,use_ssl:Optional[bool]=None,force_starttls:Optional[bool]=None,disable_starttls:Optional[bool]=None)->None:"""Connect to the server. :param host: The name of the desired server for the connection. Defaults to :attr:`server_host`. :param port: Port to connect to on the server. Defauts to :attr:`server_port`. :param use_ssl: Flag indicating if SSL should be used by connecting directly to a port using SSL. :param force_starttls: UNUSED :param disable_starttls: UNUSED """ifhostisnotNone:self.server_host=hostifport:self.server_port=portself.server_name=self.boundjid.hostlog.debug("Connecting to %s:%s",host,port)XMLStream.connect(self,host=self.server_host,port=self.server_port,use_ssl=use_ssl)
[docs]defincoming_filter(self,xml):""" Pre-process incoming XML stanzas by converting any ``'jabber:client'`` namespaced elements to the component's default namespace. :param xml: The XML stanza to pre-process. """ifxml.tag.startswith('{jabber:client}'):xml.tag=xml.tag.replace('jabber:client',self.default_ns)returnxml
[docs]defstart_stream_handler(self,xml):""" Once the streams are established, attempt to handshake with the server to be accepted as a component. :param xml: The incoming stream's root element. """BaseXMPP.start_stream_handler(self,xml)# Construct a hash of the stream ID and the component secret.sid=xml.get('id','')pre_hash=bytes('%s%s'%(sid,self.secret),'utf-8')handshake=Handshake()handshake['value']=hashlib.sha1(pre_hash).hexdigest().lower()self.send(handshake)
def_handle_handshake(self,xml):"""The handshake has been accepted. :param xml: The reply handshake stanza. """self.session_bind_event.set()self.sessionstarted=Trueself.event('session_bind',self.boundjid)self.event('session_start')def_handle_probe(self,pres):self.roster[pres['to']][pres['from']].handle_probe(pres)