diff --git a/src/plugins/plugin_xep_0065.py b/src/plugins/plugin_xep_0065.py
index c49f7652..6b6505a0 100755
--- a/src/plugins/plugin_xep_0065.py
+++ b/src/plugins/plugin_xep_0065.py
@@ -296,8 +296,12 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.loseConnection()
return
- self.state = STATE_TARGET_READY
- self.factory.activateCb(self.sid, self.factory.iq_id)
+ if self.factory.proxy:
+ self.state = STATE_READY
+ self.factory.activateCb(self.sid, self.factory.iq_id, self.startTransfer)
+ else:
+ self.state = STATE_TARGET_READY
+ self.factory.activateCb(self.sid, self.factory.iq_id)
except struct.error, why:
return None
@@ -405,12 +409,20 @@ class Socks5ServerFactory(protocol.ServerFactory):
class Socks5ClientFactory(protocol.ClientFactory):
protocol = SOCKSv5
- def __init__(self, current_stream, sid, iq_id, activateCb, finishedCb):
+ def __init__(self, current_stream, sid, iq_id, activateCb, finishedCb, proxy=False):
+ """Init the Client Factory
+ @param current_stream: current streams data
+ @param sid: Session ID
+ @param iq_id: iq id used to initiate the stream
+ @param activateCb: method to call to activate the stream
+ @param finishedCb: method to call when the stream session is finished
+ @param proxy: True if we are connecting throught a proxy (and we are a requester)"""
self.data = current_stream[sid]
self.sid = sid
self.iq_id = iq_id
self.activateCb = activateCb
self.finishedCb = finishedCb
+ self.proxy = proxy
def startedConnecting(self, connector):
debug (_("Socks 5 client connection started"))
@@ -432,6 +444,13 @@ class XEP_0065():
+
+
+
+
+
+
+
"""
@@ -503,19 +522,6 @@ class XEP_0065():
else:
failure_cb(sid, file_obj, NS_BS, failure_reason)
- def setData(self, data, id):
- self.data = data
- self.transfer_id = id
-
- def sendFile(self, id, filepath, size):
- #lauching socks5 requester
- debug(_("Launching socks5 requester"))
- self.server_factory.protocol.mode = "requester"
- self.server_factory.protocol.filepath = filepath
- self.server_factory.protocol.filesize = size
- self.server_factory.protocol.transfer_id = id
-
-
def startStream(self, file_obj, to_jid, sid, length, successCb, failureCb, size = None, profile='@NONE@'):
"""Launch the stream workflow
@param file_obj: file_obj to send
@@ -529,9 +535,14 @@ class XEP_0065():
error(_('stream length not managed yet'))
return;
profile_jid, xmlstream = self.host.getJidNStream(profile)
+ if not profile_jid or not xmlstream:
+ error(_("Unknown profile, this should not happen"))
+ return;
data = self.current_stream[sid] = {}
+ data["profile"] = profile
data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid)
data["file_obj"] = file_obj
+ data["from"] = profile_jid
data["to"] = to_jid
data["success_cb"] = successCb
data["failure_cb"] = failureCb
@@ -547,10 +558,18 @@ class XEP_0065():
query_elt = iq_elt.addElement('query', NS_BS)
query_elt['mode'] = 'tcp'
query_elt['sid'] = sid
+ #first streamhost: direct connection
streamhost = query_elt.addElement('streamhost')
- streamhost['host'] = "127.0.0.1" #self.host.memory.getParamA("IP", "File Transfer")
+ streamhost['host'] = self.host.memory.getParamA("IP", "File Transfer")
streamhost['port'] = self.host.memory.getParamA("Port", "File Transfer")
streamhost['jid'] = profile_jid.full()
+
+ #second streamhost: mediated connection, using proxy
+ streamhost = query_elt.addElement('streamhost')
+ streamhost['host'] = self.host.memory.getParamA("Proxy host", "File Transfer", profile_key=profile)
+ streamhost['port'] = self.host.memory.getParamA("Proxy port", "File Transfer", profile_key=profile)
+ streamhost['jid'] = self.host.memory.getParamA("Proxy", "File Transfer", profile_key=profile)
+
iq_elt.addCallback(self.iqResult, sid)
iq_elt.send()
@@ -560,20 +579,60 @@ class XEP_0065():
warning(_("Transfer failed"))
return
- try:
+ try:
data = self.current_stream[sid]
- callback = data["start_transfer_cb"]
file_obj = data["file_obj"]
timer = data["timer"]
+ profile = data["profile"]
except KeyError:
error(_("Internal error, can't do transfer"))
return
if timer.active():
timer.cancel()
+
+ profile_jid, xmlstream = self.host.getJidNStream(profile)
+ query_elt = iq_elt.firstChildElement()
+ streamhost_elts = filter(lambda elt: elt.name == 'streamhost-used', query_elt.elements())
+ if not streamhost_elts:
+ warning(_("No streamhost found in stream query"))
+ return
- callback(file_obj)
+ streamhost_jid = streamhost_elts[0]['jid']
+ if streamhost_jid != profile_jid.full():
+ debug(_("A proxy server is used"))
+ proxy_host = self.host.memory.getParamA("Proxy host", "File Transfer", profile_key=profile)
+ proxy_port = self.host.memory.getParamA("Proxy port", "File Transfer", profile_key=profile)
+ proxy_jid = self.host.memory.getParamA("Proxy", "File Transfer", profile_key=profile)
+ if proxy_jid != streamhost_jid:
+ warning(_("Proxy jid is not the same as in parameters, this should not happen"))
+ return
+ factory = Socks5ClientFactory(self.current_stream, sid, None, self.activateProxyStream, self._killId, True)
+ reactor.connectTCP(proxy_host, int(proxy_port), factory)
+ else:
+ data["start_transfer_cb"](file_obj) #We now activate the stream
+ def activateProxyStream(self, sid, iq_id, start_transfer_cb):
+ debug(_("activating stream"))
+ data = self.current_stream[sid]
+ profile = data['profile']
+ profile_jid, xmlstream = self.host.getJidNStream(profile)
+
+ iq_elt = client.IQ(xmlstream,'set')
+ iq_elt["from"] = profile_jid.full()
+ iq_elt["to"] = self.host.memory.getParamA("Proxy", "File Transfer", profile_key=profile)
+ query_elt = iq_elt.addElement('query', NS_BS)
+ query_elt['sid'] = sid
+ query_elt.addElement('activate', content=data['to'].full())
+ iq_elt.addCallback(self.proxyResult, sid, start_transfer_cb, data['file_obj'])
+ iq_elt.send()
+
+ def proxyResult(self, sid, start_transfer_cb, file_obj, iq_elt):
+ if iq_elt['type'] == 'error':
+ warning(_("Can't activate the proxy stream"))
+ return
+ else:
+ start_transfer_cb(file_obj)
def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb):
"""Called when a bytestream is imminent
@@ -680,9 +739,47 @@ class XEP_0065_handler(XMPPHandler):
def __init__(self, plugin_parent):
self.plugin_parent = plugin_parent
self.host = plugin_parent.host
+
+ def _proxyDataResult(self, iq_elt):
+ """Called with the informations about proxy according to XEP-0065 #4
+ Params should be filled with these infos"""
+ if iq_elt["type"] == "error":
+ warning(_("Can't determine proxy informations"))
+ return
+ query_elt = iq_elt.firstChildElement()
+ if query_elt.name != "query":
+ warning(_("Bad answer received from proxy"))
+ return
+ streamhost_elts = filter(lambda elt: elt.name == 'streamhost', query_elt.elements())
+ if not streamhost_elts:
+ warning(_("No streamhost found in stream query"))
+ return
+ if len(streamhost_elts) != 1:
+ warning(_("Multiple streamhost elements in proxy not managed, keeping only the first one"))
+ streamhost_elt = streamhost_elts[0]
+ proxy = self.host.memory.setParam("Proxy", streamhost_elt.getAttribute("jid",""), "File Transfer", self.parent.profile)
+ proxy = self.host.memory.setParam("Proxy host", streamhost_elt.getAttribute("host",""), "File Transfer", self.parent.profile)
+ proxy = self.host.memory.setParam("Proxy port", streamhost_elt.getAttribute("port",""), "File Transfer", self.parent.profile)
+
def connectionInitialized(self):
+ def after_init(ignore):
+ proxy_ent = self.host.memory.getServerServiceEntity("proxy", "bytestreams", self.parent.profile)
+ if not proxy_ent:
+ debug(_("No proxy found on this server"))
+ return
+ iq_elt = client.IQ(self.parent.xmlstream,'get')
+ iq_elt["to"] = proxy_ent.full()
+ query_elt = iq_elt.addElement('query', NS_BS)
+ iq_elt.addCallback(self._proxyDataResult)
+ iq_elt.send()
+
+
self.xmlstream.addObserver(BS_REQUEST, self.plugin_parent.streamQuery, profile = self.parent.profile)
+ proxy = self.host.memory.getParamA("Proxy", "File Transfer", profile_key = self.parent.profile)
+ if not proxy:
+ self.parent.client_initialized.addCallback(after_init)
+
def getDiscoInfo(self, requestor, target, nodeIdentifier=''):
diff --git a/src/plugins/plugin_xep_0095.py b/src/plugins/plugin_xep_0095.py
index 0fab4151..389fc13e 100644
--- a/src/plugins/plugin_xep_0095.py
+++ b/src/plugins/plugin_xep_0095.py
@@ -75,13 +75,13 @@ class XEP_0095():
info (_("XEP-0095 Stream initiation"))
iq_el.handled=True
si_el = iq_el.firstChildElement()
- si_id = iq_el.getAttribute('id')
+ si_id = si_el.getAttribute('id')
si_mime_type = iq_el.getAttribute('mime-type', 'application/octet-stream')
si_profile = si_el.getAttribute('profile')
si_profile_key = si_profile[len(SI_PROFILE_HEADER):] if si_profile.startswith(SI_PROFILE_HEADER) else si_profile
if self.si_profiles.has_key(si_profile_key):
#We know this SI profile, we call the callback
- self.si_profiles[si_profile_key](iq_el['from'], si_id, si_mime_type, si_el, profile)
+ self.si_profiles[si_profile_key](iq_el['id'], iq_el['from'], si_id, si_mime_type, si_el, profile)
else:
#We don't know this profile, we send an error
self.sendBadProfileError(iq_el['id'], iq_el['from'], profile)
diff --git a/src/plugins/plugin_xep_0096.py b/src/plugins/plugin_xep_0096.py
index f28e3467..054256df 100755
--- a/src/plugins/plugin_xep_0096.py
+++ b/src/plugins/plugin_xep_0096.py
@@ -70,8 +70,9 @@ class XEP_0096():
except KeyError:
warning(_("kill id called on a non existant approval id"))
- def transferRequest(self, from_jid, si_id, si_mime_type, si_el, profile):
+ def transferRequest(self, iq_id, from_jid, si_id, si_mime_type, si_el, profile):
"""Called when a file transfer is requested
+ @param iq_id: id of the iq request
@param from_jid: jid of the sender
@param si_id: Stream Initiation session id
@param si_mime_type: Mime type of the file (or default "application/octet-stream" if unknown)
@@ -102,7 +103,7 @@ class XEP_0096():
can_range = True
else:
warning(_("No file element found"))
- self.host.plugins["XEP-0095"].sendBadRequestError(si_id, from_jid, profile)
+ self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
return
if feature_elts:
@@ -112,19 +113,19 @@ class XEP_0096():
stream_method = self.host.plugins["XEP-0020"].negociate(feature_el, 'stream-method',self.managed_stream_m)
except KeyError:
warning(_("No stream method found"))
- self.host.plugins["XEP-0095"].sendBadRequestError(si_id, from_jid, profile)
+ self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
return
if not stream_method:
warning(_("Can't find a valid stream method"))
- self.host.plugins["XEP-0095"].sendFailedError(si_id, from_jid, profile)
+ self.host.plugins["XEP-0095"].sendFailedError(iq_id, from_jid, profile)
return
else:
warning(_("No feature element found"))
- self.host.plugins["XEP-0095"].sendBadRequestError(si_id, from_jid, profile)
+ self.host.plugins["XEP-0095"].sendBadRequestError(iq_id, from_jid, profile)
return
#if we are here, the transfer can start, we just need user's agreement
- data={ "filename":filename, "from":from_jid, "size":file_size, "date":file_date, "hash":file_hash, "desc":file_desc, "can_range": str(can_range) }
+ data={ "filename":filename, "id": iq_id, "from":from_jid, "size":file_size, "date":file_date, "hash":file_hash, "desc":file_desc, "can_range": str(can_range) }
self._waiting_for_approval[si_id] = [data, reactor.callLater(300, self._kill_id, si_id), stream_method, [], profile]
self.host.askConfirmation(si_id, "FILE_TRANSFER", data, self.confirmationCB)
@@ -137,12 +138,12 @@ class XEP_0096():
@return: File Object"""
return open(dest_path, "ab" if can_range else "wb")
- def confirmationCB(self, id, accepted, frontend_data):
+ def confirmationCB(self, sid, accepted, frontend_data):
"""Called on confirmation answer
- @param id: file transfer session id
+ @param sid: file transfer session id
@param accepted: True if file transfer is accepted
@param frontend_data: data sent by frontend"""
- data, timeout, stream_method, failed_methods, profile = self._waiting_for_approval[id]
+ data, timeout, stream_method, failed_methods, profile = self._waiting_for_approval[sid]
can_range = data['can_range'] == "True"
range_offset = 0
if accepted:
@@ -152,20 +153,19 @@ class XEP_0096():
dest_path = frontend_data['dest_path']
except KeyError:
error(_('dest path not found in frontend_data'))
- del(self._waiting_for_approval[id])
+ del(self._waiting_for_approval[sid])
return
if stream_method == self.host.plugins["XEP-0065"].NAMESPACE:
file_obj = self._getFileObject(dest_path, can_range)
range_offset = file_obj.tell()
- self.host.plugins["XEP-0065"].prepareToReceive(jid.JID(data['from']), id, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed)
- #self.host.plugins["XEP-0065"].setData(data, id)
+ self.host.plugins["XEP-0065"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed)
elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE:
file_obj = self._getFileObject(dest_path, can_range)
range_offset = file_obj.tell()
- self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), id, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed)
+ self.host.plugins["XEP-0047"].prepareToReceive(jid.JID(data['from']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed)
else:
error(_("Unknown stream method, this should not happen at this stage, cancelling transfer"))
- del(self._waiting_for_approval[id])
+ del(self._waiting_for_approval[sid])
return
#we can send the iq result
@@ -177,11 +177,11 @@ class XEP_0096():
range_elt['offset'] = str(range_offset)
#TODO: manage range length
misc_elts.append(range_elt)
- self.host.plugins["XEP-0095"].acceptStream(id, data['from'], feature_elt, misc_elts, profile)
+ self.host.plugins["XEP-0095"].acceptStream(data["id"], data['from'], feature_elt, misc_elts, profile)
else:
- debug (_("Transfer [%s] refused"), id)
- self.host.plugins["XEP-0095"].sendRejectedError (id, data['from'], profile=profile)
- del(self._waiting_for_approval[id])
+ debug (_("Transfer [%s] refused"), sid)
+ self.host.plugins["XEP-0095"].sendRejectedError (data["id"], data['from'], profile=profile)
+ del(self._waiting_for_approval[sid])
def _transferSucceeded(self, sid, file_obj, stream_method):
"""Called by the stream method when transfer successfuly finished
@@ -289,4 +289,4 @@ class XEP_0096():
def sendFailureCb(self, sid, file_obj, stream_method, reason):
file_obj.close()
- warning(_('Transfer %(id)s failed with stream method %(s_method)s') % { 'id': sid, s_method: stream_method })
+ warning(_('Transfer %(id)s failed with stream method %(s_method)s') % { 'id': sid, "s_method": stream_method })
diff --git a/src/tools/memory.py b/src/tools/memory.py
index 9ca908f3..687a2a3f 100644
--- a/src/tools/memory.py
+++ b/src/tools/memory.py
@@ -667,7 +667,6 @@ class Memory:
"""Return list of contacts for given profile
@param profile_key: profile key
@return list of [contact, attr, groups]"""
- debug ("Memory getContact OK (%s)", self.contacts)
profile = self.getProfileName(profile_key)
if not profile:
error(_('Asking contacts for a non-existant profile'))