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()