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