bumper/__init__.py

@@ -13,6 +13,7 @@ import os
 import logging
 from base64 import b64decode, b64encode
 from tinydb import TinyDB, Query
+import json
 from tinydb.storages import MemoryStorage
 
 bumper_users_var = contextvars.ContextVar("bumper_users", default=[])
@@ -62,16 +63,24 @@ def os_db_path():
 
 
 def db_get():
-    # Will create the database if it doesn't exist
+    try:
-    db = TinyDB(db_file())
+        # Will create the database if it doesn't exist
+        db = TinyDB(db_file())
 
-    # Will create the tables if they don't exist
+        # Will create the tables if they don't exist
-    db.table("users", cache_size=0)
+        db.table("users", cache_size=0)
-    db.table("clients", cache_size=0)
+        db.table("clients", cache_size=0)
-    db.table("bots", cache_size=0)
+        db.table("bots", cache_size=0)
-    db.table("tokens", cache_size=0)
+        db.table("tokens", cache_size=0)
 
-    return db
+        return db
+    
+    
+    except json.decoder.JSONDecodeError as jerr:
+        bumperlog.error("JsonErr: {} - Doc: {}".format(jerr.msg, jerr.doc))
+
+    except Exception as ex:
+        bumperlog.error(ex)
 
 
 class BumperUser(object):

bumper/mqttserver.py

@@ -73,6 +73,10 @@ class MQTTHelperBot:
     async def start_helper_bot(self):
 
         try:
+            self.Client = MQTTClient(
+                client_id=self.client_id, config={"check_hostname": False}
+            )
+
             await self.Client.connect(
                 "mqtts://{}:{}/".format(self.address[0], self.address[1]),
                 cafile=bumper.ca_cert,
@@ -84,6 +88,8 @@ class MQTTHelperBot:
                 ]
             )
 
+            asyncio.ensure_future(self.get_msg())
+
         except Exception as e:
             helperbotlog.exception("{}".format(e))
 

bumper/xmpp_old_client.py

