libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0065.py @ 2927:69e4716d6268

plugins (jingle) file transfer: use initial "from" attribute as local jid instead of client.jid: while client.jid is fine in a client context, for components it's not the right jid to use: it is the jid of the component itself while the file transfer/jingle session entity may be established with this jid + a local part (e.g. if client is files.example.net, session may be established with louise@files.example.net, in which case "from" is louise@files.example.net, while client.jid will be files.example.net). As a consequence, using client.jid was causing trouble with components. This patch fixes it for jingle and plugins linked to file transfer by keeping a "local_jid" variable in the session, where the jid from the original "from" attribute is used.

author	Goffi <goffi@goffi.org>
date	Sun, 28 Apr 2019 08:55:13 +0200
parents	003b8b4b56a7
children	ab2696e34d29

comparison

equal deleted inserted replaced

-:4cd7545c4ebb
+:69e4716d6268
 multiplier = 10
 else:
 raise exceptions.InternalError(u"Unknown {} type !".format(self.type))
 return 2 ** 16 * multiplier + self._local_priority
-def activate(self, sid, peer_jid, client):
+def activate(self, client, sid, peer_jid, local_jid):
 """Activate the proxy candidate
 Send activation request as explained in XEP-0065 § 6.3.5
 Must only be used with proxy candidates
 @param sid(unicode): session id (same as for getSessionHash)
 @param peer_jid(jid.JID): jid of the other peer
 @return (D(domish.Element)): IQ result (or error)
 """
 assert self.type == XEP_0065.TYPE_PROXY
 iq_elt = client.IQ()
+iq_elt["from"] = local_jid.full()
 iq_elt["to"] = self.jid.full()
 query_elt = iq_elt.addElement((NS_BS, "query"))
 query_elt["sid"] = sid
 query_elt.addElement("activate", content=peer_jid.full())
 return iq_elt.send()
 )
 )
 return self._server_factory
 @defer.inlineCallbacks
-def getProxy(self, client):
+def getProxy(self, client, local_jid):
 """Return the proxy available for this profile
 cache is used between clients using the same server
+@param local_jid(jid.JID): same as for [getCandidates]
 @return ((D)(ProxyInfos, None)): Found proxy infos,
 or None if not acceptable proxy is found
+@raise exceptions.NotFound: no Proxy found
 """
 def notFound(server):
 log.info(u"No proxy found on this server")
 self._cache_proxies[server] = None
-defer.returnValue(None)
+raise exceptions.NotFound
-server = client.jid.host
+server = client.host if client.is_component else client.jid.host
 try:
 defer.returnValue(self._cache_proxies[server])
 except KeyError:
 pass
 try:
 yield self.host.findServiceEntities(client, "proxy", "bytestreams")
 ).pop()
 except (defer.CancelledError, StopIteration, KeyError):
 notFound(server)
 iq_elt = client.IQ("get")
+iq_elt["from"] = local_jid.full()
 iq_elt["to"] = proxy.full()
 iq_elt.addElement((NS_BS, "query"))
 try:
 result_elt = yield iq_elt.send()
 self._external_port = ext_port
 defer.returnValue((local_port, self._external_port, local_ips, external_ip))
 @defer.inlineCallbacks
-def getCandidates(self, client):
+def getCandidates(self, client, local_jid):
 """Return a list of our stream candidates
+@param local_jid(jid.JID): jid to use as local jid
+This is needed for client which can be addressed with a different jid than
+client.jid if a local part is used (e.g. piotr@file.example.net where
+client.jid would be file.example.net)
 @return (D(list[Candidate])): list of candidates, ordered by priority
 """
 server_factory = yield self.getSocks5ServerFactory()
 local_port, ext_port, local_ips, external_ip = yield self._getNetworkData(client)
-proxy = yield self.getProxy(client)
+try:
+proxy = yield self.getProxy(client, local_jid)
+except exceptions.NotFound:
+proxy = None
 # its time to gather the candidates
 candidates = []
 # first the direct ones
 Candidate(
 ip,
 local_port,
 XEP_0065.TYPE_DIRECT,
 PRIORITY_BEST_DIRECT,
-client.jid,
+local_jid,
 priority_local=True,
 factory=server_factory,
 )
 )
 for ip in local_ips:
 Candidate(
 ip,
 local_port,
 XEP_0065.TYPE_DIRECT,
 PRIORITY_DIRECT,
-client.jid,
+local_jid,
 priority_local=True,
 factory=server_factory,
 )
 )
 Candidate(
 external_ip,
 ext_port,
 XEP_0065.TYPE_ASSISTED,
 PRIORITY_ASSISTED,
-client.jid,
+local_jid,
 priority_local=True,
 factory=server_factory,
 )
 )
 except (internet_error.AlreadyCalled, internet_error.AlreadyCancelled):
 pass
 return failure_
