libervia-backend: sat/plugins/plugin_xep

comparison sat/plugins/plugin_xep_0065.py @ 3028:ab2696e34d29

Python 3 port: /!\ this is a huge commit /!\ starting from this commit, SàT is needs Python 3.6+ /!\ SàT maybe be instable or some feature may not work anymore, this will improve with time This patch port backend, bridge and frontends to Python 3. Roughly this has been done this way: - 2to3 tools has been applied (with python 3.7) - all references to python2 have been replaced with python3 (notably shebangs) - fixed files not handled by 2to3 (notably the shell script) - several manual fixes - fixed issues reported by Python 3 that where not handled in Python 2 - replaced "async" with "async_" when needed (it's a reserved word from Python 3.7) - replaced zope's "implements" with @implementer decorator - temporary hack to handle data pickled in database, as str or bytes may be returned, to be checked later - fixed hash comparison for password - removed some code which is not needed anymore with Python 3 - deactivated some code which needs to be checked (notably certificate validation) - tested with jp, fixed reported issues until some basic commands worked - ported Primitivus (after porting dependencies like urwid satext) - more manual fixes

author	Goffi <goffi@goffi.org>
date	Tue, 13 Aug 2019 19:08:41 +0200
parents	69e4716d6268
children	fee60f17ebac

comparison

equal deleted inserted replaced

-:ff5bcb12ae60
+:ab2696e34d29
-#!/usr/bin/env python2
+#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 # SAT plugin for managing xep-0065
 # Copyright (C)
 from collections import namedtuple
 import struct
 import hashlib
 import uuid
-from zope.interface import implements
+from zope.interface import implementer
 try:
 from twisted.words.protocols.xmlstream import XMPPHandler
 except ImportError:
 from wokkel.subprotocols import XMPPHandler
 STATE_READY,
 STATE_AUTH_USERPASS,
 STATE_CLIENT_INITIAL,
 STATE_CLIENT_AUTH,
 STATE_CLIENT_REQUEST,
-) = xrange(8)
+) = range(8)
 SOCKS5_VER = 0x05
 ADDR_IPV4 = 0x01
 ADDR_DOMAINNAME = 0x03
 @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())
+self.id = id_ if id_ is not None else str(uuid.uuid4())
 if priority_local:
 self._local_priority = int(priority)
 self._priority = self.calculatePriority()
 else:
 self._local_priority = 0
 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))
+log.debug("Discarding {}".format(self))
 try:
 self.factory.discard()
 except AttributeError:
 pass  # no discard for Socks5ServerFactory
 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(
+id=" id={}".format(self.id if self.id is not None else "").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(
+return "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"")
+self, id=" id={}".format(self.id if self.id is not None else "")
 )
 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
 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))
+raise exceptions.InternalError("Unknown {} type !".format(self.type))
 return 2 ** 16 * multiplier + self._local_priority
 def activate(self, client, sid, peer_jid, local_jid):
 """Activate the proxy candidate
 @param session_hash(str): hash of the session
 must only be used in client mode
 """
 self.connection = defer.Deferred()  # called when connection/auth is done
 if session_hash is not None:
+assert isinstance(session_hash, str)
 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.buf = b""
 self.supportedAuthMechs = [AUTHMECH_ANON]
 self.supportedAddrs = [ADDR_DOMAINNAME]
 self.enabledCommands = [CMD_CONNECT]
 self.peersock = None
 self.addressType = 0
 # 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
-log.warning(u"Unsupported authentication mechanism")
+log.warning("Unsupported authentication mechanism")
 self.transport.write(struct.pack("!BB", SOCKS5_VER, AUTHMECH_INVALID))
 self.transport.loseConnection()
 except struct.error:
 pass
 # 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])
+nlen = 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)
 except struct.error:
 # The buffer is probably not complete, we need to wait more
 return None
 def _makeRequest(self):
