erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
libervia-backend/src/plugins/plugin_xep_0065.py
Goffi 5a00dc9503 core: i18n refactoring: 
- _() is no more installed in __builtin__
    - instead, there is a new sat.core.i18n module
    - added D_() method for deferred translation
    - languageSwitch method allow to dynamically change translation language
    - import gettext is tested against ImportError, and dummy methods are used when not available (mainly useful for Libervia)
2013-12-29 17:06:01 +01:00

830 lines
33 KiB
Python
Executable file
Raw Blame History

#!/usr/bin/python
#-*- coding: utf-8 -*-
# SAT plugin for managing xep-0065
# Copyright (C)
# 2002, 2003, 2004 Dave Smith (dizzyd@jabber.org)
# 2007, 2008 Fabio Forno (xmpp:ff@jabber.bluendo.com)
# 2009, 2010, 2011, 2012, 2013 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# --
# This program 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.
# --
# Here is a copy of the original license:
# Copyright (C)
# 2002-2004 Dave Smith (dizzyd@jabber.org)
# 2007-2008 Fabio Forno (xmpp:ff@jabber.bluendo.com)
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from sat.core.i18n import _
from logging import debug, info, warning, error
from twisted.internet import protocol, reactor
from twisted.internet import error as jab_error
from twisted.words.protocols.jabber import jid, client as jabber_client
from twisted.protocols.basic import FileSender
from twisted.words.xish import domish
from twisted.web.client import getPage
from sat.core.exceptions import ProfileNotInCacheError
import struct
import hashlib
from zope.interface import implements
try:
from twisted.words.protocols.xmlstream import XMPPHandler
except ImportError:
from wokkel.subprotocols import XMPPHandler
from wokkel import disco, iwokkel
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
PLUGIN_INFO = {
"name": "XEP 0065 Plugin",
"import_name": "XEP-0065",
"type": "XEP",
"protocols": ["XEP-0065"],
"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
STATE_CONNECT_PENDING = STATE_LAST + 1
SOCKS5_VER = 0x05
ADDR_IPV4 = 0x01
ADDR_DOMAINNAME = 0x03
ADDR_IPV6 = 0x04
CMD_CONNECT = 0x01
CMD_BIND = 0x02
CMD_UDPASSOC = 0x03
AUTHMECH_ANON = 0x00
AUTHMECH_USERPASS = 0x02
AUTHMECH_INVALID = 0xFF
REPLY_SUCCESS = 0x00
REPLY_GENERAL_FAILUR = 0x01
REPLY_CONN_NOT_ALLOWED = 0x02
REPLY_NETWORK_UNREACHABLE = 0x03
REPLY_HOST_UNREACHABLE = 0x04
REPLY_CONN_REFUSED = 0x05
REPLY_TTL_EXPIRED = 0x06
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)"""
return hashlib.sha1((sid + from_jid.full() + to_jid.full()).encode('utf-8')).hexdigest()
class SOCKSv5(protocol.Protocol, FileSender):
def __init__(self):
debug(_("Protocol init"))
self.state = STATE_INITIAL
self.buf = ""
self.supportedAuthMechs = [AUTHMECH_ANON]
self.supportedAddrs = [ADDR_DOMAINNAME]
self.enabledCommands = [CMD_CONNECT]
self.peersock = None
self.addressType = 0
self.requestType = 0
def _startNegotiation(self):
debug("_startNegotiation")
self.state = STATE_TARGET_AUTH
self.transport.write(struct.pack('!3B', SOCKS5_VER, 1, AUTHMECH_ANON))
def _parseNegotiation(self):
debug("_parseNegotiation")
try:
# Parse out data
ver, nmethod = struct.unpack('!BB', self.buf[:2])
methods = struct.unpack('%dB' % nmethod, self.buf[2:nmethod + 2])
# Ensure version is correct
if ver != 5:
self.transport.write(struct.pack('!BB', SOCKS5_VER, AUTHMECH_INVALID))
self.transport.loseConnection()
return
# Trim off front of the buffer
self.buf = self.buf[nmethod + 2:]
# Check for supported auth mechs
for m in self.supportedAuthMechs:
if m in methods:
# Update internal state, according to selected method
if m == AUTHMECH_ANON:
self.state = STATE_REQUEST
elif m == AUTHMECH_USERPASS:
self.state = STATE_AUTH_USERPASS
# Complete negotiation w/ this method
self.transport.write(struct.pack('!BB', SOCKS5_VER, m))
return
# No supported mechs found, notify client and close the connection
self.transport.write(struct.pack('!BB', SOCKS5_VER, AUTHMECH_INVALID))
self.transport.loseConnection()
except struct.error:
pass
def _parseUserPass(self):
debug("_parseUserPass")
try:
# Parse out data
ver, ulen = struct.unpack('BB', self.buf[:2])
uname, = struct.unpack('%ds' % ulen, self.buf[2:ulen + 2])
plen, = struct.unpack('B', self.buf[ulen + 2])
password, = struct.unpack('%ds' % plen, self.buf[ulen + 3:ulen + 3 + plen])
# Trim off fron of the buffer
self.buf = self.buf[3 + ulen + plen:]
# Fire event to authenticate user
if self.authenticateUserPass(uname, password):
# Signal success
self.state = STATE_REQUEST
self.transport.write(struct.pack('!BB', SOCKS5_VER, 0x00))
else:
# Signal failure
self.transport.write(struct.pack('!BB', SOCKS5_VER, 0x01))
self.transport.loseConnection()
except struct.error:
pass
def sendErrorReply(self, errorcode):
debug("sendErrorReply")
# Any other address types are not supported
result = struct.pack('!BBBBIH', SOCKS5_VER, errorcode, 0, 1, 0, 0)
self.transport.write(result)
self.transport.loseConnection()
def _parseRequest(self):
debug("_parseRequest")
try:
# Parse out data and trim buffer accordingly
ver, cmd, rsvd, self.addressType = struct.unpack('!BBBB', self.buf[:4])
# Ensure we actually support the requested address type
if self.addressType not in self.supportedAddrs:
self.sendErrorReply(REPLY_ADDR_NOT_SUPPORTED)
return
# Deal with addresses
if self.addressType == ADDR_IPV4:
addr, port = struct.unpack('!IH', self.buf[4:10])
self.buf = self.buf[10:]
elif self.addressType == ADDR_DOMAINNAME:
nlen = ord(self.buf[4])
addr, port = struct.unpack('!%dsH' % nlen, self.buf[5:])
self.buf = self.buf[7 + len(addr):]
else:
# Any other address types are not supported
self.sendErrorReply(REPLY_ADDR_NOT_SUPPORTED)
return
# Ensure command is supported
if cmd not in self.enabledCommands:
# Send a not supported error
self.sendErrorReply(REPLY_CMD_NOT_SUPPORTED)
return
# Process the command
if cmd == CMD_CONNECT:
self.connectRequested(addr, port)
elif cmd == CMD_BIND:
self.bindRequested(addr, port)
else:
# Any other command is not supported
self.sendErrorReply(REPLY_CMD_NOT_SUPPORTED)
except struct.error, why:
return None
def _makeRequest(self):
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)
self.transport.write(request)
def _parseRequestReply(self):
debug("_parseRequestReply")
try:
ver, rep, rsvd, self.addressType = struct.unpack('!BBBB', self.buf[:4])
# Ensure we actually support the requested address type
if self.addressType not in self.supportedAddrs:
self.sendErrorReply(REPLY_ADDR_NOT_SUPPORTED)
return
# Deal with addresses
if self.addressType == ADDR_IPV4:
addr, port = struct.unpack('!IH', self.buf[4:10])
self.buf = self.buf[10:]
elif self.addressType == ADDR_DOMAINNAME:
nlen = ord(self.buf[4])
addr, port = struct.unpack('!%dsH' % nlen, self.buf[5:])
self.buf = self.buf[7 + len(addr):]
else:
# Any other address types are not supported
self.sendErrorReply(REPLY_ADDR_NOT_SUPPORTED)
return
# Ensure reply is OK
if rep != REPLY_SUCCESS:
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)
except struct.error, why:
return None
def connectionMade(self):
debug("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
self._startNegotiation()
def connectRequested(self, addr, port):
debug("connectRequested")
# Check that this session is expected
if addr not in self.factory.hash_sid_map:
#no: we refuse it
self.sendErrorReply(REPLY_CONN_REFUSED)
return
self.sid, self.profile = self.factory.hash_sid_map[addr]
client = self.factory.host.getClient(self.profile)
if not client:
raise ProfileNotInCacheError
client.xep_0065_current_stream[self.sid]["start_transfer_cb"] = self.startTransfer
self.connectCompleted(addr, 0)
self.transport.stopReading()
def startTransfer(self, file_obj):
"""Callback called when the result iq is received"""
d = self.beginFileTransfer(file_obj, self.transport)
d.addCallback(self.fileTransfered)
def fileTransfered(self, d):
info(_("File transfer completed, closing connection"))
self.transport.loseConnection()
self.factory.finishedCb(self.sid, True, self.profile)
def connectCompleted(self, remotehost, remoteport):
debug("connectCompleted")
if self.addressType == ADDR_IPV4:
result = struct.pack('!BBBBIH', SOCKS5_VER, REPLY_SUCCESS, 0, 1, remotehost, remoteport)
elif self.addressType == ADDR_DOMAINNAME:
result = struct.pack('!BBBBB%dsH' % len(remotehost), SOCKS5_VER, REPLY_SUCCESS, 0,
ADDR_DOMAINNAME, len(remotehost), remotehost, remoteport)
self.transport.write(result)
self.state = STATE_READY
def bindRequested(self, addr, port):
pass
def authenticateUserPass(self, user, passwd):
debug("User/pass: %s/%s", user, passwd)
return True
def dataReceived(self, buf):
if self.state == STATE_TARGET_READY:
self.data["file_obj"].write(buf)
return
self.buf = self.buf + buf
if self.state == STATE_INITIAL:
self._parseNegotiation()
if self.state == STATE_AUTH_USERPASS:
self._parseUserPass()
if self.state == STATE_REQUEST:
self._parseRequest()
if self.state == STATE_TARGET_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):
debug("clientConnectionLost")
self.transport.loseConnection()
def connectionLost(self, reason):
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
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 startedConnecting(self, connector):
debug(_("Socks 5 server connection started"))
def clientConnectionLost(self, connector, reason):
debug(_("Socks 5 server connection lost (reason: %s)"), reason)
class Socks5ClientFactory(protocol.ClientFactory):
protocol = SOCKSv5
def __init__(self, current_stream, sid, iq_id, activateCb, finishedCb, proxy=False, profile=None):
"""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):
debug(_("Socks 5 client connection started"))
def clientConnectionLost(self, connector, reason):
debug(_("Socks 5 client connection lost (reason: %s)"), reason)
self.finishedCb(self.sid, reason.type == jab_error.ConnectionDone, self.profile) # TODO: really check if the state is actually successful
class XEP_0065(object):
NAMESPACE = NS_BS
params = """
<params>
<general>
<category name="File Transfer">
<param name="IP" value='0.0.0.0' default_cb='yes' type="string" />
<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>
"""
def __init__(self, host):
info(_("Plugin XEP_0065 initialization"))
#session data
self.hash_sid_map = {} # key: hash of the transfer session, value: (session id, profile)
self.host = host
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"))
info(_("Launching Socks5 Stream server on port %d"), port)
reactor.listenTCP(port, self.server_factory)
def getHandler(self, profile):
return XEP_0065_handler(self)
def profileConnected(self, profile):
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
client.xep_0065_current_stream = {} # key: stream_id, value: data(dict)
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 getProgress(self, sid, data, profile):
"""Fill data with position of current transfer"""
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
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:
pass
def _timeOut(self, sid, profile):
"""Delecte current_stream id, called after timeout
@param id: id of client.xep_0065_current_stream"""
info(_("Socks5 Bytestream: TimeOut reached for id %(sid)s [%(profile)s]")
% {"sid": sid, "profile": profile})
self._killId(sid, False, "TIMEOUT", profile)
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 not client:
warning(_("Client no more in cache"))
return
if sid not in client.xep_0065_current_stream:
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']
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]
if success:
success_cb(sid, file_obj, NS_BS, profile)
else:
failure_cb(sid, file_obj, NS_BS, failure_reason, profile)
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
@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)
client = self.host.getClient(profile)
if not client:
error(_("Unknown profile, this should not happen"))
raise ProfileNotInCacheError
if length is not None:
error(_('stream length not managed yet'))
return
profile_jid = client.jid
xmlstream = client.xmlstream
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["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()
#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, profile)
iq_elt.send()
def iqResult(self, sid, profile, iq_elt):
"""Called when the result of open iq is received"""
if iq_elt["type"] == "error":
warning(_("Transfer failed"))
return
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
try:
data = client.xep_0065_current_stream[sid]
file_obj = data["file_obj"]
timer = data["timer"]
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
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(client.xep_0065_current_stream, sid, None, self.activateProxyStream, lambda sid, success, profile: self._killId(sid, success, profile=profile), True, profile)
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, profile):
debug(_("activating stream"))
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
data = client.xep_0065_current_stream[sid]
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, 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"""
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
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
def streamQuery(self, iq_elt, profile):
"""Get file using byte stream"""
debug(_("BS stream query"))
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
xmlstream = client.xmlstream
iq_elt.handled = True
query_elt = iq_elt.firstChildElement()
sid = query_elt.getAttribute("sid")
streamhost_elts = filter(lambda elt: elt.name == 'streamhost', query_elt.elements())
if not sid in client.xep_0065_current_stream:
warning(_("Ignoring unexpected BS transfer: %s" % sid))
self.sendNotAcceptableError(iq_elt['id'], iq_elt['from'], xmlstream)
return
client.xep_0065_current_stream[sid]['timer'].cancel()
client.xep_0065_current_stream[sid]["to"] = jid.JID(iq_elt["to"])
client.xep_0065_current_stream[sid]["xmlstream"] = xmlstream
if not streamhost_elts:
warning(_("No streamhost found in stream query %s" % sid))
self.sendBadRequestError(iq_elt['id'], iq_elt['from'], xmlstream)
return
streamhost_elt = streamhost_elts[0] # TODO: manage several streamhost elements case
sh_host = streamhost_elt.getAttribute("host")
sh_port = streamhost_elt.getAttribute("port")
sh_jid = streamhost_elt.getAttribute("jid")
if not sh_host or not sh_port or not sh_jid:
warning(_("incomplete streamhost element"))
self.sendBadRequestError(iq_elt['id'], iq_elt['from'], xmlstream)
return
client.xep_0065_current_stream[sid]["streamhost"] = (sh_host, sh_port, sh_jid)
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)
reactor.connectTCP(sh_host, int(sh_port), factory)
def activateStream(self, sid, iq_id, profile):
client = self.host.getClient(profile)
if not client:
raise ProfileNotInCacheError
debug(_("activating stream"))
result = domish.Element((None, 'iq'))
data = client.xep_0065_current_stream[sid]
result['type'] = 'result'
result['id'] = iq_id
result['from'] = data["to"].full()
result['to'] = 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)
def sendNotAcceptableError(self, iq_id, to_jid, xmlstream):
"""Not acceptable error used when the stream is not expected or something is going wrong
@param iq_id: IQ id
@param to_jid: addressee
@param xmlstream: XML stream to use to send the error"""
result = domish.Element((None, 'iq'))
result['type'] = 'result'
result['id'] = iq_id
result['to'] = to_jid
error_el = result.addElement('error')
error_el['type'] = 'modify'
error_el.addElement(('urn:ietf:params:xml:ns:xmpp-stanzas', 'not-acceptable'))
xmlstream.send(result)
def sendBadRequestError(self, iq_id, to_jid, xmlstream):
"""Not acceptable error used when the stream is not expected or something is going wrong
@param iq_id: IQ id
@param to_jid: addressee
@param xmlstream: XML stream to use to send the error"""
result = domish.Element((None, 'iq'))
result['type'] = 'result'
result['id'] = iq_id
result['to'] = to_jid
error_el = result.addElement('error')
error_el['type'] = 'cancel'
error_el.addElement(('urn:ietf:params:xml:ns:xmpp-stanzas', 'bad-request'))
xmlstream.send(result)
class XEP_0065_handler(XMPPHandler):
implements(iwokkel.IDisco)
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", profile_key=self.parent.profile)
proxy = self.host.memory.setParam("Proxy host", streamhost_elt.getAttribute("host", ""),
"File Transfer", profile_key=self.parent.profile)
proxy = self.host.memory.setParam("Proxy port", streamhost_elt.getAttribute("port", ""),
"File Transfer", profile_key=self.parent.profile)
def connectionInitialized(self):
def after_init(ignore):
proxy_ent = self.host.memory.getServerServiceEntity("proxy", "bytestreams", profile=self.parent.profile)
if not proxy_ent:
debug(_("No proxy found on this server"))
return
iq_elt = jabber_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=''):
return [disco.DiscoFeature(NS_BS)]
def getDiscoItems(self, requestor, target, nodeIdentifier=''):
return []
Reference in a new issue View git blame Copy permalink