diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py
index a9c3ca1..6515c54 100644
--- a/bumper/mqttserver.py
+++ b/bumper/mqttserver.py
@@ -62,45 +62,42 @@ class MQTTHelperBot:
]
)
asyncio.create_task(self.get_msg())
-
+
except Exception as e:
helperbotlog.exception("{}".format(e))
async def get_msg(self):
- try:
- while True:
- message = await self.Client.deliver_message()
+ while True:
+ message = await self.Client.deliver_message()
- 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(),
- "topic": message.topic,
- "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"))))
+ 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(),
+ "topic": message.topic,
+ "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:
- 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))
- self.command_responses.remove(msg)
+ # Cleanup "expired messages" > 60 seconds from time
+ 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))
+ self.command_responses.remove(msg)
- except Exception as e:
- helperbotlog.exception("{}".format(e))
async def wait_for_resp(self, requestid):
try:
diff --git a/bumper/xmpp_old_client.py b/bumper/xmpp_old_client.py
deleted file mode 100644
index ea5b24b..0000000
--- a/bumper/xmpp_old_client.py
+++ /dev/null
@@ -1,769 +0,0 @@
-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 d9d4f63..e2d637c 100644
--- a/bumper/xmppserver.py
+++ b/bumper/xmppserver.py
@@ -9,50 +9,29 @@ import asyncio, functools
xmppserverlog = logging.getLogger("xmppserver")
-
-class XMPPServer:
- server_id = "ecouser.net"
- client_id = None
+class XMPPServer():
+ server_id = "ecouser.net"
clients = []
exit_flag = False
-
+
def __init__(self, address):
# Initialize bot server
self.address = address
+ self.xmpp_protocol = lambda: XMPPServer_Protocol()
- async def async_server(self):
+ async def start_async_server(self):
xmppserverlog.info(
"Starting XMPP Server at {}:{}".format(self.address[0], self.address[1])
)
- server = await asyncio.start_server(
- self.accept_client, self.address[0], self.address[1]
+
+ loop = asyncio.get_running_loop()
+
+ server = await loop.create_server(
+ self.xmpp_protocol, host=self.address[0], port=self.address[1]
)
- await server.serve_forever()
-
- # self.clients = {} # task -> (reader, writer)
-
- def accept_client(self, client_reader, client_writer):
- try:
- aclient = XMPPAsyncClient(client_reader, client_writer)
- task = asyncio.Task(aclient.handle_async_client())
- aclient._async_task = task
- self.clients.append(aclient)
-
- def client_done(aclient, task):
- try:
- self.clients.remove(aclient)
- xmppserverlog.debug("End Connection for ({}:{} | {})".format(client_writer.get_extra_info("peername")[0], client_writer.get_extra_info("peername")[1], aclient.bumper_jid))
- client_writer.close()
- except Exception as e:
- xmppserverlog.error("{}".format(e))
-
- clientaddr = client_writer.get_extra_info("peername")
- xmppserverlog.debug("New Connection from {}:{}".format(clientaddr[0],clientaddr[1]))
- task.add_done_callback(functools.partial(client_done, aclient))
-
- except Exception as e:
- xmppserverlog.error("{}".format(e))
+ async with server:
+ await server.serve_forever()
def disconnect(self):
try:
@@ -66,6 +45,36 @@ class XMPPServer:
except Exception as e:
xmppserverlog.error("{}".format(e))
+class XMPPServer_Protocol(asyncio.Protocol):
+ client_id = None
+ exit_flag = False
+ server_cert = "./certs/cert.pem"
+ server_key = "./certs/key.pem"
+ ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+ ssl_ctx.load_cert_chain(server_cert, server_key)
+ aclient = None
+
+ def connection_made(self, transport):
+ if self.aclient: #Existing client... upgrading to TLS
+ xmppserverlog.debug("Upgraded connection for {}".format(self.aclient.address))
+ self.aclient.transport = transport
+ else:
+ aclient = XMPPAsyncClient(transport)
+ self.aclient = aclient
+ XMPPServer.clients.append(aclient)
+ self.aclient.state = getattr(aclient, "CONNECT")
+ xmppserverlog.debug("New Connection from {}".format(aclient.address))
+
+
+ def connection_lost(self, error):
+ XMPPServer.clients.remove(self.aclient)
+ self.aclient._set_state("DISCONNECT")
+ xmppserverlog.debug("End Connection for ({}:{} | {})".format(self.aclient.address[0], self.aclient.address[1], self.aclient.bumper_jid))
+
+ def data_received(self, data):
+ self.aclient._parse_data(data)
+
+
class XMPPAsyncClient:
IDLE = 0
CONNECT = 1
@@ -76,70 +85,32 @@ class XMPPAsyncClient:
UNKNOWN = 0
BOT = 1
CONTROLLER = 2
- _async_task = None
+ TLSUpgraded = False
- def __init__(self, client_reader, client_writer):
+ def __init__(self, transport):
self.type = self.UNKNOWN
self.state = self.IDLE
- self.address = client_writer.get_extra_info("peername")
- self.client_reader = client_reader
- self.client_writer = client_writer
+ self.address = transport.get_extra_info("peername")
+ self.transport = transport
self.clientresource = ""
self.devclass = ""
self.bumper_jid = ""
self.uid = ""
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))
- await self._set_state("CONNECT")
- while True:
- await asyncio.sleep(0.05)
- if not self.state == self.DISCONNECT:
- data = await self.client_reader.read(4096)
- if data is None or data == b'':
- xmppserverlog.debug("Received no data")
- # exit loop and disconnect
- return
- else:
- await self._parse_data(data)
- else:
- break
-
- # exit loop and disconnect
- return
-
- async def send(self, command):
+ def send(self, command):
try:
- # if not self.connection._closed:
if self.log_sent_message:
xmppserverlog.debug("send to ({}:{} | {}) - {}".format(self.address[0], self.address[1], self.bumper_jid, command))
- self.client_writer.write(command.encode())
- await self.client_writer.drain()
-
- except BrokenPipeError as e:
- #xmppserverlog.debug("{}".format(e))
- await self._set_state("DISCONNECT")
-
- except ConnectionResetError as e:
- #xmppserverlog.debug("{}".format(e))
- await self._set_state("DISCONNECT")
-
- except ConnectionAbortedError as e:
- #xmppserverlog.debug("{}".format(e))
- await self._set_state("DISCONNECT")
-
- except OSError as e:
- xmppserverlog.error("{}".format(e))
-
+ self.transport.write(command.encode())
+
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _disconnect(self):
+ def _disconnect(self):
try:
bot = bumper.bot_get(self.uid)
@@ -150,12 +121,12 @@ class XMPPAsyncClient:
if client:
bumper.client_set_xmpp(client["resource"], False)
- self.client_writer.close()
+ self.transport.close()
except Exception as e:
xmppserverlog.error("{}".format(e))
- async def _tag_strip_uri(self, tag):
+ def _tag_strip_uri(self, tag):
try:
if tag[0] == "{":
_, _, tag = tag[1:].partition("}")
@@ -164,7 +135,7 @@ class XMPPAsyncClient:
except Exception as e:
xmppserverlog.error("{}".format(e))
- async def _set_state(self, state):
+ def _set_state(self, state):
try:
new_state = getattr(XMPPAsyncClient, state)
if self.state > new_state:
@@ -179,17 +150,17 @@ class XMPPAsyncClient:
self.state = new_state
if new_state == 5:
- await self._disconnect()
+ self._disconnect()
except Exception as e:
xmppserverlog.error("{}".format(e))
- async def _handle_ctl(self, xml, data):
+ def _handle_ctl(self, xml, data):
try:
if "roster" in data:
# Return not-implemented for roster
- await self.send(
+ self.send(
''.format(
xml.get("id")
)
@@ -198,7 +169,7 @@ class XMPPAsyncClient:
if "disco#items" in data:
# Return not-implemented for disco#items
- await self.send(
+ self.send(
''.format(
xml.get("id")
))
@@ -206,7 +177,7 @@ class XMPPAsyncClient:
if "disco#info" in data:
# Return not-implemented for disco#info
- await self.send(
+ self.send(
''.format(
xml.get("id")
)
@@ -218,7 +189,7 @@ class XMPPAsyncClient:
if (
"com:sf" in data and xml.get("to") == "rl.ecorobot.net"
): # Android bind? Not sure what this does yet.
- await self.send(
+ self.send(
''.format(
xml.get("id"),
self.uid,
@@ -257,12 +228,12 @@ class XMPPAsyncClient:
xmppserverlog.debug(
"Sending ctl to bot: {}".format(rxmlstring)
)
- await client.send(rxmlstring)
+ client.send(rxmlstring)
except Exception as e:
xmppserverlog.error("{}".format(e))
- async def _handle_ping(self, xml, data):
+ def _handle_ping(self, xml, data):
try:
if xml.get("to").find("@") == -1: # No to address
# Ping to server - respond
@@ -270,7 +241,7 @@ class XMPPAsyncClient:
xml.get("id"), xml.get("to")
)
# xmppserverlog.debug("Server Ping resp: {}".format(pingresp))
- await self.send(pingresp)
+ self.send(pingresp)
else:
pingto = xml.get("to")
@@ -291,7 +262,7 @@ class XMPPAsyncClient:
and client.state == client.READY
):
if client.uid.lower() in pingto.lower():
- await client.send(pingstring)
+ client.send(pingstring)
except Exception as e:
xmppserverlog.exception("{}".format(e))
@@ -300,11 +271,11 @@ class XMPPAsyncClient:
async def schedule_ping(self, time):
if not self.state == 5: #disconnected
pingstring = "".format(XMPPServer.server_id, self.bumper_jid)
- await self.send(pingstring)
+ self.send(pingstring)
await asyncio.sleep(time)
asyncio.Task(self.schedule_ping(time))
- async def _handle_result(self, xml, data):
+ def _handle_result(self, xml, data):
try:
ctl_to = xml.get("to")
if not "from" in xml.attrib:
@@ -329,17 +300,17 @@ class XMPPAsyncClient:
uuid.uuid4(), adminuser, self.bumper_jid, newuser
)
xmppserverlog.debug("Add User: {}".format(adduser))
- await self.send(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))
- await self.send(adduseracs)
+ self.send(adduseracs)
# GetUserInfo - Just to confirm it set correctly
- await self.send(
+ self.send(
''.format(
uuid.uuid4(), adminuser, self.bumper_jid
)
@@ -360,7 +331,7 @@ class XMPPAsyncClient:
)
)
for client in XMPPServer.clients:
- await client.send(rxmlstring)
+ client.send(rxmlstring)
if xml.get("to").find("@") == -1: # No to address
ctl_to = xml.get("to")
@@ -374,7 +345,7 @@ class XMPPAsyncClient:
):
if not "@" in ctl_to: # No user@, send to all clients?
# TODO: Revisit later, this may be wrong
- await client.send(rxmlstring)
+ client.send(rxmlstring)
elif (
client.uid.lower() in ctl_to.lower()
@@ -384,12 +355,12 @@ class XMPPAsyncClient:
self.uid, client.uid, rxmlstring
)
)
- await client.send(rxmlstring)
+ client.send(rxmlstring)
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_connect(self, data, xml=None):
+ def _handle_connect(self, data, xml=None):
try:
if self.state == self.CONNECT:
@@ -401,33 +372,36 @@ class XMPPAsyncClient:
if ec > -1:
self.devclass = data.decode("utf-8")[sc + 4 : ec]
# ack jabbr:client
- # no STARTTLS
- await self.send(
+ # Send stream tag to client, acknowledging connection
+ self.send(
''.format(
XMPPServer.server_id
)
)
- # with STARTTLS
- # await self.send(''.format(XMPPServer.server_id))
- await asyncio.sleep(0.25)
- #time.sleep(0.25)
- # send authentication support for iq-auth (fallback) and SASL
- await self.send(
- 'PLAIN'
- )
- # await self.send('')
+ # Send STARTTLS to client with auth mechanisms
+ if self.TLSUpgraded == False:
+ #With STARTTLS #https://xmpp.org/rfcs/rfc3920.html
+ self.send(
+ 'PLAIN'
+ )
+
+ else:
+ # Already using TLS send authentication support for iq-auth (fallback) and SASL
+ self.send(
+ 'PLAIN'
+ )
else:
- await self.send("")
+ self.send("")
else:
if "jabber:iq:auth" in xml.tag: # Handle iq-auth
- await self._handle_iq_auth(xml)
+ self._handle_iq_auth(xml)
elif (
"urn:ietf:params:xml:ns:xmpp-sasl" in xml.tag
): # Handle SASL Auth
- await self._handle_sasl_auth(xml)
+ self._handle_sasl_auth(xml)
else:
xmppserverlog.error("Couldn't handle: {}".format(xml))
@@ -436,34 +410,57 @@ class XMPPAsyncClient:
# Client getting session after authentication
if data.decode("utf-8").find("jabber:client") > -1:
# ack jabbr:client
- await self.send(
+ self.send(
''.format(
XMPPServer.server_id
)
)
- await asyncio.sleep(0.25)
- #time.sleep(0.25)
- # session
- await self.send(
+
+ self.send(
''
)
else: # Handle init bind
if len(xml):
- child = await self._tag_strip_uri(xml[0].tag)
+ child = self._tag_strip_uri(xml[0].tag)
else:
child = None
if xml.tag == "iq":
if child == "bind":
- await self._handle_bind(xml)
+ self._handle_bind(xml)
else:
xmppserverlog.error("Couldn't handle: {}".format(xml))
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_iq_auth(self, data):
+
+ async def _handle_starttls(self, data):
+ try:
+ if self.TLSUpgraded == False:
+ self.TLSUpgraded = True #Set TLSUpgraded true to prevent further attempts to upgrade connection
+ xmppserverlog.debug("Upgrading connection with STARTTLS for {}:{}".format(self.address[0],self.address[1]))
+ self.send("") #send process to client
+
+ # After proceed the connection should be upgraded to TLS
+ loop = asyncio.get_event_loop()
+ transport = self.transport
+ protocol = self.transport.get_protocol()
+
+ ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+ ssl_ctx.load_cert_chain(bumper.server_cert, bumper.server_key)
+ ssl_ctx.load_verify_locations(cafile=bumper.ca_cert)
+
+ new_transport = await loop.start_tls(transport , protocol, ssl_ctx, server_side=True)
+ protocol.connection_made(new_transport)
+
+
+ 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]
@@ -474,7 +471,7 @@ class XMPPAsyncClient:
and "auth}username" in ctl.tag
and self.type == self.UNKNOWN
):
- await self.send(
+ self.send(
''.format(
xml.get("id")
)
@@ -508,10 +505,10 @@ class XMPPAsyncClient:
xmppserverlog.debug("bot authenticated {}".format(self.uid))
# Client authenticated, move to next state
- await self._set_state("INIT")
+ self._set_state("INIT")
# Successful auth
- await self.send(''.format(xml.get("id")))
+ self.send(''.format(xml.get("id")))
else:
auth = False
@@ -525,16 +522,16 @@ class XMPPAsyncClient:
xmppserverlog.debug("client authenticated {}".format(self.uid))
# Client authenticated, move to next state
- await self._set_state("INIT")
+ self._set_state("INIT")
# Successful auth
- await self.send(
+ self.send(
''.format(xml.get("id"))
)
else:
# Failed auth
- await self.send(
+ self.send(
''.format(
xml.get("id")
)
@@ -557,7 +554,7 @@ class XMPPAsyncClient:
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_sasl_auth(self, xml):
+ def _handle_sasl_auth(self, xml):
try:
saslauth = base64.b64decode(xml.text).decode("utf-8").split("/")
@@ -580,12 +577,12 @@ class XMPPAsyncClient:
self.type = self.BOT
xmppserverlog.debug("bot authenticated {}".format(self.uid))
# Send response
- await self.send(
+ self.send(
''
) # Success
# Client authenticated, move to next state
- await self._set_state("INIT")
+ self._set_state("INIT")
else:
auth = False
@@ -600,23 +597,23 @@ class XMPPAsyncClient:
xmppserverlog.debug("client authenticated {}".format(self.uid))
# Client authenticated, move to next state
- await self._set_state("INIT")
+ self._set_state("INIT")
# Send response
- await self.send(
+ self.send(
''
) # Success
else:
# Failed to authenticate
- await self.send(
+ self.send(
''
) # Fail
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_bind(self, xml):
+ def _handle_bind(self, xml):
try:
bot = bumper.bot_get(self.uid)
@@ -656,24 +653,24 @@ class XMPPAsyncClient:
xml.get("id"), self.bumper_jid
)
- await self._set_state("BIND")
- await self.send(res)
+ self._set_state("BIND")
+ self.send(res)
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_session(self, xml):
+ def _handle_session(self, xml):
try:
res = ''.format(xml.get("id"))
- await self._set_state("READY")
- await self.send(res)
+ self._set_state("READY")
+ self.send(res)
asyncio.Task(self.schedule_ping(30))
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_presence(self, xml):
+ def _handle_presence(self, xml):
try:
if len(xml) and xml[0].tag == "status":
@@ -683,7 +680,7 @@ class XMPPAsyncClient:
# Most likely a bot, possibly hello world in text
# Send dummy return
- await self.send(
+ self.send(
' dummy '.format(self.bumper_jid)
)
@@ -692,7 +689,7 @@ class XMPPAsyncClient:
# If it is a BOT, send extras
if self.type == self.BOT:
# get device info
- await self.send(
+ self.send(
''.format(
self.bumper_jid, XMPPServer.server_id
)
@@ -712,7 +709,7 @@ class XMPPAsyncClient:
)
# Send dummy return
- await self.send(
+ self.send(
' dummy '.format(self.bumper_jid)
)
elif xml.get("type") == "unavailable":
@@ -721,7 +718,7 @@ class XMPPAsyncClient:
ET.tostring(xml, encoding="utf-8").decode("utf-8"))
)
- await self._set_state("DISCONNECT")
+ self._set_state("DISCONNECT")
else:
# Sometimes the android app sends these
xmppserverlog.debug(
@@ -730,14 +727,14 @@ class XMPPAsyncClient:
)
)
# Send dummy return
- await self.send(
+ self.send(
' dummy '.format(self.bumper_jid)
)
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _parse_data(self, data):
+ def _parse_data(self, data):
if data.decode("utf-8").startswith(
"" in newdata:
xmppserverlog.error("xml parse error - {} - {}".format(newdata, e))
else:
- await self.send("") # Close stream
+ self.send("") # Close stream
else:
if "" in newdata:
xmppserverlog.error(
"xml parse error - {} - {}".format(newdata, e)
)
else:
- await self.send("") # Close stream
- await self._set_state("DISCONNECT")
+ self.send("") # Close stream
+ self._set_state("DISCONNECT")
except Exception as e:
xmppserverlog.exception("{}".format(e))
- async def _handle_iq(self, xml, data):
+ def _handle_iq(self, xml, data):
try:
if len(xml):
- child = await self._tag_strip_uri(xml[0].tag)
+ child = self._tag_strip_uri(xml[0].tag)
else:
child = None
if xml.tag == "iq":
if child == "bind":
- await self._handle_bind(xml)
+ self._handle_bind(xml)
elif child == "session":
- await self._handle_session(xml)
+ self._handle_session(xml)
elif child == "ping":
- await self._handle_ping(xml, data)
+ self._handle_ping(xml, data)
elif child == "query":
if self.type == self.BOT:
- await self._handle_result(xml, data)
+ self._handle_result(xml, data)
else:
- await self._handle_ctl(xml, data)
+ self._handle_ctl(xml, data)
elif xml.get("type") == "result":
if self.type == self.BOT:
- await self._handle_result(xml, data)
+ self._handle_result(xml, data)
else:
- await self._handle_result(xml, data)
+ self._handle_result(xml, data)
elif xml.get("type") == "set":
if self.type == self.BOT:
- await self._handle_result(xml, data)
+ self._handle_result(xml, data)
else:
- await self._handle_result(xml, data)
+ self._handle_result(xml, data)
except Exception as e:
xmppserverlog.exception("{}".format(e))
diff --git a/start_bumper.py b/start_bumper.py
index a01cd81..0ddf4f0 100644
--- a/start_bumper.py
+++ b/start_bumper.py
@@ -18,7 +18,7 @@ async def main():
args = sys.argv
listen_host = ""
-
+
if len(args) > 0:
if "--debug" in args:
logging.basicConfig(
@@ -41,7 +41,7 @@ async def main():
listen_host = "0.0.0.0"
else:
listen_host = socket.gethostbyname(socket.gethostname())
-
+
conf_address_443 = (listen_host, 443)
conf_address_8007 = (listen_host, 8007)
xmpp_address = (listen_host, 5223)
@@ -58,46 +58,47 @@ async def main():
conf_server_2 = bumper.ConfServer(
conf_address_8007, usessl=False, helperbot=mqtt_helperbot
)
+ xmpp_server = bumper.XMPPServer(xmpp_address)
- # Start web servers
- conf_server.confserver_app()
- task_conf_server = asyncio.create_task(conf_server.start_server())
- bumper.bumperlog.debug("task_conf_server added")
- await task_conf_server
+ try:
+ # Start web servers
+ conf_server.confserver_app()
+ asyncio.create_task(conf_server.start_server())
+
+ conf_server_2.confserver_app()
+ asyncio.create_task(conf_server_2.start_server())
+
+ # Start MQTT Server
+ asyncio.create_task(mqtt_server.broker_coro())
+
+ # Start MQTT Helperbot
+ asyncio.create_task(mqtt_helperbot.start_helper_bot())
+
+ # Start XMPP Server
+ asyncio.create_task(xmpp_server.start_async_server())
+
+ maintain = asyncio.create_task(maintenance_tasks())
+ await maintain #Keeps the loop running until this exits
- conf_server_2.confserver_app()
- task_conf_server2 = asyncio.create_task(conf_server_2.start_server())
- bumper.bumperlog.debug("task_conf_server2 added")
- await task_conf_server2
-
- # Start MQTT Server
- task_mqtt_server = asyncio.create_task(mqtt_server.broker_coro())
- bumper.bumperlog.debug("task_mqtt_server added")
- await task_mqtt_server
-
- # Start MQTT Helperbot
- task_mqtt_helperbot = asyncio.create_task(mqtt_helperbot.start_helper_bot())
- bumper.bumperlog.debug("task_mqtt_helperbot added")
- await task_mqtt_helperbot
-
- # Start XMPP Server
- task_xmpp_server = asyncio.create_task(xmpp_server.async_server())
- bumper.bumperlog.debug("task_xmpp_server added")
- await task_xmpp_server
+ finally:
+ # Cleanup and close tasks/loop
+ for task in asyncio.Task.all_tasks():
+ task.cancel()
+ loop.close()
+async def maintenance_tasks():
while True:
- try:
- await asyncio.sleep(30)
- bumper.revoke_expired_tokens()
- #disconnected_clients = bumper.get_disconnected_xmpp_clients()
- #for client in disconnected_clients:
- # xmpp_server.remove_client_byuid(client["userid"])
-
- except KeyboardInterrupt:
- bumper.bumperlog.info("Bumper Exiting - Keyboard Interrupt")
- print("Bumper Exiting")
- exit(0)
-
+ await asyncio.sleep(30) # Sleep 30 seconds
+ bumper.revoke_expired_tokens()
+
if __name__ == "__main__":
- asyncio.run(main())
+ try:
+ asyncio.run(main())
+
+ except KeyboardInterrupt:
+ bumper.bumperlog.info("Keyboard Interrupt!")
+ pass
+ finally:
+ bumper.bumperlog.info("Bumper Exiting!")
+