From f7d91aa650f7a24f4312189495311bd55d996c03 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Sun, 3 Feb 2019 16:16:29 -0500 Subject: [PATCH] Split bots and clients, Fix XMPP threads not exiting Split bots and clients, Fix XMPP threads not exiting --- bumper.py | 8 +-- bumper/__init__.py | 14 +++++ bumper/confserver.py | 34 +++++++----- bumper/mqttserver.py | 74 ++++++++++++++++++------- bumper/xmppserver.py | 125 +++++++++++++++++++++++++++++-------------- 5 files changed, 177 insertions(+), 78 deletions(-) diff --git a/bumper.py b/bumper.py index ae4b6a8..1a98928 100644 --- a/bumper.py +++ b/bumper.py @@ -24,18 +24,18 @@ mqtt_address = ("0.0.0.0", 8883) # A default bot could be set here to automatically add it as available # dbot = bumper.VacBotDevice("did", "class", "resource", "name","nick" ) -# bclient = bumper.bumper_clients_var +# bclient = bumper.bumper_bots_var # bclienttemp = bclient.get() # bclienttemp.append(dbot.asdict()) # bclient.set(bclienttemp) # start mqtt server (async) -mqtt_server = bumper.MQTTServer(mqtt_address, run_async=True,bumper_clients=bumper.bumper_clients_var) +mqtt_server = bumper.MQTTServer(mqtt_address, run_async=True,bumper_bots=bumper.bumper_bots_var,bumper_clients=bumper.bumper_clients_var) time.sleep(1.5) #Wait for broker startup # start mqtt server (async) -mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address, run_async=True,bumper_clients=bumper.bumper_clients_var) +mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address, run_async=True,bumper_bots=bumper.bumper_bots_var,bumper_clients=bumper.bumper_clients_var) # start conf server (async) -conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True, bumper_clients=bumper.bumper_clients_var, helperbot=mqtt_helperbot) +conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True,bumper_bots=bumper.bumper_bots_var,bumper_clients=bumper.bumper_clients_var, helperbot=mqtt_helperbot) # start xmpp server (sync) xmpp_server = bumper.XMPPServer(xmpp_address) diff --git a/bumper/__init__.py b/bumper/__init__.py index fd9349c..bfcdb9f 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -9,6 +9,10 @@ import contextvars import time 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' def get_milli_time(timetoconvert): return int(round(timetoconvert * 1000)) @@ -26,3 +30,13 @@ class VacBotDevice(object): def asdict(self): return {"class": self.vac_bot_device_class, "company": self.company, "did": self.did, "name": self.name, "nick": self.nick, "resource": self.resource} + +class VacBotUser(object): + def __init__(self,userid="",realm="",token=""): + self.userid = userid + self.realm = realm + self.resource = token + + + def asdict(self): + return {"userid": self.userid,"realm": self.realm,"resource": self.resource} diff --git a/bumper/confserver.py b/bumper/confserver.py index 398f9b2..ec6b70a 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -13,8 +13,10 @@ from aiohttp import web class ConfServer(): bumper_clients = contextvars.ContextVar + bumper_bots = contextvars.ContextVar - def __init__(self, address, usessl=False, run_async=True, bumper_clients=contextvars.ContextVar, helperbot=None): + def __init__(self, address, usessl=False, run_async=True, bumper_bots=contextvars.ContextVar, bumper_clients=contextvars.ContextVar, helperbot=None): + self.bumper_bots = bumper_bots self.bumper_clients = bumper_clients self.helperbot = helperbot self.usessl = usessl @@ -54,15 +56,16 @@ class ConfServer(): app.add_routes([ web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/login', self.handle_login), - web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout', self.handle_login), - web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode', self.handle_getAuthCode), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkLogin', self.handle_login), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/logout', self.handle_logout), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/getAuthCode', self.handle_getAuthCode), web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/user/checkAgreement', self.handle_checkAgreement), - web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/getAuthCode', self.handle_checkVersion), + web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/common/checkVersion', self.handle_checkVersion), web.get('/{apiversion}/private/{country}/{language}/{devid}/{apptype}/{appversion}/{devtype}/{aid}/campaign/homePageAlert', self.handle_homePageAlert), web.post('/api/users/user.do', self.handle_usersapi), web.post('/api/pim/product/getProductIotMap', self.handle_getProductIotMap), - web.post('/api/iot/devmanager.do', self.handle_devmanager) + web.post('/api/iot/devmanager.do', self.handle_devmanager_botcommand) ]) @@ -71,7 +74,7 @@ class ConfServer(): if self.usessl: ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - ssl_ctx.load_cert_chain('./certs/cert.pem', './certs/key.pem') + ssl_ctx.load_cert_chain(bumper.server_cert,bumper.server_key) site = web.TCPSite(runner, host=self.address[0], port=self.address[1],ssl_context=ssl_ctx) else: @@ -89,8 +92,8 @@ class ConfServer(): "accessToken": "tempaccesstoken", #Random chars 32 length "country": countrycode, "email": "null@null.com", - "uid": "fuid_1", #Date(14)_RandomChars(32) - "username": "fusername_1" #Random chars 8 + "uid": "fuid_{}".format(''.join(random.sample(string.ascii_letters,6))), #Date(14)_RandomChars(32) + "username": "fusername_{}".format(''.join(random.sample(string.ascii_letters,6))) #Random chars 8 }, "msg": "操作成功", "time": bumper.get_milli_time(time.time()) @@ -100,7 +103,8 @@ class ConfServer(): async def handle_logout(self, request): body = {"code": "0000","data": None,"msg": "操作成功", "time": bumper.get_milli_time(time.time())} - + #TODO - when logging out close out any other connections MQTT/XMPP + return web.json_response(body) async def handle_getAuthCode(self, request): @@ -109,13 +113,13 @@ class ConfServer(): "code": "0000", "data": { "authCode": "{}_tempauthcode".format(countrycode), #countrycode_randomchars(32) - "ecovacsUid": "fuid_1" #Date(14)_RandomChars(32) + "ecovacsUid": "fuid_{}".format(''.join(random.sample(string.ascii_letters,6))) #Date(14)_RandomChars(32) }, "msg": "操作成功", "time": bumper.get_milli_time(time.time()) } - return web.json_response(body) + return web.json_response(body) async def handle_checkVersion(self, request): body = { @@ -170,7 +174,9 @@ class ConfServer(): body = {"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]} return web.json_response(body) - async def handle_usersapi(self, request): + async def handle_usersapi(self, request): + body = {} + json_body = json.loads(await request.text()) todo = json_body['todo'] if todo == 'FindBest': @@ -188,7 +194,7 @@ class ConfServer(): "userId": json_body["userId"] #RandomChar(16) } elif todo == 'GetDeviceList': - active_bots = self.bumper_clients.get() + active_bots = self.bumper_bots.get() body = { "devices": active_bots, "result": "ok", @@ -197,7 +203,7 @@ class ConfServer(): return web.json_response(body) - async def handle_devmanager(self, request): + async def handle_devmanager_botcommand(self, request): json_body = json.loads(await request.text()) randomid = ''.join(random.sample(string.ascii_letters,6)) retcmd = await self.helperbot.send_command(json_body, randomid) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 3db6866..83ce588 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -19,7 +19,7 @@ from datetime import datetime, timedelta class MQTTHelperBot(): Client = MQTTClient() - def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar): + def __init__(self, address, run_async=False, bumper_bots=contextvars.ContextVar, bumper_clients=contextvars.ContextVar): self.address = address self.client_id = "helper1@bumper/helper1" @@ -51,7 +51,7 @@ class MQTTHelperBot(): async def start_helper_bot(self): try: - await self.Client.connect('mqtts://{}:{}/'.format(self.address[0], self.address[1]), cafile='./certs/CA/cacert.pem') + await self.Client.connect('mqtts://{}:{}/'.format(self.address[0], self.address[1]), cafile=bumper.ca_cert) await self.Client.subscribe([ ('iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+',QOS_0), ('iot/p2p/+',QOS_0) @@ -125,6 +125,7 @@ class MQTTHelperBot(): class MQTTServer(): default_config = {} bumper_clients = [] + bumper_bots = [] async def broker_coro(self): broker = hbmqtt.broker.Broker(config=self.default_config) @@ -133,16 +134,16 @@ class MQTTServer(): async def active_bot_listing(self): while True: await asyncio.sleep(5) - logging.debug('Connected bots: %s' % self.bumper_clients.get()) + logging.debug('Connected bots: %s' % self.bumper_bots.get()) - def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar): + def __init__(self, address, run_async=False, bumper_bots=contextvars.ContextVar, bumper_clients=contextvars.ContextVar): #The below adds a plugin to the hbmqtt.broker.plugins without having to futz with setup.py distribution = pkg_resources.Distribution("hbmqtt.broker.plugins") bumper_plugin = pkg_resources.EntryPoint.parse('bumper = bumper.mqttserver:BumperMQTTServer_Plugin', dist=distribution) distribution._ep_map = {"hbmqtt.broker.plugins": {"bumper": bumper_plugin}} pkg_resources.working_set.add(distribution) - + self.bumper_bots = bumper_bots self.bumper_clients = bumper_clients try: # Initialize bot server @@ -154,8 +155,8 @@ class MQTTServer(): 'tls1': { 'bind': "{}:{}".format(address[0], address[1]), 'ssl': 'on', - 'certfile': './certs/cert.pem', - 'keyfile': './certs/key.pem', + 'certfile': bumper.server_cert, + 'keyfile': bumper.server_key, }, }, 'sys_interval': 10, @@ -169,8 +170,10 @@ class MQTTServer(): 'topic-check': { 'enabled': False }, - 'bots':{ - 'connected_bots': bumper_clients + 'clients':{ + 'connected_bots': bumper_bots, + 'connected_clients': bumper_clients + } } if run_async: @@ -197,15 +200,17 @@ class MQTTServer(): class BumperMQTTServer_Plugin: def __init__(self, context): self.context = context - try: - self.bots = self.context.config['bots'] + try: + self.clients = self.context.config['clients'] except KeyError: - self.context.logger.warning("'bots' section not found in context configuration") + self.context.logger.warning("'clients' section not found in context configuration") async def on_broker_client_connected(self, client_id): + logging.debug('Bumper Connection: %s connected' % client_id) - connected_bots = self.bots['connected_bots'].get() + connected_bots = self.clients['connected_bots'].get() + connected_clients = self.clients['connected_clients'].get() didsplit = str(client_id).split("@") #If this isn't a fake user (fuid) then add as a bot if not (str(didsplit[0]).startswith("fuid") or str(didsplit[0]).startswith("helper")): @@ -221,20 +226,51 @@ class BumperMQTTServer_Plugin: if botactive == False: connected_bots.append(newbot.asdict()) + logging.info("Adding bot to list: {}".format(newbot.asdict())) - self.bots['connected_bots'].set(connected_bots) + self.clients['connected_bots'].set(connected_bots) + else: + tmpuserdetail = str(didsplit[1]).split("/") + newuser = bumper.VacBotUser() + newuser.userid = didsplit[0] + newuser.realm = tmpuserdetail[0] + newuser.resource = tmpuserdetail[1] + + clientactive = False + for client in connected_clients: + if client['userid'] == newuser.userid: + clientactive = True + + if clientactive == False: + connected_clients.append(newuser.asdict()) + logging.info("Adding client to list: {}".format(newuser.asdict())) + + self.clients['connected_clients'].set(connected_clients) + + + logging.debug('Connected Bots: %s' %self.clients['connected_bots'].get()) + logging.debug('Connected Clients: %s' %self.clients['connected_clients'].get()) + - logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get()) async def on_broker_client_disconnected(self, client_id): logging.debug('Bumper Connection: %s disconnected' % client_id) - connected_bots = self.bots['connected_bots'].get() + connected_bots = self.clients['connected_bots'].get() + connected_clients = self.clients['connected_clients'].get() didsplit = str(client_id).split("@") #If the did is in the list, remove it for bot in connected_bots: if didsplit[0] == bot['did']: - logging.debug("Removing bot from list: {}".format(bot)) + logging.info("Removing bot from list: {}".format(bot['did'])) connected_bots.remove(bot) - self.bots['connected_bots'].set(connected_bots) + self.clients['connected_bots'].set(connected_bots) - logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get()) \ No newline at end of file + logging.debug('Connected Bots: %s' %self.clients['connected_bots'].get()) + + for client in connected_clients: + if didsplit[0] == client['userid']: + logging.info("Removing client from list: {}".format(client['userid'])) + connected_clients.remove(client) + self.clients['connected_clients'].set(connected_clients) + + logging.debug('Connected Clients: %s' %self.clients['connected_clients'].get()) \ No newline at end of file diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index e958516..9dde3fa 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -23,8 +23,9 @@ class XMPPServer(): connection, client_address = self.socket.accept() # disconnect any clients with this ip for client in self.clients: - if client.address == client_address[0]: + if client.address == client_address[0]: client.disconnect() + client.join() thread_id = uuid.uuid4() client = Client(thread_id, connection, client_address) @@ -40,12 +41,15 @@ class XMPPServer(): logging.info('XMPPServer: bye') def disconnect(self): - logging.info('XMPPServer: waiting for all client threads to exit') - for client in self.clients: - client.disconnect() - client.join() - self.exit_flag = True - logging.info('XMPPServer: shutting down...') + try: + logging.info('XMPPServer: waiting for all client threads to exit') + for client in self.clients: + client.disconnect() + client.join() + self.exit_flag = True + logging.info('XMPPServer: shutting down...') + except Exception as e: + logging.exception("Exception: {}".format(e)) class Client(threading.Thread): @@ -62,49 +66,72 @@ class Client(threading.Thread): def __init__(self, thread_id, connection, client_address): threading.Thread.__init__(self) self.id = thread_id - self.name = "XMPP Thread {}".format(self.id) + self.name = "XMPP Thread {}".format(client_address[0]) self.type = self.UNKNOWN self.state = self.IDLE self.connection = connection self.address = client_address[0] + self.clientresource = "" def send(self, command): - logging.debug('to {}: {}'.format(self.address, command)) - self.connection.send(command.encode()) + try: + logging.debug('to {}: {}'.format(self.address, command)) + self.connection.send(command.encode()) + except OSError as e: + logging.error('XMPPServer: {}'.format(e)) + except Exception as e: + logging.exception("Exception: {}".format(e)) + + def disconnect(self): - logging.info('{} disconnecting'.format(self.address)) - self.connection.close() - self._set_state('DISCONNECT') + try: + logging.info('{} with resource {} disconnecting'.format(self.address, self.clientresource)) + self.connection.close() + self._set_state('DISCONNECT') + + except Exception as e: + logging.exception("Exception: {}".format(e)) def _tag_strip_uri(self, tag): - if tag[0] == '{': - uri, ignore, tag = tag[1:].partition('}') - return tag + try: + if tag[0] == '{': + uri, ignore, tag = tag[1:].partition('}') + return tag + except Exception as e: + logging.exception("Exception: {}".format(e)) def _set_state(self, state): - new_state = getattr(Client, state) - if self.state > new_state: - raise Exception('{} illegal state change {}->{}'.format(self.address, self.state, new_state)) - logging.info('{} state: {}'.format(self.address, state)) - self.state = new_state + try: + new_state = getattr(Client, state) + if self.state > new_state: + raise Exception('{} illegal state change {}->{}'.format(self.address, self.state, new_state)) + logging.debug('{} state: {}'.format(self.address, state)) + self.state = new_state + if new_state == '5': + self.join() + except Exception as e: + logging.exception("Exception: {}".format(e)) def _handle_ctl(self, xml, data): - ctl = xml[0][0] - if ctl.get('admin') and self.type == self.BOT: - logging.info('admin username received from bot: {}'.format(ctl.get('admin'))) - XMPPServer.client_id = ctl.get('admin') - return - # forward - for client in XMPPServer.clients: - if client.address != self.address and client.state == client.READY: - if client.type == self.BOT: - data = data.decode('utf-8') - id_index = data.find('id') - if id_index > -1: - data = data[:id_index] + 'from="' + XMPPServer.client_id + '" ' + data[id_index:] - data = data.encode() - client.send(data.decode('utf-8')) + try: + ctl = xml[0][0] + if ctl.get('admin') and self.type == self.BOT: + logging.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.address != self.address and client.state == client.READY: + if client.type == self.BOT: + data = data.decode('utf-8') + id_index = data.find('id') + if id_index > -1: + data = data[:id_index] + 'from="' + XMPPServer.client_id + '" ' + data[id_index:] + data = data.encode() + client.send(data.decode('utf-8')) + except Exception as e: + logging.exception("Exception: {}".format(e)) def _handle_ping(self, xml, data): @@ -118,14 +145,18 @@ class Client(threading.Thread): def _handle_result(self, data): # forward - for client in XMPPServer.clients: - if client.address != self.address and client.state == client.READY: - client.send(data.decode('utf-8')) + try: + for client in XMPPServer.clients: + if client.address != self.address and client.state == client.READY: + client.send(data.decode('utf-8')) + except Exception as e: + logging.exception("Exception: {}".format(e)) def run(self): try: logging.info('client connected: {}'.format(self.address)) self._set_state('CONNECT') + data = "" while True: time.sleep(0.2) if not self.connection._closed: @@ -143,6 +174,7 @@ class Client(threading.Thread): self.send('') continue xml = ET.fromstring(data) + logging.debug("XMPPXML: {}".format(data)) if len(xml): child = self._tag_strip_uri(xml[0].tag) else: @@ -150,6 +182,11 @@ class Client(threading.Thread): if xml.tag == 'iq': res = None if child == 'bind': + clientbindxml = xml.getchildren() + clientresourcexml = clientbindxml[0].getchildren() + self.clientresource = clientresourcexml[0].text + self.name = "XMPP Thread {}".format(self.clientresource) + logging.info("XMPP Client {} using resource {}".format(self.address, self.clientresource)) res = '{}'.format(xml.get('id'), XMPPServer.bot_id) self._set_state('BIND') elif child == 'session': @@ -167,12 +204,12 @@ class Client(threading.Thread): if len(xml) and xml[0].tag == 'status': # bot announcing arrival self.type = self.BOT - logging.info('{} type set to BOT (based on presence tag)'.format(self.address)) + logging.debug('{} type set to BOT (based on presence tag)'.format(self.address)) # send a command from an unknown user - the response will contain the correct admin username self.send(''.format(uuid.uuid4(), 'unknown@ecouser.net', XMPPServer.bot_id)) elif xml.get('type') == 'available': self.type = self.CONTROLLER - logging.info('{} type set to CONTROLLER (based on presence tag)'.format(self.address)) + logging.debug('{} type set to CONTROLLER (based on presence tag)'.format(self.address)) except ET.ParseError as e: logging.debug('error: {}'.format(e)) except Exception as e: @@ -181,8 +218,14 @@ class Client(threading.Thread): except OSError as e: logging.error('XMPPServer: {}'.format(e)) self._set_state('DISCONNECT') + + except ConnectionResetError as e: + logging.error('XMPPServer: {}'.format(e)) + self._set_state('DISCONNECT') + except Exception as e: logging.error('XMPPServer: {}'.format(e)) self._set_state('DISCONNECT') + finally: self.disconnect()