-hash_ = self._session_hash
+hash_ = self._session_hash.encode('utf-8')
 request = struct.pack(
 "!5B%dsH" % len(hash_),
 SOCKS5_VER,
 CMD_CONNECT,
 0,
 # 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])
+nlen = 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)
 # The buffer is probably not complete, we need to wait more
 return None
 def connectionMade(self):
 log.debug(
-u"Socks5 connectionMade (mode = {})".format(
+"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):
 # Check that this session is expected
-if not self.factory.addToSession(addr, self):
+if not self.factory.addToSession(addr.decode('utf-8'), self):
 self.sendErrorReply(REPLY_CONN_REFUSED)
 log.warning(
-u"Unexpected connection request received from {host}".format(
+"Unexpected connection request received from {host}".format(
 host=self.transport.getPeer().host
 )
 )
 return
-self._session_hash = addr
+self._session_hash = addr.decode('utf-8')
 self.connectCompleted(addr, 0)
 def startTransfer(self, chunk_size):
 """Callback called when the result iq is received
 @param chunk_size(None, int): size of the buffer, or None for default
 """
 self.active = True
 if chunk_size is not None:
 self.CHUNK_SIZE = chunk_size
-log.debug(u"Starting file transfer")
+log.debug("Starting file transfer")
 d = self.stream_object.startStream(self.transport)
 d.addCallback(self.streamFinished)
 def streamFinished(self, d):
 log.info(_("File transfer completed, closing connection"))
 def bindRequested(self, addr, port):
 pass
 def authenticateUserPass(self, user, passwd):
 # FIXME: implement authentication and remove the debug printing a password
-log.debug(u"User/pass: %s/%s" % (user, passwd))
+log.debug("User/pass: %s/%s" % (user, passwd))
 return True
 def dataReceived(self, buf):
 if self.state == STATE_READY:
 # Everything is set, we just have to write the incoming data
 self.transport.loseConnection()
 else:
 self._makeRequest()
 def connectionLost(self, reason):
-log.debug(u"Socks5 connection lost: {}".format(reason.value))
+log.debug("Socks5 connection lost: {}".format(reason.value))
 if self.state != STATE_READY:
 self.connection.errback(reason)
 if self.server_mode:
 self.factory.removeFromSession(self._session_hash, self, reason)
 def startTransfer(self, session_hash, chunk_size=None):
 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")
+log.error("Can't start file transfer, can't find protocol")
 else:
 session[TIMER_KEY].cancel()
 protocol.startTransfer(chunk_size)
 def addToSession(self, session_hash, protocol):
 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
 """
+assert isinstance(session_hash, str)
 try:
 session_data = self.getSession(session_hash)
 except KeyError:
 return False
 else:
 """
 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")
+log.error("Protocol not found in session while it should be there")
 else:
 if protocol.active:
 # The active protocol has been removed, session is finished
 if reason.check(internet_error.ConnectionDone):
 self.getSession(session_hash)[DEFER_KEY].callback(None)
 def startTransfer(self, __=None, chunk_size=None):
 self.session[TIMER_KEY].cancel()
 self._protocol_instance.startTransfer(chunk_size)
 def clientConnectionFailed(self, connector, reason):
-log.debug(u"Connection failed")
+log.debug("Connection failed")
 self.connection.errback(reason)
 def clientConnectionLost(self, connector, reason):
-log.debug(_(u"Socks 5 client connection lost (reason: %s)") % reason.value)
+log.debug(_("Socks 5 client connection lost (reason: %s)") % reason.value)
 if self._protocol_instance.active:
 # 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)
 # 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")
+log.debug("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)
 self._server_factory_port is set on server creation
 """
 if self._server_factory is None:
 self._server_factory = Socks5ServerFactory(self)
-for port in xrange(SERVER_STARTING_PORT, 65356):
+for port in range(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(
+"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),
+err_num=" (error code: {})".format(e.socketError.errno),
 )
 )
 else:
 self._server_factory_port = listening_port.getHost().port
 break
 or None if not acceptable proxy is found
 @raise exceptions.NotFound: no Proxy found
 """
 def notFound(server):
-log.info(u"No proxy found on this server")
+log.info("No proxy found on this server")
 self._cache_proxies[server] = None
 raise exceptions.NotFound
 server = client.host if client.is_component else client.jid.host
 try:
 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(
+"Error while requesting proxy info on {jid}: {error}".format(
 proxy.full(), failure
 )
 )
 notFound(server)
 try:
-query_elt = result_elt.elements(NS_BS, "query").next()
+query_elt = next(result_elt.elements(NS_BS, "query"))
-streamhost_elt = query_elt.elements(NS_BS, "streamhost").next()
+streamhost_elt = next(query_elt.elements(NS_BS, "streamhost"))
 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, AttributeError):
-log.warning(u"Invalid proxy data received from {}".format(proxy.full()))
+log.warning("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))
+log.info("Proxy found: {}".format(proxy_infos))
 defer.returnValue(proxy_infos)
 @defer.inlineCallbacks
 def _getNetworkData(self, client):
 """Retrieve information about network
 external_ip = yield self._ip.getExternalIP(client)
 local_ips = yield self._ip.getLocalIPs(client)
 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")
+log.info("We are probably behind a NAT")
 if self._np is None:
-log.warning(u"NAT port plugin not available, we can't map port")
+log.warning("NAT port plugin not available, we can't map port")
 else:
 ext_port = yield self._np.mapPort(
-local_port, desc=u"SaT socks5 stream"
+local_port, desc="SaT socks5 stream"
 )
 if ext_port is None:
-log.warning(u"Can't map NAT port")
+log.warning("Can't map NAT port")
 else:
 self._external_port = ext_port
 defer.returnValue((local_port, self._external_port, local_ips, external_ip))
 @return (D(None, Candidate)): best candidate or None if none can connect
 """
 defer_candidates = None
 def connectionCb(client, candidate):
-log.info(u"Connection of {} successful".format(unicode(candidate)))
+log.info("Connection of {} successful".format(str(candidate)))
 for idx, other_candidate in enumerate(candidates):
 try:
 if other_candidate.priority < candidate.priority:
-log.debug(u"Cancelling {}".format(other_candidate))
+log.debug("Cancelling {}".format(other_candidate))
 defer_candidates[idx].cancel()
 except AttributeError:
 assert other_candidate is None
 def connectionEb(failure, client, candidate):
 if failure.check(defer.CancelledError):
-log.debug(u"Connection of {} has been cancelled".format(candidate))
+log.debug("Connection of {} has been cancelled".format(candidate))
 else:
 log.info(
-u"Connection of {candidate} Failed: {error}".format(
+"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")
+log.debug("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(
 client,
 """Called when stream was not started quickly enough
 @param session_hash(str): hash as returned by getSessionHash
 @param client: %(doc_client)s
 """
-log.info(u"Socks5 Bytestream: TimeOut reached")
+log.info("Socks5 Bytestream: TimeOut reached")
 session = self.getSession(client, session_hash)
 session[DEFER_KEY].errback(exceptions.TimeOutError)
 def killSession(self, failure_, session_hash, sid, client):
 """Clean the current session
 @param client: %(doc_client)s
 @param failure_(None, failure.Failure): None if eveything was fine, a failure else
 @return (None, failure.Failure): failure_ is returned
 """
 log.debug(
-u"Cleaning session with hash {hash}{id}: {reason}".format(
+"Cleaning session with hash {hash}{id}: {reason}".format(
 hash=session_hash,
 reason="" if failure_ is None else failure_.value,
-id="" if sid is None else u" (id: {})".format(sid),
+id="" if sid is None else " (id: {})".format(sid),
 )
 )
 try:
 assert self.hash_clients_map[session_hash] == client
 if sid is not None:
 try:
 del client.xep_0065_sid_session[sid]
 except KeyError:
-log.warning(u"Session id {} is unknown".format(sid))
+log.warning("Session id {} is unknown".format(sid))
 try:
 session_data = client._s5b_sessions[session_hash]
 except KeyError:
-log.warning(u"There is no session with this hash")
+log.warning("There is no session with this hash")
 return
 else:
 del client._s5b_sessions[session_hash]
 try:
 for candidate in candidates:
 streamhost = query_elt.addElement("streamhost")
 streamhost["host"] = candidate.host
 streamhost["port"] = str(candidate.port)
 streamhost["jid"] = candidate.jid.full()
-log.debug(u"Candidate proposed: {}".format(candidate))
+log.debug("Candidate proposed: {}".format(candidate))
 d = iq_elt.send()
 args = [client, session_data, local_jid]
 d.addCallbacks(self._IQNegotiationCb, self._IQNegotiationEb, args, None, args)
 @param session_data(dict): data of the session
 @param client: %(doc_client)s
 @param iq_elt(domish.Element): <iq> result
 """
 try:
-query_elt = iq_elt.elements(NS_BS, "query").next()
+query_elt = next(iq_elt.elements(NS_BS, "query"))
-streamhost_used_elt = query_elt.elements(NS_BS, "streamhost-used").next()
+streamhost_used_elt = next(query_elt.elements(NS_BS, "streamhost-used"))
 except StopIteration:
-log.warning(u"No streamhost found in stream query")
+log.warning("No streamhost found in stream query")
 # FIXME: must clean session
 return
 streamhost_jid = jid.JID(streamhost_used_elt["jid"])
 try:
-candidate = (
+candidate = next((
 c for c in session_data["candidates"] if c.jid == streamhost_jid
-).next()
+))
 except StopIteration:
 log.warning(
-u"Candidate [{jid}] is unknown !".format(jid=streamhost_jid.full())
+"Candidate [{jid}] is unknown !".format(jid=streamhost_jid.full())
 )
 return
 else:
-log.info(u"Candidate choosed by target: {}".format(candidate))
+log.info("Candidate choosed by target: {}".format(candidate))
 if candidate.type == XEP_0065.TYPE_PROXY:
-log.info(u"A Socks5 proxy is used")
+log.info("A Socks5 proxy is used")
 d = self.connectCandidate(client, candidate, session_data["hash"])
 d.addCallback(
 lambda __: candidate.activate(
 client, session_data["id"], session_data["peer_jid"], local_jid
 )
 d = defer.succeed(None)
 d.addCallback(lambda __: candidate.startTransfer(session_data["hash"]))
 def _activationEb(self, failure):
-log.warning(u"Proxy activation error: {}".format(failure.value))
+log.warning("Proxy activation error: {}".format(failure.value))
 def _IQNegotiationEb(self, stanza_err, client, session_data, local_jid):
-log.warning(u"Socks5 transfer failed: {}".format(stanza_err.value))
+log.warning("Socks5 transfer failed: {}".format(stanza_err.value))
 # FIXME: must clean session
 def createSession(self, *args, **kwargs):
 """like [_createSession] but return the session deferred instead of the whole session
 @param sid(unicode): session id
 @param initiator(bool): if True, this session is create by initiator
 @return (dict): session data
 """
 if sid in client.xep_0065_sid_session:
-raise exceptions.ConflictError(u"A session with this id already exists !")
+raise exceptions.ConflictError("A session with this id already exists !")
 if requester:
 session_hash = getSessionHash(local_jid, to_jid, sid)
 session_data = self._registerHash(client, session_hash, stream_object)
 else:
 session_hash = getSessionHash(to_jid, local_jid, sid)
 for incoming request received by Socks5ServerFactory). None must
 only be used by Socks5ServerFactory.
 See comments below for details
 @return (dict): session data
 """
+assert isinstance(session_hash, str)
 if client is None:
 try:
 client = self.hash_clients_map[session_hash]
 except KeyError as e:
-log.warning(u"The requested session doesn't exists !")
+log.warning("The requested session doesn't exists !")
 raise e
 return client._s5b_sessions[session_hash]
 def registerHash(self, *args, **kwargs):
 """like [_registerHash] but return the session deferred instead of the whole session
 session_data = self.getSession(client, session_hash)
 assert "stream_object" not in session_data
 session_data["stream_object"] = stream_object
 def streamQuery(self, iq_elt, client):
-log.debug(u"BS stream query")
+log.debug("BS stream query")
 iq_elt.handled = True
-query_elt = iq_elt.elements(NS_BS, "query").next()
+query_elt = next(iq_elt.elements(NS_BS, "query"))
 try:
 sid = query_elt["sid"]
 except KeyError:
-log.warning(u"Invalid bystreams request received")
+log.warning("Invalid bystreams request received")
 return client.sendError(iq_elt, "bad-request")
 streamhost_elts = list(query_elt.elements(NS_BS, "streamhost"))
 if not streamhost_elts:
 return client.sendError(iq_elt, "bad-request")
 try:
 session_data = client.xep_0065_sid_session[sid]
 except KeyError:
-log.warning(u"Ignoring unexpected BS transfer: {}".format(sid))
+log.warning("Ignoring unexpected BS transfer: {}".format(sid))
 return client.sendError(iq_elt, "not-acceptable")
 peer_jid = session_data["peer_jid"] = jid.JID(iq_elt["from"])
 candidates = []
 nb_sh = len(streamhost_elts)
 for idx, sh_elt in enumerate(streamhost_elts):
 try:
 host, port, jid_ = sh_elt["host"], sh_elt["port"], jid.JID(sh_elt["jid"])
 except KeyError:
-log.warning(u"malformed streamhost element")
+log.warning("malformed streamhost element")
 return client.sendError(iq_elt, "bad-request")
 priority = nb_sh - idx
 if jid_.userhostJID() != peer_jid.userhostJID():
 type_ = XEP_0065.TYPE_PROXY
 else:
 type_ = XEP_0065.TYPE_DIRECT
 candidates.append(Candidate(host, port, type_, priority, jid_))
 for candidate in candidates:
-log.info(u"Candidate proposed: {}".format(candidate))
+log.info("Candidate proposed: {}".format(candidate))
 d = self.getBestCandidate(client, candidates, session_data["hash"])
 d.addCallback(self._ackStream, iq_elt, session_data, client)
 def _ackStream(self, candidate, iq_elt, session_data, client):
 if candidate is None:
 log.info("No streamhost candidate worked, we have to end negotiation")
 return client.sendError(iq_elt, "item-not-found")
-log.info(u"We choose: {}".format(candidate))
+log.info("We choose: {}".format(candidate))
 result_elt = xmlstream.toResponse(iq_elt, "result")
 query_elt = result_elt.addElement((NS_BS, "query"))
 query_elt["sid"] = session_data["id"]
 streamhost_used_elt = query_elt.addElement("streamhost-used")
 streamhost_used_elt["jid"] = candidate.jid.full()
 client.send(result_elt)
+@implementer(iwokkel.IDisco)
 class XEP_0065_handler(XMPPHandler):
-implements(iwokkel.IDisco)
 def __init__(self, plugin_parent):
 self.plugin_parent = plugin_parent
 self.host = plugin_parent.host

Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0065.py @ 3028:ab2696e34d29