jshiffer/slixmpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adapted the xep_0065 plugin to be compatible with all kind of others XMPP

Browse Source 
client.
Sent a 'socks_connected' xmpp event when the streamer is connected.
This commit is contained in:
Sandro Munda 2012-11-04 11:38:57 +01:00
parent 3a7569e3ea
commit 032d41dbb8
1 changed files with 32 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
sleekxmpp/plugins/xep_0065/proxy.py
View File

@ -1,12 +1,6 @@
import sys
import logging import logging
import struct
import pickle
import socket
from threading import Thread, Event
from hashlib import sha1 from hashlib import sha1
from select import select
from uuid import uuid4 from uuid import uuid4
from sleekxmpp.plugins.xep_0065 import stanza from sleekxmpp.plugins.xep_0065 import stanza
@ -32,7 +26,7 @@ class XEP_0065(base_plugin):
# A dict contains for each SID, the proxy thread currently # A dict contains for each SID, the proxy thread currently
# running. # running.
proxy_threads = {} proxies = {}
def plugin_init(self): def plugin_init(self):
""" Initializes the xep_0065 plugin and all event callbacks. """ Initializes the xep_0065 plugin and all event callbacks.
@ -57,9 +51,9 @@ class XEP_0065(base_plugin):
""" Returns the socket associated to the SID. """ Returns the socket associated to the SID.
""" """
proxy = self.proxy_threads.get(sid) proxy = self.proxies.get(sid)
if proxy: if proxy:
return proxy.s return proxy
def handshake(self, to, streamer=None): def handshake(self, to, streamer=None):
""" Starts the handshake to establish the socks5 bytestreams """ Starts the handshake to establish the socks5 bytestreams
@ -138,15 +132,11 @@ class XEP_0065(base_plugin):
# Next the Target attempts to open a standard TCP socket on # Next the Target attempts to open a standard TCP socket on
# the network address of the Proxy. # the network address of the Proxy.
self.proxy_thread = Proxy(sid, requester, target, self.proxy_host, self.proxy = self._connect_proxy(sid, requester, target,
self.proxy_port, self.on_recv) self.proxy_host, self.proxy_port)
self.proxy_thread.start()
# Registers the new thread in the proxy_thread dict. # Registers the new proxy to the proxies dict.
self.proxy_threads[sid] = self.proxy_thread self.proxies[sid] = self.proxy
# Wait until the proxy is connected
self.proxy_thread.connected.wait()
# Replies to the incoming iq with a streamhost-used stanza. # Replies to the incoming iq with a streamhost-used stanza.
res_iq = iq.reply() res_iq = iq.reply()
@ -167,19 +157,14 @@ class XEP_0065(base_plugin):
# The Requester will establish a connection to the SOCKS5 # The Requester will establish a connection to the SOCKS5
# proxy in the same way the Target did. # proxy in the same way the Target did.
self.proxy_thread = Proxy(sid, requester, target, self.proxy_host, self.proxy = self._connect_proxy(sid, requester, target,
self.proxy_port, self.on_recv) self.proxy_host, self.proxy_port)
self.proxy_thread.start()
# Registers the new thread in the proxy_thread dict. # Registers the new thread in the proxy_thread dict.
self.proxy_threads[sid] = self.proxy_thread self.proxies[sid] = self.proxy
# Wait until the proxy is connected # Requester sends IQ-set to StreamHost requesting that StreamHost
self.proxy_thread.connected.wait() # activate the bytestream associated with the StreamID.
# Requester sends IQ-set to StreamHost requesting that
# StreamHost activate the bytestream associated with the
# StreamID.
self.activate(iq['socks']['sid'], target) self.activate(iq['socks']['sid'], target)
def activate(self, sid, to): def activate(self, sid, to):
@ -199,197 +184,63 @@ class XEP_0065(base_plugin):
""" Closes the Proxy thread associated to this SID. """ Closes the Proxy thread associated to this SID.
""" """
proxy = self.proxy_threads.get(sid) proxy = self.proxies.get(sid)
if proxy: if proxy:
proxy.s.close() proxy.s.close()
del self.proxy_threads[sid] del self.proxies[sid]
def close(self): def close(self):
""" Closes all Proxy threads. """ Closes all Proxy threads.
""" """
for sid, proxy in self.proxy_threads.items(): for sid, proxy in self.proxies.items():
proxy.s.close() proxy.close()
del self.proxy_threads[sid] del self.proxies[sid]
def send(self, sid, data): def send(self, sid, data):
""" Sends the data over the Proxy socket associated to the """ Sends the data over the Proxy socket associated to the
SID. SID.
""" """
proxy = self.proxy_threads.get(sid) proxy = self.get_socket(sid)
if proxy: if proxy:
proxy.send(data) proxy.sendall(data)
def on_recv(self, sid, data): def _connect_proxy(self, sid, requester, target, proxy, proxy_port):
""" Calls when data is recv from the Proxy socket associated """ Establishes a connection between the client and the server-side
to the SID. Socks5 proxy.
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> sid : The StreamID. <str>
requester : The JID of the requester. <str> requester : The JID of the requester. <str>
target : The JID of the target. <str> target : The JID of the target. <str>
proxy_host : The hostname or the IP of the proxy. <str> proxy_host : The hostname or the IP of the proxy. <str>
proxy_port : The port of the proxy. <str> or <int> 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, # Because the xep_0065 plugin uses the proxy_port as string,
# the Proxy class accepts the proxy_port argument as a 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. # or an integer. Here, we force to use the port as an integer.
proxy_port = int(proxy_port) 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 # Creates the socks5 proxy socket
self.s = socksocket() sock = socksocket()
self.s.setproxy(PROXY_TYPE_SOCKS5, self.proxy, port=self.proxy_port) sock.setproxy(PROXY_TYPE_SOCKS5, proxy, port=proxy_port)
# The hostname MUST be SHA1(SID + Requester JID + Target JID) # The hostname MUST be SHA1(SID + Requester JID + Target JID)
# where the output is hexadecimal-encoded (not binary). # where the output is hexadecimal-encoded (not binary).
digest = sha1() digest = sha1()
digest.update(self.sid) # SID digest.update(sid) # SID
digest.update(self.requester) # Requester JID digest.update(requester) # Requester JID
digest.update(self.target) # Target JID digest.update(target) # Target JID
# Computes the digest in hex. # Computes the digest in hex.
dest = '%s' % digest.hexdigest() dest = '%s' % digest.hexdigest()
# The port MUST be 0. # The port MUST be 0.
self.s.connect((dest, 0)) sock.connect((dest, 0))
log.info('Socket connected.') log.info('Socket connected.')
self.connected.set()
# Blocks until the socket need to be closed. # Send the XMPP event.
self.listen() self.xmpp.event('socks_connected', sid)
# Closes the socket. return sock
self.s.close()
log.info('Socket closed.')
def send(self, data):
""" Send data through the socket.
"""
try:
packed_data = self._pack(data)
self.s.sendall(packed_data)
except pickle.PickleError as err:
log.error(err)
def _pack(self, data):
""" Packs the data.
"""
# The data format is: `len_data`+`data`. Useful to receive all the data
# at once (avoid splitted data) thanks to the recv_size method.
data = pickle.dumps(data)
return struct.pack('>i', len(data)) + data
def _unpack(self, data):
""" Unpacks the data. On error, log an error message and returns None.
"""
try:
return pickle.loads(data)
except Exception as err:
log.error(err)
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 socket.error as (errno, err):
# 9 means the socket is closed. It can be normal. Otherwise,
# log the error.
if errno != 9:
log.debug('Socket error: %s' % err)
break
except Exception as e:
log.debug(e)
break
for s in ins:
data = self.recv_size(self.s)
if not data:
socket_open = False
else:
unpacked_data = self._unpack(data)
if unpacked_data:
self.on_recv(self.sid, unpacked_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)