diff --git a/src/plugins/plugin_xep_0065.py b/src/plugins/plugin_xep_0065.py
index 6252580e..eaac9a1e 100755
--- a/src/plugins/plugin_xep_0065.py
+++ b/src/plugins/plugin_xep_0065.py
@@ -23,7 +23,7 @@
# --
-# This program is based on proxy65 (http://code.google.com/p/proxy65),
+# This module is based on proxy65 (http://code.google.com/p/proxy65),
# originaly written by David Smith and modified by Fabio Forno.
# It is sublicensed under AGPL v3 (or any later version) as allowed by the original
# license.
@@ -57,15 +57,23 @@
from sat.core.i18n import _
from sat.core.log import getLogger
log = getLogger(__name__)
-from twisted.internet import protocol, reactor
-from twisted.internet import error
+from sat.core.constants import Const as C
+from sat.core import exceptions
+from sat.tools import sat_defer
+from twisted.internet import protocol
+from twisted.internet import reactor
+from twisted.internet import error as internet_error
from twisted.words.protocols.jabber import jid, client as jabber_client
+from twisted.words.protocols.jabber import error as jabber_error
from twisted.protocols.basic import FileSender
from twisted.words.xish import domish
-from twisted.web.client import getPage
+from twisted.internet import defer
+from twisted.python import failure
from sat.core.exceptions import ProfileNotInCacheError
+from collections import namedtuple
import struct
import hashlib
+import uuid
from zope.interface import implements
@@ -80,29 +88,58 @@ IQ_SET = '/iq[@type="set"]'
NS_BS = 'http://jabber.org/protocol/bytestreams'
BS_REQUEST = IQ_SET + '/query[@xmlns="' + NS_BS + '"]'
TIMEOUT = 60 # timeout for workflow
+DEFER_KEY = 'finished' # key of the deferred used to track session end
+SERVER_STARTING_PORT = 0 # starting number for server port search (0 to ask automatic attribution)
+
+# priorities are candidates local priorities, must be a int between 0 and 65535
+PRIORITY_BEST_DIRECT = 10000
+PRIORITY_DIRECT = 5000
+PRIORITY_ASSISTED = 1000
+PRIORITY_PROXY = 0.2 # proxy is the last option for s5b
+CANDIDATE_DELAY = 0.2 # see XEP-0260 §4
+CANDIDATE_DELAY_PROXY = 0.2 # additional time for proxy types (see XEP-0260 §4 note 3)
PLUGIN_INFO = {
"name": "XEP 0065 Plugin",
"import_name": "XEP-0065",
"type": "XEP",
"protocols": ["XEP-0065"],
+ "dependencies": ["IP"],
+ "recommendations": ["NAT-PORT"],
"main": "XEP_0065",
"handler": "yes",
"description": _("""Implementation of SOCKS5 Bytestreams""")
}
-STATE_INITIAL = 0
-STATE_AUTH = 1
-STATE_REQUEST = 2
-STATE_READY = 3
-STATE_AUTH_USERPASS = 4
-STATE_TARGET_INITIAL = 5
-STATE_TARGET_AUTH = 6
-STATE_TARGET_REQUEST = 7
-STATE_TARGET_READY = 8
-STATE_LAST = 9
+# XXX: by default eveything is automatic
+# TODO: use these params to force use of specific proxy/port/IP
+# PARAMS = """
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+#
+# """
-STATE_CONNECT_PENDING = STATE_LAST + 1
+(STATE_INITIAL,
+STATE_AUTH,
+STATE_REQUEST,
+STATE_READY,
+STATE_AUTH_USERPASS,
+STATE_CLIENT_INITIAL,
+STATE_CLIENT_AUTH,
+STATE_CLIENT_REQUEST,
+) = xrange(8)
SOCKS5_VER = 0x05
@@ -129,19 +166,128 @@ REPLY_CMD_NOT_SUPPORTED = 0x07
REPLY_ADDR_NOT_SUPPORTED = 0x08
-def calculateHash(from_jid, to_jid, sid):
- """Calculate SHA1 Hash according to XEP-0065
- @param from_jid: jid of the requester
- @param to_jid: jid of the target
- @param sid: session id
- @return: hash (string)"""
+ProxyInfos = namedtuple("ProxyInfos", ['host', 'jid', 'port'])
+
+
+class Candidate(object):
+
+ def __init__(self, host, port, type_, priority, jid_, id_=None, priority_local=False, factory=None):
+ """
+ @param host(unicode): host IP or domain
+ @param port(int): port
+ @param type_(unicode): stream type (one of XEP_0065.TYPE_*)
+ @param priority(int): priority
+ @param jid_(jid.JID): jid
+ @param id_(None, id_): Candidate ID, or None to generate
+ @param priority_local(bool): if True, priority is used as local priority,
+ else priority is used as global one (and local priority is set to 0)
+ """
+ assert isinstance(jid_, jid.JID)
+ self.host, self.port, self.type, self.jid = (
+ host, int(port), type_, jid_)
+ self.id = id_ if id_ is not None else unicode(uuid.uuid4())
+ if priority_local:
+ self._local_priority = int(priority)
+ self._priority = self.calculatePriority()
+ else:
+ self._local_priority = 0
+ self._priority = int(priority)
+ self.factory = factory
+
+ def discard(self):
+ """Disconnect a candidate if it is connected
+
+ Used to disconnect tryed client when they are discarded
+ """
+ log.debug(u"Discarding {}".format(self))
+ try:
+ self.factory.discard()
+ except AttributeError:
+ pass # no discard for Socks5ServerFactory
+
+ @property
+ def local_priority(self):
+ return self._local_priority
+
+ @property
+ def priority(self):
+ return self._priority
+
+ def __str__(self):
+ # similar to __unicode__ but we don't show jid and we encode id
+ return "Candidate ({0.priority}): host={0.host} port={0.port} type={0.type}{id}".format(
+ self,
+ id=u" id={}".format(self.id if self.id is not None else u'').encode('utf-8', 'ignore'),
+ )
+
+ def __unicode__(self):
+ return u"Candidate ({0.priority}): host={0.host} port={0.port} jid={0.jid} type={0.type}{id}".format(
+ self,
+ id=u" id={}".format(self.id if self.id is not None else u''),
+ )
+
+ def __eq__(self, other):
+ # self.id is is not used in __eq__ as the same candidate can have
+ # different ids if proposed by initiator or responder
+ try:
+ return (self.host == other.host and
+ self.port == other.port and
+ self.jid == other.jid)
+ except (AttributeError, TypeError):
+ return False
+
+ def __ne__(self, other):
+ return not self.__eq__(other)
+
+ def calculatePriority(self):
+ """Calculate candidate priority according to XEP-0260 §2.2
+
+
+ @return (int): priority
+ """
+ if self.type == XEP_0065.TYPE_DIRECT:
+ multiplier = 126
+ elif self.type == XEP_0065.TYPE_ASSISTED:
+ multiplier = 120
+ elif self.type == XEP_0065.TYPE_TUNEL:
+ multiplier = 110
+ elif self.type == XEP_0065.TYPE_PROXY:
+ multiplier = 10
+ else:
+ raise exceptions.InternalError(u"Unknown {} type !".format(self.type))
+ return 2**16 * multiplier + self._local_priority
+
+ def startTransfer(self, session_hash=None):
+ self.factory.startTransfer(session_hash)
+
+
+def getSessionHash(from_jid, to_jid, sid):
+ """Calculate SHA1 Hash according to XEP-0065 §5.3.2
+
+ @param from_jid(jid.JID): jid of the requester
+ @param to_jid(jid.JID): jid of the target
+ @param sid(unicode): session id
+ @return (str): hash
+ """
return hashlib.sha1((sid + from_jid.full() + to_jid.full()).encode('utf-8')).hexdigest()
class SOCKSv5(protocol.Protocol, FileSender):
- def __init__(self):
+
+ def __init__(self, session_hash=None):
+ """
+ @param session_hash(str): hash of the session
+ must only be used in client mode
+ """
log.debug(_("Protocol init"))
- self.state = STATE_INITIAL
+ self.connection = defer.Deferred() # called when connection/auth is done
+ if session_hash is not None:
+ self.server_mode = False
+ self._session_hash = session_hash
+ self.state = STATE_CLIENT_INITIAL
+ else:
+ self.server_mode = True
+ self.state = STATE_INITIAL
self.buf = ""
self.supportedAuthMechs = [AUTHMECH_ANON]
self.supportedAddrs = [ADDR_DOMAINNAME]
@@ -149,10 +295,20 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.peersock = None
self.addressType = 0
self.requestType = 0
+ self._file_obj = None
+
+ @property
+ def file_obj(self):
+ if self._file_obj is None:
+ if self.server_mode:
+ self._file_obj = self.factory.getSession(self._session_hash)["file"]
+ else:
+ self._file_obj = self.factory.getSession()['file']
+ return self._file_obj
def _startNegotiation(self):
- log.debug("_startNegotiation")
- self.state = STATE_TARGET_AUTH
+ log.debug("starting negotiation (client mode)")
+ self.state = STATE_CLIENT_AUTH
self.transport.write(struct.pack('!3B', SOCKS5_VER, 1, AUTHMECH_ANON))
def _parseNegotiation(self):
@@ -184,6 +340,7 @@ class SOCKSv5(protocol.Protocol, FileSender):
return
# No supported mechs found, notify client and close the connection
+ log.warning(u"Unsupported authentication mechanism")
self.transport.write(struct.pack('!BB', SOCKS5_VER, AUTHMECH_INVALID))
self.transport.loseConnection()
except struct.error:
@@ -258,14 +415,16 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.sendErrorReply(REPLY_CMD_NOT_SUPPORTED)
except struct.error:
+ # The buffer is probably not complete, we need to wait more
return None
def _makeRequest(self):
log.debug("_makeRequest")
- self.state = STATE_TARGET_REQUEST
- sha1 = calculateHash(self.data["from"], self.data["to"], self.sid)
- request = struct.pack('!5B%dsH' % len(sha1), SOCKS5_VER, CMD_CONNECT, 0, ADDR_DOMAINNAME, len(sha1), sha1, 0)
+ # sha1 = getSessionHash(self.data["from"], self.data["to"], self.sid)
+ hash_ = self._session_hash
+ request = struct.pack('!5B%dsH' % len(hash_), SOCKS5_VER, CMD_CONNECT, 0, ADDR_DOMAINNAME, len(hash_), hash_, 0)
self.transport.write(request)
+ self.state = STATE_CLIENT_REQUEST
def _parseRequestReply(self):
log.debug("_parseRequestReply")
@@ -294,49 +453,47 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.loseConnection()
return
- if self.factory.proxy:
- self.state = STATE_READY
- self.factory.activateCb(self.sid, self.factory.iq_id, self.startTransfer, self.profile)
- else:
- self.state = STATE_TARGET_READY
- self.factory.activateCb(self.sid, self.factory.iq_id, self.profile)
+ # if self.factory.proxy:
+ # self.state = STATE_READY
+ # self.factory.activateCb(self.sid, self.factory.iq_id, self.startTransfer, self.profile)
+ # else:
+ self.state = STATE_READY
+ self.connection.callback(None)
+ # self.factory.activateCb(self.sid, self.factory.iq_id, self.profile)
except struct.error:
+ # The buffer is probably not complete, we need to wait more
return None
def connectionMade(self):
- log.debug(u"connectionMade (mode = %s)" % "requester" if isinstance(self.factory, Socks5ServerFactory) else "target")
-
- if isinstance(self.factory, Socks5ClientFactory):
- self.sid = self.factory.sid
- self.profile = self.factory.profile
- self.data = self.factory.data
- self.state = STATE_TARGET_INITIAL
+ log.debug(u"Socks5 connectionMade (mode = {})".format("server" if self.state == STATE_INITIAL else "client"))
+ if self.state == STATE_CLIENT_INITIAL:
self._startNegotiation()
def connectRequested(self, addr, port):
log.debug("connectRequested")
# Check that this session is expected
- if addr not in self.factory.hash_sid_map:
- #no: we refuse it
+ if not self.factory.addToSession(addr, self):
self.sendErrorReply(REPLY_CONN_REFUSED)
+ log.warning(u"Unexpected connection request received from {host}"
+ .format(host=self.transport.getPeer().host))
return
- self.sid, self.profile = self.factory.hash_sid_map[addr]
- client = self.factory.host.getClient(self.profile)
- client.xep_0065_current_stream[self.sid]["start_transfer_cb"] = self.startTransfer
+ self._session_hash = addr
+ # self.sid, self.profile = self.factory.hash_profiles_map[addr]
+ # client = self.factory.host.getClient(self.profile)
+ # client.xep_0065_current_stream[self.sid]["start_transfer_cb"] = self.startTransfer
self.connectCompleted(addr, 0)
- self.transport.stopReading()
- def startTransfer(self, file_obj):
+ def startTransfer(self):
"""Callback called when the result iq is received"""
- d = self.beginFileTransfer(file_obj, self.transport)
+ log.debug(u"Starting file transfer")
+ d = self.beginFileTransfer(self.file_obj, self.transport)
d.addCallback(self.fileTransfered)
def fileTransfered(self, d):
log.info(_("File transfer completed, closing connection"))
self.transport.loseConnection()
- self.factory.finishedCb(self.sid, True, self.profile)
def connectCompleted(self, remotehost, remoteport):
log.debug("connectCompleted")
@@ -357,8 +514,9 @@ class SOCKSv5(protocol.Protocol, FileSender):
return True
def dataReceived(self, buf):
- if self.state == STATE_TARGET_READY:
- self.data["file_obj"].write(buf)
+ if self.state == STATE_READY:
+ # Everything is set, we just have to write the incoming data
+ self.file_obj.write(buf)
return
self.buf = self.buf + buf
@@ -368,216 +526,483 @@ class SOCKSv5(protocol.Protocol, FileSender):
self._parseUserPass()
if self.state == STATE_REQUEST:
self._parseRequest()
- if self.state == STATE_TARGET_AUTH:
+ if self.state == STATE_CLIENT_REQUEST:
+ self._parseRequestReply()
+ if self.state == STATE_CLIENT_AUTH:
ver, method = struct.unpack('!BB', buf)
self.buf = self.buf[2:]
if ver != SOCKS5_VER or method != AUTHMECH_ANON:
self.transport.loseConnection()
else:
self._makeRequest()
- if self.state == STATE_TARGET_REQUEST:
- self._parseRequestReply()
-
- def clientConnectionLost(self, reason):
- log.debug("clientConnectionLost")
- self.transport.loseConnection()
def connectionLost(self, reason):
- log.debug("connectionLost")
- if self.state != STATE_CONNECT_PENDING:
- self.transport.unregisterProducer()
- if self.peersock is not None:
- self.peersock.peersock = None
- self.peersock.transport.unregisterProducer()
- self.peersock = None
+ log.debug(u"Socks5 connection lost: {}".format(reason.value))
+ # self.transport.unregisterProducer()
+ # if self.peersock is not None:
+ # self.peersock.peersock = None
+ # self.peersock.transport.unregisterProducer()
+ # self.peersock = None
+ if self.state != STATE_READY:
+ self.connection.errback(reason)
+ if self.server_mode :
+ self.factory.removeFromSession(self._session_hash, self, reason)
class Socks5ServerFactory(protocol.ServerFactory):
protocol = SOCKSv5
- def __init__(self, host, hash_sid_map, finishedCb):
- self.host = host
- self.hash_sid_map = hash_sid_map
- self.finishedCb = finishedCb
+ def __init__(self, parent):
+ """
+ @param parent(XEP_0065): XEP_0065 parent instance
+ """
+ self.parent = parent
- def startedConnecting(self, connector):
- log.debug(_("Socks 5 server connection started"))
+ def getSession(self, session_hash):
+ return self.parent.getSession(session_hash, None)
- def clientConnectionLost(self, connector, reason):
- log.debug(_(u"Socks 5 server connection lost (reason: %s)") % reason)
+ def startTransfer(self, session_hash):
+ session = self.getSession(session_hash)
+ try:
+ protocol = session['protocols'][0]
+ except (KeyError, IndexError):
+ log.error(u"Can't start file transfer, can't find protocol")
+ else:
+ protocol.startTransfer()
+
+ def addToSession(self, session_hash, protocol):
+ """Check is session_hash is valid, and associate protocol with it
+
+ the session will be associated to the corresponding candidate
+ @param session_hash(str): hash of the session
+ @param protocol(SOCKSv5): protocol instance
+ @param return(bool): True if hash was valid (i.e. expected), False else
+ """
+ try:
+ session_data = self.getSession(session_hash)
+ except KeyError:
+ return False
+ else:
+ session_data.setdefault('protocols', []).append(protocol)
+ return True
+
+ def removeFromSession(self, session_hash, protocol, reason):
+ """Remove a protocol from session_data
+
+ There can be several protocol instances while candidates are tried, they
+ have removed when candidate connection is closed
+ @param session_hash(str): hash of the session
+ @param protocol(SOCKSv5): protocol instance
+ @param reason(failure.Failure): reason of the removal
+ """
+ try:
+ protocols = self.getSession(session_hash)['protocols']
+ protocols.remove(protocol)
+ except (KeyError, ValueError):
+ log.error(u"Protocol not found in session while it should be there")
+ else:
+ if not protocols:
+ # The last protocol has been removed, session is finished
+ if reason.check(internet_error.ConnectionDone):
+ self.getSession(session_hash)[DEFER_KEY].callback(None)
+ else:
+ self.getSession(session_hash)[DEFER_KEY].errback(reason)
class Socks5ClientFactory(protocol.ClientFactory):
protocol = SOCKSv5
- def __init__(self, current_stream, sid, iq_id, activateCb, finishedCb, proxy=False, profile=None):
+ # def __init__(self, stream_data, sid, iq_id, activateCb, finishedCb, proxy=False, profile=C.PROF_KEY_NONE):
+ def __init__(self, parent, session_hash, profile):
"""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)
- @param profile: %(doc_profile)s"""
- assert(profile)
- self.data = current_stream[sid]
- self.sid = sid
- self.iq_id = iq_id
- self.activateCb = activateCb
- self.finishedCb = finishedCb
- self.proxy = proxy
- self.profile = profile
- def startedConnecting(self, connector):
- log.debug(_("Socks 5 client connection started"))
+ @param session_hash(unicode): hash of the session
+ hash is the same as hostname computer in XEP-0065 § 5.3.2 #1
+ @param profile(unciode): %(doc_profile)s
+ """
+ self.session = parent.getSession(session_hash, profile)
+ self.session_hash = session_hash
+ self.profile = profile
+ self.connection = defer.Deferred()
+ self._protocol_instance = None
+ self.connector = None
+ self._discarded = False
+ # self.data = stream_data[sid]
+ # self.sid = sid
+ # self.iq_id = iq_id
+ # self.activateCb = activateCb
+ # self.finishedCb = finishedCb
+ # self.proxy = proxy
+ # self.profile = profile
+
+ def discard(self):
+ """Disconnect the client
+
+ Also set a discarded flag, which avoid to call the session Deferred
+ """
+ self.connector.disconnect()
+ self._discarded = True
+
+ def getSession(self):
+ return self.session
+
+ def startTransfer(self, dummy=None):
+ self._protocol_instance.startTransfer()
+
+ def clientConnectionFailed(self, connector, reason):
+ log.debug(u"Connection failed")
+ self.connection.errback(reason)
def clientConnectionLost(self, connector, reason):
- log.debug(_(u"Socks 5 client connection lost (reason: %s)") % reason)
- self.finishedCb(self.sid, reason.type == error.ConnectionDone, self.profile) # TODO: really check if the state is actually successful
+ log.debug(_(u"Socks 5 client connection lost (reason: %s)") % reason.value)
+ self._protocol_instance = None
+ if not self._discarded:
+ # This one was used for the transfer, than mean that
+ # the Socks5 session is finished
+ if reason.check(internet_error.ConnectionDone):
+ self.getSession()[DEFER_KEY].callback(None)
+ else:
+ self.getSession()[DEFER_KEY].errback(reason)
+ # self.finishedCb(self.sid, reason.type == internet_error.ConnectionDone, self.profile) # TODO: really check if the state is actually successful
+
+ def buildProtocol(self, addr):
+ log.debug(("Socks 5 client connection started"))
+ p = self.protocol(session_hash=self.session_hash)
+ p.factory = self
+ p.connection.chainDeferred(self.connection)
+ self._protocol_instance = p
+ return p
class XEP_0065(object):
-
NAMESPACE = NS_BS
-
- params = """
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- """
+ TYPE_DIRECT = 'direct'
+ TYPE_ASSISTED = 'assisted'
+ TYPE_TUNEL = 'tunel'
+ TYPE_PROXY = 'proxy'
+ Candidate = Candidate
def __init__(self, host):
log.info(_("Plugin XEP_0065 initialization"))
-
- #session data
- self.hash_sid_map = {} # key: hash of the transfer session, value: (session id, profile)
-
self.host = host
- log.debug(_("registering"))
- self.server_factory = Socks5ServerFactory(host, self.hash_sid_map, lambda sid, success, profile: self._killId(sid, success, profile=profile))
- #parameters
- host.memory.updateParams(XEP_0065.params)
- host.memory.setDefault("IP", "File Transfer", self.getExternalIP)
- port = int(self.host.memory.getParamA("Port", "File Transfer"))
+ # session data
+ self.hash_profiles_map = {} # key: hash of the transfer session, value: session data
+ self._cache_proxies = {} # key: server jid, value: proxy data
- log.info(_("Launching Socks5 Stream server on port %d") % port)
- reactor.listenTCP(port, self.server_factory)
+ # misc data
+ self._server_factory = None
+ self._external_port = None
+
+ # plugins shortcuts
+ self._ip = self.host.plugins['IP']
+ try:
+ self._np = self.host.plugins['NAT-PORT']
+ except KeyError:
+ log.debug(u"NAT Port plugin not available")
+ self._np = None
+
+ # parameters
+ # XXX: params are not used for now, but they may be used in the futur to force proxy/IP
+ # host.memory.updateParams(PARAMS)
def getHandler(self, profile):
return XEP_0065_handler(self)
def profileConnected(self, profile):
client = self.host.getClient(profile)
- client.xep_0065_current_stream = {} # key: stream_id, value: data(dict)
+ client.xep_0065_current_stream = {} # key: stream_id, value: session_data(dict)
+ client._s5b_sessions = {}
- def getExternalIP(self):
- """Return IP visible from outside, by asking to a website"""
- return getPage("http://www.goffi.org/sat_tools/get_ip.php")
+ def getSessionHash(self, from_jid, to_jid, sid):
+ return getSessionHash(from_jid, to_jid, sid)
- def getProgress(self, sid, data, profile):
- """Fill data with position of current transfer"""
+ def getSocks5ServerFactory(self):
+ """Return server factory
+
+ The server is created if it doesn't exists yet
+ self._server_factory_port is set on server creation
+ """
+
+ if self._server_factory is None:
+ # self._server_factory = Socks5ServerFactory(self.host, self.hash_profiles_map, lambda sid, client: self._killSession(sid, client))
+ self._server_factory = Socks5ServerFactory(self)
+ for port in xrange(SERVER_STARTING_PORT, 65356):
+ try:
+ listening_port = reactor.listenTCP(port, self._server_factory)
+ except internet_error.CannotListenError as e:
+ log.debug(u"Cannot listen on port {port}: {err_msg}{err_num}".format(
+ port=port,
+ err_msg=e.socketError.strerror,
+ err_num=u' (error code: {})'.format(e.socketError.errno),
+ ))
+ else:
+ self._server_factory_port = listening_port.getHost().port
+ break
+
+ log.info(_("Socks5 Stream server launched on port {}").format(self._server_factory_port))
+ return self._server_factory
+
+ @defer.inlineCallbacks
+ def getProxy(self, profile):
+ """Return the proxy available for this profile
+
+ cache is used between profiles using the same server
+ @param profile: %(doc_profile)s
+ @return ((D)(ProxyInfos, None)): Found proxy infos,
+ or None if not acceptable proxy is found
+ """
+ def notFound(server):
+ log.info(u"No proxy found on this server")
+ self._cache_proxies[server] = None
+ defer.returnValue(None)
client = self.host.getClient(profile)
+ server = client.jid.host
try:
- file_obj = client.xep_0065_current_stream[sid]["file_obj"]
- data["position"] = str(file_obj.tell())
- data["size"] = str(client.xep_0065_current_stream[sid]["size"])
- except:
+ defer.returnValue(self._cache_proxies[server])
+ except KeyError:
pass
+ try:
+ proxy = (yield self.host.findServiceEntities('proxy', 'bytestreams', profile=profile)).pop()
+ except (exceptions.CancelError, StopIteration):
+ notFound(server)
+ iq_elt = client.IQ('get')
+ iq_elt['to'] = proxy.full()
+ iq_elt.addElement('query', NS_BS)
- def _timeOut(self, sid, profile):
+ try:
+ result_elt = yield iq_elt.send()
+ except jabber_error.StanzaError as failure:
+ log.warning(u"Error while requesting proxy info on {jid}: {error}"
+ .format(proxy.full(), failure))
+ notFound(server)
+
+ try:
+ query_elt = result_elt.elements(NS_BS, 'query').next()
+ streamhost_elt = query_elt.elements(NS_BS, 'streamhost').next()
+ host = streamhost_elt['host']
+ jid_ = streamhost_elt['jid']
+ port = streamhost_elt['port']
+ if not all((host, jid, port)):
+ raise KeyError
+ jid_ = jid.JID(jid_)
+ except (StopIteration, KeyError, RuntimeError, jid.InvalidFormat):
+ log.warning(u"Invalid proxy data received from {}".format(proxy.full()))
+ notFound(server)
+
+ proxy_infos = self._cache_proxies[server] = ProxyInfos(host, jid_, port)
+ log.info(u"Proxy found: {}".format(proxy_infos))
+ defer.returnValue(proxy_infos)
+
+ @defer.inlineCallbacks
+ def _getNetworkData(self, client):
+ """Retrieve information about network
+
+ @param client: %(doc_client)s
+ @return (D(tuple[local_port, external_port, local_ips, external_ip])): network data
+ """
+ self.getSocks5ServerFactory()
+ local_port = self._server_factory_port
+ external_ip = yield self._ip.getExternalIP(client.profile)
+ local_ips = yield self._ip.getLocalIPs(client.profile)
+
+ if not local_ips:
+ log.warning(u"Can't find local IPs, we can't do direct connection")
+ else:
+ if external_ip is not None and self._external_port is None:
+ if external_ip != local_ips[0]:
+ log.info(u"We are probably behind a NAT")
+ if self._np is None:
+ log.warning(u"NAT port plugin not available, we can't map port")
+ else:
+ ext_port = yield self._np.mapPort(local_port, desc=u"SaT socks5 stream")
+ if ext_port is None:
+ log.warning(u"Can't map NAT port")
+ else:
+ self._external_port = ext_port
+
+ defer.returnValue((local_port, self._external_port, local_ips, external_ip))
+
+ @defer.inlineCallbacks
+ def getCandidates(self, profile):
+ """Return a list of our stream candidates
+
+ @param profile: %(doc_profile)s
+ @return (D(list[Candidate])): list of candidates, ordered by priority
+ """
+ client = self.host.getClient(profile)
+ server_factory = yield self.getSocks5ServerFactory()
+ local_port, ext_port, local_ips, external_ip = yield self._getNetworkData(client)
+ proxy = yield self.getProxy(profile)
+
+ # its time to gather the candidates
+ candidates = []
+
+ # first the direct ones
+ if local_ips:
+ # the preferred direct connection
+ ip = local_ips.pop(0)
+ candidates.append(Candidate(ip, local_port, XEP_0065.TYPE_DIRECT, PRIORITY_BEST_DIRECT, client.jid, priority_local=True, factory=server_factory))
+ for ip in local_ips:
+ candidates.append(Candidate(ip, local_port, XEP_0065.TYPE_DIRECT, PRIORITY_DIRECT, client.jid, priority_local=True, factory=server_factory))
+
+ # then the assisted one
+ if ext_port is not None:
+ candidates.append(Candidate(external_ip, ext_port, XEP_0065.TYPE_ASSISTED, PRIORITY_ASSISTED, client.jid, priority_local=True, factory=server_factory))
+
+ # finally the proxy
+ if proxy:
+ candidates.append(Candidate(proxy.host, proxy.port, XEP_0065.TYPE_PROXY, PRIORITY_PROXY, proxy.jid, priority_local=True))
+
+ # should be already sorted, but just in case the priorities get weird
+ candidates.sort(key=lambda c: c.priority, reverse=True)
+
+ defer.returnValue(candidates)
+
+ def _addConnector(self, connector, candidate):
+ """Add connector used to connect to candidate, and return client factory's connection Deferred
+
+ the connector can be used to disconnect the candidate, and returning the factory's connection Deferred allow to wait for connection completion
+ @param connector: a connector implementing IConnector
+ @param candidate(Candidate): candidate linked to the connector
+ @return (D): Deferred fired when factory connection is done or has failed
+ """
+ candidate.factory.connector = connector
+ return candidate.factory.connection
+
+ def tryCandidates(self, candidates, session_hash, connection_cb=None, connection_eb=None, profile=C.PROF_KEY_NONE):
+ defers_list = []
+
+ for candidate in candidates:
+ factory = Socks5ClientFactory(self, session_hash, profile)
+ candidate.factory = factory
+ delay = CANDIDATE_DELAY * len(defers_list)
+ if candidate.type == XEP_0065.TYPE_PROXY:
+ delay += CANDIDATE_DELAY_PROXY
+ d = sat_defer.DelayedDeferred(delay, candidate.host)
+ d.addCallback(reactor.connectTCP, candidate.port, factory)
+ d.addCallback(self._addConnector, candidate)
+ if connection_cb is not None:
+ d.addCallback(lambda dummy, candidate=candidate, profile=profile: connection_cb(candidate, profile))
+ if connection_eb is not None:
+ d.addErrback(connection_eb, candidate, profile)
+ defers_list.append(d)
+
+ return defers_list
+
+ def getBestCandidate(self, candidates, session_hash, profile=C.PROF_KEY_NONE):
+ defer_candidates = None
+
+ def connectionCb(candidate, profile):
+ log.info(u"Connection of {} successful".format(unicode(candidate)))
+ for idx, other_candidate in enumerate(candidates):
+ try:
+ if other_candidate.priority < candidate.priority:
+ log.debug(u"Cancelling {}".format(other_candidate))
+ defer_candidates[idx].cancel()
+ except AttributeError:
+ assert other_candidate is None
+
+ def connectionEb(failure, candidate, profile):
+ if failure.check(defer.CancelledError):
+ log.debug(u"Connection of {} has been cancelled".format(candidate))
+ else:
+ log.info(u"Connection of {candidate} Failed: {error}".format(
+ candidate = candidate,
+ error = failure.value))
+ candidates[candidates.index(candidate)] = None
+
+ def allTested(self):
+ log.debug(u"All candidates have been tested")
+ good_candidates = [c for c in candidates if c]
+ return good_candidates[0] if good_candidates else None
+
+ defer_candidates = self.tryCandidates(candidates, session_hash, connectionCb, connectionEb, profile)
+ d_list = defer.DeferredList(defer_candidates)
+ d_list.addCallback(allTested)
+ return d_list
+
+ def _timeOut(self, sid, client):
"""Delecte current_stream id, called after timeout
@param id: id of client.xep_0065_current_stream"""
- log.info(_("Socks5 Bytestream: TimeOut reached for id %(sid)s [%(profile)s]")
- % {"sid": sid, "profile": profile})
- self._killId(sid, False, "TIMEOUT", profile)
+ log.info(_("Socks5 Bytestream: TimeOut reached for id {sid} [{profile}]").format(
+ sid=sid, profile=client.profile))
+ self._killSession(sid, client, u"TIMEOUT")
- def _killId(self, sid, success=False, failure_reason="UNKNOWN", profile=None):
- """Delete an current_stream id, clean up associated observers
- @param sid: id of client.xep_0065_current_stream"""
- assert(profile)
- client = self.host.getClient(profile)
- if sid not in client.xep_0065_current_stream:
+ def _killSession(self, sid, client, failure_reason=None):
+ """Delete a current_stream id, clean up associated observers
+
+ @param sid(unicode): session id
+ @param client: %(doc_client)s
+ @param failure_reason(None, unicode): if None the session is successful
+ else, will be used to call failure_cb
+ """
+ try:
+ session = client.xep_0065_current_stream[sid]
+ except KeyError:
log.warning(_("kill id called on a non existant id"))
return
- if "observer_cb" in client.xep_0065_current_stream[sid]:
- xmlstream = client.xep_0065_current_stream[sid]["xmlstream"]
- xmlstream.removeObserver(client.xep_0065_current_stream[sid]["event_data"], client.xep_0065_current_stream[sid]["observer_cb"])
- if client.xep_0065_current_stream[sid]['timer'].active():
- client.xep_0065_current_stream[sid]['timer'].cancel()
- if "size" in client.xep_0065_current_stream[sid]:
- self.host.removeProgressCB(sid, profile)
- file_obj = client.xep_0065_current_stream[sid]['file_obj']
- success_cb = client.xep_0065_current_stream[sid]['success_cb']
- failure_cb = client.xep_0065_current_stream[sid]['failure_cb']
+ try:
+ observer_cb = session['observer_cb']
+ except KeyError:
+ pass
+ else:
+ client.xmlstream.removeObserver(session["event_data"], observer_cb)
+
+ if session['timer'].active():
+ session['timer'].cancel()
- session_hash = client.xep_0065_current_stream[sid].get('hash')
del client.xep_0065_current_stream[sid]
- if session_hash in self.hash_sid_map:
- #FIXME: check that self.hash_sid_map is correctly cleaned in all cases (timeout, normal flow, etc).
- del self.hash_sid_map[session_hash]
+
+ # FIXME: to check
+ try:
+ session_hash = session.get['hash']
+ del self.hash_profiles_map[session_hash]
+ # FIXME: check that self.hash_profiles_map is correctly cleaned in all cases (timeout, normal flow, etc).
+ except KeyError:
+ log.debug(u"Not hash found for this session")
+ pass
+
+ success = failure_reason is None
+ stream_d = session[DEFER_KEY]
if success:
- success_cb(sid, file_obj, NS_BS, profile)
+ stream_d.callback(None)
else:
- failure_cb(sid, file_obj, NS_BS, failure_reason, profile)
+ stream_d.errback(failure.Failure(exceptions.DataError(failure_reason)))
- def startStream(self, file_obj, to_jid, sid, length, successCb, failureCb, size=None, profile=None):
+ def startStream(self, file_obj, to_jid, sid, profile=C.PROF_KEY_NONE):
"""Launch the stream workflow
+
@param file_obj: file_obj to send
@param to_jid: JID of the recipient
@param sid: Stream session id
- @param length: number of byte to send, or None to send until the end
@param successCb: method to call when stream successfuly finished
@param failureCb: method to call when something goes wrong
- @param profile: %(doc_profile)s"""
- assert(profile)
+ @param profile: %(doc_profile)s
+ """
client = self.host.getClient(profile)
+ session_data = self._createSession(file_obj, to_jid, sid, client.profile)
- if length is not None:
- log.error(_('stream length not managed yet'))
- return
+ session_data["to"] = to_jid
+ session_data["xmlstream"] = client.xmlstream
+ hash_ = session_data["hash"] = getSessionHash(client.jid, to_jid, sid)
- profile_jid = client.jid
- xmlstream = client.xmlstream
+ self.hash_profiles_map[hash_] = (sid, profile)
- data = client.xep_0065_current_stream[sid] = {}
- data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid, profile)
- data["file_obj"] = file_obj
- data["from"] = profile_jid
- data["to"] = to_jid
- data["success_cb"] = successCb
- data["failure_cb"] = failureCb
- data["xmlstream"] = xmlstream
- data["hash"] = calculateHash(profile_jid, to_jid, sid)
- self.hash_sid_map[data["hash"]] = (sid, profile)
- if size:
- data["size"] = size
- self.host.registerProgressCB(sid, self.getProgress, profile)
- iq_elt = jabber_client.IQ(xmlstream, 'set')
- iq_elt["from"] = profile_jid.full()
+ iq_elt = jabber_client.IQ(client.xmlstream, 'set')
+ iq_elt["from"] = client.jid.full()
iq_elt["to"] = to_jid.full()
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'] = self.host.memory.getParamA("IP", "File Transfer")
streamhost['port'] = self.host.memory.getParamA("Port", "File Transfer")
- streamhost['jid'] = profile_jid.full()
+ streamhost['jid'] = client.jid.full()
#second streamhost: mediated connection, using proxy
streamhost = query_elt.addElement('streamhost')
@@ -585,60 +1010,68 @@ class XEP_0065(object):
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, profile)
+ iq_elt.addCallback(self._IQOpen, session_data, client)
iq_elt.send()
+ return session_data[DEFER_KEY]
- def iqResult(self, sid, profile, iq_elt):
- """Called when the result of open iq is received"""
+ def _IQOpen(self, session_data, client, iq_elt):
+ """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): result
+ """
+ sid = session_data['id']
if iq_elt["type"] == "error":
- log.warning(_("Transfer failed"))
+ log.warning(_("Socks5 transfer failed"))
+ # FIXME: must clean session
return
- client = self.host.getClient(profile)
+
try:
- data = client.xep_0065_current_stream[sid]
- file_obj = data["file_obj"]
- timer = data["timer"]
+ session_data = client.xep_0065_current_stream[sid]
+ file_obj = session_data["file_obj"]
+ timer = session_data["timer"]
except KeyError:
- log.error(_("Internal error, can't do transfer"))
- return
+ raise exceptions.InternalError
- if timer.active():
- timer.cancel()
+ timer.reset(TIMEOUT)
+
+ query_elt = iq_elt.elements(NS_BS, 'query').next()
+ streamhost_elts = list(query_elt.elements(NS_BS, 'streamhost-used'))
- 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:
log.warning(_("No streamhost found in stream query"))
+ # FIXME: must clean session
return
+ # FIXME: must be cleaned !
+
streamhost_jid = streamhost_elts[0]['jid']
- if streamhost_jid != profile_jid.full():
+ if streamhost_jid != client.jid.full():
log.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)
+ proxy_host = self.host.memory.getParamA("Proxy host", "File Transfer", profile_key=client.profile)
+ proxy_port = self.host.memory.getParamA("Proxy port", "File Transfer", profile_key=client.profile)
+ proxy_jid = self.host.memory.getParamA("Proxy", "File Transfer", profile_key=client.profile)
if proxy_jid != streamhost_jid:
log.warning(_("Proxy jid is not the same as in parameters, this should not happen"))
return
- factory = Socks5ClientFactory(client.xep_0065_current_stream, sid, None, self.activateProxyStream, lambda sid, success, profile: self._killId(sid, success, profile=profile), True, profile)
+ factory = Socks5ClientFactory(client.xep_0065_current_stream, sid, None, self.activateProxyStream, lambda sid, success, profile: self._killSession(sid, client), True, client.profile)
reactor.connectTCP(proxy_host, int(proxy_port), factory)
else:
- data["start_transfer_cb"](file_obj) # We now activate the stream
+ session_data["start_transfer_cb"](file_obj) # We now activate the stream
def activateProxyStream(self, sid, iq_id, start_transfer_cb, profile):
log.debug(_("activating stream"))
client = self.host.getClient(profile)
- data = client.xep_0065_current_stream[sid]
- profile_jid, xmlstream = self.host.getJidNStream(profile)
+ session_data = client.xep_0065_current_stream[sid]
- iq_elt = client.IQ(xmlstream, 'set')
- iq_elt["from"] = profile_jid.full()
+ iq_elt = client.IQ(client.xmlstream, 'set')
+ iq_elt["from"] = client.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'])
+ query_elt.addElement('activate', content=session_data['to'].full())
+ iq_elt.addCallback(self.proxyResult, sid, start_transfer_cb, session_data['file_obj'])
iq_elt.send()
def proxyResult(self, sid, start_transfer_cb, file_obj, iq_elt):
@@ -648,26 +1081,95 @@ class XEP_0065(object):
else:
start_transfer_cb(file_obj)
- def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb, profile):
+ 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, file_obj, to_jid, sid, profile):
"""Called when a bytestream is imminent
- @param from_jid: jid of the sender
- @param sid: Stream id
- @param file_obj: File object where data will be written
- @param size: full size of the data, or None if unknown
- @param success_cb: method to call when successfuly finished
- @param failure_cb: method to call when something goes wrong
- @param profile: %(doc_profile)s"""
+
+ @param file_obj(file): File object where data will be written
+ @param to_jid(jid.JId): jid of the other peer
+ @param sid(unicode): session id
+ @param profile: %(doc_profile)s
+ @return (dict): session data
+ """
client = self.host.getClient(profile)
- data = client.xep_0065_current_stream[sid] = {}
- data["from"] = from_jid
- data["file_obj"] = file_obj
- data["seq"] = -1
- if size:
- data["size"] = size
- self.host.registerProgressCB(sid, self.getProgress, profile)
- data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid, profile)
- data["success_cb"] = success_cb
- data["failure_cb"] = failure_cb
+ if sid in client.xep_0065_current_stream:
+ raise exceptions.ConflictError(u'A session with this id already exists !')
+ session_data = client.xep_0065_current_stream[sid] = \
+ {'id': sid,
+ DEFER_KEY: defer.Deferred(),
+ 'to': to_jid,
+ 'file_obj': file_obj,
+ 'seq': -1, # FIXME: to check
+ 'timer': reactor.callLater(TIMEOUT, self._timeOut, sid, client),
+ }
+
+ return session_data
+
+ def getSession(self, session_hash, profile):
+ """Return session data
+
+ @param session_hash(unicode): hash of the session
+ hash is the same as hostname computer in XEP-0065 § 5.3.2 #1
+ @param profile(None, unicode): profile of the peer
+ None is used only if profile is unknown (this is only the case
+ for incoming request received by Socks5ServerFactory). None must
+ only be used by Socks5ServerFactory.
+ See comments below for details
+ @return (dict): session data
+ """
+ if profile is None:
+ try:
+ profile = self.hash_profiles_map[session_hash]
+ except KeyError as e:
+ log.warning(u"The requested session doesn't exists !")
+ raise e
+ client = self.host.getClient(profile)
+ return client._s5b_sessions[session_hash]
+
+ def registerHash(self, *args, **kwargs):
+ """like [_registerHash] but resutrn the session deferred instead of the whole session
+ session deferred is fired when transfer is finished
+ """
+ return self._registerHash(*args, **kwargs)[DEFER_KEY]
+
+ def _registerHash(self, session_hash, file_obj, profile):
+ """Create a session_data associated to hash
+
+ @param session_hash(str): hash of the session
+ @param file_obj(file): file-like object
+ @param profile: %(doc_profile)s
+ return (dict): session data
+ """
+ client = self.host.getClient(profile)
+ assert session_hash not in client._s5b_sessions
+ session_data = client._s5b_sessions[session_hash] = {
+ "file": file_obj,
+ DEFER_KEY: defer.Deferred(),
+ }
+ if session_hash in self.hash_profiles_map:
+ # The only case when 2 profiles want to register the same hash
+ # is when they are on the same instance
+ log.info(u"Both Socks5 peers are on the same instance")
+ # XXX:If both peers are on the same instance, they'll register the same
+ # session_hash, so we'll have 2 profiles for the same hash. The first
+ # one will be the responder (and so the second one the initiator).
+ # As we'll keep the initiator choosed candidate (see XEP-0260 § 2.4 #4),
+ # responder will handle the Socks5 server. Only the server will use
+ # self.hash_profiles_map to get the profile, so we can ignore the second
+ # one (the initiator profile).
+ # There is no easy way to known if the incoming connection
+ # to the Socks5Server is from initiator or responder, so this seams a
+ # reasonable workaround.
+ else:
+ self.hash_profiles_map[session_hash] = profile
+
+ return session_data
def streamQuery(self, iq_elt, profile):
"""Get file using byte stream"""
@@ -710,23 +1212,23 @@ class XEP_0065(object):
client.xep_0065_current_stream[sid]["streamhost"] = (sh_host, sh_port, sh_jid)
log.info(_("Stream proposed: host=[%(host)s] port=[%(port)s]") % {'host': sh_host, 'port': sh_port})
- factory = Socks5ClientFactory(client.xep_0065_current_stream, sid, iq_elt["id"], self.activateStream, lambda sid, success, profile: self._killId(sid, success, profile=profile), profile=profile)
+ factory = Socks5ClientFactory(client.xep_0065_current_stream, sid, iq_elt["id"], self.activateStream, lambda sid, success, profile: self._killSession(sid, client), profile=profile)
reactor.connectTCP(sh_host, int(sh_port), factory)
def activateStream(self, sid, iq_id, profile):
client = self.host.getClient(profile)
log.debug(_("activating stream"))
result = domish.Element((None, 'iq'))
- data = client.xep_0065_current_stream[sid]
+ session_data = client.xep_0065_current_stream[sid]
result['type'] = 'result'
result['id'] = iq_id
- result['from'] = data["to"].full()
- result['to'] = data["from"].full()
+ result['from'] = session_data["to"].full()
+ result['to'] = session_data["from"].full()
query = result.addElement('query', NS_BS)
query['sid'] = sid
streamhost = query.addElement('streamhost-used')
- streamhost['jid'] = data["streamhost"][2]
- data["xmlstream"].send(result)
+ streamhost['jid'] = session_data["streamhost"][2]
+ session_data["xmlstream"].send(result)
def sendNotAcceptableError(self, iq_id, to_jid, xmlstream):
"""Not acceptable error used when the stream is not expected or something is going wrong
@@ -764,50 +1266,8 @@ class XEP_0065_handler(XMPPHandler):
self.plugin_parent = plugin_parent
self.host = plugin_parent.host
- def _proxyDataResult(self, iq_elt):
- """Called with the information about proxy according to XEP-0065 #4
- Params should be filled with these infos"""
- if iq_elt["type"] == "error":
- log.warning(_("Can't determine proxy information"))
- return
- query_elt = iq_elt.firstChildElement()
- if query_elt.name != "query":
- log.warning(_("Bad answer received from proxy"))
- return
- streamhost_elts = filter(lambda elt: elt.name == 'streamhost', query_elt.elements())
- if not streamhost_elts:
- log.warning(_("No streamhost found in stream query"))
- return
- if len(streamhost_elts) != 1:
- log.warning(_("Multiple streamhost elements in proxy not managed, keeping only the first one"))
- streamhost_elt = streamhost_elts[0]
- self.host.memory.setParam("Proxy", streamhost_elt.getAttribute("jid", ""),
- "File Transfer", profile_key=self.parent.profile)
- self.host.memory.setParam("Proxy host", streamhost_elt.getAttribute("host", ""),
- "File Transfer", profile_key=self.parent.profile)
- self.host.memory.setParam("Proxy port", streamhost_elt.getAttribute("port", ""),
- "File Transfer", profile_key=self.parent.profile)
-
def connectionInitialized(self):
- def connection_ok(dummy):
- 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:
- def proxiesFound(entities):
- try:
- proxy_ent = entities.pop()
- except KeyError:
- log.info(_("No proxy found on this server"))
- return
- iq_elt = jabber_client.IQ(self.parent.xmlstream, 'get')
- iq_elt["to"] = proxy_ent.full()
- iq_elt.addElement('query', NS_BS)
- iq_elt.addCallback(self._proxyDataResult)
- iq_elt.send()
- d = self.host.findServiceEntities("proxy", "bytestreams", profile_key=self.parent.profile)
- d.addCallback(proxiesFound)
- self.parent.getConnectionDeferred().addCallback(connection_ok)
-
+ self.xmlstream.addObserver(BS_REQUEST, self.plugin_parent.streamQuery, profile=self.parent.profile)
def getDiscoInfo(self, requestor, target, nodeIdentifier=''):
return [disco.DiscoFeature(NS_BS)]
diff --git a/src/plugins/plugin_xep_0096.py b/src/plugins/plugin_xep_0096.py
index f6919a15..b890acad 100755
--- a/src/plugins/plugin_xep_0096.py
+++ b/src/plugins/plugin_xep_0096.py
@@ -155,23 +155,23 @@ class XEP_0096(object):
dest_path = frontend_data['dest_path']
except KeyError:
log.error(_('dest path not found in frontend_data'))
- del(client._xep_0096_waiting_for_approval[sid])
+ del client._xep_0096_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']), sid, file_obj, int(data["size"]), self._transferSucceeded, self._transferFailed, profile)
+ plugin = self.host.plugins["XEP-0065"]
elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE:
- file_obj = self._getFileObject(dest_path, can_range)
- range_offset = file_obj.tell()
- d = self.host.plugins["XEP-0047"].createSession(file_obj, jid.JID(data['from']), sid, int(data["size"]), profile)
- d.addCallback(self._transferSucceeded, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile)
- d.addErrback(self._transferFailed, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile)
+ plugin = self.host.plugins["XEP-0047"]
else:
log.error(_("Unknown stream method, this should not happen at this stage, cancelling transfer"))
- del(client._xep_0096_waiting_for_approval[sid])
+ del client._xep_0096_waiting_for_approval[sid]
return
+ file_obj = self._getFileObject(dest_path, can_range)
+ range_offset = file_obj.tell()
+ d = plugin.createSession(file_obj, jid.JID(data['from']), sid, int(data["size"]), profile)
+ d.addCallback(self._transferSucceeded, sid, file_obj, stream_method, profile)
+ d.addErrback(self._transferFailed, sid, file_obj, stream_method, profile)
+
#we can send the iq result
feature_elt = self.host.plugins["XEP-0020"].chooseOption({'stream-method': stream_method})
misc_elts = []
@@ -249,27 +249,27 @@ class XEP_0096(object):
return
range_offset = 0
- range_length = None
+ # range_length = None
range_elts = filter(lambda elt: elt.name == 'range', si_elt.elements())
if range_elts:
range_elt = range_elts[0]
range_offset = range_elt.getAttribute("offset", 0)
- range_length = range_elt.getAttribute("length")
+ # range_length = range_elt.getAttribute("length")
if stream_method == self.host.plugins["XEP-0065"].NAMESPACE:
- file_obj = open(filepath, 'r')
- if range_offset:
- file_obj.seek(range_offset)
- self.host.plugins["XEP-0065"].startStream(file_obj, jid.JID(IQ['from']), sid, range_length, self._sendSuccessCb, self._sendFailureCb, size, profile)
+ plugin = self.host.plugins["XEP-0065"]
elif stream_method == self.host.plugins["XEP-0047"].NAMESPACE:
- file_obj = open(filepath, 'r')
- if range_offset:
- file_obj.seek(range_offset)
- d = self.host.plugins["XEP-0047"].startStream(file_obj, jid.JID(IQ['from']), sid, profile=profile)
- d.addCallback(self.sendSuccessCb, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile)
- d.addErrback(self.sendFailureCb, sid, file_obj, self.host.plugins["XEP-0047"].NAMESPACE, profile)
+ plugin = self.host.plugins["XEP-0047"]
else:
- log.warning(_("Invalid stream method received"))
+ log.error(u"Invalid stream method received")
+ return
+
+ file_obj = open(filepath, 'r')
+ if range_offset:
+ file_obj.seek(range_offset)
+ d = plugin.startStream(file_obj, jid.JID(IQ['from']), sid, profile=profile)
+ d.addCallback(self.sendSuccessCb, sid, file_obj, stream_method, profile)
+ d.addErrback(self.sendFailureCb, sid, file_obj, stream_method, profile)
def sendFile(self, to_jid, filepath, data={}, profile_key=C.PROF_KEY_NONE):
"""send a file using XEP-0096