@@ -0,0 +1,769 @@
+class Client(threading.Thread):
+    IDLE = 0
+    CONNECT = 1
+    INIT = 2
+    BIND = 3
+    READY = 4
+    DISCONNECT = 5
+    UNKNOWN = 0
+    BOT = 1
+    CONTROLLER = 2
+
+    def __init__(self, thread_id, connection, client_address):
+        threading.Thread.__init__(self)
+        self.id = thread_id
+        self.name = "XMPP_Client_{}".format(client_address[0])
+        self.type = self.UNKNOWN
+        self.state = self.IDLE
+        self.connection = connection
+        self.address = client_address[0]
+        self.clientresource = ""
+        self.devclass = ""
+        self.bumper_jid = ""
+        self.uid = ""
+        self.log_sent_message = False  # Set to true to log sends
+        self.log_incoming_data = True  # Set to true to log sends
+
+        xmppserverlog.debug(
+            "new client thread init for client with ip {}".format(self.address)
+        )
+
+    def send(self, command):
+        try:
+            if not self.connection._closed:
+                if self.log_sent_message:
+                    xmppserverlog.debug("send {} - {}".format(self.address, command))
+                self.connection.send(command.encode())
+
+        except BrokenPipeError as e:
+            xmppserverlog.debug("{}".format(e))
+            self._set_state("DISCONNECT")
+
+        except ConnectionResetError as e:
+            xmppserverlog.debug("{}".format(e))
+            self._set_state("DISCONNECT")
+
+        except ConnectionAbortedError as e:
+            xmppserverlog.debug("{}".format(e))
+            self._set_state("DISCONNECT")
+
+        except OSError as e:
+            xmppserverlog.debug("{}".format(e))
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _disconnect(self):
+        try:
+
+            bot = bumper.bot_get(self.uid)
+            if bot:
+                bumper.bot_set_xmpp(bot["did"], False)
+
+            client = bumper.client_get(self.clientresource)
+            if client:
+                bumper.client_set_xmpp(client["resource"], False)
+
+            self.connection.close()
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _tag_strip_uri(self, tag):
+        try:
+            if tag[0] == "{":
+                _, _, tag = tag[1:].partition("}")
+            return tag
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _set_state(self, state):
+        try:
+            new_state = getattr(Client, state)
+            if self.state > new_state:
+                raise Exception(
+                    "{} illegal state change {}->{}".format(
+                        self.address, self.state, new_state
+                    )
+                )
+
+            xmppserverlog.debug("{} state: {}".format(self.address, state))
+
+            self.state = new_state
+
+            if new_state == 5:
+                self._disconnect()
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_ctl(self, xml, data):
+        try:
+
+            if "roster" in data:
+                # Return not-implemented for roster
+                self.send(
+                    '<iq type="error" id="{}"><error type="cancel" code="501"><feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas"/></error></iq>'.format(
+                        xml.get("id")
+                    )
+                )
+                return
+
+            if xml.get("type") == "set":
+                if (
+                    "com:sf" in data and xml.get("to") == "rl.ecorobot.net"
+                ):  # Android bind? Not sure what this does yet.
+                    self.send(
+                        '<iq id="{}" to="{}@{}/{}" from="rl.ecorobot.net" type="result"/>'.format(
+                            xml.get("id"),
+                            self.uid,
+                            XMPPServer.server_id,
+                            self.clientresource,
+                        )
+                    )
+
+            if xml[0][0]:
+                ctl = xml[0][0]
+                if ctl.get("admin") and self.type == self.BOT:
+                    xmppserverlog.debug(
+                        "admin username received from bot: {}".format(ctl.get("admin"))
+                    )
+                    XMPPServer.client_id = ctl.get("admin")
+                    return
+
+            # forward
+            for client in XMPPServer.clients:
+                if (
+                    client.bumper_jid != self.bumper_jid
+                    and client.state == client.READY
+                ):
+                    ctl_to = xml.get("to")
+                    xml.attrib["from"] = "{}".format(self.bumper_jid)
+                    rxmlstring = ET.tostring(xml).decode("utf-8")
+                    # clean up string to remove namespaces added by ET
+                    rxmlstring = rxmlstring.replace("xmlns:ns0=", "xmlns=")
+                    rxmlstring = rxmlstring.replace("ns0:", "")
+                    rxmlstring = rxmlstring.replace('iq xmlns="com:ctl"', "iq")
+                    rxmlstring = rxmlstring.replace("<query", '<query xmlns="com:ctl"')
+
+                    if client.type == self.BOT:
+                        if client.uid.lower() in ctl_to.lower():
+                            xmppserverlog.info(
+                                "Sending ctl to bot: {}".format(rxmlstring)
+                            )
+                            client.send(rxmlstring)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_ping(self, xml, data):
+        try:
+            if xml.get("to").find("@") == -1:  # No to address
+                # Ping to server - respond
+                pingresp = '<iq type="result" id="{}" from="{}" />'.format(
+                    xml.get("id"), xml.get("to")
+                )
+                # xmppserverlog.debug("Server Ping resp: {}".format(pingresp))
+                self.send(pingresp)
+
+            else:
+                pingto = xml.get("to")
+                pingfrom = self.bumper_jid
+
+                xml.attrib["from"] = pingfrom
+                pingstring = ET.tostring(xml).decode("utf-8")
+                # clean up string to remove namespaces added by ET
+                pingstring = pingstring.replace("xmlns:ns0=", "xmlns=")
+                pingstring = pingstring.replace("ns0:", "")
+                pingstring = pingstring.replace('iq xmlns="com:ctl"', "iq")
+                pingstring = pingstring.replace("<query", '<query xmlns="com:ctl"')
+
+                for client in XMPPServer.clients:
+                    if (
+                        client.bumper_jid != self.bumper_jid
+                        and client.state == client.READY
+                    ):
+                        if pingto.lower() in client.bumper_jid.lower():
+                            pingstring = '<iq type="result" id="{}" from="{}" to="{}" />'.format(
+                                xml.get("id"), pingfrom, pingto
+                            )
+                            xmppserverlog.debug(
+                                "ping from {} to {}".format(pingfrom, pingto)
+                            )
+                            client.send(pingstring)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_result(self, xml, data):
+        try:
+            ctl_to = xml.get("to")
+            xml.attrib["from"] = self.bumper_jid
+            if (
+                "errno='103' error='permission denied," in data
+            ):  # No permissions, usually if bot was last on Ecovac network
+                if self.type == self.BOT:
+                    xquery = xml.getchildren()
+                    ctl = xquery[0].getchildren()
+                    ctlerr = ctl[0].attrib["error"]
+                    adminuser = ctlerr.replace("permission denied, please contact ", "")
+                    adminuser = adminuser.replace(" ", "")
+                    if not (
+                        adminuser.startswith("fuid_") or bumper.use_auth
+                    ):  # if not fuid_ then its ecovacs OR ignore bumper auth
+                        # TODO: Implement auth later, should this user have access to bot?
+
+                        # Add user jid to bot
+                        newuser = ctl_to.split("/")[0]
+                        adduser = '<iq type="set" id="{}" from="{}" to="{}"><query xmlns="com:ctl"><ctl td="AddUser" id="0000" jid="{}" /></query></iq>'.format(
+                            uuid.uuid4(), adminuser, self.bumper_jid, newuser
+                        )
+                        xmppserverlog.debug("Add User: {}".format(adduser))
+                        self.send(adduser)
+
+                        # Add user ACs - Manage users, settings, and clean (full access)
+                        adduseracs = '<iq type="set" id="{}" from="{}" to="{}"><query xmlns="com:ctl"><ctl td="SetAC" id="1111" jid="{}"><acs><ac name="userman" allow="1"/><ac name="setting" allow="1"/><ac name="clean" allow="1"/></acs></ctl></query></iq>'.format(
+                            uuid.uuid4(), adminuser, self.bumper_jid, newuser
+                        )
+                        xmppserverlog.debug("Add User ACs: {}".format(adduseracs))
+                        self.send(adduseracs)
+
+                        # GetUserInfo - Just to confirm it set correctly
+                        self.send(
+                            '<iq type="set" id="{}" from="{}" to="{}"><query xmlns="com:ctl"><ctl td="GetUserInfo" id="4444" /><UserInfos/></query></iq>'.format(
+                                uuid.uuid4(), adminuser, self.bumper_jid
+                            )
+                        )
+
+            else:
+                rxmlstring = ET.tostring(xml).decode("utf-8")
+                # clean up string to remove namespaces added by ET
+                rxmlstring = rxmlstring.replace("xmlns:ns0=", "xmlns=")
+                rxmlstring = rxmlstring.replace("ns0:", "")
+                rxmlstring = rxmlstring.replace('iq xmlns="com:ctl"', "iq")
+                rxmlstring = rxmlstring.replace("<query", '<query xmlns="com:ctl"')
+                if self.type == self.BOT:
+                    if ctl_to == "de.ecorobot.net":  # Send to all clients
+                        xmppserverlog.debug(
+                            "Sending to all clients because of de: {}".format(
+                                rxmlstring
+                            )
+                        )
+                        for client in XMPPServer.clients:
+                            client.send(rxmlstring)
+
+                if xml.get("to").find("@") == -1:  # No to address
+                    ctl_to = xml.get("to")
+                else:
+                    ctl_to = "{}@ecouser.net".format(ctl_to.split("@")[0])
+
+                for client in XMPPServer.clients:
+                    if (
+                        client.bumper_jid != self.bumper_jid
+                        and client.state == client.READY
+                    ):
+                        if not "@" in ctl_to:  # No user@, send to all clients?
+                            # TODO: Revisit later, this may be wrong
+                            client.send(rxmlstring)
+
+                        elif (
+                            client.uid.lower() in ctl_to.lower()
+                        ):  # If client matches TO=
+                            xmppserverlog.debug(
+                                "Sending from {} to client {}: {}".format(
+                                    self.uid, client.uid, rxmlstring
+                                )
+                            )
+                            client.send(rxmlstring)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_connect(self, data, xml=None):
+        try:
+
+            if self.state == self.CONNECT:
+                if xml == None:
+                    # Client first connecting, send our features
+                    if data.decode("utf-8").find("jabber:client") > -1:
+                        sc = data.decode("utf-8").find("to=")
+                        ec = data.decode("utf-8").find(".ecorobot.net")
+                        if ec > -1:
+                            self.devclass = data.decode("utf-8")[sc + 4 : ec]
+                        # ack jabbr:client
+                        # no STARTTLS
+                        self.send(
+                            '<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(
+                                XMPPServer.server_id
+                            )
+                        )
+                        # with STARTTLS
+                        # self.send('<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns:tls="http://www.ietf.org/rfc/rfc2595.txt" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(XMPPServer.server_id))
+                        time.sleep(0.25)
+                        # send authentication support for iq-auth (fallback) and SASL
+                        self.send(
+                            '<stream:features><auth xmlns="http://jabber.org/features/iq-auth"/><mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl"><mechanism>PLAIN</mechanism></mechanisms></stream:features>'
+                        )
+                        # self.send('<stream:features><auth xmlns="http://jabber.org/features/iq-auth"/></stream:features>')
+
+                    else:
+                        self.send("</stream>")
+
+                else:
+                    if "jabber:iq:auth" in xml.tag:  # Handle iq-auth
+                        self._handle_iq_auth(xml)
+                    elif (
+                        "urn:ietf:params:xml:ns:xmpp-sasl" in xml.tag
+                    ):  # Handle SASL Auth
+                        self._handle_sasl_auth(xml)
+                    else:
+                        xmppserverlog.error("Couldn't handle: {}".format(xml))
+
+            elif self.state == self.INIT:
+                if xml == None:
+                    # Client getting session after authentication
+                    if data.decode("utf-8").find("jabber:client") > -1:
+                        # ack jabbr:client
+                        self.send(
+                            '<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(
+                                XMPPServer.server_id
+                            )
+                        )
+                        time.sleep(0.25)
+                        # session
+                        self.send(
+                            '<stream:features><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"/><session xmlns="urn:ietf:params:xml:ns:xmpp-session"/></stream:features>'
+                        )
+
+                else:  # Handle init bind
+                    if len(xml):
+                        child = self._tag_strip_uri(xml[0].tag)
+                    else:
+                        child = None
+
+                    if xml.tag == "iq":
+                        if child == "bind":
+                            self._handle_bind(xml)
+                    else:
+                        xmppserverlog.error("Couldn't handle: {}".format(xml))
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_iq_auth(self, data):
+        try:
+            xml = ET.fromstring(data.decode("utf-8"))
+            ctl = xml[0][0]
+            xmppserverlog.info("IQ AUTH XML: {}".format(xml))
+            # Received username and auth tag, send username/password requirement
+            if (
+                xml.get("type") == "get"
+                and "auth}username" in ctl.tag
+                and self.type == self.UNKNOWN
+            ):
+                self.send(
+                    '<iq type="result" id="{}"><query xmlns="jabber:iq:auth"><username/><password/></query></iq>'.format(
+                        xml.get("id")
+                    )
+                )
+
+            # Received username, password, resource - Handle auth here and return pass or fail
+            if (
+                xml.get("type") == "set"
+                and "auth}username" in ctl.tag
+                and self.type == self.UNKNOWN
+            ):
+                xmlauth = xml[0].getchildren()
+                # uid = ""
+                password = ""
+                resource = ""
+                for aitem in xmlauth:
+                    if "username" in aitem.tag:
+                        self.uid = aitem.text
+
+                    elif "password" in aitem.tag:
+                        password = aitem.text.split("/")[2]
+                        authcode = password
+
+                    elif "resource" in aitem.tag:
+                        self.clientresource = aitem.text
+                        resource = self.clientresource
+
+                if not self.uid.startswith("fuid"):
+
+                    # Need sample data to see details here
+                    bumper.bot_add("", self.uid, "", resource, "eco-legacy")
+                    xmppserverlog.info("bot authenticated {}".format(self.uid))
+
+                    # Client authenticated, move to next state
+                    self._set_state("INIT")
+
+                    # Successful auth
+                    self.send('<iq type="result" id="{}"/>'.format(xml.get("id")))
+
+                else:
+                    auth = False
+                    if bumper.check_authcode(self.uid, authcode):
+                        auth = True
+                    elif bumper.use_auth == False:
+                        auth = True
+
+                    if auth:
+                        bumper.client_add(self.uid, "bumper", self.clientresource)
+                        xmppserverlog.debug("client authenticated {}".format(self.uid))
+
+                        # Client authenticated, move to next state
+                        self._set_state("INIT")
+
+                        # Successful auth
+                        self.send('<iq type="result" id="{}"/>'.format(xml.get("id")))
+
+                    else:
+                        # Failed auth
+                        self.send(
+                            '<iq type="error" id="{}"><error code="401" type="auth"><not-authorized xmlns="urn:ietf:params:xml:ns:xmpp-stanzas"/></error></iq>'.format(
+                                xml.get("id")
+                            )
+                        )
+
+        except ET.ParseError as e:
+            if "no element found" in e.msg:
+                xmppserverlog.debug(
+                    "xml parse error - {} - {}".format(data.decode("utf-8"), e)
+                )
+            elif "not well-formed (invalid token)" in e.msg:
+                xmppserverlog.debug(
+                    "xml parse error - {} - {}".format(data.decode("utf-8"), e)
+                )
+            else:
+                xmppserverlog.debug(
+                    "xml parse error - {} - {}".format(data.decode("utf-8"), e)
+                )
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_sasl_auth(self, xml):
+        try:
+
+            saslauth = base64.b64decode(xml.text).decode("utf-8").split("/")
+            username = saslauth[0]
+            username = saslauth[0].split("\x00")[1]
+            self.uid = username
+            if len(saslauth) > 1:
+                resource = saslauth[1]
+                self.clientresource = resource
+            elif len(saslauth[0].split("\x00")) > 2:
+                resource = saslauth[0].split("\x00")[2]
+                self.clientresource = resource
+
+            if len(saslauth) > 2:
+                authcode = saslauth[2]
+
+            if not self.uid.startswith("fuid"):
+                # Need sample data to see details here
+                bumper.bot_add(self.uid, self.uid, self.devclass, "atom", "eco-legacy")
+                self.type = self.BOT
+                xmppserverlog.info("bot authenticated {}".format(self.uid))
+                # Send response
+                self.send(
+                    '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
+                )  # Success
+
+                # Client authenticated, move to next state
+                self._set_state("INIT")
+
+            else:
+                auth = False
+                if bumper.check_authcode(self.uid, authcode):
+                    auth = True
+                elif bumper.use_auth == False:
+                    auth = True
+
+                if auth:
+                    self.type = self.CONTROLLER
+                    bumper.client_add(self.uid, "bumper", self.clientresource)
+                    xmppserverlog.debug("client authenticated {}".format(self.uid))
+
+                    # Client authenticated, move to next state
+                    self._set_state("INIT")
+
+                    # Send response
+                    self.send(
+                        '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
+                    )  # Success
+
+                else:
+                    # Failed to authenticate
+                    self.send(
+                        '<response xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
+                    )  # Fail
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_bind(self, xml):
+        try:
+
+            bot = bumper.bot_get(self.uid)
+            if bot:
+                bumper.bot_set_xmpp(bot["did"], True)
+
+            client = bumper.client_get(self.clientresource)
+            if client:
+                bumper.client_set_xmpp(client["resource"], True)
+
+            clientbindxml = xml.getchildren()
+            clientresourcexml = clientbindxml[0].getchildren()
+            if self.devclass:  # its a bot
+                self.name = "XMPP_Client_{}_{}".format(self.uid, self.devclass)
+                self.bumper_jid = "{}@{}.ecorobot.net/atom".format(
+                    self.uid, self.devclass
+                )
+                xmppserverlog.debug("new bot {}".format(self.uid))
+                res = '<iq type="result" id="{}"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><jid>{}</jid></bind></iq>'.format(
+                    xml.get("id"), self.bumper_jid
+                )
+            elif len(clientresourcexml) > 0:
+                self.clientresource = clientresourcexml[0].text
+                self.name = "XMPP_Client_{}".format(self.clientresource)
+                self.bumper_jid = "{}@{}/{}".format(
+                    self.uid, XMPPServer.server_id, self.clientresource
+                )
+                xmppserverlog.debug(
+                    "new client {} using resource {}".format(
+                        self.uid, self.clientresource
+                    )
+                )
+                res = '<iq type="result" id="{}"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><jid>{}</jid></bind></iq>'.format(
+                    xml.get("id"), self.bumper_jid
+                )
+            else:
+                self.name = "XMPP_Client_{}_{}".format(self.uid, self.address)
+                self.bumper_jid = "{}@{}".format(self.uid, XMPPServer.server_id)
+                xmppserverlog.debug("new client {}".format(self.uid))
+                res = '<iq type="result" id="{}"><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"><jid>{}</jid></bind></iq>'.format(
+                    xml.get("id"), self.bumper_jid
+                )
+
+            self._set_state("BIND")
+            self.send(res)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_session(self, xml):
+        try:
+            res = '<iq type="result" id="{}" />'.format(xml.get("id"))
+            self._set_state("READY")
+            self.send(res)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_presence(self, xml):
+        try:
+
+            if len(xml) and xml[0].tag == "status":
+                xmppserverlog.debug(
+                    "bot presence {} ".format(ET.tostring(xml, encoding="utf-8"))
+                )
+                # Most likely a bot, possibly hello world in text
+
+                # Send dummy return
+                self.send(
+                    '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
+                )
+
+                # If it is a BOT, send extras
+                if self.type == self.BOT:
+                    # get device info
+                    self.send(
+                        '<iq type="set" id="14" to="{}" from="{}"><query xmlns="com:ctl"><ctl td="GetDeviceInfo"/></query></iq>'.format(
+                            self.bumper_jid, XMPPServer.server_id
+                        )
+                    )
+
+            else:
+                xmppserverlog.debug(
+                    "client presence - {} ".format(ET.tostring(xml, encoding="utf-8"))
+                )
+
+                if xml.get("type") == "available":
+                    xmppserverlog.debug(
+                        "client presence available - {} ".format(
+                            ET.tostring(xml, encoding="utf-8")
+                        )
+                    )
+                    # Send dummy return
+                    self.send(
+                        '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
+                    )
+                elif xml.get("type") == "unavailable":
+                    xmppserverlog.debug(
+                        "client presence unavailable (DISCONNECT) - {} ".format(
+                            ET.tostring(xml, encoding="utf-8")
+                        )
+                    )
+
+                    self._set_state("DISCONNECT")
+                else:
+                    # Sometimes the android app sends these
+                    xmppserverlog.debug(
+                        "client presence (UNKNOWN) - {} ".format(
+                            ET.tostring(xml, encoding="utf-8")
+                        )
+                    )
+                    # Send dummy return
+                    self.send(
+                        '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
+                    )
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _parse_data(self, data):
+
+        if data.decode("utf-8").startswith(
+            "<?xml"
+        ):  # Strip <?xml and add artificial root
+            newdata = (
+                re.sub(r"(<\?xml[^>]+\?>)", r"<root>", data.decode("utf-8")) + "</root>"
+            )
+
+        else:
+            newdata = "<root>{}</root>".format(
+                data.decode("utf-8")
+            )  # Add artificial root
+
+        try:
+            root = ET.fromstring(newdata)
+            for item in root.iter():
+                if item.tag != "root":
+                    if item.tag == "iq":
+                        if self.log_incoming_data:
+                            xmppserverlog.debug(
+                                "from {} - {}".format(
+                                    self.address,
+                                    str(
+                                        ET.tostring(item, encoding="utf-8").decode(
+                                            "utf-8"
+                                        )
+                                    ).replace("ns0:", ""),
+                                )
+                            )
+                        self._handle_iq(item, newdata)
+                        item.clear()
+
+                    elif "auth" in item.tag:
+                        if "urn:ietf:params:xml:ns:xmpp-sasl" in item.tag:  # SASL Auth
+                            self._handle_sasl_auth(item)
+                            item.clear()
+
+                    elif "presence" in item.tag:
+                        self._handle_presence(item)
+                        item.clear()
+
+                    else:
+                        if self.log_incoming_data:
+                            xmppserverlog.debug(
+                                "Unparsed Item - {}".format(
+                                    str(
+                                        ET.tostring(item, encoding="utf-8").decode(
+                                            "utf-8"
+                                        )
+                                    ).replace("ns0:", "")
+                                )
+                            )
+
+        except ET.ParseError as e:
+            if (
+                "no element found" in e.msg
+            ):  # Element not closed or not all bytes received
+                # Happens wth connect stream often
+                if "<stream:stream " in newdata:
+                    if self.state == self.CONNECT or self.state == self.INIT:
+                        self._handle_connect(newdata.encode("utf-8"))
+                else:
+                    if not (newdata == "" or newdata == " "):
+                        xmppserverlog.error(
+                            "xml parse error - {} - {}".format(newdata, e)
+                        )
+
+            elif "not well-formed (invalid token)" in e.msg:
+                # If a lone </stream:stream> - client is signalling end of session/disconnect
+                if not "</stream:stream>" in newdata:
+                    xmppserverlog.error("xml parse error - {} - {}".format(newdata, e))
+                else:
+                    self.send("</stream:stream>")  # Close stream
+
+            else:
+                if "<stream:stream" in newdata:  # Handle start stream and connect
+                    if self.state == self.CONNECT or self.state == self.INIT:
+                        xmppserverlog.debug(
+                            "Handling connect data - {}".format(newdata)
+                        )
+                        self._handle_connect(newdata.encode("utf-8"))
+                else:
+                    if not "</stream:stream>" in newdata:
+                        xmppserverlog.error(
+                            "xml parse error - {} - {}".format(newdata, e)
+                        )
+                    else:
+                        self.send("</stream:stream>")  # Close stream
+                        self._set_state("DISCONNECT")
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def _handle_iq(self, xml, data):
+        try:
+            if len(xml):
+                child = self._tag_strip_uri(xml[0].tag)
+            else:
+                child = None
+
+            if xml.tag == "iq":
+                if child == "bind":
+                    self._handle_bind(xml)
+                elif child == "session":
+                    self._handle_session(xml)
+                elif child == "ping":
+                    self._handle_ping(xml, data)
+                elif child == "query":
+                    if self.type == self.BOT:
+                        self._handle_result(xml, data)
+                    else:
+                        self._handle_ctl(xml, data)
+                elif xml.get("type") == "result":
+                    if self.type == self.BOT:
+                        self._handle_result(xml, data)
+                    else:
+                        self._handle_result(xml, data)
+                elif xml.get("type") == "set":
+                    if self.type == self.BOT:
+                        self._handle_result(xml, data)
+                    else:
+                        self._handle_result(xml, data)
+
+        except Exception as e:
+            xmppserverlog.exception("{}".format(e))
+
+    def run(self):
+        # xmppserverlog.info('client connected - {}'.format(self.address))
+        await self._set_state("CONNECT")
+        while not self.state == self.DISCONNECT and not self.connection._closed:
+            data = b""
+            time.sleep(0.1)
+            if not self.connection._closed:
+                try:
+                    data = self.connection.recv(4096)
+                    if data != b"":
+                        self._parse_data(data)
+                except ConnectionResetError as e:
+                    xmppserverlog.debug("{}".format(e))
+                except OSError as e:
+                    xmppserverlog.debug("{}".format(e))
+                except Exception as e:
+                    xmppserverlog.exception("{}".format(e))

