erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

File transfer:

Browse source 
- proxy managed in XEP-0065 (Socks5 bytestream)
- bug: fixed a bad id used during stream negociation
This commit is contained in:
Goffi 2011-10-05 16:49:57 +02:00
commit dc936be42b
4 changed files with 138 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

137
src/plugins/plugin_xep_0065.py
View file

@ -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():
<param name="Port" value="28915" type="string" />
</category>
</general>
<individual>
<category name="File Transfer">
<param name="Proxy" value="" type="string" />
<param name="Proxy host" value="" type="string" />
<param name="Proxy port" value="" type="string" />
</category>
</individual>
</params>
"""
@ -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=''):

4
src/plugins/plugin_xep_0095.py
View file

@ -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)

38
src/plugins/plugin_xep_0096.py
View file

@ -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 })

1
src/tools/memory.py
View file

@ -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'))