From a296f038a7d617fbc4d9d091c1e84757de3d440b Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Thu, 21 Mar 2019 21:19:29 -0400 Subject: [PATCH 01/11] 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: From 437417929eb8b0dbe0cfd100c1aa55f2d619248e Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 22 Mar 2019 09:49:15 -0400 Subject: [PATCH 02/11] 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 From b8f95d5b9ce95cb2564f8c4f9ba28421c28c916f Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 22 Mar 2019 09:58:03 -0400 Subject: [PATCH 03/11] 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: From bde8a7331e04682d86af6800cf65593f9abc921a Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Tue, 16 Apr 2019 20:42:41 -0400 Subject: [PATCH 04/11] 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( '' From 17d7ee5f667426d7d7c1cf459ba9580145b3453c Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Fri, 3 May 2019 09:00:05 -0400 Subject: [PATCH 05/11] Individual log files Log files per service - Closes #21 --- .gitignore | 3 ++- bumper/__init__.py | 28 +++++++++++++++++++++++++++- 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index 3595135..ebef324 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ __pycache__ .noseids nosetests.xml tests/report -tests/tmp.db \ No newline at end of file +tests/tmp.db +logs/ \ No newline at end of file diff --git a/bumper/__init__.py b/bumper/__init__.py index 81f4667..88fd4fe 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -9,8 +9,9 @@ import contextvars import time from datetime import datetime, timedelta import platform -import os +import os, sys import logging +from logging.handlers import RotatingFileHandler from base64 import b64decode, b64encode from tinydb import TinyDB, Query import json @@ -29,17 +30,42 @@ token_validity_seconds = 3600 # 1 hour db = None # Logs +os.makedirs("logs", exist_ok=True) #Ensure logs directory exists or create +# Set format for all logs +logformat = logging.Formatter("[%(asctime)s] :: %(levelname)s :: %(name)s :: %(module)s :: %(funcName)s :: %(lineno)d :: %(message)s") + bumperlog = logging.getLogger("bumper") +bumper_rotate = RotatingFileHandler("logs/bumper.log", maxBytes=5000000, backupCount=5) +bumper_rotate.setFormatter(logformat) +bumperlog.addHandler(bumper_rotate) +# Override the logging level +# bumperlog.setLevel(logging.INFO) + confserverlog = logging.getLogger("confserver") +conf_rotate = RotatingFileHandler("logs/confserver.log", maxBytes=5000000, backupCount=5) +conf_rotate.setFormatter(logformat) +confserverlog.addHandler(conf_rotate) # Override the logging level # confserverlog.setLevel(logging.INFO) + mqttserverlog = logging.getLogger("mqttserver") +mqtt_rotate = RotatingFileHandler("logs/mqttserver.log", maxBytes=5000000, backupCount=5) +mqtt_rotate.setFormatter(logformat) +mqttserverlog.addHandler(mqtt_rotate) # Override the logging level # mqttserverlog.setLevel(logging.INFO) + helperbotlog = logging.getLogger("helperbot") +helperbot_rotate = RotatingFileHandler("logs/helperbot.log", maxBytes=5000000, backupCount=5) +helperbot_rotate.setFormatter(logformat) +helperbotlog.addHandler(helperbot_rotate) # Override the logging level # helperbotlog.setLevel(logging.INFO) + xmppserverlog = logging.getLogger("xmppserver") +xmpp_rotate = RotatingFileHandler("logs/xmppserver.log", maxBytes=5000000, backupCount=5) +xmpp_rotate.setFormatter(logformat) +xmppserverlog.addHandler(xmpp_rotate) # Override the logging level # xmppserverlog.setLevel(logging.INFO) From 2683421c34cdcedc63721af697682f1723c8f3fc Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sun, 12 May 2019 15:20:10 -0400 Subject: [PATCH 06/11] Fix startup and create db path Fix for listen args and create db path if it doesn't exist --- bumper/__init__.py | 8 ++++++-- start_bumper.py | 13 ++++++------- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/bumper/__init__.py b/bumper/__init__.py index 88fd4fe..cd21a05 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -82,11 +82,15 @@ def db_file(): def os_db_path(): + db_path = "" if platform.system() == "Windows": - return os.path.join(os.getenv("APPDATA"), "bumper.db") + + db_path = os.path.join(os.getenv("APPDATA")) else: - return os.path.expanduser("~/.config/bumper.db") + db_path = os.path.expanduser("~/.config") + os.makedirs(db_path, exist_ok=True) #Ensure db_path directory exists or create + return os.path.join(db_path, "bumper.db") def db_get(): try: diff --git a/start_bumper.py b/start_bumper.py index 1b0c9d4..a6796f8 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -10,6 +10,7 @@ import asyncio def main(): args = sys.argv + listen_host = "" if len(args) > 0: if "--debug" in args: @@ -24,16 +25,14 @@ def main(): ) # format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(module)s :: %(funcName)s :: %(lineno)d :: %(message)s") - listen_host = args.index("--listen") - if (len(args) - 1) >= (listen_host + 1): - listen_host = args[listen_host+1] - else: + if "--listen" in args: + listen_host = args[args.index("--listen") + 1] + + if listen_host == "": if platform.system() == "Darwin": # If a Mac, use 0.0.0.0 for listening listen_host = "0.0.0.0" else: - listen_host = socket.gethostbyname(socket.gethostname()) - #listen_host = "localhost" # Try this if the above doesn't work - + listen_host = socket.gethostbyname(socket.gethostname()) conf_address_443 = (listen_host, 443) conf_address_8007 = (listen_host, 8007) From d6816a02ec9a9bbca37bf01be98b2ffd34a3727b Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sun, 12 May 2019 15:31:52 -0400 Subject: [PATCH 07/11] Remove vars Start removal --- bumper/__init__.py | 5 ----- bumper/confserver.py | 1 - bumper/mqttserver.py | 8 ++++---- bumper/xmppserver.py | 1 - start_bumper.py | 8 -------- 5 files changed, 4 insertions(+), 19 deletions(-) diff --git a/bumper/__init__.py b/bumper/__init__.py index cd21a05..9abbbce 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -5,7 +5,6 @@ from .mqttserver import MQTTServer from .mqttserver import MQTTHelperBot from .xmppserver import XMPPServer import asyncio -import contextvars import time from datetime import datetime, timedelta import platform @@ -17,10 +16,6 @@ from tinydb import TinyDB, Query import json from tinydb.storages import MemoryStorage -bumper_users_var = contextvars.ContextVar("bumper_users", default=[]) -bumper_clients_var = contextvars.ContextVar("bumper_clients", default=[]) -bumper_bots_var = contextvars.ContextVar("bumper_bots", default=[]) - ca_cert = "./certs/CA/cacert.pem" server_cert = "./certs/cert.pem" server_key = "./certs/key.pem" diff --git a/bumper/confserver.py b/bumper/confserver.py index 375e931..bd5f42f 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -8,7 +8,6 @@ import bumper import time from datetime import datetime, timedelta import asyncio -import contextvars from aiohttp import web import uuid diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 40cf783..859fd69 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -8,7 +8,7 @@ from hbmqtt.broker import Broker from hbmqtt.client import MQTTClient from hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2 import pkg_resources -import contextvars +#import contextvars import time from threading import Thread import ssl @@ -40,7 +40,7 @@ class MQTTHelperBot: ): self.address = address self.client_id = "helper1@bumper/helper1" - self.command_responses = contextvars.ContextVar("command_responses", default=[]) + self.command_responses = [] # = contextvars.ContextVar("command_responses", default=[]) self.helperthread = None def run(self, run_async=False): @@ -99,7 +99,7 @@ class MQTTHelperBot: message = await self.Client.deliver_message() # helperbotlog.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8")))) - cresp = self.command_responses.get() + cresp = self.command_responses #.get() if str(message.topic).split("/")[6] == "helper1": cresp.append( @@ -119,7 +119,7 @@ class MQTTHelperBot: # helperbotlog.debug("Pruning Message Time: {}, MsgTime: {}, MsgTime+60: {}".format(time.time(), msg['time'], expire_time)) cresp.remove(msg) - self.command_responses.set(cresp) + self.command_responses = cresp #.set(cresp) # helperbotlog.debug("MQTT Command Response List Count: %s" %len(cresp)) except Exception as e: diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 6d86d09..02471c4 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -4,7 +4,6 @@ from threading import Thread import sys, socket, threading, re, time, logging, uuid, xml.etree.ElementTree as ET import base64 import ssl -import contextvars import bumper import asyncio diff --git a/start_bumper.py b/start_bumper.py index a6796f8..0f777d8 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -51,14 +51,6 @@ def main(): conf_address_8007, usessl=False, helperbot=mqtt_helperbot ) - # add user - # users = bumper.bumper_users_var.get() - # user1 = bumper.BumperUser('user1') - # user1.add_device('devid') - # user1.add_bot('bot_did') - # users.append(user1) - # bumper.bumper_users_var.set(users) - try: loop = asyncio.get_event_loop() except: From 652a0b0e9beb8e7a98636ac456ac2103b11b7fd2 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sun, 12 May 2019 15:42:47 -0400 Subject: [PATCH 08/11] Fix command_responses Final removal of contextvars --- bumper/mqttserver.py | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 859fd69..2cb775b 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -8,7 +8,6 @@ from hbmqtt.broker import Broker from hbmqtt.client import MQTTClient from hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2 import pkg_resources -#import contextvars import time from threading import Thread import ssl @@ -40,7 +39,7 @@ class MQTTHelperBot: ): self.address = address self.client_id = "helper1@bumper/helper1" - self.command_responses = [] # = contextvars.ContextVar("command_responses", default=[]) + self.command_responses = [] self.helperthread = None def run(self, run_async=False): @@ -99,10 +98,9 @@ class MQTTHelperBot: message = await self.Client.deliver_message() # helperbotlog.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8")))) - cresp = self.command_responses #.get() if str(message.topic).split("/")[6] == "helper1": - cresp.append( + self.command_responses.append( { "time": time.time(), "topic": message.topic, @@ -111,15 +109,14 @@ class MQTTHelperBot: ) # Cleanup "expired messages" > 60 seconds from time - for msg in cresp: + for msg in self.command_responses: expire_time = ( datetime.fromtimestamp(msg["time"]) + timedelta(seconds=10) ).timestamp() if time.time() > expire_time: # helperbotlog.debug("Pruning Message Time: {}, MsgTime: {}, MsgTime+60: {}".format(time.time(), msg['time'], expire_time)) - cresp.remove(msg) + self.command_responses.remove(msg) - self.command_responses = cresp #.set(cresp) # helperbotlog.debug("MQTT Command Response List Count: %s" %len(cresp)) except Exception as e: @@ -131,10 +128,9 @@ class MQTTHelperBot: t_end = (datetime.now() + timedelta(seconds=10)).timestamp() while time.time() < t_end: - await asyncio.sleep(0.1) - responses = self.command_responses.get() - if len(responses) > 0: - for msg in responses: + await asyncio.sleep(0.1) + if len(self.command_responses) > 0: + for msg in self.command_responses: topic = str(msg["topic"]).split("/") if topic[6] == "helper1" and topic[10] == requestid: # helperbotlog.debug('VacBot MQTT Response: Topic: %s Payload: %s' % (msg['topic'], msg['payload'])) @@ -142,10 +138,8 @@ class MQTTHelperBot: resppayload = json.loads(msg["payload"]) else: resppayload = str(msg["payload"]) - resp = {"id": requestid, "ret": "ok", "resp": resppayload} - cresp = self.command_responses.get() - cresp.remove(msg) - self.command_responses.set(cresp) + resp = {"id": requestid, "ret": "ok", "resp": resppayload} + self.command_responses.remove(msg) return resp return {"id": requestid, "errno": "timeout", "ret": "fail"} From c4900be511046628bdd9924f1887e8b43f15ab89 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sun, 12 May 2019 15:49:08 -0400 Subject: [PATCH 09/11] Fix db_path and test Earlier commit broke the db_path and tests..... --- bumper/__init__.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/bumper/__init__.py b/bumper/__init__.py index 9abbbce..618f1ab 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -77,15 +77,12 @@ def db_file(): def os_db_path(): - db_path = "" - if platform.system() == "Windows": - - db_path = os.path.join(os.getenv("APPDATA")) + if platform.system() == "Windows": + os.makedirs(os.getenv("APPDATA"), exist_ok=True) #Ensure db_path directory exists or create + return os.path.join(os.getenv("APPDATA"), "bumper.db") else: - db_path = os.path.expanduser("~/.config") - - os.makedirs(db_path, exist_ok=True) #Ensure db_path directory exists or create - return os.path.join(db_path, "bumper.db") + os.makedirs(os.path.expanduser("~/.config"), exist_ok=True) #Ensure db_path directory exists or create + return os.path.expanduser("~/.config/bumper.db") def db_get(): try: From d238ded6c1ca5de4a8825a03ed2bc0ff663f8dc5 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sat, 18 May 2019 08:35:08 -0400 Subject: [PATCH 10/11] Better MQTT logging and asyncio debug on Better MQTT logging and asyncio debug on --- bumper/confserver.py | 2 +- bumper/mqttserver.py | 20 ++++++++++++++++---- start_bumper.py | 4 ++++ 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/bumper/confserver.py b/bumper/confserver.py index bd5f42f..a9aa00d 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -31,7 +31,7 @@ class aiohttp_filter(logging.Filter): confserverlog = logging.getLogger("confserver") -logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1) # Ignore this logger +#logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1) # Ignore this logger logging.getLogger("aiohttp.access").addFilter(aiohttp_filter()) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 2cb775b..0458787 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -84,6 +84,7 @@ class MQTTHelperBot: [ ("iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+", QOS_0), ("iot/p2p/+", QOS_0), + ("iot/atr/+", QOS_0), ] ) @@ -97,9 +98,9 @@ class MQTTHelperBot: while True: message = await self.Client.deliver_message() - # helperbotlog.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8")))) - if str(message.topic).split("/")[6] == "helper1": + #Response to command + helperbotlog.debug("Received Response - Topic: {} - Message: {}".format(message.topic, str(message.data.decode("utf-8")))) self.command_responses.append( { "time": time.time(), @@ -107,6 +108,14 @@ class MQTTHelperBot: "payload": str(message.data.decode("utf-8")), } ) + elif str(message.topic).split("/")[3] == "helper1": + #Helperbot sending command + helperbotlog.debug("Send Command - Topic: {} - Message: {}".format(message.topic, str(message.data.decode("utf-8")))) + elif str(message.topic).split("/")[1] == "atr": + #Broadcast message received on atr + helperbotlog.debug("Received Broadcast - Topic: {} - Message: {}".format(message.topic, str(message.data.decode("utf-8")))) + else: + helperbotlog.debug("Received Message - Topic: {} - Message: {}".format(message.topic, str(message.data.decode("utf-8")))) # Cleanup "expired messages" > 60 seconds from time for msg in self.command_responses: @@ -114,7 +123,7 @@ class MQTTHelperBot: datetime.fromtimestamp(msg["time"]) + timedelta(seconds=10) ).timestamp() if time.time() > expire_time: - # helperbotlog.debug("Pruning Message Time: {}, MsgTime: {}, MsgTime+60: {}".format(time.time(), msg['time'], expire_time)) + helperbotlog.debug("Pruning Message Time: {}, MsgTime: {}, MsgTime+60: {}".format(time.time(), msg['time'], expire_time)) self.command_responses.remove(msg) # helperbotlog.debug("MQTT Command Response List Count: %s" %len(cresp)) @@ -128,7 +137,7 @@ class MQTTHelperBot: t_end = (datetime.now() + timedelta(seconds=10)).timestamp() while time.time() < t_end: - await asyncio.sleep(0.1) + await asyncio.sleep(0.1) if len(self.command_responses) > 0: for msg in self.command_responses: topic = str(msg["topic"]).split("/") @@ -145,8 +154,10 @@ class MQTTHelperBot: return {"id": requestid, "errno": "timeout", "ret": "fail"} except asyncio.CancelledError as e: helperbotlog.debug("wait_for_resp cancelled by asyncio") + return {"id": requestid, "errno": "timeout", "ret": "fail"} except Exception as e: helperbotlog.exception("{}".format(e)) + return {"id": requestid, "errno": "timeout", "ret": "fail"} async def send_command(self, cmdjson, requestid): try: @@ -171,6 +182,7 @@ class MQTTHelperBot: except Exception as e: helperbotlog.exception("{}".format(e)) + return {} class MQTTServer: diff --git a/start_bumper.py b/start_bumper.py index 0f777d8..2503f44 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -5,6 +5,8 @@ import bumper import sys, socket import time import platform +import os +os.environ['PYTHONASYNCIODEBUG'] = '1' import asyncio @@ -57,6 +59,7 @@ def main(): loop = asyncio.new_event_loop() # Start web servers + loop.set_debug(True) conf_server.confserver_app() conf_server_2.confserver_app() asyncio.ensure_future(conf_server.start_server(),loop=loop) @@ -73,6 +76,7 @@ def main(): loop.run_forever() + # start xmpp server on port 5223 (sync) #xmpp_server.run(run_async=True) # Start in new thread From 9509163464b8c6cf2430117529d74038e4f561dc Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sat, 18 May 2019 15:55:00 -0400 Subject: [PATCH 11/11] Updates for XMPP Updates for XMPP, seems to fix an issue with holding the rest of the asyncio loops hostage after a disconnect. --- bumper/xmppserver.py | 63 +++++++++++++++++++++++++++++++++----------- 1 file changed, 47 insertions(+), 16 deletions(-) diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 02471c4..5ad8e22 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -41,8 +41,8 @@ class XMPPServer: def client_done(task): del self.aclients[task] - client_writer.close() - xmppserverlog.info("End Connection") + xmppserverlog.info("End Connection for {}".format(client_writer.get_extra_info("peername"))) + client_writer.close() clientaddr = client_writer.get_extra_info("peername") xmppserverlog.info("New Connection from {}".format(clientaddr)) @@ -197,24 +197,29 @@ class XMPPAsyncClient: self.devclass = "" self.bumper_jid = "" self.uid = "" - self.log_sent_message = False # Set to true to log sends + self.log_sent_message = True # Set to true to log sends self.log_incoming_data = True # Set to true to log sends xmppserverlog.debug("new client with ip {}".format(self.address)) async def handle_async_client(self): - # xmppserverlog.info('client connected - {}'.format(self.address)) + # xmppserverlog.info('client connected - {}'.format(self.address)) + #await self._set_state("READY") 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 + #asyncio.Task(self.send_ping(30)) + while True: + await asyncio.sleep(0.1) + if 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) + await self._parse_data(data) + else: + break # exit loop and disconnect return @@ -302,6 +307,24 @@ class XMPPAsyncClient: ) ) return + + if "disco#items" in data: + # Return not-implemented for disco#items + await self.send( + ''.format( + xml.get("id") + )) + return + + if "disco#info" in data: + # Return not-implemented for disco#info + await self.send( + ''.format( + xml.get("id") + ) + ) + return + if xml.get("type") == "set": if ( @@ -316,7 +339,7 @@ class XMPPAsyncClient: ) ) - if xml[0][0]: + if len(xml[0]) > 0: ctl = xml[0][0] if ctl.get("admin") and self.type == self.BOT: xmppserverlog.debug( @@ -580,6 +603,7 @@ class XMPPAsyncClient: xmlauth = xml[0].getchildren() # uid = "" password = "" + authcode = "" resource = "" for aitem in xmlauth: if "username" in aitem.tag: @@ -655,6 +679,7 @@ class XMPPAsyncClient: saslauth = base64.b64decode(xml.text).decode("utf-8").split("/") username = saslauth[0] username = saslauth[0].split("\x00")[1] + authcode = "" self.uid = username if len(saslauth) > 1: resource = saslauth[1] @@ -761,8 +786,10 @@ class XMPPAsyncClient: async def _handle_session(self, xml): try: res = ''.format(xml.get("id")) - await self._set_state("READY") + await self._set_state("READY") await self.send(res) + asyncio.Task(self.send_ping(30)) + except Exception as e: xmppserverlog.exception("{}".format(e)) @@ -780,6 +807,8 @@ class XMPPAsyncClient: await self.send( ' dummy '.format(self.bumper_jid) ) + + # If it is a BOT, send extras if self.type == self.BOT: @@ -790,11 +819,13 @@ class XMPPAsyncClient: ) ) + + else: xmppserverlog.debug( "client presence - {} ".format(ET.tostring(xml, encoding="utf-8")) ) - + if xml.get("type") == "available": xmppserverlog.debug( "client presence available - {} ".format(