-def startStream(self, client, stream_object, to_jid, sid):
+def startStream(self, client, stream_object, local_jid, to_jid, sid):
 """Launch the stream workflow
 @param streamProducer: stream_object to use
+@param local_jid(jid.JID): same as for [getCandidates]
 @param to_jid: JID of the recipient
 @param sid: Stream session id
 @param successCb: method to call when stream successfuly finished
 @param failureCb: method to call when something goes wrong
 @return (D): Deferred fired when session is finished
 """
-session_data = self._createSession(client, stream_object, to_jid, sid, True)
+session_data = self._createSession(
+client, stream_object, local_jid, to_jid, sid, True)
 session_data[client] = client
 def gotCandidates(candidates):
 session_data["candidates"] = candidates
 iq_elt = client.IQ()
-iq_elt["from"] = client.jid.full()
+iq_elt["from"] = local_jid.full()
 iq_elt["to"] = to_jid.full()
 query_elt = iq_elt.addElement((NS_BS, "query"))
 query_elt["mode"] = "tcp"
 query_elt["sid"] = sid
 streamhost["port"] = str(candidate.port)
 streamhost["jid"] = candidate.jid.full()
 log.debug(u"Candidate proposed: {}".format(candidate))
 d = iq_elt.send()
-args = [session_data, client]
+args = [client, session_data, local_jid]
 d.addCallbacks(self._IQNegotiationCb, self._IQNegotiationEb, args, None, args)
-self.getCandidates(client).addCallback(gotCandidates)
+self.getCandidates(client, local_jid).addCallback(gotCandidates)
 return session_data[DEFER_KEY]
-def _IQNegotiationCb(self, iq_elt, session_data, client):
+def _IQNegotiationCb(self, iq_elt, client, session_data, local_jid):
 """Called when the result of open iq is received
 @param session_data(dict): data of the session
 @param client: %(doc_client)s
 @param iq_elt(domish.Element): <iq> result
 if candidate.type == XEP_0065.TYPE_PROXY:
 log.info(u"A Socks5 proxy is used")
 d = self.connectCandidate(client, candidate, session_data["hash"])
 d.addCallback(
 lambda __: candidate.activate(
-session_data["id"], session_data["peer_jid"], client
+client, session_data["id"], session_data["peer_jid"], local_jid
 )
 )
 d.addErrback(self._activationEb)
 else:
 d = defer.succeed(None)
 d.addCallback(lambda __: candidate.startTransfer(session_data["hash"]))
 def _activationEb(self, failure):
 log.warning(u"Proxy activation error: {}".format(failure.value))
-def _IQNegotiationEb(self, stanza_err, session_data, client):
+def _IQNegotiationEb(self, stanza_err, client, session_data, local_jid):
 log.warning(u"Socks5 transfer failed: {}".format(stanza_err.value))
 # FIXME: must clean session
 def createSession(self, *args, **kwargs):
 """like [_createSession] but return the session deferred instead of the whole session
 session deferred is fired when transfer is finished
 """
 return self._createSession(*args, **kwargs)[DEFER_KEY]
-def _createSession(self, client, stream_object, to_jid, sid, requester=False):
+def _createSession(self, client, stream_object, local_jid, to_jid, sid,
+requester=False):
 """Called when a bytestream is imminent
-@param stream_object(iface.IStreamProducer): File object where data will be written
+@param stream_object(iface.IStreamProducer): File object where data will be
+written
 @param to_jid(jid.JId): jid of the other peer
 @param sid(unicode): session id
 @param initiator(bool): if True, this session is create by initiator
 @return (dict): session data
 """
 if sid in client.xep_0065_sid_session:
 raise exceptions.ConflictError(u"A session with this id already exists !")
 if requester:
-session_hash = getSessionHash(client.jid, to_jid, sid)
+session_hash = getSessionHash(local_jid, to_jid, sid)
 session_data = self._registerHash(client, session_hash, stream_object)
 else:
-session_hash = getSessionHash(to_jid, client.jid, sid)
+session_hash = getSessionHash(to_jid, local_jid, sid)
 session_d = defer.Deferred()
 session_d.addBoth(self.killSession, session_hash, sid, client)
 session_data = client._s5b_sessions[session_hash] = {
 DEFER_KEY: session_d,
 TIMER_KEY: reactor.callLater(
 }
 client.xep_0065_sid_session[sid] = session_data
 session_data.update(
 {
 "id": sid,
+"local_jid": local_jid,
 "peer_jid": to_jid,
 "stream_object": stream_object,
 "hash": session_hash,
 }
 )

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0065.py @ 2927:69e4716d6268