libervia-backend: src/plugins/plugin_xep

comparison src/plugins/plugin_xep_0065.py @ 587:952322b1d490

Remove trailing whitespaces.

author	Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
date	Fri, 18 Jan 2013 17:55:34 +0100
parents	9902ec2d8d9b
children	beaf6bec2fcd

comparison

equal deleted inserted replaced

-:6a718ede8be1
+:952322b1d490
 --
 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
 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:
 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])
 # 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
 # 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
 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 not self.factory.hash_sid_map.has_key(addr):
 #no: we refuse it
 self.sendErrorReply(REPLY_CONN_REFUSED)
 return
 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)
 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):
 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():
 NAMESPACE = NS_BS
 params = """
 <params>
 <general>
 </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.importParams(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)
 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 %s [%s]") % (sid, 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)
 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 client.xep_0065_current_stream[sid].has_key("size"):
 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]
 assert(profile)
 client = self.host.getClient(profile)
 if not client:
 error(_("Unknown profile, this should not happen"))
 raise ProfileNotInCacheError
 if length != 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
 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"))
 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]["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"))
 @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)
 @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):
 streamhost_elt = streamhost_elts[0]
 proxy = self.host.memory.setParam("Proxy", streamhost_elt.getAttribute("jid",""), "File Transfer", self.parent.profile)
 proxy = self.host.memory.setParam("Proxy host", streamhost_elt.getAttribute("host",""), "File Transfer", self.parent.profile)
 proxy = self.host.memory.setParam("Proxy port", streamhost_elt.getAttribute("port",""), "File Transfer", self.parent.profile)
 def connectionInitialized(self):
 def after_init(ignore):
 proxy_ent = self.host.memory.getServerServiceEntity("proxy", "bytestreams", self.parent.profile)
 if not proxy_ent:
 debug(_("No proxy found on this server"))
 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)]

Mercurial > libervia-backend

comparison src/plugins/plugin_xep_0065.py @ 587:952322b1d490