bumper/xmppserver.py

@@ -6,6 +6,7 @@ import base64
 import ssl
 import contextvars
 import bumper
+import asyncio
 
 xmppserverlog = logging.getLogger("xmppserver")
 
@@ -19,88 +20,116 @@ class XMPPServer:
     def __init__(self, address):
         # Initialize bot server
         self.address = address
+        self.aclients = {}  # task -> (reader, writer)
 
-    def run(self, run_async=False):
+    async def async_server(self):
-        if run_async:
+        xmppserverlog.debug(
-            xmppserverlog.debug("Starting XMPPServer Thread: 1")
+            "listening on {}:{}".format(self.address[0], self.address[1])
-            self.xmppthread = Thread(name="XMPPServer_Thread", target=self.run_server)
+        )
-            self.xmppthread.setDaemon(True)
+        server = await asyncio.start_server(
-            self.xmppthread.start()
+            self.accept_client, self.address[0], self.address[1]
-
-        else:
-            try:
-                self.run_server()
-            except KeyboardInterrupt:
-                self.disconnect()
-
-    def run_server(self):
-        logging.info("Starting XMPP Server at {}".format(self.address))
-        print("Starting XMPP Server at {}".format(self.address))
-
-        # xmppserverlog.setLevel(logging.DEBUG)
-
-        # Set SSL Context
-        self.ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-        self.ssl_ctx.load_cert_chain(
-            certfile=bumper.server_cert, keyfile=bumper.server_key
         )
 
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        await server.serve_forever()
-        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
-        try:
+    # self.clients = {}  # task -> (reader, writer)
-            self.socket.bind(self.address)
-            self.socket.listen(5)
 
-            xmppserverlog.debug(
+    def accept_client(self, client_reader, client_writer):
-                "listening on {}:{}".format(self.address[0], self.address[1])
+        # task = asyncio.Task(self.handle_client(client_reader, client_writer))
-            )
+        aclient = XMPPAsyncClient(client_reader, client_writer)
-            while not self.exit_flag:
+        task = asyncio.Task(aclient.handle_async_client())
-                connection, client_address = self.socket.accept()
+        self.aclients[task] = (client_reader, client_writer)
 
-                # disconnect any clients with this ip
+        def client_done(task):
-                for client in self.clients:
+            del self.aclients[task]
-                    if client.address == client_address[0]:
+            client_writer.close()
-                        xmppserverlog.debug(
+            xmppserverlog.info("End Connection")
-                            "disconnecting existing client {} with resource {}".format(
-                                client.address, client.clientresource
-                            )
-                        )
-                        client._disconnect()
-                        self.remove_client_byip(client.address)
 
-                xmppserverlog.debug(
+        clientaddr = client_writer.get_extra_info("peername")
-                    "starting new client with ip {}".format(client_address[0])
+        xmppserverlog.info("New Connection from {}".format(clientaddr))
-                )
+        task.add_done_callback(client_done)
-                thread_id = uuid.uuid4()
-                client = Client(thread_id, connection, client_address)
-                client.setDaemon(True)
-                client.start()
-                self.clients.append(client)
 
-        except PermissionError as e:
+    # def run(self, run_async=False):
-            if "bind" in e.strerror:
+    #     if run_async:
-                xmppserverlog.exception(
+    #         xmppserverlog.debug("Starting XMPPServer Thread: 1")
-                    "Error binding XMPPServer, exiting. Try using a different hostname or IP - {}".format(
+    #         self.xmppthread = Thread(name="XMPPServer_Thread", target=self.run_server)
-                        e
+    #         self.xmppthread.setDaemon(True)
-                    )
+    #         self.xmppthread.start()
-                )
-                exit(1)
 
-        except Exception as e:
+    #     else:
-            xmppserverlog.exception("{}".format(e))
+    #         try:
-            exit(1)
+    #             self.run_server()
+    #         except KeyboardInterrupt:
+    #             self.disconnect()
 
-        except KeyboardInterrupt as e:
+    # def run_server(self):
-            xmppserverlog.exception("{}".format(e))
+    #     logging.info("Starting XMPP Server at {}".format(self.address))
+    #     print("Starting XMPP Server at {}".format(self.address))
 
-        finally:
+    #     # xmppserverlog.setLevel(logging.DEBUG)
-            connection.shutdown(socket.SHUT_RDWR)
-            connection.close()
-            self.disconnect()
-            xmppserverlog.info("disconnecting")
 
-        self.socket.close()
+    #     # Set SSL Context
+    #     self.ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    #     self.ssl_ctx.load_cert_chain(
+    #         certfile=bumper.server_cert, keyfile=bumper.server_key
+    #     )
+
+    #     self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    #     self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+    #     try:
+    #         self.socket.bind(self.address)
+    #         self.socket.listen(5)
+
+    #         xmppserverlog.debug(
+    #             "listening on {}:{}".format(self.address[0], self.address[1])
+    #         )
+    #         while not self.exit_flag:
+    #             connection, client_address = self.socket.accept()
+
+    #             # disconnect any clients with this ip
+    #             for client in self.clients:
+    #                 if client.address == client_address[0]:
+    #                     xmppserverlog.debug(
+    #                         "disconnecting existing client {} with resource {}".format(
+    #                             client.address, client.clientresource
+    #                         )
+    #                     )
+    #                     client._disconnect()
+    #                     self.remove_client_byip(client.address)
+
+    #             xmppserverlog.debug(
+    #                 "starting new client with ip {}".format(client_address[0])
+    #             )
+    #             thread_id = uuid.uuid4()
+    #             client = XMPPAsyncClient(thread_id, connection, client_address)
+    #             client.setDaemon(True)
+    #             client.start()
+    #             self.clients.append(client)
+
+    #     except PermissionError as e:
+    #         if "bind" in e.strerror:
+    #             xmppserverlog.exception(
+    #                 "Error binding XMPPServer, exiting. Try using a different hostname or IP - {}".format(
+    #                     e
+    #                 )
+    #             )
+    #             exit(1)
+
+    #     except Exception as e:
+    #         xmppserverlog.exception("{}".format(e))
+    #         exit(1)
+
+    #     except KeyboardInterrupt as e:
+    #         xmppserverlog.exception("{}".format(e))
+
+    #     finally:
+    #         connection.shutdown(socket.SHUT_RDWR)
+    #         connection.close()
+    #         self.disconnect()
+    #         xmppserverlog.info("disconnecting")
+
+    #     self.socket.close()
 
     def disconnect(self):
         try:
@@ -148,7 +177,7 @@ class XMPPServer:
                 self.clients.remove(client)
 
 
-class Client(threading.Thread):
+class XMPPAsyncClient:
     IDLE = 0
     CONNECT = 1
     INIT = 2
@@ -159,14 +188,12 @@ class Client(threading.Thread):
     BOT = 1
     CONTROLLER = 2
 
-    def __init__(self, thread_id, connection, client_address):
+    def __init__(self, client_reader, client_writer):
-        threading.Thread.__init__(self)
-        self.id = thread_id
-        self.name = "XMPP_Client_{}".format(client_address[0])
         self.type = self.UNKNOWN
         self.state = self.IDLE
-        self.connection = connection
+        self.address = client_writer.get_extra_info("peername")
-        self.address = client_address[0]
+        self.client_reader = client_reader
+        self.client_writer = client_writer
         self.clientresource = ""
         self.devclass = ""
         self.bumper_jid = ""
@@ -174,28 +201,45 @@ class Client(threading.Thread):
         self.log_sent_message = False  # Set to true to log sends
         self.log_incoming_data = True  # Set to true to log sends
 
-        xmppserverlog.debug(
+        xmppserverlog.debug("new client with ip {}".format(self.address))
-            "new client thread init for client with ip {}".format(self.address)
-        )
 
-    def send(self, command):
+    async def handle_async_client(self):
+        # xmppserverlog.info('client connected - {}'.format(self.address))
+        await self._set_state("CONNECT")
+        pingtask = asyncio.Task(self.send_ping(30))        
+        while not self.state == self.DISCONNECT:
+            data = await self.client_reader.read(4096)
+            # data = await asyncio.wait_for(client_reader.readline(), timeout=10.0)
+            if data is None:
+                xmppserverlog.warning("Received no data")
+                # exit loop and disconnect
+                return
+
+            await self._parse_data(data)
+
+        # exit loop and disconnect
+        return
+
+    async def send(self, command):
         try:
-            if not self.connection._closed:
+            # if not self.connection._closed:
-                if self.log_sent_message:
+            if self.log_sent_message:
-                    xmppserverlog.debug("send {} - {}".format(self.address, command))
+                xmppserverlog.debug("send {} - {}".format(self.address, command))
-                self.connection.send(command.encode())
+            # self.connection.send(command.encode())
+            self.client_writer.write(command.encode())
+            await self.client_writer.drain()
 
         except BrokenPipeError as e:
             xmppserverlog.debug("{}".format(e))
-            self._set_state("DISCONNECT")
+            await self._set_state("DISCONNECT")
 
         except ConnectionResetError as e:
             xmppserverlog.debug("{}".format(e))
-            self._set_state("DISCONNECT")
+            await self._set_state("DISCONNECT")
 
         except ConnectionAbortedError as e:
             xmppserverlog.debug("{}".format(e))
-            self._set_state("DISCONNECT")
+            await self._set_state("DISCONNECT")
 
         except OSError as e:
             xmppserverlog.debug("{}".format(e))
@@ -203,7 +247,7 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _disconnect(self):
+    async def _disconnect(self):
         try:
 
             bot = bumper.bot_get(self.uid)
@@ -214,12 +258,12 @@ class Client(threading.Thread):
             if client:
                 bumper.client_set_xmpp(client["resource"], False)
 
-            self.connection.close()
+            self.client_writer.close()
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _tag_strip_uri(self, tag):
+    async def _tag_strip_uri(self, tag):
         try:
             if tag[0] == "{":
                 _, _, tag = tag[1:].partition("}")
@@ -228,9 +272,9 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _set_state(self, state):
+    async def _set_state(self, state):
         try:
-            new_state = getattr(Client, state)
+            new_state = getattr(XMPPAsyncClient, state)
             if self.state > new_state:
                 raise Exception(
                     "{} illegal state change {}->{}".format(
@@ -243,17 +287,17 @@ class Client(threading.Thread):
             self.state = new_state
 
             if new_state == 5:
-                self._disconnect()
+                await self._disconnect()
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_ctl(self, xml, data):
+    async def _handle_ctl(self, xml, data):
         try:
 
             if "roster" in data:
                 # Return not-implemented for roster
-                self.send(
+                await self.send(
                     '<iq type="error" id="{}"><error type="cancel" code="501"><feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas"/></error></iq>'.format(
                         xml.get("id")
                     )
@@ -264,7 +308,7 @@ class Client(threading.Thread):
                 if (
                     "com:sf" in data and xml.get("to") == "rl.ecorobot.net"
                 ):  # Android bind? Not sure what this does yet.
-                    self.send(
+                    await self.send(
                         '<iq id="{}" to="{}@{}/{}" from="rl.ecorobot.net" type="result"/>'.format(
                             xml.get("id"),
                             self.uid,
@@ -307,7 +351,7 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_ping(self, xml, data):
+    async def _handle_ping(self, xml, data):
         try:
             if xml.get("to").find("@") == -1:  # No to address
                 # Ping to server - respond
@@ -315,7 +359,7 @@ class Client(threading.Thread):
                     xml.get("id"), xml.get("to")
                 )
                 # xmppserverlog.debug("Server Ping resp: {}".format(pingresp))
-                self.send(pingresp)
+                await self.send(pingresp)
 
             else:
                 pingto = xml.get("to")
@@ -346,7 +390,14 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_result(self, xml, data):
+    async def send_ping(self, time):
+        if not self.state == 5: #disconnected
+            pingstring = "<iq from='{}' to='{}' id='s2c1' type='get'><ping xmlns='urn:xmpp:ping'/></iq>".format(XMPPServer.server_id, self.bumper_jid)
+            await self.send(pingstring)
+            await asyncio.sleep(time)
+            asyncio.Task(self.send_ping(time))
+        
+    async def _handle_result(self, xml, data):
         try:
             ctl_to = xml.get("to")
             xml.attrib["from"] = self.bumper_jid
@@ -370,17 +421,17 @@ class Client(threading.Thread):
                             uuid.uuid4(), adminuser, self.bumper_jid, newuser
                         )
                         xmppserverlog.debug("Add User: {}".format(adduser))
-                        self.send(adduser)
+                        await self.send(adduser)
 
                         # Add user ACs - Manage users, settings, and clean (full access)
                         adduseracs = '<iq type="set" id="{}" from="{}" to="{}"><query xmlns="com:ctl"><ctl td="SetAC" id="1111" jid="{}"><acs><ac name="userman" allow="1"/><ac name="setting" allow="1"/><ac name="clean" allow="1"/></acs></ctl></query></iq>'.format(
                             uuid.uuid4(), adminuser, self.bumper_jid, newuser
                         )
                         xmppserverlog.debug("Add User ACs: {}".format(adduseracs))
-                        self.send(adduseracs)
+                        await self.send(adduseracs)
 
                         # GetUserInfo - Just to confirm it set correctly
-                        self.send(
+                        await self.send(
                             '<iq type="set" id="{}" from="{}" to="{}"><query xmlns="com:ctl"><ctl td="GetUserInfo" id="4444" /><UserInfos/></query></iq>'.format(
                                 uuid.uuid4(), adminuser, self.bumper_jid
                             )
@@ -430,7 +481,7 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_connect(self, data, xml=None):
+    async def _handle_connect(self, data, xml=None):
         try:
 
             if self.state == self.CONNECT:
@@ -443,30 +494,32 @@ class Client(threading.Thread):
                             self.devclass = data.decode("utf-8")[sc + 4 : ec]
                         # ack jabbr:client
                         # no STARTTLS
-                        self.send(
+                        await self.send(
                             '<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(
                                 XMPPServer.server_id
                             )
                         )
                         # with STARTTLS
-                        # self.send('<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns:tls="http://www.ietf.org/rfc/rfc2595.txt" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(XMPPServer.server_id))
+                        # await self.send('<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns:tls="http://www.ietf.org/rfc/rfc2595.txt" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(XMPPServer.server_id))
-                        time.sleep(0.25)
+                        
+                        await asyncio.sleep(0.25)
+                        #time.sleep(0.25)
                         # send authentication support for iq-auth (fallback) and SASL
-                        self.send(
+                        await self.send(
                             '<stream:features><auth xmlns="http://jabber.org/features/iq-auth"/><mechanisms xmlns="urn:ietf:params:xml:ns:xmpp-sasl"><mechanism>PLAIN</mechanism></mechanisms></stream:features>'
                         )
-                        # self.send('<stream:features><auth xmlns="http://jabber.org/features/iq-auth"/></stream:features>')
+                        # await self.send('<stream:features><auth xmlns="http://jabber.org/features/iq-auth"/></stream:features>')
 
                     else:
-                        self.send("</stream>")
+                        await self.send("</stream>")
 
                 else:
                     if "jabber:iq:auth" in xml.tag:  # Handle iq-auth
-                        self._handle_iq_auth(xml)
+                        await self._handle_iq_auth(xml)
                     elif (
                         "urn:ietf:params:xml:ns:xmpp-sasl" in xml.tag
                     ):  # Handle SASL Auth
-                        self._handle_sasl_auth(xml)
+                        await self._handle_sasl_auth(xml)
                     else:
                         xmppserverlog.error("Couldn't handle: {}".format(xml))
 
@@ -475,33 +528,34 @@ class Client(threading.Thread):
                     # Client getting session after authentication
                     if data.decode("utf-8").find("jabber:client") > -1:
                         # ack jabbr:client
-                        self.send(
+                        await self.send(
                             '<stream:stream xmlns:stream="http://etherx.jabber.org/streams" xmlns="jabber:client" version="1.0" id="1" from="{}">'.format(
                                 XMPPServer.server_id
                             )
                         )
-                        time.sleep(0.25)
+                        await asyncio.sleep(0.25)
+                        #time.sleep(0.25)
                         # session
-                        self.send(
+                        await self.send(
                             '<stream:features><bind xmlns="urn:ietf:params:xml:ns:xmpp-bind"/><session xmlns="urn:ietf:params:xml:ns:xmpp-session"/></stream:features>'
                         )
 
                 else:  # Handle init bind
                     if len(xml):
-                        child = self._tag_strip_uri(xml[0].tag)
+                        child = await self._tag_strip_uri(xml[0].tag)
                     else:
                         child = None
 
                     if xml.tag == "iq":
                         if child == "bind":
-                            self._handle_bind(xml)
+                            await self._handle_bind(xml)
                     else:
                         xmppserverlog.error("Couldn't handle: {}".format(xml))
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_iq_auth(self, data):
+    async def _handle_iq_auth(self, data):
         try:
             xml = ET.fromstring(data.decode("utf-8"))
             ctl = xml[0][0]
@@ -512,7 +566,7 @@ class Client(threading.Thread):
                 and "auth}username" in ctl.tag
                 and self.type == self.UNKNOWN
             ):
-                self.send(
+                await self.send(
                     '<iq type="result" id="{}"><query xmlns="jabber:iq:auth"><username/><password/></query></iq>'.format(
                         xml.get("id")
                     )
@@ -547,10 +601,10 @@ class Client(threading.Thread):
                     xmppserverlog.info("bot authenticated {}".format(self.uid))
 
                     # Client authenticated, move to next state
-                    self._set_state("INIT")
+                    await self._set_state("INIT")
 
                     # Successful auth
-                    self.send('<iq type="result" id="{}"/>'.format(xml.get("id")))
+                    await self.send('<iq type="result" id="{}"/>'.format(xml.get("id")))
 
                 else:
                     auth = False
@@ -564,14 +618,16 @@ class Client(threading.Thread):
                         xmppserverlog.debug("client authenticated {}".format(self.uid))
 
                         # Client authenticated, move to next state
-                        self._set_state("INIT")
+                        await self._set_state("INIT")
 
                         # Successful auth
-                        self.send('<iq type="result" id="{}"/>'.format(xml.get("id")))
+                        await self.send(
+                            '<iq type="result" id="{}"/>'.format(xml.get("id"))
+                        )
 
                     else:
                         # Failed auth
-                        self.send(
+                        await self.send(
                             '<iq type="error" id="{}"><error code="401" type="auth"><not-authorized xmlns="urn:ietf:params:xml:ns:xmpp-stanzas"/></error></iq>'.format(
                                 xml.get("id")
                             )
@@ -594,7 +650,7 @@ class Client(threading.Thread):
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_sasl_auth(self, xml):
+    async def _handle_sasl_auth(self, xml):
         try:
 
             saslauth = base64.b64decode(xml.text).decode("utf-8").split("/")
@@ -617,12 +673,12 @@ class Client(threading.Thread):
                 self.type = self.BOT
                 xmppserverlog.info("bot authenticated {}".format(self.uid))
                 # Send response
-                self.send(
+                await self.send(
                     '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
                 )  # Success
 
                 # Client authenticated, move to next state
-                self._set_state("INIT")
+                await self._set_state("INIT")
 
             else:
                 auth = False
@@ -637,23 +693,23 @@ class Client(threading.Thread):
                     xmppserverlog.debug("client authenticated {}".format(self.uid))
 
                     # Client authenticated, move to next state
-                    self._set_state("INIT")
+                    await self._set_state("INIT")
 
                     # Send response
-                    self.send(
+                    await self.send(
                         '<success xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
                     )  # Success
 
                 else:
                     # Failed to authenticate
-                    self.send(
+                    await self.send(
                         '<response xmlns="urn:ietf:params:xml:ns:xmpp-sasl"/>'
                     )  # Fail
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_bind(self, xml):
+    async def _handle_bind(self, xml):
         try:
 
             bot = bumper.bot_get(self.uid)
@@ -697,22 +753,22 @@ class Client(threading.Thread):
                     xml.get("id"), self.bumper_jid
                 )
 
-            self._set_state("BIND")
+            await self._set_state("BIND")
-            self.send(res)
+            await self.send(res)
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_session(self, xml):
+    async def _handle_session(self, xml):
         try:
             res = '<iq type="result" id="{}" />'.format(xml.get("id"))
-            self._set_state("READY")
+            await self._set_state("READY")
-            self.send(res)
+            await self.send(res)
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_presence(self, xml):
+    async def _handle_presence(self, xml):
         try:
 
             if len(xml) and xml[0].tag == "status":
@@ -722,14 +778,14 @@ class Client(threading.Thread):
                 # Most likely a bot, possibly hello world in text
 
                 # Send dummy return
-                self.send(
+                await self.send(
                     '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
                 )
 
                 # If it is a BOT, send extras
                 if self.type == self.BOT:
                     # get device info
-                    self.send(
+                    await self.send(
                         '<iq type="set" id="14" to="{}" from="{}"><query xmlns="com:ctl"><ctl td="GetDeviceInfo"/></query></iq>'.format(
                             self.bumper_jid, XMPPServer.server_id
                         )
@@ -747,7 +803,7 @@ class Client(threading.Thread):
                         )
                     )
                     # Send dummy return
-                    self.send(
+                    await self.send(
                         '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
                     )
                 elif xml.get("type") == "unavailable":
@@ -757,7 +813,7 @@ class Client(threading.Thread):
                         )
                     )
 
-                    self._set_state("DISCONNECT")
+                    await self._set_state("DISCONNECT")
                 else:
                     # Sometimes the android app sends these
                     xmppserverlog.debug(
@@ -766,14 +822,14 @@ class Client(threading.Thread):
                         )
                     )
                     # Send dummy return
-                    self.send(
+                    await self.send(
                         '<presence to="{}"> dummy </presence>'.format(self.bumper_jid)
                     )
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _parse_data(self, data):
+    async def _parse_data(self, data):
 
         if data.decode("utf-8").startswith(
             "<?xml"
@@ -803,16 +859,16 @@ class Client(threading.Thread):
                                     ).replace("ns0:", ""),
                                 )
                             )
-                        self._handle_iq(item, newdata)
+                        await self._handle_iq(item, newdata)
                         item.clear()
 
                     elif "auth" in item.tag:
                         if "urn:ietf:params:xml:ns:xmpp-sasl" in item.tag:  # SASL Auth
-                            self._handle_sasl_auth(item)
+                            await self._handle_sasl_auth(item)
                             item.clear()
 
                     elif "presence" in item.tag:
-                        self._handle_presence(item)
+                        await self._handle_presence(item)
                         item.clear()
 
                     else:
@@ -834,7 +890,7 @@ class Client(threading.Thread):
                 # Happens wth connect stream often
                 if "<stream:stream " in newdata:
                     if self.state == self.CONNECT or self.state == self.INIT:
-                        self._handle_connect(newdata.encode("utf-8"))
+                        await self._handle_connect(newdata.encode("utf-8"))
                 else:
                     if not (newdata == "" or newdata == " "):
                         xmppserverlog.error(
@@ -846,7 +902,7 @@ class Client(threading.Thread):
                 if not "</stream:stream>" in newdata:
                     xmppserverlog.error("xml parse error - {} - {}".format(newdata, e))
                 else:
-                    self.send("</stream:stream>")  # Close stream
+                    await self.send("</stream:stream>")  # Close stream
 
             else:
                 if "<stream:stream" in newdata:  # Handle start stream and connect
@@ -854,67 +910,48 @@ class Client(threading.Thread):
                         xmppserverlog.debug(
                             "Handling connect data - {}".format(newdata)
                         )
-                        self._handle_connect(newdata.encode("utf-8"))
+                        await self._handle_connect(newdata.encode("utf-8"))
                 else:
                     if not "</stream:stream>" in newdata:
                         xmppserverlog.error(
                             "xml parse error - {} - {}".format(newdata, e)
                         )
                     else:
-                        self.send("</stream:stream>")  # Close stream
+                        await self.send("</stream:stream>")  # Close stream
-                        self._set_state("DISCONNECT")
+                        await self._set_state("DISCONNECT")
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
 
-    def _handle_iq(self, xml, data):
+    async def _handle_iq(self, xml, data):
         try:
             if len(xml):
-                child = self._tag_strip_uri(xml[0].tag)
+                child = await self._tag_strip_uri(xml[0].tag)
             else:
                 child = None
 
             if xml.tag == "iq":
                 if child == "bind":
-                    self._handle_bind(xml)
+                    await self._handle_bind(xml)
                 elif child == "session":
-                    self._handle_session(xml)
+                    await self._handle_session(xml)
                 elif child == "ping":
-                    self._handle_ping(xml, data)
+                    await self._handle_ping(xml, data)
                 elif child == "query":
                     if self.type == self.BOT:
-                        self._handle_result(xml, data)
+                        await self._handle_result(xml, data)
                     else:
-                        self._handle_ctl(xml, data)
+                        await self._handle_ctl(xml, data)
                 elif xml.get("type") == "result":
                     if self.type == self.BOT:
-                        self._handle_result(xml, data)
+                        await self._handle_result(xml, data)
                     else:
-                        self._handle_result(xml, data)
+                        await self._handle_result(xml, data)
                 elif xml.get("type") == "set":
                     if self.type == self.BOT:
-                        self._handle_result(xml, data)
+                        await self._handle_result(xml, data)
                     else:
-                        self._handle_result(xml, data)
+                        await self._handle_result(xml, data)
 
         except Exception as e:
             xmppserverlog.exception("{}".format(e))
-
-    def run(self):
-        # xmppserverlog.info('client connected - {}'.format(self.address))
-        self._set_state("CONNECT")
-        while not self.state == self.DISCONNECT and not self.connection._closed:
-            data = b""
-            time.sleep(0.1)
-            if not self.connection._closed:
-                try:
-                    data = self.connection.recv(4096)
-                    if data != b"":
-                        self._parse_data(data)
-                except ConnectionResetError as e:
-                    xmppserverlog.debug("{}".format(e))
-                except OSError as e:
-                    xmppserverlog.debug("{}".format(e))
-                except Exception as e:
-                    xmppserverlog.exception("{}".format(e))
-

start_bumper.py

@@ -5,6 +5,7 @@ import bumper
 import sys, socket
 import time
 import platform
+import asyncio
 
 
 def main():
@@ -59,22 +60,44 @@ def main():
     # users.append(user1)
     # bumper.bumper_users_var.set(users)
 
+    try:
+        loop = asyncio.get_event_loop()
+    except:
+        loop = asyncio.new_event_loop()
+
+    # Start web servers
+    conf_server.confserver_app()
+    conf_server_2.confserver_app()
+    asyncio.ensure_future(conf_server.start_server(),loop=loop)
+    asyncio.ensure_future(conf_server_2.start_server(),loop=loop)
+
+    # Start MQTT Server
+    asyncio.ensure_future(mqtt_server.broker_coro())
+
+    # Start MQTT Helperbot
+    asyncio.ensure_future(mqtt_helperbot.start_helper_bot())
+    
+    # Start XMPP Server
+    asyncio.ensure_future(xmpp_server.async_server())
+
+    loop.run_forever()
+
     # start xmpp server on port 5223 (sync)
-    xmpp_server.run(run_async=True)  # Start in new thread
+    #xmpp_server.run(run_async=True)  # Start in new thread
 
     # start mqtt server on port 8883 (async)
-    mqtt_server.run(run_async=True)  # Start in new thread
+    #mqtt_server.run(run_async=True)  # Start in new thread
 
-    time.sleep(1.5)  # Wait for broker startup
+    #time.sleep(1.5)  # Wait for broker startup
 
     # start mqtt_helperbot (async)
-    mqtt_helperbot.run(run_async=True)  # Start in new thread
+    #mqtt_helperbot.run(run_async=True)  # Start in new thread
 
     # start conf server on port 443 (async) - Used for most https calls
-    conf_server.run(run_async=True)  # Start in new thread
+    #conf_server.run(run_async=True)  # Start in new thread
 
     # start conf server on port 8007 (async) - Used for a load balancer request
-    conf_server_2.run(run_async=True)  # Start in new thread
+    #conf_server_2.run(run_async=True)  # Start in new thread
 
     while True:
         try: