From a296f038a7d617fbc4d9d091c1e84757de3d440b Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Thu, 21 Mar 2019 21:19:29 -0400 Subject: [PATCH 1/4] rearrange start to use asyncio Create an asyncio loop then start confservers, mqtt, and mqtthelper in loop. TODO: Re-work xmpp for asyncio --- bumper/__init__.py | 25 +++++++++++++++++-------- bumper/mqttserver.py | 6 ++++++ start_bumper.py | 30 +++++++++++++++++++++++++----- 3 files changed, 48 insertions(+), 13 deletions(-) diff --git a/bumper/__init__.py b/bumper/__init__.py index db1a44a..81f4667 100644 --- a/bumper/__init__.py +++ b/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 - db = TinyDB(db_file()) + try: + # Will create the database if it doesn't exist + db = TinyDB(db_file()) - # Will create the tables if they don't exist - db.table("users", cache_size=0) - db.table("clients", cache_size=0) - db.table("bots", cache_size=0) - db.table("tokens", cache_size=0) + # Will create the tables if they don't exist + db.table("users", cache_size=0) + db.table("clients", cache_size=0) + db.table("bots", 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): diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 115e71d..40cf783 100644 --- a/bumper/mqttserver.py +++ b/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)) diff --git a/start_bumper.py b/start_bumper.py index 1975e80..6e41a82 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -5,6 +5,7 @@ import bumper import sys, socket import time import platform +import asyncio def main(): @@ -59,22 +60,41 @@ 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()) + + loop.run_forever() + # start xmpp server on port 5223 (sync) 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: -- 2.39.5 From 437417929eb8b0dbe0cfd100c1aa55f2d619248e Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 22 Mar 2019 09:49:15 -0400 Subject: [PATCH 2/4] WIP - Convert XMPP to Async WIP - Convert XMPP to Async --- bumper/xmpp_old_client.py | 769 ++++++++++++++++++++++++++++++++++++++ bumper/xmppserver.py | 391 ++++++++++--------- start_bumper.py | 5 +- 3 files changed, 985 insertions(+), 180 deletions(-) create mode 100644 bumper/xmpp_old_client.py diff --git a/bumper/xmpp_old_client.py b/bumper/xmpp_old_client.py new file mode 100644 index 0000000..ea5b24b --- /dev/null +++ b/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( + ''.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( + ''.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("'.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 = ''.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( + ''.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(" -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( + ''.format( + XMPPServer.server_id + ) + ) + # with STARTTLS + # self.send(''.format(XMPPServer.server_id)) + time.sleep(0.25) + # send authentication support for iq-auth (fallback) and SASL + self.send( + 'PLAIN' + ) + # self.send('') + + else: + self.send("") + + 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( + ''.format( + XMPPServer.server_id + ) + ) + time.sleep(0.25) + # session + self.send( + '' + ) + + 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( + ''.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(''.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(''.format(xml.get("id"))) + + else: + # Failed auth + self.send( + ''.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 + + # 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 + + else: + # Failed to authenticate + self.send( + '' + ) # 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 = '{}'.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 = '{}'.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 = '{}'.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 = ''.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( + ' dummy '.format(self.bumper_jid) + ) + + # If it is a BOT, send extras + if self.type == self.BOT: + # get device info + self.send( + ''.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( + ' dummy '.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( + ' dummy '.format(self.bumper_jid) + ) + + except Exception as e: + xmppserverlog.exception("{}".format(e)) + + def _parse_data(self, data): + + if data.decode("utf-8").startswith( + "]+\?>)", r"", data.decode("utf-8")) + "" + ) + + else: + newdata = "{}".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 " - client is signalling end of session/disconnect + if not "" in newdata: + xmppserverlog.error("xml parse error - {} - {}".format(newdata, e)) + else: + self.send("") # Close stream + + else: + if "" in newdata: + xmppserverlog.error( + "xml parse error - {} - {}".format(newdata, e) + ) + else: + self.send("") # 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)) diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index e044790..4c2a898 100644 --- a/bumper/xmppserver.py +++ b/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): - if run_async: - xmppserverlog.debug("Starting XMPPServer Thread: 1") - self.xmppthread = Thread(name="XMPPServer_Thread", target=self.run_server) - self.xmppthread.setDaemon(True) - self.xmppthread.start() - - 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 + async def async_server(self): + xmppserverlog.debug( + "listening on {}:{}".format(self.address[0], self.address[1]) + ) + server = await asyncio.start_server( + self.accept_client, self.address[0], self.address[1] ) - self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + await server.serve_forever() - try: - self.socket.bind(self.address) - self.socket.listen(5) + # self.clients = {} # task -> (reader, writer) - xmppserverlog.debug( - "listening on {}:{}".format(self.address[0], self.address[1]) - ) - while not self.exit_flag: - connection, client_address = self.socket.accept() + def accept_client(self, client_reader, client_writer): + # task = asyncio.Task(self.handle_client(client_reader, client_writer)) + aclient = XMPPAsyncClient(client_reader, client_writer) + task = asyncio.Task(aclient.handle_async_client()) + self.aclients[task] = (client_reader, client_writer) - # 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) + def client_done(task): + del self.aclients[task] + client_writer.close() + xmppserverlog.info("End Connection") - xmppserverlog.debug( - "starting new client with ip {}".format(client_address[0]) - ) - thread_id = uuid.uuid4() - client = Client(thread_id, connection, client_address) - client.setDaemon(True) - client.start() - self.clients.append(client) + clientaddr = client_writer.get_extra_info("peername") + xmppserverlog.info("New Connection from {}".format(clientaddr)) + task.add_done_callback(client_done) - 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) + # def run(self, run_async=False): + # if run_async: + # xmppserverlog.debug("Starting XMPPServer Thread: 1") + # self.xmppthread = Thread(name="XMPPServer_Thread", target=self.run_server) + # self.xmppthread.setDaemon(True) + # self.xmppthread.start() - except Exception as e: - xmppserverlog.exception("{}".format(e)) - exit(1) + # else: + # try: + # self.run_server() + # except KeyboardInterrupt: + # self.disconnect() - except KeyboardInterrupt as e: - xmppserverlog.exception("{}".format(e)) + # def run_server(self): + # logging.info("Starting XMPP Server at {}".format(self.address)) + # print("Starting XMPP Server at {}".format(self.address)) - finally: - connection.shutdown(socket.SHUT_RDWR) - connection.close() - self.disconnect() - xmppserverlog.info("disconnecting") + # # xmppserverlog.setLevel(logging.DEBUG) - 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): - threading.Thread.__init__(self) - self.id = thread_id - self.name = "XMPP_Client_{}".format(client_address[0]) + def __init__(self, client_reader, client_writer): self.type = self.UNKNOWN self.state = self.IDLE - self.connection = connection - self.address = client_address[0] + self.address = client_writer.get_extra_info("peername") + 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( - "new client thread init for client with ip {}".format(self.address) - ) + xmppserverlog.debug("new 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 self.log_sent_message: - xmppserverlog.debug("send {} - {}".format(self.address, command)) - self.connection.send(command.encode()) + # if not self.connection._closed: + if self.log_sent_message: + xmppserverlog.debug("send {} - {}".format(self.address, command)) + # 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( ''.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( ''.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,13 @@ class Client(threading.Thread): except Exception as e: xmppserverlog.exception("{}".format(e)) - def _handle_result(self, xml, data): + async def send_ping(self, time): + pingstring = "".format(XMPPServer.server_id, self.bumper_jid) + await self.send(pingstring) + await asyncio.sleep(time) + pingtask = 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 +420,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 = ''.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( ''.format( uuid.uuid4(), adminuser, self.bumper_jid ) @@ -430,7 +480,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 +493,30 @@ class Client(threading.Thread): self.devclass = data.decode("utf-8")[sc + 4 : ec] # ack jabbr:client # no STARTTLS - self.send( + await self.send( ''.format( XMPPServer.server_id ) ) # with STARTTLS - # self.send(''.format(XMPPServer.server_id)) + # await self.send(''.format(XMPPServer.server_id)) time.sleep(0.25) # send authentication support for iq-auth (fallback) and SASL - self.send( + await self.send( 'PLAIN' ) - # self.send('') + # await self.send('') else: - self.send("") + await self.send("") 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 +525,33 @@ 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( ''.format( XMPPServer.server_id ) ) time.sleep(0.25) # session - self.send( + await self.send( '' ) 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 +562,7 @@ class Client(threading.Thread): and "auth}username" in ctl.tag and self.type == self.UNKNOWN ): - self.send( + await self.send( ''.format( xml.get("id") ) @@ -547,10 +597,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(''.format(xml.get("id"))) + await self.send(''.format(xml.get("id"))) else: auth = False @@ -564,14 +614,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(''.format(xml.get("id"))) + await self.send( + ''.format(xml.get("id")) + ) else: # Failed auth - self.send( + await self.send( ''.format( xml.get("id") ) @@ -594,7 +646,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 +669,12 @@ class Client(threading.Thread): self.type = self.BOT xmppserverlog.info("bot authenticated {}".format(self.uid)) # Send response - self.send( + await self.send( '' ) # Success # Client authenticated, move to next state - self._set_state("INIT") + await self._set_state("INIT") else: auth = False @@ -637,23 +689,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 else: # Failed to authenticate - self.send( + await self.send( '' ) # 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 +749,22 @@ class Client(threading.Thread): xml.get("id"), self.bumper_jid ) - self._set_state("BIND") - self.send(res) + await self._set_state("BIND") + 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 = ''.format(xml.get("id")) - self._set_state("READY") - self.send(res) + await self._set_state("READY") + 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 +774,14 @@ class Client(threading.Thread): # Most likely a bot, possibly hello world in text # Send dummy return - self.send( + await self.send( ' dummy '.format(self.bumper_jid) ) # If it is a BOT, send extras if self.type == self.BOT: # get device info - self.send( + await self.send( ''.format( self.bumper_jid, XMPPServer.server_id ) @@ -747,7 +799,7 @@ class Client(threading.Thread): ) ) # Send dummy return - self.send( + await self.send( ' dummy '.format(self.bumper_jid) ) elif xml.get("type") == "unavailable": @@ -757,7 +809,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 +818,14 @@ class Client(threading.Thread): ) ) # Send dummy return - self.send( + await self.send( ' dummy '.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( "" in newdata: xmppserverlog.error("xml parse error - {} - {}".format(newdata, e)) else: - self.send("") # Close stream + await self.send("") # Close stream else: if "" in newdata: xmppserverlog.error( "xml parse error - {} - {}".format(newdata, e) ) else: - self.send("") # Close stream - self._set_state("DISCONNECT") + await self.send("") # Close stream + 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)) - diff --git a/start_bumper.py b/start_bumper.py index 6e41a82..1b0c9d4 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -77,10 +77,13 @@ def main(): # 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 -- 2.39.5 From b8f95d5b9ce95cb2564f8c4f9ba28421c28c916f Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 22 Mar 2019 09:58:03 -0400 Subject: [PATCH 3/4] dont schedule pings if client disconnected dont schedule pings if client disconnected --- bumper/xmppserver.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 4c2a898..933b92b 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -391,10 +391,11 @@ class XMPPAsyncClient: xmppserverlog.exception("{}".format(e)) async def send_ping(self, time): - pingstring = "".format(XMPPServer.server_id, self.bumper_jid) - await self.send(pingstring) - await asyncio.sleep(time) - pingtask = asyncio.Task(self.send_ping(time)) + if not self.state == 5: #disconnected + pingstring = "".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: -- 2.39.5 From bde8a7331e04682d86af6800cf65593f9abc921a Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Tue, 16 Apr 2019 20:42:41 -0400 Subject: [PATCH 4/4] xmpp sleep to async xmpp sleep to async --- bumper/xmppserver.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 933b92b..6d86d09 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -501,7 +501,9 @@ class XMPPAsyncClient: ) # with STARTTLS # await self.send(''.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 await self.send( 'PLAIN' @@ -531,7 +533,8 @@ class XMPPAsyncClient: XMPPServer.server_id ) ) - time.sleep(0.25) + await asyncio.sleep(0.25) + #time.sleep(0.25) # session await self.send( '' -- 2.39.5