plugin XEP-0065: refactored and misc stuff fixed. Still not finished

plugins XEP-0096: XEP-0065 (Socks5 stream method) managed
This commit is contained in:
Goffi 2011-10-03 18:05:15 +02:00
commit 1225131aec

View file

@ -55,8 +55,10 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""
from logging import debug, info, error
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 client, jid
from twisted.protocols.basic import FileSender
from twisted.words.xish import domish
from twisted.web.client import getPage
@ -76,6 +78,7 @@ 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
@ -127,6 +130,13 @@ 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()
@ -141,9 +151,6 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.peersock = None
self.addressType = 0
self.requestType = 0
self.activeConns = {}
self.pendingConns = {}
self.transfered = 0 #nb of bytes already copied
def _startNegotiation(self):
debug("_startNegotiation")
@ -213,27 +220,6 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.transport.write(result)
self.transport.loseConnection()
def addConnection(self, address, connection):
info(_("Adding connection: %(address)s, %(connection)s") % {'address':address, 'connection':connection})
olist = self.pendingConns.get(address, [])
if len(olist) <= 1:
olist.append(connection)
self.pendingConns[address] = olist
return True
else:
return False
def removePendingConnection(self, address, connection):
olist = self.pendingConns[address]
if len(olist) == 1:
del self.pendingConns[address]
else:
olist.remove(connection)
self.pendingConns[address] = olist
def removeActiveConnection(self, address):
del self.activeConns[address]
def _parseRequest(self):
debug("_parseRequest")
try:
@ -279,7 +265,7 @@ class SOCKSv5(protocol.Protocol, FileSender):
def _makeRequest(self):
debug("_makeRequest")
self.state = STATE_TARGET_REQUEST
sha1 = hashlib.sha1(self.sid + self.initiator_jid + self.target_jid).hexdigest()
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)
@ -310,67 +296,43 @@ class SOCKSv5(protocol.Protocol, FileSender):
self.loseConnection()
return
debug(_("Saving file in %s."), self.data["dest_path"])
self.dest_file = open(self.data["dest_path"], 'w')
self.state = STATE_TARGET_READY
self.activateCB(self.target_jid, self.initiator_jid, self.sid, self.IQ_id, self.xmlstream)
self.factory.activateCb(self.sid, self.factory.iq_id)
except struct.error, why:
return None
def connectionMade(self):
debug("connectionMade (mode = %s)" % self.mode)
self.host.registerProgressCB(self.transfert_id, self.getProgress)
if self.mode == "target":
debug("connectionMade (mode = %s)" % "requester" if isinstance(self.factory, Socks5ServerFactory) else "target")
if isinstance(self.factory, Socks5ClientFactory):
self.sid = self.factory.sid
self.data = self.factory.data
self.state = STATE_TARGET_INITIAL
self._startNegotiation()
def connectRequested(self, addr, port):
debug("connectRequested")
# Check for special connect to the namespace -- this signifies that the client
# is just checking to ensure it can connect to the streamhost
if addr == "http://jabber.org/protocol/bytestreams":
self.connectCompleted(addr, 0)
self.transport.loseConnection()
return
# Save addr, for cleanup
self.addr = addr
# Check to see if the requested address is already
# activated -- send an error if so
if addr in self.activeConns:
self.sendErrorReply(socks5.REPLY_CONN_NOT_ALLOWED)
return
# Add this address to the pending connections
if self.addConnection(addr, self):
self.connectCompleted(addr, 0)
self.transport.stopReading()
else:
# Check that this session if expected
if not self.factory.hash_sid_map.has_key(addr):
#no: we refuse it
self.sendErrorReply(socks5.REPLY_CONN_REFUSED)
return
self.sid = self.factory.hash_sid_map[addr]
self.factory.current_stream[self.sid]["start_transfer_cb"] = self.startTransfer
self.connectCompleted(addr, 0)
self.transport.stopReading()
def getProgress(self, data):
"""Fill data with position of current transfert"""
try:
data["position"] = str(self.dest_file.tell())
data["size"] = self.filesize
except (ValueError, AttributeError):
pass
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()
try:
self.dest_file.close()
except:
pass
def updateTransfered(self, data):
self.transfered+=len(data)
return data
self.factory.finishedCb(self.sid, True)
def connectCompleted(self, remotehost, remoteport):
debug("connectCompleted")
@ -381,9 +343,6 @@ class SOCKSv5(protocol.Protocol, FileSender):
ADDR_DOMAINNAME, len(remotehost), remotehost, remoteport)
self.transport.write(result)
self.state = STATE_READY
self.dest_file=open(self.filepath)
d=self.beginFileTransfer(self.dest_file, self.transport, self.updateTransfered)
d.addCallback(self.fileTransfered)
def bindRequested(self, addr, port):
pass
@ -394,8 +353,7 @@ class SOCKSv5(protocol.Protocol, FileSender):
def dataReceived(self, buf):
if self.state == STATE_TARGET_READY:
self.dest_file.write(buf)
self.transfered+=len(buf)
self.data["file_obj"].write(buf)
return
self.buf = self.buf + buf
@ -422,21 +380,21 @@ class SOCKSv5(protocol.Protocol, FileSender):
def connectionLost(self, reason):
debug("connectionLost")
self.host.removeProgressCB(self.transfert_id)
if self.state == STATE_CONNECT_PENDING:
self.removePendingConnection(self.addr, self)
else:
if self.state != STATE_CONNECT_PENDING:
self.transport.unregisterProducer()
if self.peersock != None:
self.peersock.peersock = None
self.peersock.transport.unregisterProducer()
self.peersock = None
self.removeActiveConnection(self.addr)
class Socks5ServerFactory(protocol.ServerFactory):
protocol = SOCKSv5
protocol.mode = "initiator" #FIXME: Q&D way, fix it
def __init__(self, current_stream, hash_sid_map, finishedCb):
self.current_stream = current_stream
self.hash_sid_map = hash_sid_map
self.finishedCb = finishedCb
def startedConnecting(self, connector):
debug (_("Socks 5 server connection started"))
@ -446,21 +404,30 @@ class Socks5ServerFactory(protocol.ServerFactory):
class Socks5ClientFactory(protocol.ClientFactory):
protocol = SOCKSv5
protocol.mode = "target" #FIXME: Q&D way, fix it
def __init__(self, current_stream, sid, iq_id, activateCb, finishedCb):
self.data = current_stream[sid]
self.sid = sid
self.iq_id = iq_id
self.activateCb = activateCb
self.finishedCb = finishedCb
def startedConnecting(self, connector):
debug (_("Socks 5 client connection started"))
def clientConnectionLost(self, connector, reason):
debug (_("Socks 5 client connection lost (reason: %s)"), reason)
debug (_("Socks 5 client connection lost"))
self.finishedCb(self.sid, reason.type == jab_error.ConnectionDone) #TODO: really check if the state is actually successful
class XEP_0065():
NAMESPACE = NS_BS
params = """
<params>
<general>
<category name="File Transfert">
<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>
@ -470,17 +437,20 @@ class XEP_0065():
def __init__(self, host):
info(_("Plugin XEP_0065 initialization"))
#session data
self.current_stream = {} #key: stream_id, value: data(dict)
self.hash_sid_map = {} #key: hash of the transfer session, value: session id
self.host = host
debug(_("registering"))
self.server_factory = Socks5ServerFactory()
self.server_factory.protocol.host = self.host #needed for progress CB
self.client_factory = Socks5ClientFactory()
self.server_factory = Socks5ServerFactory(self.current_stream, self.hash_sid_map, self._killId)
#parameters
host.memory.importParams(XEP_0065.params)
host.memory.setDefault("IP", "File Transfert", self.getExternalIP)
host.memory.setDefault("IP", "File Transfer", self.getExternalIP)
port = int(self.host.memory.getParamA("Port", "File Transfer"))
port = int(self.host.memory.getParamA("Port", "File Transfert"))
info(_("Launching Socks5 Stream server on port %d"), port)
reactor.listenTCP(port, self.server_factory)
@ -491,53 +461,217 @@ class XEP_0065():
"""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):
"""Fill data with position of current transfer"""
try:
file_obj = self.current_stream[sid]["file_obj"]
data["position"] = str(file_obj.tell())
data["size"] = str(self.current_stream[sid]["size"])
except:
pass
def _timeOut(self, sid):
"""Delecte current_stream id, called after timeout
@param id: id of self.current_stream"""
info(_("Socks5 Bytestream: TimeOut reached for id %s") % sid);
self._killId(sid, False, "TIMEOUT")
def _killId(self, sid, success=False, failure_reason="UNKNOWN"):
"""Delete an current_stream id, clean up associated observers
@param sid: id of self.current_stream"""
if not self.current_stream.has_key(sid):
warning(_("kill id called on a non existant id"))
return
if self.current_stream[sid].has_key("observer_cb"):
xmlstream = self.current_stream[sid]["xmlstream"]
xmlstream.removeObserver(self.current_stream[sid]["event_data"], self.current_stream[sid]["observer_cb"])
if self.current_stream[sid]['timer'].active():
self.current_stream[sid]['timer'].cancel()
if self.current_stream[sid].has_key("size"):
self.host.removeProgressCB(sid)
file_obj = self.current_stream[sid]['file_obj']
success_cb = self.current_stream[sid]['success_cb']
failure_cb = self.current_stream[sid]['failure_cb']
del self.current_stream[sid]
if self.hash_sid_map.has_key(sid):
del self.hash_sid_map[sid]
if success:
success_cb(sid, file_obj, NS_BS)
else:
failure_cb(sid, file_obj, NS_BS, failure_reason)
def setData(self, data, id):
self.data = data
self.transfert_id = id
self.transfer_id = id
def sendFile(self, id, filepath, size):
#lauching socks5 initiator
debug(_("Launching socks5 initiator"))
self.server_factory.protocol.mode = "initiator"
#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.transfert_id = id
self.server_factory.protocol.transfer_id = id
def getFile(self, iq, profile_key='@DEFAULT@'):
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"""
if length != None:
error(_('stream length not managed yet'))
return;
profile_jid, xmlstream = self.host.getJidNStream(profile)
data = self.current_stream[sid] = {}
data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid)
data["file_obj"] = file_obj
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
if size:
data["size"] = size
self.host.registerProgressCB(sid, self.getProgress)
iq_elt = 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
streamhost = query_elt.addElement('streamhost')
streamhost['host'] = "127.0.0.1" #self.host.memory.getParamA("IP", "File Transfer")
streamhost['port'] = self.host.memory.getParamA("Port", "File Transfer")
streamhost['jid'] = profile_jid.full()
iq_elt.addCallback(self.iqResult, sid)
iq_elt.send()
def iqResult(self, sid, iq_elt):
"""Called when the result of open iq is received"""
if iq_elt["type"] == "error":
warning(_("Transfer failed"))
return
try:
data = self.current_stream[sid]
callback = data["start_transfer_cb"]
file_obj = data["file_obj"]
timer = data["timer"]
except KeyError:
error(_("Internal error, can't do transfer"))
return
if timer.active():
timer.cancel()
callback(file_obj)
def prepareToReceive(self, from_jid, sid, file_obj, size, success_cb, failure_cb):
"""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"""
data = self.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)
data["timer"] = reactor.callLater(TIMEOUT, self._timeOut, sid)
data["success_cb"] = success_cb
data["failure_cb"] = failure_cb
def streamQuery(self, iq_elt, profile):
"""Get file using byte stream"""
client = self.host.getClient(profile_key)
assert(client)
iq.handled = True
SI_elem = iq.firstChildElement()
IQ_id = iq['id']
for element in SI_elem.elements():
if element.name == "streamhost":
info (_("Stream proposed: host=[%(host)s] port=[%(port)s]") % {'host':element['host'], 'port':element['port']})
factory = self.client_factory
self.server_factory.protocol.mode = "target"
factory.protocol.host = self.host #needed for progress CB
factory.protocol.xmlstream = client.xmlstream
factory.protocol.data = self.data
factory.protocol.transfert_id = self.transfert_id
factory.protocol.filesize = self.data["size"]
factory.protocol.sid = SI_elem['sid']
factory.protocol.initiator_jid = element['jid']
factory.protocol.target_jid = client.jid.full()
factory.protocol.IQ_id = IQ_id
factory.protocol.activateCB = self.activateStream
reactor.connectTCP(element['host'], int(element['port']), factory)
debug(_("BS stream query"))
profile_jid, xmlstream = self.host.getJidNStream(profile)
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 self.current_stream:
warning(_("Ignoring unexpected BS transfer: %s" % sid))
self.sendNotAcceptableError(iq_elt['id'], iq_elt['from'], xmlstream)
return
self.current_stream[sid]["to"] = jid.JID(iq_elt["to"])
self.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
self.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(self.current_stream, sid, iq_elt["id"], self.activateStream, self._killId)
reactor.connectTCP(sh_host, int(sh_port), factory)
def activateStream(self, from_jid, to_jid, sid, IQ_id, xmlstream):
def activateStream(self, sid, iq_id):
debug(_("activating stream"))
result = domish.Element(('', 'iq'))
data = self.current_stream[sid]
result['type'] = 'result'
result['id'] = IQ_id
result['from'] = from_jid
result['to'] = to_jid
query = result.addElement('query', 'http://jabber.org/protocol/bytestreams')
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'] = to_jid #FIXME: use real streamhost
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(('', '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(('', '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):
@ -548,7 +682,7 @@ class XEP_0065_handler(XMPPHandler):
self.host = plugin_parent.host
def connectionInitialized(self):
self.xmlstream.addObserver(BS_REQUEST, self.plugin_parent.getFile)
self.xmlstream.addObserver(BS_REQUEST, self.plugin_parent.streamQuery, profile = self.parent.profile)
def getDiscoInfo(self, requestor, target, nodeIdentifier=''):