diff options
Diffstat (limited to 'sleekxmpp/plugins')
| -rw-r--r-- | sleekxmpp/plugins/__init__.py | 1 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0030/disco.py | 10 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0065/__init__.py | 5 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0065/proxy.py | 359 | ||||
| -rw-r--r-- | sleekxmpp/plugins/xep_0065/stanza.py | 41 |
5 files changed, 414 insertions, 2 deletions
diff --git a/sleekxmpp/plugins/__init__.py b/sleekxmpp/plugins/__init__.py index 59d716eb..c6617a87 100644 --- a/sleekxmpp/plugins/__init__.py +++ b/sleekxmpp/plugins/__init__.py @@ -30,6 +30,7 @@ __all__ = [ 'xep_0054', # vcard-temp 'xep_0059', # Result Set Management 'xep_0060', # Pubsub (Client) + 'xep_0065', # SOCKS5 Bytestreams 'xep_0066', # Out of Band Data 'xep_0077', # In-Band Registration # 'xep_0078', # Non-SASL auth. Don't automatically load diff --git a/sleekxmpp/plugins/xep_0030/disco.py b/sleekxmpp/plugins/xep_0030/disco.py index 278b4a34..8a397923 100644 --- a/sleekxmpp/plugins/xep_0030/disco.py +++ b/sleekxmpp/plugins/xep_0030/disco.py @@ -324,6 +324,8 @@ class XEP_0030(BasePlugin): callback -- Optional callback to execute when a reply is received instead of blocking and waiting for the reply. + timeout_callback -- Optional callback to execute when no result + has been received in timeout seconds. """ if local is None: if jid is not None and not isinstance(jid, JID): @@ -364,7 +366,8 @@ class XEP_0030(BasePlugin): iq['disco_info']['node'] = node if node else '' return iq.send(timeout=kwargs.get('timeout', None), block=kwargs.get('block', True), - callback=kwargs.get('callback', None)) + callback=kwargs.get('callback', None), + timeout_callback=kwargs.get('timeout_callback', None)) def set_info(self, jid=None, node=None, info=None): """ @@ -405,6 +408,8 @@ class XEP_0030(BasePlugin): iterator -- If True, return a result set iterator using the XEP-0059 plugin, if the plugin is loaded. Otherwise the parameter is ignored. + timeout_callback -- Optional callback to execute when no result + has been received in timeout seconds. """ if local or local is None and jid is None: items = self.api['get_items'](jid, node, @@ -423,7 +428,8 @@ class XEP_0030(BasePlugin): else: return iq.send(timeout=kwargs.get('timeout', None), block=kwargs.get('block', True), - callback=kwargs.get('callback', None)) + callback=kwargs.get('callback', None), + timeout_callback=kwargs.get('timeout_callback', None)) def set_items(self, jid=None, node=None, **kwargs): """ diff --git a/sleekxmpp/plugins/xep_0065/__init__.py b/sleekxmpp/plugins/xep_0065/__init__.py new file mode 100644 index 00000000..c577d859 --- /dev/null +++ b/sleekxmpp/plugins/xep_0065/__init__.py @@ -0,0 +1,5 @@ +from sleekxmpp.plugins.base import register_plugin +from sleekxmpp.plugins.xep_0065.proxy import XEP_0065 + + +register_plugin(XEP_0065) diff --git a/sleekxmpp/plugins/xep_0065/proxy.py b/sleekxmpp/plugins/xep_0065/proxy.py new file mode 100644 index 00000000..b027e4e0 --- /dev/null +++ b/sleekxmpp/plugins/xep_0065/proxy.py @@ -0,0 +1,359 @@ +import sys +import logging +import struct + +from threading import Thread, Event +from hashlib import sha1 +from select import select +from uuid import uuid4 + +from sleekxmpp.plugins.xep_0065 import stanza + +from sleekxmpp.plugins.base import base_plugin +from sleekxmpp.xmlstream.handler import Callback +from sleekxmpp.xmlstream.matcher import StanzaPath +from sleekxmpp.thirdparty.socks import socksocket, PROXY_TYPE_SOCKS5 + +# Registers the sleekxmpp logger +log = logging.getLogger(__name__) + + +class XEP_0065(base_plugin): + """ + XEP-0065 Socks5 Bytestreams + """ + + description = "Socks5 Bytestreams" + dependencies = set(['xep_0030', ]) + xep = '0065' + name = 'xep_0065' + + # A dict contains for each SID, the proxy thread currently + # running. + proxy_threads = {} + + def plugin_init(self): + """ Initializes the xep_0065 plugin and all event callbacks. + """ + + # Shortcuts to access to the xep_0030 plugin. + self.disco = self.xmpp['xep_0030'] + + # Handler for the streamhost stanza. + self.xmpp.registerHandler( + Callback('Socks5 Bytestreams', + StanzaPath('iq@type=set/socks/streamhost'), + self._handle_streamhost)) + + # Handler for the streamhost-used stanza. + self.xmpp.registerHandler( + Callback('Socks5 Bytestreams', + StanzaPath('iq@type=result/socks/streamhost-used'), + self._handle_streamhost_used)) + + def get_socket(self, sid): + """ Returns the socket associated to the SID. + """ + + proxy = self.proxy_threads.get(sid) + if proxy: + return proxy.s + + def handshake(self, to, streamer=None): + """ Starts the handshake to establish the socks5 bytestreams + connection. + """ + + # Discovers the proxy. + self.streamer = streamer or self.discover_proxy() + + # Requester requests network address from the proxy. + streamhost = self.get_network_address(self.streamer) + self.proxy_host = streamhost['socks']['streamhost']['host'] + self.proxy_port = streamhost['socks']['streamhost']['port'] + + # Generates the SID for this new handshake. + sid = uuid4().hex + + # Requester initiates S5B negotation with Target by sending + # IQ-set that includes the JabberID and network address of + # StreamHost as well as the StreamID (SID) of the proposed + # bytestream. + iq = self.xmpp.Iq(sto=to, stype='set') + iq['socks']['sid'] = sid + iq['socks']['streamhost']['jid'] = self.streamer + iq['socks']['streamhost']['host'] = self.proxy_host + iq['socks']['streamhost']['port'] = self.proxy_port + + # Sends the new IQ. + return iq.send() + + def discover_proxy(self): + """ Auto-discovers (using XEP 0030) the available bytestream + proxy on the XMPP server. + + Returns the JID of the proxy. + """ + + # Gets all disco items. + disco_items = self.disco.get_items(self.xmpp.server) + + for item in disco_items['disco_items']['items']: + # For each items, gets the disco info. + disco_info = self.disco.get_info(item[0]) + + # Gets and verifies if the identity is a bytestream proxy. + identities = disco_info['disco_info']['identities'] + for identity in identities: + if identity[0] == 'proxy' and identity[1] == 'bytestreams': + # Returns when the first occurence is found. + return '%s' % disco_info['from'] + + def get_network_address(self, streamer): + """ Gets the streamhost information of the proxy. + + streamer : The jid of the proxy. + """ + + iq = self.xmpp.Iq(sto=streamer, stype='get') + iq['socks'] # Adds the query eleme to the iq. + + return iq.send() + + def _handle_streamhost(self, iq): + """ Handles all streamhost stanzas. + """ + + # Registers the streamhost info. + self.streamer = iq['socks']['streamhost']['jid'] + self.proxy_host = iq['socks']['streamhost']['host'] + self.proxy_port = iq['socks']['streamhost']['port'] + + # Sets the SID, the requester and the target. + sid = iq['socks']['sid'] + requester = '%s' % iq['from'] + target = '%s' % self.xmpp.boundjid + + # Next the Target attempts to open a standard TCP socket on + # the network address of the Proxy. + self.proxy_thread = Proxy(sid, requester, target, self.proxy_host, + self.proxy_port, self.on_recv) + self.proxy_thread.start() + + # Registers the new thread in the proxy_thread dict. + self.proxy_threads[sid] = self.proxy_thread + + # Wait until the proxy is connected + self.proxy_thread.connected.wait() + + # Replies to the incoming iq with a streamhost-used stanza. + res_iq = iq.reply() + res_iq['socks']['sid'] = sid + res_iq['socks']['streamhost-used']['jid'] = self.streamer + + # Sends the IQ + return res_iq.send() + + def _handle_streamhost_used(self, iq): + """ Handles all streamhost-used stanzas. + """ + + # Sets the SID, the requester and the target. + sid = iq['socks']['sid'] + requester = '%s' % self.xmpp.boundjid + target = '%s' % iq['from'] + + # The Requester will establish a connection to the SOCKS5 + # proxy in the same way the Target did. + self.proxy_thread = Proxy(sid, requester, target, self.proxy_host, + self.proxy_port, self.on_recv) + self.proxy_thread.start() + + # Registers the new thread in the proxy_thread dict. + self.proxy_threads[sid] = self.proxy_thread + + # Wait until the proxy is connected + self.proxy_thread.connected.wait() + + # Requester sends IQ-set to StreamHost requesting that + # StreamHost activate the bytestream associated with the + # StreamID. + self.activate(iq['socks']['sid'], target) + + def activate(self, sid, to): + """ IQ-set to StreamHost requesting that StreamHost activate + the bytestream associated with the StreamID. + """ + + # Creates the activate IQ. + act_iq = self.xmpp.Iq(sto=self.streamer, stype='set') + act_iq['socks']['sid'] = sid + act_iq['socks']['activate'] = to + + # Send the IQ. + act_iq.send() + + def deactivate(self, sid): + """ Closes the Proxy thread associated to this SID. + """ + + proxy = self.proxy_threads.get(sid) + if proxy: + proxy.s.close() + del self.proxy_threads[sid] + + def close(self): + """ Closes all Proxy threads. + """ + + for sid, proxy in self.proxy_threads.items(): + proxy.s.close() + del self.proxy_threads[sid] + + def send(self, sid, data): + """ Sends the data over the Proxy socket associated to the + SID. + """ + + proxy = self.proxy_threads.get(sid) + if proxy: + proxy.s.sendall(data) + + def on_recv(self, sid, data): + """ Calls when data is recv from the Proxy socket associated + to the SID. + + Triggers a socks_closed event if the socket is closed. The sid + is passed to this event. + + Triggers a socks_recv event if there's available data. A dict + that contains the sid and the data is passed to this event. + """ + + proxy = self.proxy_threads.get(sid) + if proxy: + if not data: + self.xmpp.event('socks_closed', sid) + else: + self.xmpp.event('socks_recv', {'sid': sid, 'data': data}) + + +class Proxy(Thread): + """ Establishes in a thread a connection between the client and + the server-side Socks5 proxy. + """ + + def __init__(self, sid, requester, target, proxy, proxy_port, + on_recv): + """ Initializes the proxy thread. + + sid : The StreamID. <str> + requester : The JID of the requester. <str> + target : The JID of the target. <str> + proxy_host : The hostname or the IP of the proxy. <str> + proxy_port : The port of the proxy. <str> or <int> + on_recv : A callback called when data are received from the + socket. <Callable> + """ + + # Initializes the thread. + Thread.__init__(self) + + # Because the xep_0065 plugin uses the proxy_port as string, + # the Proxy class accepts the proxy_port argument as a string + # or an integer. Here, we force to use the port as an integer. + proxy_port = int(proxy_port) + + # Creates a connected event to warn when to proxy is + # connected. + self.connected = Event() + + # Registers the arguments. + self.sid = sid + self.requester = requester + self.target = target + self.proxy = proxy + self.proxy_port = proxy_port + self.on_recv = on_recv + + def run(self): + """ Starts the thread. + """ + + # Creates the socks5 proxy socket + self.s = socksocket() + self.s.setproxy(PROXY_TYPE_SOCKS5, self.proxy, port=self.proxy_port) + + # The hostname MUST be SHA1(SID + Requester JID + Target JID) + # where the output is hexadecimal-encoded (not binary). + digest = sha1() + digest.update(self.sid) # SID + digest.update(self.requester) # Requester JID + digest.update(self.target) # Target JID + + # Computes the digest in hex. + dest = '%s' % digest.hexdigest() + + # The port MUST be 0. + self.s.connect((dest, 0)) + log.info('Socket connected.') + self.connected.set() + + # Blocks until the socket need to be closed. + self.listen() + + # Closes the socket. + self.s.close() + log.info('Socket closed.') + + def listen(self): + """ Listen for data on the socket. When receiving data, call + the callback on_recv callable. + """ + + socket_open = True + while socket_open: + ins = [] + try: + # Wait any read available data on socket. Timeout + # after 5 secs. + ins, out, err = select([self.s, ], [], [], 5) + except Exception as e: + # There's an error with the socket (maybe the socket + # has been closed and the file descriptor is bad). + log.debug('Socket error: %s' % e) + break + + for s in ins: + data = self.recv_size(self.s) + if not data: + socket_open = False + + self.on_recv(self.sid, data) + + def recv_size(self, the_socket): + total_len = 0 + total_data = [] + size = sys.maxint + size_data = sock_data = '' + recv_size = 8192 + + while total_len < size: + sock_data = the_socket.recv(recv_size) + if not sock_data: + return ''.join(total_data) + + if not total_data: + if len(sock_data) > 4: + size_data += sock_data + size = struct.unpack('>i', size_data[:4])[0] + recv_size = size + if recv_size > 524288: + recv_size = 524288 + total_data.append(size_data[4:]) + else: + size_data += sock_data + else: + total_data.append(sock_data) + total_len = sum([len(i) for i in total_data]) + return ''.join(total_data) diff --git a/sleekxmpp/plugins/xep_0065/stanza.py b/sleekxmpp/plugins/xep_0065/stanza.py new file mode 100644 index 00000000..ae57aba8 --- /dev/null +++ b/sleekxmpp/plugins/xep_0065/stanza.py @@ -0,0 +1,41 @@ +from sleekxmpp import Iq +from sleekxmpp.xmlstream import ElementBase, register_stanza_plugin + + +# The protocol namespace defined in the Socks5Bytestream (0065) spec. +namespace = 'http://jabber.org/protocol/bytestreams' + + +class StreamHost(ElementBase): + """ The streamhost xml element. + """ + + namespace = namespace + name = 'streamhost' + plugin_attrib = 'streamhost' + interfaces = set(('host', 'jid', 'port')) + + +class StreamHostUsed(ElementBase): + """ The streamhost-used xml element. + """ + + namespace = namespace + name = 'streamhost-used' + plugin_attrib = 'streamhost-used' + interfaces = set(('jid',)) + + +class Socks5(ElementBase): + """ The query xml element. + """ + + namespace = namespace + name = 'query' + plugin_attrib = 'socks' + interfaces = set(('sid', 'activate')) + sub_interfaces = set(('activate',)) + +register_stanza_plugin(Iq, Socks5) +register_stanza_plugin(Socks5, StreamHost) +register_stanza_plugin(Socks5, StreamHostUsed) |